CaiWuXiTong/app/Console/Commands/SyncStoreViolations.php

<?php

namespace App\Console\Commands;

use App\Business\StoreViolationAlertCalculator;
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,
        StoreViolationAlertCalculator $calculator,
    ): 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;

        // 违规原因/处置方式：上游视图补齐后自动参与同步；未提供时不触碰这两列，
        // 避免每日同步把导入补充的数据清空。
        $extraColumns = $this->availableExtraColumns($pdo);
        $updateColumns = $extraColumns === []
            ? StoreViolationRecordRepository::CRAWL_UPDATE_COLUMNS
            : array_merge(StoreViolationRecordRepository::CRAWL_UPDATE_COLUMNS, array_keys($extraColumns));
        // 预警派生字段（级别/责任人/处理截止等）每次都随同步重算写入。
        $updateColumns = array_values(array_unique(array_merge(
            $updateColumns,
            StoreViolationRecordRepository::ALERT_DERIVED_COLUMNS,
        )));

        try {
            while (true) {
                // FreeTDS 可能按连接区域返回 date（例如 "Oct  4 2026 12:00:00:AM"），
                // 与镜像同步器保持一致：在源端转成 ISO 字符串再交给 PHP 解析。
                $selectedColumns = ' id, shop_name, violation_no,'
                    . ' CONVERT(varchar(10), violation_date, 23) AS violation_date,'
                    . ' CONVERT(varchar(10), deadline, 23) AS deadline,'
                    . ' violation_content, status, window_name, platform,'
                    . ' CONVERT(varchar(10), crawl_date, 23) AS crawl_date';
                foreach ($extraColumns as $target => $sourceColumn) {
                    $selectedColumns .= ', ' . $sourceColumn . ' AS ' . $target;
                }

                $sql = 'SELECT TOP ' . $batchSize . $selectedColumns
                    . ' 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;
                    }

                    $mapped = array_merge($mapped, $calculator->calculate(
                        (string) $mapped['platform'],
                        $mapped['shop_name'],
                        $mapped['violation_type'],
                        $mapped['violation_date'],
                        $mapped['appeal_deadline'],
                    ));

                    $payload[] = $mapped + [
                        'synced_at' => $now,
                        'created_at' => $now,
                        'updated_at' => $now,
                    ];
                }

                $upserted += $repository->upsertRows($payload, $updateColumns);
            }
        } 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;
    }

    /** @return array<string,string> 统一表字段 => 上游视图列名 */
    private function availableExtraColumns(PDO $pdo): array
    {
        try {
            $statement = $pdo->query(
                "SELECT name FROM sys.columns WHERE object_id = OBJECT_ID('rpt.crawl_shop_violation')"
            );
            $names = $statement->fetchAll(PDO::FETCH_COLUMN);
            $statement->closeCursor();
        } catch (Throwable) {
            return [];
        }

        return self::pickExtraColumns(array_map('strval', $names));
    }

    /**
     * 从上游可用列名中挑选可同步的扩展列；列名以数据平台实际提供的为准。
     *
     * @param  list<string>  $availableColumns
     * @return array<string,string>
     */
    public static function pickExtraColumns(array $availableColumns): array
    {
        $candidates = [
            'violation_reason' => ['violation_reason', 'reason'],
            'disposition' => ['disposition', 'handling_method', 'disposal'],
        ];

        $available = array_map(static fn (string $name): string => strtolower($name), $availableColumns);
        $selected = [];
        foreach ($candidates as $target => $names) {
            foreach ($names as $name) {
                if (in_array($name, $available, true)) {
                    $selected[$target] = $name;
                    break;
                }
            }
        }

        return $selected;
    }
}