136 lines
4.8 KiB
PHP
136 lines
4.8 KiB
PHP
<?php
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use App\Business\StoreViolationMapper;
|
||
use Illuminate\Console\Command;
|
||
use Modules\EcommerceFinance\Repositories\StoreViolationRecordRepository;
|
||
use Modules\EcommerceFinance\Services\UpstreamOdbcConnection;
|
||
use PDO;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 店铺违规信息预警：从经营分析副本库 rpt.crawl_shop_violation 同步到统一表。
|
||
*
|
||
* 上游只读；按 id 游标分批读取，幂等写入 store_violation_records。
|
||
* 本地补充字段（扣分、处罚日期、处理状态）不会被覆盖。
|
||
*/
|
||
class SyncStoreViolations extends Command
|
||
{
|
||
protected $signature = 'store-violations:sync
|
||
{--platform= : 只同步指定平台（taobao/alibaba/jd/jingxi/pinduoduo/douyin）}
|
||
{--batch=1000 : 每批读取行数}';
|
||
|
||
protected $description = '同步店铺违规信息（operations_replica.rpt.crawl_shop_violation → store_violation_records）';
|
||
|
||
public function handle(StoreViolationMapper $mapper, StoreViolationRecordRepository $repository): int
|
||
{
|
||
if (! $repository->available()) {
|
||
$this->error('统一表 store_violation_records 尚未建立，请先执行数据库迁移。');
|
||
|
||
return self::FAILURE;
|
||
}
|
||
|
||
$platform = trim((string) $this->option('platform'));
|
||
$platforms = (array) config('store_violations.platforms', []);
|
||
if ($platform !== '' && ! array_key_exists($platform, $platforms)) {
|
||
$this->error('未知平台：' . $platform);
|
||
|
||
return self::FAILURE;
|
||
}
|
||
|
||
$sourceValues = [];
|
||
foreach ($platforms as $slug => $definition) {
|
||
if ($platform !== '' && $slug !== $platform) {
|
||
continue;
|
||
}
|
||
foreach ($definition['source_values'] ?? [] as $value) {
|
||
$sourceValues[] = (string) $value;
|
||
}
|
||
}
|
||
$sourceValues = array_values(array_unique($sourceValues));
|
||
|
||
$batchSize = max(min((int) $this->option('batch'), 5000), 100);
|
||
|
||
try {
|
||
$pdo = UpstreamOdbcConnection::create('operations_replica');
|
||
} catch (Throwable $exception) {
|
||
report($exception);
|
||
$repository->recordSyncState('failed', $repository->totalRows(), 'connect_failed');
|
||
$this->error('无法连接经营分析副本库：' . $exception->getMessage());
|
||
|
||
return self::FAILURE;
|
||
}
|
||
|
||
$now = now();
|
||
$lastId = 0;
|
||
$upserted = 0;
|
||
$skipped = 0;
|
||
$read = 0;
|
||
|
||
try {
|
||
while (true) {
|
||
$sql = 'SELECT TOP ' . $batchSize
|
||
. ' id, shop_name, violation_no, violation_date, deadline, violation_content,'
|
||
. ' status, window_name, platform, crawl_date'
|
||
. ' FROM rpt.crawl_shop_violation WHERE id > ?';
|
||
$params = [$lastId];
|
||
|
||
if ($sourceValues !== []) {
|
||
$sql .= ' AND platform IN (' . implode(', ', array_fill(0, count($sourceValues), '?')) . ')';
|
||
$params = array_merge($params, $sourceValues);
|
||
}
|
||
|
||
$sql .= ' ORDER BY id ASC';
|
||
|
||
$statement = $pdo->prepare($sql);
|
||
$statement->execute($params);
|
||
$rows = $statement->fetchAll(PDO::FETCH_ASSOC);
|
||
$statement->closeCursor();
|
||
|
||
if ($rows === []) {
|
||
break;
|
||
}
|
||
|
||
$payload = [];
|
||
foreach ($rows as $row) {
|
||
$read++;
|
||
$lastId = max($lastId, (int) ($row['id'] ?? 0));
|
||
$mapped = $mapper->mapCrawlRow($row);
|
||
if ($mapped === null || $mapped['violation_no'] === '') {
|
||
$skipped++;
|
||
continue;
|
||
}
|
||
|
||
$payload[] = $mapped + [
|
||
'synced_at' => $now,
|
||
'created_at' => $now,
|
||
'updated_at' => $now,
|
||
];
|
||
}
|
||
|
||
$upserted += $repository->upsertRows($payload, StoreViolationRecordRepository::CRAWL_UPDATE_COLUMNS);
|
||
}
|
||
} catch (Throwable $exception) {
|
||
report($exception);
|
||
$repository->recordSyncState('failed', $repository->totalRows(), 'sync_failed');
|
||
$this->error('同步失败：' . $exception->getMessage());
|
||
|
||
return self::FAILURE;
|
||
}
|
||
|
||
$total = $repository->totalRows();
|
||
$repository->recordSyncState('succeeded', $total);
|
||
|
||
$this->info(sprintf(
|
||
'同步完成：读取 %d 行，写入/更新 %d 行，跳过 %d 行，统一表共 %d 行。',
|
||
$read,
|
||
$upserted,
|
||
$skipped,
|
||
$total,
|
||
));
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
}
|