WuYongJie
561c42d830
- 店铺维度报表按期初数据+系统主体+爬虫新主体生成主体期间，一行一期间，修复流程历史重复行 - 变更日当天归新主体、老主体失效显示变更日前一天；关店和其他非正常状态两列显示 - - 新增店铺维度报表逻辑配置页（R1-R8）、主体商标数据逻辑配置页和预警页面 - 爬虫店铺主体按税号归一，首次抓到的新主体写入期间变更表并自动生成新期间 - 店铺基础信息、商标展示按爬虫优先；新增爬虫数据质量检查与预警处理
260 lines
10 KiB
PHP
260 lines
10 KiB
PHP
<?php
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use Illuminate\Console\Command;
|
||
use Illuminate\Support\Facades\DB;
|
||
use Modules\EcommerceFinance\Repositories\DataWarningMirrorRepository;
|
||
use Modules\EcommerceFinance\Services\CrawlerDataGuard;
|
||
use Modules\EcommerceFinance\Services\StoreReportPeriodBuilder;
|
||
/**
|
||
* 将爬虫抓到的店铺主体写入期间变更表。
|
||
*
|
||
* 逻辑对应「店铺主体商标数据逻辑配置」页面的 R5、R6：
|
||
* - 只在主体真的发生变化时新增一条变更记录，生效日期取 crawl_date；
|
||
* - 同一天流程表也记录了同店铺变更但主体不同时，以爬虫为准，同时生成
|
||
* 待人工复审的预警。
|
||
*/
|
||
class SyncCrawlerShopSubjects extends Command
|
||
{
|
||
protected $signature = 'store-report:sync-crawler-subjects
|
||
{--dry-run : 只输出将要记录的变更，不写库}';
|
||
|
||
protected $description = '按爬虫抓取结果生成店铺主体期间变更，并登记同日冲突预警';
|
||
|
||
public function handle(DataWarningMirrorRepository $warnings, CrawlerDataGuard $guard): int
|
||
{
|
||
$connection = DB::connection((string) config('upstream_mirrors.connection', 'business_mirror'));
|
||
$schema = $connection->getSchemaBuilder();
|
||
|
||
if (! $schema->hasTable('ops_crawl_shop_company')) {
|
||
$this->error('爬虫店铺主体镜像尚未完成首次同步，本次不生成期间变更。');
|
||
|
||
return self::FAILURE;
|
||
}
|
||
|
||
if (! $schema->hasTable('store_report_shop_period_changes')) {
|
||
$this->error('期间变更表尚未迁移，请先执行数据库迁移。');
|
||
|
||
return self::FAILURE;
|
||
}
|
||
|
||
$official = collect($connection->table('ops_fys_company_info')
|
||
->select(['unifiedSocialCreditCode', 'businessLicenseName'])
|
||
->get())
|
||
->keyBy(fn (object $row): string => trim((string) $row->unifiedSocialCreditCode));
|
||
|
||
$existing = $connection->table('store_report_shop_period_changes')
|
||
->select(['erp_store_name', 'company_name', 'observed_on'])
|
||
->orderBy('observed_on')
|
||
->orderBy('id')
|
||
->get();
|
||
$latestChange = [];
|
||
$changesByStore = [];
|
||
foreach ($existing as $row) {
|
||
$name = trim((string) $row->erp_store_name);
|
||
$latestChange[$name] = (string) $row->company_name;
|
||
$changesByStore[$name][] = $row;
|
||
}
|
||
|
||
// 已知主体基线：优先用期间构建器的当前期间（期初数据 + 系统主体 + 历史爬虫变更），
|
||
// 保证只为“真的发生变化”的店铺记录新期间。
|
||
$baseline = $this->baselineCompanies($connection);
|
||
$flow = $this->flowCompanies($connection);
|
||
$dryRun = (bool) $this->option('dry-run');
|
||
$created = 0;
|
||
$conflicts = 0;
|
||
$skipped = 0;
|
||
$newWarnings = [];
|
||
|
||
foreach ($this->latestCrawlRows($connection) as $row) {
|
||
$storeName = trim((string) ($row->shop_name ?? ''));
|
||
if ($storeName === '') {
|
||
continue;
|
||
}
|
||
|
||
$company = $this->resolveCompany($row, $official);
|
||
$observedOn = trim((string) ($row->crawl_date ?? ''));
|
||
if ($company === null || $observedOn === '') {
|
||
$skipped++;
|
||
continue;
|
||
}
|
||
|
||
$known = $latestChange[$storeName] ?? ($baseline[$storeName] ?? null);
|
||
if ($known !== null && mb_strtolower(trim($known)) === mb_strtolower($company)) {
|
||
$skipped++;
|
||
continue;
|
||
}
|
||
|
||
// 同一天流程表给出不同主体：以爬虫为准，同时留待人工复审。
|
||
$flowCompany = $flow[$storeName][$observedOn] ?? null;
|
||
if ($flowCompany !== null && mb_strtolower(trim($flowCompany)) !== mb_strtolower($company)) {
|
||
$conflicts++;
|
||
$newWarnings[] = [
|
||
'warning_key' => 'subject-conflict:' . md5($storeName . '|' . $observedOn . '|' . $company),
|
||
'warning_type' => CrawlerDataGuard::TYPE_SUBJECT_CONFLICT,
|
||
'risk_level' => 'high',
|
||
'related_object' => $storeName,
|
||
'risk_description' => sprintf('店铺「%s」%s 流程记录主体为「%s」，爬虫抓取为「%s」，系统已按爬虫调整期间。', $storeName, $observedOn, $flowCompany, $company),
|
||
'evidence' => sprintf('flow=%s，crawl=%s，crawl_date=%s', $flowCompany, $company, $observedOn),
|
||
'impact' => '主体期间与收入归属',
|
||
'trigger_rule' => 'R6',
|
||
'due_at' => now()->addDays(3),
|
||
];
|
||
}
|
||
|
||
$businessKey = hash('sha256', $storeName . '|' . $company . '|' . $observedOn);
|
||
if (! $dryRun) {
|
||
$connection->table('store_report_shop_period_changes')->updateOrInsert(
|
||
['business_key' => $businessKey],
|
||
[
|
||
'erp_store_name' => $storeName,
|
||
'company_name' => $company,
|
||
'unified_social_credit_code' => trim((string) ($row->tax_no ?? '')) ?: null,
|
||
'observed_on' => $observedOn,
|
||
'source' => 'crawl_shop_company',
|
||
'updated_at' => now(),
|
||
'created_at' => now(),
|
||
],
|
||
);
|
||
}
|
||
|
||
$latestChange[$storeName] = $company;
|
||
$created++;
|
||
$this->line(sprintf('%s：%s 起主体变更为 %s%s', $storeName, $observedOn, $company, $dryRun ? '（演练）' : ''));
|
||
}
|
||
|
||
if (! $dryRun && $newWarnings !== []) {
|
||
foreach ($newWarnings as $warning) {
|
||
$warning['responsible_person'] = $guard->responsiblePersonFor((string) $warning['warning_type']);
|
||
$warnings->upsert($warning);
|
||
}
|
||
}
|
||
|
||
$this->info(sprintf(
|
||
'%s：新增主体变更 %d 条，同日冲突预警 %d 条，跳过 %d 家店铺。',
|
||
$dryRun ? '演练完成' : '同步完成',
|
||
$created,
|
||
$conflicts,
|
||
$skipped,
|
||
));
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
|
||
/**
|
||
* 当前系统认可的主体：把期间构建器的当前期间（最后一条）作为比较基线。
|
||
*
|
||
* @return array<string,string>
|
||
*/
|
||
private function baselineCompanies($connection): array
|
||
{
|
||
if (! $connection->getSchemaBuilder()->hasTable('ops_fys_shop_info')) {
|
||
return [];
|
||
}
|
||
|
||
$stores = $connection->table('ops_fys_shop_info')
|
||
->select(['Id', 'erpShopName', 'unifiedSocialCreditCode'])
|
||
->get()
|
||
->map(static fn (object $row): array => (array) $row)
|
||
->all();
|
||
$companies = $connection->table('ops_fys_company_info')
|
||
->select(['Id', 'unifiedSocialCreditCode', 'businessLicenseName'])
|
||
->get()
|
||
->map(static fn (object $row): array => (array) $row)
|
||
->all();
|
||
$byCreditCode = [];
|
||
foreach ($companies as $company) {
|
||
$code = trim((string) ($company['unifiedSocialCreditCode'] ?? ''));
|
||
if ($code !== '') {
|
||
$byCreditCode[$code] = (string) $company['businessLicenseName'];
|
||
}
|
||
}
|
||
foreach ($stores as &$store) {
|
||
$code = trim((string) ($store['unifiedSocialCreditCode'] ?? ''));
|
||
$store = [
|
||
'erp_store_name' => (string) ($store['erpShopName'] ?? ''),
|
||
'company_name' => $byCreditCode[$code] ?? null,
|
||
'system_company_name' => $byCreditCode[$code] ?? null,
|
||
];
|
||
}
|
||
unset($store);
|
||
|
||
$periods = app(StoreReportPeriodBuilder::class)->build($stores);
|
||
$baseline = [];
|
||
foreach ($periods as $name => $list) {
|
||
$last = end($list);
|
||
$company = trim((string) ($last['company_name'] ?? ''));
|
||
if ($company !== '') {
|
||
$baseline[$name] = $company;
|
||
}
|
||
}
|
||
|
||
return $baseline;
|
||
}
|
||
|
||
/** @return list<object> */
|
||
private function latestCrawlRows($connection): array
|
||
{
|
||
return $connection->table('ops_crawl_shop_company')
|
||
->orderBy('shop_name')
|
||
->orderByDesc('crawl_date')
|
||
->orderByDesc('id')
|
||
->get()
|
||
->groupBy(static fn (object $row): string => (string) $row->shop_name)
|
||
->map(static fn ($rows): object => $rows->first())
|
||
->values()
|
||
->all();
|
||
}
|
||
|
||
/**
|
||
* 主体名称以 tax_no 对应的工商主体名为准，保证与主体报表合并键一致；
|
||
* 工商资料缺失时回退爬虫抓到的名称。
|
||
*/
|
||
private function resolveCompany(object $row, $official): ?string
|
||
{
|
||
$taxNo = trim((string) ($row->tax_no ?? ''));
|
||
if ($taxNo !== '' && $official->has($taxNo)) {
|
||
$name = trim((string) $official->get($taxNo)->businessLicenseName);
|
||
if ($name !== '') {
|
||
return $name;
|
||
}
|
||
}
|
||
|
||
$name = trim((string) ($row->company_name ?? ''));
|
||
|
||
return $name === '' ? null : $name;
|
||
}
|
||
|
||
/** @return array<string,array<string,string>> */
|
||
private function flowCompanies($connection): array
|
||
{
|
||
if (! $connection->getSchemaBuilder()->hasTable('store_report_lifecycle_rows')) {
|
||
return [];
|
||
}
|
||
|
||
$state = $connection->table('store_report_import_states')
|
||
->where('dataset_key', 'store-report.store-lifecycles')
|
||
->first();
|
||
$generation = trim((string) ($state->active_generation ?? ''));
|
||
if ($generation === '') {
|
||
return [];
|
||
}
|
||
|
||
$result = [];
|
||
foreach ($connection->table('store_report_lifecycle_rows')
|
||
->where('generation', $generation)
|
||
->select(['erp_store_name', 'company_name', 'effective_at'])
|
||
->get() as $row) {
|
||
$name = trim((string) $row->erp_store_name);
|
||
$date = substr(trim((string) $row->effective_at), 0, 10);
|
||
if ($name === '' || $date === '') {
|
||
continue;
|
||
}
|
||
$result[$name][$date] = (string) $row->company_name;
|
||
}
|
||
|
||
return $result;
|
||
}
|
||
}
|