diff --git a/app/Console/Commands/BackfillWorkOrderWaybillLookup.php b/app/Console/Commands/BackfillWorkOrderWaybillLookup.php new file mode 100644 index 00000000..1334d148 --- /dev/null +++ b/app/Console/Commands/BackfillWorkOrderWaybillLookup.php @@ -0,0 +1,29 @@ +lock('work-order-waybill-lookup:backfill', 3600); + if (! $lock->get()) { + $this->info('已有运单索引任务运行，本轮跳过。'); + return self::SUCCESS; + } + try { + $result = $lookup->backfill((int) $this->option('batch'), max(0, (int) $this->option('max-batches')), max(0, (int) $this->option('pause-ms'))); + $this->info(json_encode($result)); + return self::SUCCESS; + } finally { + $lock->release(); + } + } +} diff --git a/app/Console/Commands/RefreshStoreReportPaidSales.php b/app/Console/Commands/RefreshStoreReportPaidSales.php new file mode 100644 index 00000000..6ba179f2 --- /dev/null +++ b/app/Console/Commands/RefreshStoreReportPaidSales.php @@ -0,0 +1,19 @@ +syncPaidSales((bool) $this->option('full')); + $this->info(json_encode($result)); + return $result['status'] === 'succeeded' ? self::SUCCESS : self::FAILURE; + } +} diff --git a/app/Console/Commands/RunAccountBillsExportQueue.php b/app/Console/Commands/RunAccountBillsExportQueue.php new file mode 100644 index 00000000..c04ba9bf --- /dev/null +++ b/app/Console/Commands/RunAccountBillsExportQueue.php @@ -0,0 +1,24 @@ +option('max-jobs')); + for ($processed = 0; $limit === 0 || $processed < $limit; $processed++) { + $task = $tasks->nextQueued(); + if ($task === null) break; + $code = $this->call('account-bills:export-run', ['token' => $task['token']]); + if ($code === 75) break; + } + return self::SUCCESS; + } +} diff --git a/app/Console/Commands/RunAccountBillsExportTask.php b/app/Console/Commands/RunAccountBillsExportTask.php index 5f9fcbf0..5ea69f23 100644 --- a/app/Console/Commands/RunAccountBillsExportTask.php +++ b/app/Console/Commands/RunAccountBillsExportTask.php @@ -28,6 +28,17 @@ class RunAccountBillsExportTask extends Command protected $description = '执行店铺收款账户账单导出任务（后台进程调用）'; public function handle(AccountBillsExportTasks $tasks, AccountLedgerRemarkMap $remarkMap): int + { + $lock = \Illuminate\Support\Facades\Cache::store('upstream_shared')->lock('account-bills:export-worker', 10800); + if (! $lock->get()) return 75; + try { + return $this->runTask($tasks, $remarkMap); + } finally { + $lock->release(); + } + } + + private function runTask(AccountBillsExportTasks $tasks, AccountLedgerRemarkMap $remarkMap): int { $token = (string) $this->argument('token'); $task = $this->findTask($tasks, $token); @@ -36,6 +47,7 @@ class RunAccountBillsExportTask extends Command return self::FAILURE; } + if (($task['status'] ?? '') !== 'queued') return self::SUCCESS; $task = $tasks->patch($task, ['status' => 'running', 'started_at' => AccountBillsExportTasks::beijingNow()]); @@ -110,7 +122,7 @@ class RunAccountBillsExportTask extends Command private function findTask(AccountBillsExportTasks $tasks, string $token): ?array { - $root = storage_path('app/account-bills-exports'); + $root = $tasks->rootDir(); foreach (glob($root . '/*/' . $token . '.json') ?: [] as $path) { $task = json_decode((string) file_get_contents($path), true); if (is_array($task) && ($task['token'] ?? null) === $token) { diff --git a/app/Console/Commands/RunAccountMappingResync.php b/app/Console/Commands/RunAccountMappingResync.php index 4e418371..f6db3081 100644 --- a/app/Console/Commands/RunAccountMappingResync.php +++ b/app/Console/Commands/RunAccountMappingResync.php @@ -13,12 +13,12 @@ use Throwable; /** * 科目映射表变更后的业绩明细自动重跑队列。 * - * 一个常驻进程依次处理队列里的「平台 × 月份」；每个实际同步放在独立 PHP 子进程中， - * 避免大平台重跑时的内存累积。也可手动执行本命令补跑。 + * 每轮按预算处理队列里的「平台 × 月份」；实际同步放在独立 PHP 子进程中， + * 避免大平台重跑时的内存累积，并让其它同步有机会取得全局锁。 */ class RunAccountMappingResync extends Command { - protected $signature = 'store-receipt:auto-resync'; + protected $signature = 'store-receipt:auto-resync {--max-jobs=1 : 每轮最多任务数，0为不限} {--max-seconds=300 : 单个任务结束后检查本轮时间预算}'; protected $description = '处理科目映射表变更触发的业绩明细重跑队列'; @@ -39,7 +39,10 @@ class RunAccountMappingResync extends Command ->where('updated_at', '<', now()->subMinutes(60)) ->update(['status' => 'pending', 'updated_at' => now()]); - while (true) { + $processed = 0; + $maxJobs = max(0, (int) $this->option('max-jobs')); + $deadline = microtime(true) + max(1, (int) $this->option('max-seconds')); + while (($maxJobs === 0 || $processed < $maxJobs) && microtime(true) < $deadline) { $job = DB::table(AccountMappingResyncDispatcher::JOBS_TABLE) ->where('status', 'pending') ->orderBy('id') @@ -47,6 +50,7 @@ class RunAccountMappingResync extends Command if ($job === null) { break; } + $processed++; DB::table(AccountMappingResyncDispatcher::JOBS_TABLE) ->where('id', $job->id) @@ -58,19 +62,7 @@ class RunAccountMappingResync extends Command ]); try { - $process = Process::fromShellCommandline( - implode(' ', [ - escapeshellarg(PHP_BINARY ?: 'php'), - 'artisan', - 'store-receipt:order-match:sync', - escapeshellarg((string) $job->month), - '--platform=' . escapeshellarg((string) $job->platform), - ]), - base_path() - ); - $process->setTimeout(7200); - $process->run(); - $output = trim($process->getOutput() . "\n" . $process->getErrorOutput()); + [$succeeded, $output] = $this->runSync($job); // 同步期间又保存了映射时，dispatchFor 已把状态改回 pending； // 这里只在仍为 running 时写 done，不能吞掉新请求。 @@ -78,9 +70,9 @@ class RunAccountMappingResync extends Command ->where('id', $job->id) ->where('status', 'running') ->update([ - 'status' => $process->isSuccessful() ? 'done' : 'failed', + 'status' => $succeeded ? 'done' : 'failed', 'message' => mb_substr( - $output !== '' ? $output : ($process->isSuccessful() ? 'OK' : '同步命令执行失败'), + $output !== '' ? $output : ($succeeded ? 'OK' : '同步命令执行失败'), 0, 1000 ), @@ -104,4 +96,17 @@ class RunAccountMappingResync extends Command return self::SUCCESS; } + + /** @return array{bool,string} */ + protected function runSync(object $job): array + { + $process = new Process([ + PHP_BINARY ?: 'php', 'artisan', 'store-receipt:order-match:sync', + (string) $job->month, '--platform=' . $job->platform, + ], base_path()); + $process->setTimeout(7200); + $process->run(); + + return [$process->isSuccessful(), trim($process->getOutput() . "\n" . $process->getErrorOutput())]; + } } diff --git a/app/Console/Commands/SyncStoreReceiptOrderMatches.php b/app/Console/Commands/SyncStoreReceiptOrderMatches.php index d469f423..9c67124e 100644 --- a/app/Console/Commands/SyncStoreReceiptOrderMatches.php +++ b/app/Console/Commands/SyncStoreReceiptOrderMatches.php @@ -183,6 +183,24 @@ class SyncStoreReceiptOrderMatches extends Command private const REMARK_PAYABLE_BASIS = '订单实付按应付/已付核对'; public function handle(): int + { + if ((bool) $this->option('dry-run')) return $this->syncRows(); + // Share the aggregate lock with the writer, including manually invoked commands. + $lock = \Illuminate\Support\Facades\Cache::store('upstream_shared')->lock('store-report:paid-sales', 10800); + if (! $lock->get()) { + $this->warn('业绩明细或销量汇总正在处理，请稍后重试。'); + return 75; + } + try { + $result = $this->syncRows(); + } finally { + $lock->release(); + } + if ($result === self::SUCCESS && ! (bool) $this->option('planting-only')) $this->refreshStoreReportPaidSales(); + return $result; + } + + private function syncRows(): int { // 全量重建时账单与订单侧映射都在内存里：拼多多 / 淘系支付宝这类大平台需要 1G 以上， // 容器默认 memory_limit=512M 会在写入阶段 OOM 中断，这里统一放宽，免去手工加 -d 参数。 @@ -674,6 +692,7 @@ class SyncStoreReceiptOrderMatches extends Command // 只重建「本账期 + 本平台」的数据，避免把同月其它平台的结果一起清掉。 $dryRun = (bool) $this->option('dry-run'); if (! $dryRun) { + app(\Modules\EcommerceFinance\Services\StoreReportPaidSalesChanges::class)->markPeriod($reportMonth, (string) $source['label']); DB::connection($connection)->table(self::TARGET_TABLE) ->where('report_month', $reportMonth) ->where('platform', (string) $source['label']) @@ -831,6 +850,7 @@ class SyncStoreReceiptOrderMatches extends Command if (count($buffer) >= 500) { if (! $dryRun) { + $this->markPaidSalesRows($buffer); $target->insertOrIgnore($buffer); } $imported += count($buffer); @@ -843,6 +863,7 @@ class SyncStoreReceiptOrderMatches extends Command $buffer[] = self::promoOnlyRow($reportMonth, (string) $source['label'], $promoGroup, $now); if (count($buffer) >= 500) { if (! $dryRun) { + $this->markPaidSalesRows($buffer); $target->insertOrIgnore($buffer); } $imported += count($buffer); @@ -852,6 +873,7 @@ class SyncStoreReceiptOrderMatches extends Command if ($buffer !== []) { if (! $dryRun) { + $this->markPaidSalesRows($buffer); $target->insertOrIgnore($buffer); } $imported += count($buffer); @@ -896,14 +918,16 @@ class SyncStoreReceiptOrderMatches extends Command // 业绩明细变了，顺手刷新店铺维度报表的销量汇总（销量＝这里的订单实付/实收金额）。 // 汇总失败不影响业绩明细本身，只提示一次，避免把已写入的账期数据判成失败。 - if (! (bool) $this->option('dry-run')) { - $this->refreshStoreReportPaidSales(); - } - return self::SUCCESS; } - /** 按最新业绩明细重建店铺维度报表的销量汇总（失败只提示，不中断）。 */ + /** 在写入前记录受影响日期；与汇总任务共用锁，避免消费未完成的变更。 */ + private function markPaidSalesRows(array $rows): void + { + app(\Modules\EcommerceFinance\Services\StoreReportPaidSalesChanges::class)->markDates(array_column($rows, 'occurred_date')); + } + + /** 按最新业绩明细刷新店铺销量汇总，失败只提示。 */ private function refreshStoreReportPaidSales(): void { try { @@ -936,6 +960,10 @@ class SyncStoreReceiptOrderMatches extends Command */ private function backfillPlantingAmounts(string $reportMonth): int { + if ((bool) $this->option('dry-run')) { + $this->info('只读模式：跳过种菜金额写入。'); + return self::SUCCESS; + } $connection = (string) config('upstream_mirrors.connection', 'business_mirror'); $cores = $this->plantingOrderCores(); if ($cores === []) { @@ -985,8 +1013,13 @@ class SyncStoreReceiptOrderMatches extends Command */ private function backfillPromoAmounts(string $month, array $source, array $groups, array $promoOnlyGroups, string $reportMonth): int { + if ((bool) $this->option('dry-run')) { + $this->info('只读计算完成；未写入金额或汇总。'); + return self::SUCCESS; + } $connection = (string) config('upstream_mirrors.connection', 'business_mirror'); $platform = (string) $source['label']; + app(\Modules\EcommerceFinance\Services\StoreReportPaidSalesChanges::class)->markPeriod($reportMonth, $platform); $target = static fn () => DB::connection($connection)->table(self::TARGET_TABLE); $existing = $target() @@ -1046,12 +1079,14 @@ class SyncStoreReceiptOrderMatches extends Command foreach ($promoOnlyGroups as $promoGroup) { $buffer[] = self::promoOnlyRow($reportMonth, $platform, $promoGroup, now()); if (count($buffer) >= 500) { + $this->markPaidSalesRows($buffer); $target()->insertOrIgnore($buffer); $promoRows += count($buffer); $buffer = []; } } if ($buffer !== []) { + $this->markPaidSalesRows($buffer); $target()->insertOrIgnore($buffer); $promoRows += count($buffer); } diff --git a/app/Http/Controllers/Business/AccountMappingActions.php b/app/Http/Controllers/Business/AccountMappingActions.php index 4b89e000..e3091446 100644 --- a/app/Http/Controllers/Business/AccountMappingActions.php +++ b/app/Http/Controllers/Business/AccountMappingActions.php @@ -89,9 +89,10 @@ class AccountMappingActions extends Controller } $values = $this->payloadValues($request); - $this->store->update($id, $values, $this->userId()); + $after = $this->store->update($id, $values, $this->userId()); - $hint = $this->dispatchResync([(string) ($before['platform'] ?? ''), (string) ($values['platform'] ?? '')]); + $hint = AccountMappingStore::requiresReceiptResync($before, $after) + ? $this->dispatchResync([(string) ($before['platform'] ?? ''), (string) ($after['platform'] ?? '')]) : ''; return response()->json(['message' => '已保存。' . $hint] + $this->payload()); } @@ -241,10 +242,7 @@ class AccountMappingActions extends Controller $parsed = $this->importer->parse($file); $result = $this->store->importRows($parsed['rows'], $request->boolean('overwrite'), $this->userId()); $message = "导入完成：识别 {$result['total']} 行，新增 {$result['created']} 条，更新 {$result['updated']} 条，跳过重复 {$result['skipped']} 条。"; - $platforms = array_values(array_unique(array_map( - static fn (array $row): string => (string) ($row['platform'] ?? ''), - $parsed['rows'] - ))); + $platforms = $result['resync_platforms']; $message .= $this->dispatchResync($platforms); return response()->json(['message' => $message, 'result' => $result] + $this->payload()); diff --git a/app/Http/Controllers/Business/Pages.php b/app/Http/Controllers/Business/Pages.php index f4379575..d4ab1ca3 100644 --- a/app/Http/Controllers/Business/Pages.php +++ b/app/Http/Controllers/Business/Pages.php @@ -4791,7 +4791,13 @@ class Pages extends Controller $amountByRow[(int) ($pair['row_number'] ?? 0)] = $pair['amount'] ?? null; } - $results = app(CourierSurchargeSettlementMirrorRepository::class)->decideImportRows($pairs); + $verificationKey = 'courier-surcharge:verify:' . company_id() . ':' . auth()->id() . ':' + . hash('sha256', json_encode($pairs, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR)); + $results = \Illuminate\Support\Facades\Cache::store('upstream_shared')->lock($verificationKey, 300) + ->get(fn () => app(CourierSurchargeSettlementMirrorRepository::class)->decideImportRows($pairs)); + if ($results === false) { + return back()->with('error', '同一份数据正在核验，请等待本次核验完成后再提交。'); + } return response()->streamDownload(function () use ($results, $amountByRow): void { $writer = new \App\Support\StreamingXlsxWriter('导入核验'); diff --git a/database/migrations/2026_10_19_170000_create_work_order_waybill_lookup.php b/database/migrations/2026_10_19_170000_create_work_order_waybill_lookup.php new file mode 100644 index 00000000..a6e7bcbc --- /dev/null +++ b/database/migrations/2026_10_19_170000_create_work_order_waybill_lookup.php @@ -0,0 +1,38 @@ +hasTable('work_order_waybill_lookup')) { + $schema->create('work_order_waybill_lookup', function (Blueprint $table): void { + $table->char('mirror_key', 64)->primary(); + $table->string('waybill', 191); + $table->unsignedBigInteger('source_id'); + $table->index(['waybill', 'source_id'], 'work_order_waybill_source_idx'); + }); + } + if (! $schema->hasTable('work_order_waybill_lookup_states')) { + $schema->create('work_order_waybill_lookup_states', function (Blueprint $table): void { + $table->unsignedTinyInteger('id')->primary(); + $table->char('cursor', 64)->nullable(); + $table->timestamp('completed_at')->nullable(); + $table->timestamp('updated_at')->nullable(); + }); + } + } + + public function down(): void + { + if (config('system_database.mode', 'bootstrap') === 'bootstrap') return; + $schema = Schema::connection((string) config('upstream_mirrors.connection', 'business_mirror')); + $schema->dropIfExists('work_order_waybill_lookup_states'); + $schema->dropIfExists('work_order_waybill_lookup'); + } +}; diff --git a/database/migrations/2026_10_19_171000_track_paid_sales_dirty_dates.php b/database/migrations/2026_10_19_171000_track_paid_sales_dirty_dates.php new file mode 100644 index 00000000..905fd3f0 --- /dev/null +++ b/database/migrations/2026_10_19_171000_track_paid_sales_dirty_dates.php @@ -0,0 +1,50 @@ +hasTable('store_report_paid_sales_dirty_dates')) { + $schema->create('store_report_paid_sales_dirty_dates', function (Blueprint $table): void { + $table->date('occurred_on')->primary(); + $table->uuid('revision'); + }); + } + if ($schema->hasTable('store_receipt_order_match_rows')) { + $indexed = collect($schema->getIndexes('store_receipt_order_match_rows')) + ->contains(fn (array $index) => ($index['columns'][0] ?? null) === 'occurred_date'); + if (! $indexed) { + $connection = $schema->getConnection(); + if ($connection->getDriverName() === 'mysql') { + // Fail quickly on metadata contention and require an online index build. + $wait = (int) $connection->selectOne('SELECT @@SESSION.lock_wait_timeout AS seconds')->seconds; + $connection->statement('SET SESSION lock_wait_timeout=5'); + try { + $table = $connection->getQueryGrammar()->wrapTable('store_receipt_order_match_rows'); + $connection->statement('ALTER TABLE ' . $table . ' ADD INDEX store_receipt_paid_sales_date_index (occurred_date), ALGORITHM=INPLACE, LOCK=NONE'); + } finally { + $connection->statement('SET SESSION lock_wait_timeout=' . $wait); + } + } else { + $schema->table('store_receipt_order_match_rows', fn (Blueprint $table) => $table->index('occurred_date', 'store_receipt_paid_sales_date_index')); + } + } + } + } + + public function down(): void + { + if (config('system_database.mode', 'bootstrap') === 'bootstrap') return; + $schema = Schema::connection((string) config('upstream_mirrors.connection', 'business_mirror')); + $schema->dropIfExists('store_report_paid_sales_dirty_dates'); + if ($schema->hasTable('store_receipt_order_match_rows') && collect($schema->getIndexes('store_receipt_order_match_rows'))->contains('name', 'store_receipt_paid_sales_date_index')) { + $schema->table('store_receipt_order_match_rows', fn (Blueprint $table) => $table->dropIndex('store_receipt_paid_sales_date_index')); + } + } +}; diff --git a/deploy/activate-incremental-release.sh b/deploy/activate-incremental-release.sh index 6ac11dbb..cd67feeb 100644 --- a/deploy/activate-incremental-release.sh +++ b/deploy/activate-incremental-release.sh @@ -209,6 +209,7 @@ CRON_TZ=Asia/Shanghai 30 2 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan finance-ledger:sync-rpt --quiet >/dev/null 2>&1 # 科目映射保存后保留自动重跑队列，每分钟在受限容器内串行消费。 * * * * * root /usr/bin/flock -n /run/caiwu-account-mapping-resync.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh store-receipt:auto-resync --quiet >/dev/null 2>&1 +* * * * * root /usr/bin/flock -n /run/caiwu-account-bills-export.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh account-bills:exports:run --quiet >/dev/null 2>&1 28 1 * * * root /bin/sh /opt/caiwuxitong/run-finance-batch.sh jushuitan:sync-shipping --days=3 >/dev/null 2>&1 # 头条放心购（店铺收款账户账单 › 抖音账户）资金流水：每天 09:30 定点把该渠道所用 # 原始账单镜像追平一次，保证上班前当日/前一日流水已进系统，而不是等每分钟轮转慢慢走。 diff --git a/deploy/activate-release.sh b/deploy/activate-release.sh index d39ce1a3..8d68be89 100644 --- a/deploy/activate-release.sh +++ b/deploy/activate-release.sh @@ -213,6 +213,7 @@ CRON_TZ=Asia/Shanghai 30 2 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan finance-ledger:sync-rpt --quiet >/dev/null 2>&1 # 科目映射保存后保留自动重跑队列，每分钟在受限容器内串行消费。 * * * * * root /usr/bin/flock -n /run/caiwu-account-mapping-resync.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh store-receipt:auto-resync --quiet >/dev/null 2>&1 +* * * * * root /usr/bin/flock -n /run/caiwu-account-bills-export.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh account-bills:exports:run --quiet >/dev/null 2>&1 28 1 * * * root /bin/sh /opt/caiwuxitong/run-finance-batch.sh jushuitan:sync-shipping --days=3 >/dev/null 2>&1 # 头条放心购（店铺收款账户账单 › 抖音账户）资金流水：每天 09:30 定点把该渠道所用 # 原始账单镜像追平一次，保证上班前当日/前一日流水已进系统，而不是等每分钟轮转慢慢走。 diff --git a/deploy/run-finance-batch.sh b/deploy/run-finance-batch.sh index 6d9fa751..5892a9f1 100644 --- a/deploy/run-finance-batch.sh +++ b/deploy/run-finance-batch.sh @@ -4,14 +4,35 @@ set -eu REMOTE_DIR="${CAIWU_REMOTE_DIR:-/opt/caiwuxitong}" LOCK_FILE="${CAIWU_UPSTREAM_LOCK:-/run/caiwu-upstream-sync.lock}" case "${1:-}" in - jushuitan:sync-shipping|store-receipt:order-match:sync|store-receipt:product-cost:sync|store-receipt:auto-resync|order-data:warm-shop-candidates) ;; + jushuitan:sync-shipping|store-receipt:order-match:sync|store-receipt:product-cost:sync|store-receipt:auto-resync|order-data:warm-shop-candidates|business-mirror:index-work-order-waybills|business-mirror:refresh-paid-sales|account-bills:exports:run) ;; *) echo 'Unsupported finance batch command' >&2; exit 64 ;; esac cd "$REMOTE_DIR" -exec 9>"$LOCK_FILE" -if ! flock -w "${CAIWU_BATCH_LOCK_WAIT:-1800}" 9; then - echo 'Upstream synchronization is busy; no batch container started' >&2 - exit 75 +if [ "$1" = 'account-bills:exports:run' ]; then + storage_root="$(docker inspect caiwu_app --format '{{range .Mounts}}{{if eq .Destination "/var/www/html/storage"}}{{.Source}}{{end}}{{end}}')" + [ -n "$storage_root" ] && [ -d "$storage_root" ] || exit 65 + # Check only tiny task metadata; do not start an idle container every minute. + if ! python3 - "$storage_root" <<'PY' +import json,pathlib,sys +for path in pathlib.Path(sys.argv[1]).glob('app/account-bills-exports/*/*.json'): + try: + task = json.loads(path.read_text()) + if isinstance(task, dict) and task.get('status') == 'queued': + sys.exit(0) + except (OSError,ValueError): + continue +sys.exit(1) +PY + then + exit 0 + fi +fi +if [ "$1" != 'account-bills:exports:run' ]; then + exec 9>"$LOCK_FILE" + if ! flock -w "${CAIWU_BATCH_LOCK_WAIT:-1800}" 9; then + echo 'Upstream synchronization is busy; no batch container started' >&2 + exit 75 + fi fi image="$(docker inspect caiwu_app --format '{{.Config.Image}}')" case "$image" in @@ -20,5 +41,5 @@ case "$image" in esac RUN_MIGRATIONS=0 export IMAGE_TAG RUN_MIGRATIONS -# Holds the same lock as scheduled sync and release activation for the full job. +# Mutating jobs hold the upstream lock; read-only exports use their own single-worker lock. docker compose -f compose.jiangling.yml --profile batch run --rm -T --no-deps batch "$@" diff --git a/docs/开发与部署SOP.md b/docs/开发与部署SOP.md index 0780c2f6..9cb55480 100644 --- a/docs/开发与部署SOP.md +++ b/docs/开发与部署SOP.md @@ -278,6 +278,14 @@ sh /opt/caiwuxitong/run-finance-batch.sh store-receipt:order-match:sync YYYY-MM 入口持有现有上游同步锁，使用 `batch` Profile 的独立容器（1 核、3GiB 内存、不使用 Swap、无 Web 健康检查）；正常发布不启动该 Profile。先备份再执行回补，失败时保留原有未完成状态，不重复并行启动任务。店铺候选平时按镜像同步时间增量刷新，02:45 使用 `--full` 全量核对。ClickHouse 系统日志的保留期由 `deploy/clickhouse-resource-policy.xml` 管理；业务表不受日志策略影响。 +查件运单索引 `work_order_waybill_lookup` 只保存原始工单镜像的稳定键、运单号与排序键；工单 JSON 仍为事实来源。升级后先通过上述批入口执行 `business-mirror:index-work-order-waybills --max-batches=0 --pause-ms=50` 完成一次可续传初始化，完成前保留现有核验路径。其后索引与原始工单批次在同一事务中更新，全量校准同时移除失效关联。不得每次导入重新扫描工单 JSON。 + +销量汇总按 `store_report_paid_sales_dirty_dates` 中的受影响日期更新。业绩明细写入与汇总共用互斥锁，改金额、替换账期和删除都必须登记旧日期及新日期。镜像人工校准后，可通过批入口执行 `business-mirror:refresh-paid-sales --full` 全量重建派生汇总；普通刷新无需全表指纹扫描。科目重算每轮默认处理一个平台月份，再释放同步锁，剩余任务由下一轮继续。 + +账单导出保存在原有文件任务队列，每分钟由 `account-bills:exports:run` 消费一个任务，任一时刻只有一个重型导出。该只读任务使用同一受限批 Profile 和独立导出锁，不长期持有上游写同步锁；空队列不启动容器。状态、取消、下载和七天文件保留沿用现有逻辑。 + +生产临时排查先使用元数据和 `EXPLAIN`，业务读取须设置合理的 SQL 执行时限，并使用已有索引缩小范围。不要在网页容器内对归档大表进行无时限计数、模糊查找或循环重试；需要全表核对时先安排受控批处理窗口。 + ### 8.3 开发者需要尚未接入的上游数据时 系统数据库保存团队配置、数据字典和同步状态，共享业务库保存已接入业务行；二者都不是 SQL Server 网络代理。开发探索与正式页面的读取方式不同： diff --git a/modules/EcommerceFinance/Repositories/CourierSurchargeSettlementMirrorRepository.php b/modules/EcommerceFinance/Repositories/CourierSurchargeSettlementMirrorRepository.php index 679bc0c7..dd629e72 100644 --- a/modules/EcommerceFinance/Repositories/CourierSurchargeSettlementMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/CourierSurchargeSettlementMirrorRepository.php @@ -279,7 +279,8 @@ class CourierSurchargeSettlementMirrorRepository } } - $contextComplete = count($unmatchedDeductWaybills) <= self::IMPORT_RETURN_CONTEXT_LIMIT; + $contextComplete = count($unmatchedDeductWaybills) <= self::IMPORT_RETURN_CONTEXT_LIMIT + || app(\Modules\EcommerceFinance\Services\WorkOrderWaybillLookup::class)->ready(); $returnContext = $this->deductOrderContext(array_keys($unmatchedDeductWaybills)); $ruleModes = $this->activeRuleModes(); $quoteRules = $this->importQuoteRules(); @@ -1292,23 +1293,33 @@ class CourierSurchargeSettlementMirrorRepository static fn (string $waybill): bool => $waybill !== '', ))); - if ($waybills === [] || count($waybills) > self::IMPORT_RETURN_CONTEXT_LIMIT) { + $lookup = app(\Modules\EcommerceFinance\Services\WorkOrderWaybillLookup::class); + $indexed = $lookup->ready(); + if ($waybills === [] || (! $indexed && count($waybills) > self::IMPORT_RETURN_CONTEXT_LIMIT)) { return []; } $context = []; - foreach (array_chunk($waybills, 50) as $chunk) { + foreach (array_chunk($waybills, $indexed ? 300 : 50) as $chunk) { try { - $rows = $this->connection->table('js_jst_work_order_info') - ->where('title', '查件') - ->where(function ($query) use ($chunk): void { - foreach ($chunk as $waybill) { - $query->orWhere('businessFieldList', 'like', '%' . str_replace(['%', '_'], '', $waybill) . '%'); - } - }) - ->orderByDesc('Id') - ->get(['businessFieldList']); + if ($indexed) { + $rows = $this->connection->table($lookup::TABLE . ' as lookup') + ->join('js_jst_work_order_info as raw', 'raw._mirror_key', '=', 'lookup.mirror_key') + ->whereIn('lookup.waybill', $chunk) + ->orderByDesc('lookup.source_id')->get(['raw.businessFieldList']); + } else { + // Retain the existing path only until the resumable initial index is ready. + $rows = $this->connection->table('js_jst_work_order_info') + ->where('title', '查件') + ->where(function ($query) use ($chunk): void { + foreach ($chunk as $waybill) { + $query->orWhere('businessFieldList', 'like', '%' . str_replace(['%', '_'], '', $waybill) . '%'); + } + }) + ->orderByDesc('Id') + ->get(['businessFieldList']); + } } catch (Throwable) { continue; } diff --git a/modules/EcommerceFinance/Repositories/GenericBusinessMirrorRepository.php b/modules/EcommerceFinance/Repositories/GenericBusinessMirrorRepository.php index e9b6aec4..6dee74a0 100644 --- a/modules/EcommerceFinance/Repositories/GenericBusinessMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/GenericBusinessMirrorRepository.php @@ -6,6 +6,7 @@ use Illuminate\Database\ConnectionInterface; use Illuminate\Database\DatabaseManager; use Illuminate\Support\Facades\Schema; use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset; +use Modules\EcommerceFinance\Services\WorkOrderWaybillLookup; use RuntimeException; class GenericBusinessMirrorRepository @@ -161,6 +162,9 @@ class GenericBusinessMirrorRepository foreach (array_chunk($mapped, $chunkSize) as $chunk) { $this->connection->table($dataset->targetTable())->upsert($chunk, ['_mirror_key'], $updates); } + if ($dataset->targetTable() === WorkOrderWaybillLookup::SOURCE) { + app(WorkOrderWaybillLookup::class)->storeBatch($mapped); + } $this->connection->table('business_mirror_table_states') ->where('dataset_key', $dataset->key()) @@ -197,6 +201,9 @@ class GenericBusinessMirrorRepository ->orWhere('_mirror_generation', '!=', $generation); }) ->delete(); + if ($dataset->targetTable() === WorkOrderWaybillLookup::SOURCE) { + app(WorkOrderWaybillLookup::class)->prune(); + } } $countState = $this->supportsExactCount() diff --git a/modules/EcommerceFinance/Repositories/JdDailyBillRepository.php b/modules/EcommerceFinance/Repositories/JdDailyBillRepository.php index 37f4a8ee..a30278a1 100644 --- a/modules/EcommerceFinance/Repositories/JdDailyBillRepository.php +++ b/modules/EcommerceFinance/Repositories/JdDailyBillRepository.php @@ -4,6 +4,7 @@ namespace Modules\EcommerceFinance\Repositories; use Illuminate\Database\DatabaseManager; use Illuminate\Database\Query\Builder; +use Illuminate\Support\Facades\Cache; use Modules\EcommerceFinance\JdBillPlatform; use Modules\EcommerceFinance\Services\LedgerRemarkFilter; use RuntimeException; @@ -28,7 +29,8 @@ class JdDailyBillRepository { $periods = $this->periods(); $query = $this->query($filters, $periods); - $summary = (clone $query)->selectRaw('COUNT(*) AS count, COALESCE(SUM(income), 0) AS income, COALESCE(SUM(expense), 0) AS expense')->first(); + $summary = Cache::remember($this->cacheKey('summary', $periods, $filters), 3600, + fn () => (clone $query)->selectRaw('COUNT(*) AS count, COALESCE(SUM(income), 0) AS income, COALESCE(SUM(expense), 0) AS expense')->first()); $total = (int) $summary->count; $page = max(1, $page); $perPage = min(100, max(1, $perPage)); @@ -64,11 +66,24 @@ class JdDailyBillRepository public function filterOptions(): array { - $q = $this->query([], $this->periods()); - return [ - 'shops' => (clone $q)->distinct()->orderBy('store_name')->pluck('store_name')->all(), - 'bill_items' => (clone $q)->distinct()->orderBy('bill_item')->pluck('bill_item')->all(), - ]; + $periods = $this->periods(); + return Cache::remember($this->cacheKey('options', $periods), 3600, function () use ($periods): array { + $q = $this->query([], $periods); + return [ + 'shops' => (clone $q)->distinct()->orderBy('store_name')->pluck('store_name')->all(), + 'bill_items' => (clone $q)->distinct()->orderBy('bill_item')->pluck('bill_item')->all(), + ]; + }); + } + + /** Imported generations are immutable; replacing a month invalidates both caches. */ + private function cacheKey(string $scope, \Illuminate\Support\Collection $periods, array $filters = []): string + { + ksort($filters); + return 'jd-daily-bill:' . $scope . ':v1:' . hash('sha256', json_encode([ + $this->db()->getDatabaseName(), $this->platform, + $periods->pluck('active_generation')->sort()->values()->all(), $filters, + ], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR)); } private function periods(): \Illuminate\Support\Collection diff --git a/modules/EcommerceFinance/Repositories/OrderDataMirrorRepository.php b/modules/EcommerceFinance/Repositories/OrderDataMirrorRepository.php index 7273cf6b..4c37ecf1 100644 --- a/modules/EcommerceFinance/Repositories/OrderDataMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/OrderDataMirrorRepository.php @@ -495,12 +495,17 @@ class OrderDataMirrorRepository $shopNames = $this->selectedShopNames($filters); if ($shopNames !== []) { - // 多店铺：每个店铺做一个模糊匹配再取并集，选中的是完整店铺名时等同于精确匹配。 - $query->where(function (Builder $shop) use ($shopNames): void { - foreach ($shopNames as $name) { - $shop->orWhereRaw("o.shop_name LIKE ? ESCAPE '!'", [$this->contains($name)]); - } - }); + $known = (array) Cache::get(self::SHOP_NAMES_CACHE_KEY, []); + if (! empty($filters['shop_names_exact']) && array_diff($shopNames, $known) === []) { + $query->whereIn('o.shop_name', $shopNames); + } else { + // Free text and legacy callers keep their existing contains semantics. + $query->where(function (Builder $shop) use ($shopNames): void { + foreach ($shopNames as $name) { + $shop->orWhereRaw("o.shop_name LIKE ? ESCAPE '!'", [$this->contains($name)]); + } + }); + } } if (($filters['sku_id'] ?? '') !== '') { // 商品编码同样支持批量：每个编码一组"商品/款式/店铺编码"匹配，多组取并集。 diff --git a/modules/EcommerceFinance/Services/AccountBillsExportTasks.php b/modules/EcommerceFinance/Services/AccountBillsExportTasks.php index 7c1b2048..43f04377 100644 --- a/modules/EcommerceFinance/Services/AccountBillsExportTasks.php +++ b/modules/EcommerceFinance/Services/AccountBillsExportTasks.php @@ -11,7 +11,7 @@ use Throwable; * 店铺收款账户账单的「导出任务」：后台生成 CSV 落盘，页面按任务列表查看状态并下载。 * * 任务与文件都存在 storage/app/account-bills-exports/{userId}/ 下（JSON 记录状态 + CSV 文件）， - * 不依赖浏览器保持连接，也不需要队列表：请求里用 detach 的 php-cli 进程跑导出。 + * 不依赖浏览器保持连接；生产由受限批处理容器逐个消费，本地由单个 PHP 进程消费。 */ class AccountBillsExportTasks { @@ -29,7 +29,12 @@ class AccountBillsExportTasks public function baseDir(int $userId): string { - return storage_path('app/account-bills-exports/' . $userId); + return $this->rootDir() . '/' . $userId; + } + + public function rootDir(): string + { + return storage_path('app/account-bills-exports'); } public function jsonPath(array $task): string @@ -167,6 +172,8 @@ class AccountBillsExportTasks /** 后台执行导出（脱离当前请求）。 */ public function dispatch(array $task): void { + // Production consumes the durable file queue from the limited batch profile. + if (config('upstream_pipeline.batch_worker_enabled', false)) return; $php = PHP_BINDIR . '/php'; if (! is_file($php)) { $php = 'php'; @@ -174,7 +181,7 @@ class AccountBillsExportTasks $artisan = base_path('artisan'); $command = escapeshellcmd($php) . ' ' . escapeshellarg($artisan) - . ' account-bills:export-run ' . escapeshellarg((string) $task['token']) + . ' account-bills:exports:run --max-jobs=0' . ' > /dev/null 2>&1 &'; try { @@ -185,4 +192,15 @@ class AccountBillsExportTasks // 起不了后台进程时保持 queued，页面会提示重试。 } } + + public function nextQueued(): ?array + { + $next = null; + foreach (glob($this->rootDir() . '/*/*.json') ?: [] as $path) { + $task = json_decode((string) file_get_contents($path), true); + if (! is_array($task) || ($task['status'] ?? '') !== 'queued') continue; + if ($next === null || strcmp((string) $task['planned_at'], (string) $next['planned_at']) < 0) $next = $task; + } + return $next; + } } diff --git a/modules/EcommerceFinance/Services/AccountMappingResyncDispatcher.php b/modules/EcommerceFinance/Services/AccountMappingResyncDispatcher.php index 13aa7922..955f6189 100644 --- a/modules/EcommerceFinance/Services/AccountMappingResyncDispatcher.php +++ b/modules/EcommerceFinance/Services/AccountMappingResyncDispatcher.php @@ -49,20 +49,20 @@ class AccountMappingResyncDispatcher } $mirror = DB::connection((string) config('upstream_mirrors.connection', 'business_mirror')); - $months = $mirror->table('store_receipt_order_match_rows') + $periods = $mirror->table('store_receipt_order_match_rows') ->whereIn('platform', array_values($targets)) - ->selectRaw("DATE_FORMAT(report_month, '%Y-%m') as ym") + ->select(['platform', 'report_month']) ->distinct() - ->orderBy('ym') - ->pluck('ym') - ->all(); - if ($months === []) { + ->orderBy('report_month') + ->get()->groupBy('platform'); + if ($periods->isEmpty()) { return ['jobs' => 0]; } $now = now(); $jobs = 0; foreach ($targets as $key => $label) { + $months = $periods->get($label, collect())->map(fn ($row) => substr((string) $row->report_month, 0, 7))->unique(); foreach ($months as $month) { // running 时再次保存映射，会把状态改回 pending；当前进程完成时发现状态已变， // 不会覆盖成 done，于是会紧接着再跑一遍，避免漏掉刚保存的新规则。 diff --git a/modules/EcommerceFinance/Services/AccountMappingStore.php b/modules/EcommerceFinance/Services/AccountMappingStore.php index d982e9c0..b8cc4765 100644 --- a/modules/EcommerceFinance/Services/AccountMappingStore.php +++ b/modules/EcommerceFinance/Services/AccountMappingStore.php @@ -188,12 +188,24 @@ class AccountMappingStore throw new RuntimeException('未找到要修改的规则。'); } - $this->persist($rules); - $this->logEvent('update', $id, $before, $updated, $userId); + if ($before !== $updated) { + $this->persist($rules); + $this->logEvent('update', $id, $before, $updated, $userId); + } return $updated; } + /** Fee memo is display-only; all other fields retain conservative invalidation. */ + public static function requiresReceiptResync(?array $before, ?array $after): bool + { + foreach (self::FIELDS as $field) { + if ($field === 'feeMemo') continue; + if (trim((string) ($before[$field] ?? '')) !== trim((string) ($after[$field] ?? ''))) return true; + } + return false; + } + public function destroy(int $id, ?int $userId = null): bool { $rules = $this->all(); @@ -222,7 +234,7 @@ class AccountMappingStore * 导入：按「平台 + 账单项目 + 财务备注」判重。 * * @param array> $rows - * @return array{created:int,updated:int,skipped:int,duplicates:array,total:int} + * @return array{created:int,updated:int,skipped:int,duplicates:array,total:int,resync_platforms:list} */ public function importRows(array $rows, bool $overwrite, ?int $userId = null): array { @@ -242,6 +254,7 @@ class AccountMappingStore $skipped = 0; $duplicates = []; + $resyncPlatforms = []; foreach ($rows as $row) { $values = $this->normalize($row); if ($this->isEmptyRow($values)) { @@ -256,7 +269,16 @@ class AccountMappingStore continue; } $index = $byKey[$key]; - $rules[$index] = ['id' => $rules[$index]['id']] + $values; + $replacement = ['id' => $rules[$index]['id']] + $values; + if ($rules[$index] === $replacement) { + $skipped++; + continue; + } + if (self::requiresReceiptResync($rules[$index], $replacement)) { + $resyncPlatforms[$rules[$index]['platform']] = true; + $resyncPlatforms[$replacement['platform']] = true; + } + $rules[$index] = $replacement; $updated++; continue; } @@ -265,6 +287,7 @@ class AccountMappingStore $byKey[$key] = count($rules) - 1; $nextId++; $created++; + $resyncPlatforms[$values['platform']] = true; } if ($created + $updated > 0) { @@ -282,6 +305,7 @@ class AccountMappingStore 'skipped' => $skipped, 'duplicates' => array_slice($duplicates, 0, 20), 'total' => count($rows), + 'resync_platforms' => array_keys($resyncPlatforms), ]; } @@ -510,6 +534,7 @@ class AccountMappingStore /** @param array> $rules */ private function persist(array $rules): void { + self::forgetMemo(); if (! $this->rulesTableReady()) { $this->persistFile($rules); diff --git a/modules/EcommerceFinance/Services/StoreReportMirrorSynchronizer.php b/modules/EcommerceFinance/Services/StoreReportMirrorSynchronizer.php index 6f4af0ba..c55fb4e3 100644 --- a/modules/EcommerceFinance/Services/StoreReportMirrorSynchronizer.php +++ b/modules/EcommerceFinance/Services/StoreReportMirrorSynchronizer.php @@ -457,6 +457,15 @@ class StoreReportMirrorSynchronizer * @return array{status:string,source_rows:int,stored_rows:int,changed:bool} */ public function syncPaidSales(bool $force = false): array + { + $result = \Illuminate\Support\Facades\Cache::store('upstream_shared')->lock('store-report:paid-sales', 600) + ->get(fn () => $this->syncPaidSalesUnlocked($force)); + return $result === false + ? ['status' => 'succeeded', 'source_rows' => 0, 'stored_rows' => 0, 'changed' => false] + : $result; + } + + private function syncPaidSalesUnlocked(bool $force): array { if (! Schema::connection($this->connection->getName()) ->hasTable(StoreReportRepository::PAID_SALES_SOURCE_TABLE)) { @@ -464,55 +473,66 @@ class StoreReportMirrorSynchronizer } $state = $this->state(StoreReportRepository::PAID_SALES_DATASET); + $changes = app(StoreReportPaidSalesChanges::class); try { - $fingerprint = $this->paidSalesFingerprint(); - $active = trim((string) ($state->active_generation ?? '')); - if (! $force && $active !== '' && ! empty($state->initial_completed_at) - && (string) ($state->source_fingerprint ?? '') === $fingerprint) { - return [ - 'status' => 'succeeded', - 'source_rows' => 0, - 'stored_rows' => (int) ($state->row_count ?? 0), - 'changed' => false, - ]; + if (! $force && $changes->available() && ! empty($state->active_generation) + && ! empty($state->initial_completed_at) + && str_starts_with((string) ($state->source_fingerprint ?? ''), StoreReportPaidSalesChanges::VERSION)) { + return $this->syncPaidSalesDates($state, $changes); } + return $this->connection->transaction(function () use ($state, $force, $changes): array { + $pending = $changes->available() ? $changes->pending() : []; + $fingerprint = $changes->available() + ? StoreReportPaidSalesChanges::VERSION . Str::uuid() : $this->paidSalesFingerprint(); + $active = trim((string) ($state->active_generation ?? '')); + if (! $force && $active !== '' && ! empty($state->initial_completed_at) + && (string) ($state->source_fingerprint ?? '') === $fingerprint) { + return [ + 'status' => 'succeeded', + 'source_rows' => 0, + 'stored_rows' => (int) ($state->row_count ?? 0), + 'changed' => false, + ]; + } - $now = now(); - $generation = (string) Str::uuid(); - $this->connection->statement( - 'INSERT INTO ' . StoreReportRepository::PAID_SALES_TABLE - . ' (generation, store_name, occurred_on, paid_amount, order_count, updated_at)' - . ' SELECT ?, store_name, occurred_date,' - . ' SUM(COALESCE(income_amount, 0) + COALESCE(discount_amount, 0)' - . ' + COALESCE(offset_amount, 0) + COALESCE(refund_amount, 0)), COUNT(*), ?' - . ' FROM ' . StoreReportRepository::PAID_SALES_SOURCE_TABLE - . " WHERE store_name IS NOT NULL AND store_name <> '' AND occurred_date IS NOT NULL" - . ' GROUP BY store_name, occurred_date', - [$generation, $now->format('Y-m-d H:i:s')], - ); - $stored = (int) $this->connection->table(StoreReportRepository::PAID_SALES_TABLE) - ->where('generation', $generation)->count(); - $sourceRows = (int) ($this->connection->table(StoreReportRepository::PAID_SALES_SOURCE_TABLE)->count()); - $this->connection->table('store_report_import_states') - ->where('dataset_key', StoreReportRepository::PAID_SALES_DATASET) - ->update([ - 'status' => 'succeeded', - 'run_mode' => 'full', - 'active_generation' => $generation, - 'pending_generation' => null, - 'row_count' => $stored, - 'source_fingerprint' => $fingerprint, - 'initial_completed_at' => $state->initial_completed_at ?? $now, - 'last_attempted_at' => $now, - 'last_succeeded_at' => $now, - 'last_error_code' => null, - 'updated_at' => $now, - ]); - // 版本切换成功后再清理上一版，避免中途失败时页面读到被清空的当前版本。 - $this->connection->table(StoreReportRepository::PAID_SALES_TABLE) - ->where('generation', '<>', $generation)->delete(); + $now = now(); + $generation = (string) Str::uuid(); + $this->connection->statement( + 'INSERT INTO ' . StoreReportRepository::PAID_SALES_TABLE + . ' (generation, store_name, occurred_on, paid_amount, order_count, updated_at)' + . ' SELECT ?, store_name, occurred_date,' + . ' SUM(COALESCE(income_amount, 0) + COALESCE(discount_amount, 0)' + . ' + COALESCE(offset_amount, 0) + COALESCE(refund_amount, 0)), COUNT(*), ?' + . ' FROM ' . StoreReportRepository::PAID_SALES_SOURCE_TABLE + . " WHERE store_name IS NOT NULL AND store_name <> '' AND occurred_date IS NOT NULL" + . ' GROUP BY store_name, occurred_date', + [$generation, $now->format('Y-m-d H:i:s')], + ); + $stored = (int) $this->connection->table(StoreReportRepository::PAID_SALES_TABLE) + ->where('generation', $generation)->count(); + $sourceRows = (int) ($this->connection->table(StoreReportRepository::PAID_SALES_SOURCE_TABLE)->count()); + $this->connection->table('store_report_import_states') + ->where('dataset_key', StoreReportRepository::PAID_SALES_DATASET) + ->update([ + 'status' => 'succeeded', + 'run_mode' => 'full', + 'active_generation' => $generation, + 'pending_generation' => null, + 'row_count' => $stored, + 'source_fingerprint' => $fingerprint, + 'initial_completed_at' => $state->initial_completed_at ?? $now, + 'last_attempted_at' => $now, + 'last_succeeded_at' => $now, + 'last_error_code' => null, + 'updated_at' => $now, + ]); + // 版本切换成功后再清理上一版，避免中途失败时页面读到被清空的当前版本。 + $this->connection->table(StoreReportRepository::PAID_SALES_TABLE) + ->where('generation', '<>', $generation)->delete(); + if ($changes->available()) $changes->acknowledge($pending); - return ['status' => 'succeeded', 'source_rows' => $sourceRows, 'stored_rows' => $stored, 'changed' => true]; + return ['status' => 'succeeded', 'source_rows' => $sourceRows, 'stored_rows' => $stored, 'changed' => true]; + }); } catch (Throwable $exception) { $this->connection->table('store_report_import_states') ->where('dataset_key', StoreReportRepository::PAID_SALES_DATASET) @@ -532,6 +552,41 @@ class StoreReportMirrorSynchronizer } } + private function syncPaidSalesDates(object $state, StoreReportPaidSalesChanges $changes): array + { + $pending = $changes->pending(); + if ($pending === []) { + return ['status' => 'succeeded', 'source_rows' => 0, 'stored_rows' => (int) $state->row_count, 'changed' => false]; + } + $dates = array_column($pending, 'occurred_on'); + $generation = (string) $state->active_generation; + return $this->connection->transaction(function () use ($changes, $pending, $dates, $generation): array { + $this->connection->table(StoreReportRepository::PAID_SALES_TABLE) + ->where('generation', $generation)->whereIn('occurred_on', $dates)->delete(); + $this->connection->statement( + 'INSERT INTO ' . StoreReportRepository::PAID_SALES_TABLE + . ' (generation, store_name, occurred_on, paid_amount, order_count, updated_at)' + . ' SELECT ?, store_name, occurred_date,' + . ' SUM(COALESCE(income_amount, 0) + COALESCE(discount_amount, 0)' + . ' + COALESCE(offset_amount, 0) + COALESCE(refund_amount, 0)), COUNT(*), ?' + . ' FROM ' . StoreReportRepository::PAID_SALES_SOURCE_TABLE + . " WHERE store_name IS NOT NULL AND store_name <> '' AND occurred_date IN (" + . implode(',', array_fill(0, count($dates), '?')) . ') GROUP BY store_name, occurred_date', + [$generation, now()->format('Y-m-d H:i:s'), ...$dates], + ); + $stored = (int) $this->connection->table(StoreReportRepository::PAID_SALES_TABLE)->where('generation', $generation)->count(); + $sourceRows = (int) $this->connection->table(StoreReportRepository::PAID_SALES_SOURCE_TABLE)->whereIn('occurred_date', $dates)->count(); + $this->connection->table('store_report_import_states') + ->where('dataset_key', StoreReportRepository::PAID_SALES_DATASET)->update([ + 'status' => 'succeeded', 'run_mode' => 'incremental', 'row_count' => $stored, + 'source_fingerprint' => StoreReportPaidSalesChanges::VERSION . Str::uuid(), + 'last_attempted_at' => now(), 'last_succeeded_at' => now(), 'last_error_code' => null, 'updated_at' => now(), + ]); + $changes->acknowledge($pending); + return ['status' => 'succeeded', 'source_rows' => $sourceRows, 'stored_rows' => $stored, 'changed' => true]; + }); + } + /** 业绩明细是否变化的指纹：行数 + 最近写入时间 + 销量口径四项金额合计。 */ private function paidSalesFingerprint(): string { diff --git a/modules/EcommerceFinance/Services/StoreReportPaidSalesChanges.php b/modules/EcommerceFinance/Services/StoreReportPaidSalesChanges.php new file mode 100644 index 00000000..17da1a4e --- /dev/null +++ b/modules/EcommerceFinance/Services/StoreReportPaidSalesChanges.php @@ -0,0 +1,62 @@ +connection = $database->connection((string) config('upstream_mirrors.connection', 'business_mirror')); + } + + public function available(): bool + { + return $this->connection->getSchemaBuilder()->hasTable(self::TABLE); + } + + public function markPeriod(string $reportMonth, string $platform): void + { + if (! $this->available()) return; + $dates = $this->connection->table(StoreReportRepository::PAID_SALES_SOURCE_TABLE) + ->where('report_month', substr($reportMonth, 0, 7) . '-01')->where('platform', $platform) + ->whereNotNull('occurred_date')->distinct()->pluck('occurred_date'); + $this->markDates($dates->all()); + } + + public function markDates(array $dates): void + { + if (! $this->available()) return; + $revision = (string) Str::uuid(); + $rows = array_map(fn ($date) => ['occurred_on' => $date, 'revision' => $revision], array_values(array_unique(array_filter($dates)))); + foreach (array_chunk($rows, 500) as $chunk) { + $this->connection->table(self::TABLE)->upsert($chunk, ['occurred_on'], ['revision']); + } + } + + public function pending(): array + { + return $this->connection->table(self::TABLE)->orderBy('occurred_on')->get()->map(fn ($row) => (array) $row)->all(); + } + + public function acknowledge(array $snapshot): void + { + foreach (array_chunk($snapshot, 100) as $chunk) { + // A write arriving during aggregation retains its newer revision for the next run. + $this->connection->table(self::TABLE)->where(function ($query) use ($chunk): void { + foreach ($chunk as $row) { + $query->orWhere(fn ($match) => $match->where('occurred_on', $row['occurred_on'])->where('revision', $row['revision'])); + } + })->delete(); + } + } +} diff --git a/modules/EcommerceFinance/Services/WorkOrderWaybillLookup.php b/modules/EcommerceFinance/Services/WorkOrderWaybillLookup.php new file mode 100644 index 00000000..aed305b9 --- /dev/null +++ b/modules/EcommerceFinance/Services/WorkOrderWaybillLookup.php @@ -0,0 +1,102 @@ +connection = $database->connection((string) config('upstream_mirrors.connection', 'business_mirror')); + } + + public function available(): bool + { + return $this->available ??= $this->connection->getSchemaBuilder()->hasTable(self::TABLE) + && $this->connection->getSchemaBuilder()->hasTable(self::STATE); + } + + public function ready(): bool + { + return $this->available() && $this->connection->table(self::STATE)->where('id', 1)->whereNotNull('completed_at')->exists(); + } + + /** Called inside the raw mirror's transaction, including changes to title/waybill. */ + public function storeBatch(array $rows): void + { + if ($rows === [] || ! $this->available()) return; + $keys = []; + $values = []; + foreach ($rows as $row) { + $row = (array) $row; + $key = (string) $row['_mirror_key']; + $keys[] = $key; + if (($row['title'] ?? '') !== '查件') continue; + $fields = json_decode((string) ($row['businessFieldList'] ?? ''), true); + if (! is_array($fields)) continue; + $waybill = ''; + // Same last-field-wins semantics as the existing settlement parser. + foreach ($fields as $field) { + if (is_array($field) && ($field['fieldId'] ?? null) === 'lIdField') { + $waybill = trim((string) ($field['value'] ?? '')); + } + } + if ($waybill !== '' && mb_strlen($waybill) <= 191) { + $values[] = ['mirror_key' => $key, 'waybill' => $waybill, 'source_id' => (int) ($row['Id'] ?? 0)]; + } + } + // Remove renamed/invalid records too, so an old waybill never survives an update. + $this->connection->table(self::TABLE)->whereIn('mirror_key', $keys)->delete(); + foreach (array_chunk($values, 500) as $chunk) { + $this->connection->table(self::TABLE)->insert($chunk); + } + } + + public function prune(): void + { + if (! $this->available()) return; + $this->connection->table(self::TABLE)->whereNotExists(function ($query): void { + $query->selectRaw('1')->from(self::SOURCE) + ->whereColumn(self::SOURCE . '._mirror_key', self::TABLE . '.mirror_key'); + })->delete(); + } + + /** Resume a bounded keyset pass; production invokes this through the global sync lock. */ + public function backfill(int $batchSize = 500, int $maxBatches = 2, int $pauseMs = 50): array + { + if (! $this->available()) throw new \RuntimeException('work_order_waybill_lookup_not_migrated'); + $state = $this->connection->table(self::STATE)->where('id', 1)->first(); + if ($state?->completed_at) return ['complete' => true, 'rows' => 0, 'batches' => 0]; + $cursor = $state?->cursor; + $rows = $batches = 0; + $batchSize = min(max(1, $batchSize), 1000); + do { + $batch = $this->connection->table(self::SOURCE)->when($cursor, fn ($query) => $query->where('_mirror_key', '>', $cursor)) + ->orderBy('_mirror_key')->limit($batchSize)->get(['_mirror_key', 'Id', 'title', 'businessFieldList']); + $complete = $batch->count() < $batchSize; + $this->connection->transaction(function () use ($batch, $complete, &$cursor): void { + $this->storeBatch($batch->all()); + $cursor = $batch->last()?->_mirror_key ?? $cursor; + if ($complete) $this->prune(); + $this->connection->table(self::STATE)->updateOrInsert(['id' => 1], [ + 'cursor' => $cursor, 'completed_at' => $complete ? now() : null, 'updated_at' => now(), + ]); + }); + $rows += $batch->count(); + $batches++; + if (! $complete && $pauseMs > 0) usleep(min($pauseMs, 1000) * 1000); + } while (! $complete && ($maxBatches === 0 || $batches < $maxBatches)); + + return compact('complete', 'rows', 'batches'); + } +} diff --git a/modules/EcommerceFinance/View/OrderDataPageData.php b/modules/EcommerceFinance/View/OrderDataPageData.php index 7c234006..88c7f2d1 100644 --- a/modules/EcommerceFinance/View/OrderDataPageData.php +++ b/modules/EcommerceFinance/View/OrderDataPageData.php @@ -81,6 +81,7 @@ class OrderDataPageData // shop_name 保留单值形态，兼容只认单值的调用方；多选统一用 shop_names。 'shop_name' => $shopNames[0] ?? '', 'shop_names' => $shopNames, + 'shop_names_exact' => ! empty($input['shop_names_exact']) || is_array($input['shop_name'] ?? null), 'sku_id' => $this->batchText($input['sku_id'] ?? null), ]; } diff --git a/tests/Feature/Business/AccountMappingPageTest.php b/tests/Feature/Business/AccountMappingPageTest.php index d15567ac..13d85352 100644 --- a/tests/Feature/Business/AccountMappingPageTest.php +++ b/tests/Feature/Business/AccountMappingPageTest.php @@ -31,6 +31,9 @@ class AccountMappingPageTest extends FeatureTestCase // 只影响 storage_path()，视图编译目录已在启动时解析，不受影响 $this->app->useStoragePath($this->storageRoot); + // Rules now live in the in-memory test database; seed it independently of local files. + app(AccountMappingStore::class)->reset(); + \Illuminate\Support\Facades\DB::table(AccountMappingStore::EVENTS_TABLE)->delete(); } protected function tearDown(): void diff --git a/tests/Unit/Deployment/FinanceBatchIsolationTest.php b/tests/Unit/Deployment/FinanceBatchIsolationTest.php index 230fb40f..f6ef17f3 100644 --- a/tests/Unit/Deployment/FinanceBatchIsolationTest.php +++ b/tests/Unit/Deployment/FinanceBatchIsolationTest.php @@ -35,16 +35,39 @@ class FinanceBatchIsolationTest extends TestCase private function runBatch(string $directory, array $arguments): Process { file_put_contents($directory . '/docker', '#!/bin/sh' . "\n" - . 'if [ "$1" = inspect ]; then echo caiwu_akaunting:123456789abc; exit 0; fi' . "\n" + . 'if [ "$1" = inspect ]; then case "$4" in *Mounts*) echo "$CAIWU_STORAGE_ROOT" ;; *) echo caiwu_akaunting:123456789abc ;; esac; exit 0; fi' . "\n" . 'printf "%s\n" "image=$IMAGE_TAG" "$@" > "$CAIWU_CAPTURE"' . "\n"); chmod($directory . '/docker', 0700); $process = new Process(array_merge(['sh', dirname(__DIR__, 3) . '/deploy/run-finance-batch.sh'], $arguments), $directory, [ 'PATH' => $directory . ':' . getenv('PATH'), 'CAIWU_REMOTE_DIR' => $directory, 'CAIWU_UPSTREAM_LOCK' => $directory . '/sync.lock', 'CAIWU_BATCH_LOCK_WAIT' => '0.1', 'CAIWU_CAPTURE' => $directory . '/called', + 'CAIWU_STORAGE_ROOT' => $directory . '/storage', ]); $process->run(); return $process; } + + public function test_exports_skip_idle_containers_and_can_read_while_upstream_writer_holds_lock(): void + { + $directory = sys_get_temp_dir() . '/caiwu-export-batch-' . bin2hex(random_bytes(6)); + mkdir($directory . '/storage/app/account-bills-exports/1', 0777, true); + $lock = fopen($directory . '/sync.lock', 'c'); + flock($lock, LOCK_EX); + $task = $directory . '/storage/app/account-bills-exports/1/task.json'; + try { + $this->assertTrue($this->runBatch($directory, ['account-bills:exports:run'])->isSuccessful()); + $this->assertFileDoesNotExist($directory . '/called'); + file_put_contents($task, json_encode(['status' => 'queued'])); + $this->assertTrue($this->runBatch($directory, ['account-bills:exports:run'])->isSuccessful()); + $this->assertFileExists($directory . '/called'); + } finally { + fclose($lock); + if (is_file($task)) unlink($task); + foreach (['/storage/app/account-bills-exports/1', '/storage/app/account-bills-exports', '/storage/app', '/storage'] as $path) rmdir($directory . $path); + foreach (glob($directory . '/*') as $path) unlink($path); + rmdir($directory); + } + } } diff --git a/tests/Unit/EcommerceFinance/AccountBillsExportQueueTest.php b/tests/Unit/EcommerceFinance/AccountBillsExportQueueTest.php new file mode 100644 index 00000000..c65a4909 --- /dev/null +++ b/tests/Unit/EcommerceFinance/AccountBillsExportQueueTest.php @@ -0,0 +1,46 @@ + ['driver' => 'array'], 'upstream_pipeline.batch_worker_enabled' => true]); + $directory = sys_get_temp_dir() . '/caiwu-export-queue-' . bin2hex(random_bytes(6)); + $tasks = new class($directory) extends AccountBillsExportTasks { + public function __construct(private string $directory) {} + public function rootDir(): string { return $this->directory; } + }; + $this->app->instance(AccountBillsExportTasks::class, $tasks); + $a = $tasks->create(11, 'test', '测试', [], [['field' => 'test', 'label' => '测试']], 0); + $b = $tasks->create(11, 'test', '测试', [], [['field' => 'test', 'label' => '测试']], 0); + $a = $tasks->patch($a, ['planned_at' => '2026-10-01 00:00:00', 'cancel' => true]); + $b = $tasks->patch($b, ['planned_at' => '2026-10-02 00:00:00', 'cancel' => true]); + $lock = Cache::store('upstream_shared')->lock('account-bills:export-worker', 10800); + try { + $tasks->dispatch($a); + $this->assertSame('queued', $tasks->find(11, $a['token'])['status']); + $this->assertTrue($lock->get()); + $this->artisan('account-bills:exports:run')->assertExitCode(0); + $this->assertSame('queued', $tasks->find(11, $a['token'])['status']); + $this->assertSame('queued', $tasks->find(11, $b['token'])['status']); + $lock->release(); + $this->artisan('account-bills:exports:run')->assertExitCode(0); + $this->assertSame('cancelled', $tasks->find(11, $a['token'])['status']); + $this->assertSame('queued', $tasks->find(11, $b['token'])['status']); + $this->artisan('account-bills:exports:run')->assertExitCode(0); + $this->assertSame('cancelled', $tasks->find(11, $b['token'])['status']); + $this->assertNull($tasks->nextQueued()); + } finally { + $lock->release(); + foreach (glob($directory . '/11/*') ?: [] as $file) unlink($file); + rmdir($directory . '/11'); + rmdir($directory); + } + } +} diff --git a/tests/Unit/EcommerceFinance/AccountMappingResyncBudgetTest.php b/tests/Unit/EcommerceFinance/AccountMappingResyncBudgetTest.php new file mode 100644 index 00000000..bf2d78f2 --- /dev/null +++ b/tests/Unit/EcommerceFinance/AccountMappingResyncBudgetTest.php @@ -0,0 +1,50 @@ + ['driver' => 'array']]); + $table = AccountMappingResyncDispatcher::JOBS_TABLE; + Schema::dropIfExists($table); + Schema::create($table, function (Blueprint $table): void { + $table->increments('id'); + $table->string('platform'); + $table->string('month'); + $table->string('status'); + $table->unsignedInteger('attempts')->default(0); + $table->text('message')->nullable(); + $table->timestamps(); + }); + foreach (['2026-09', '2026-10'] as $month) { + DB::table($table)->insert(['platform' => 'jingxi', 'month' => $month, 'status' => 'pending', 'created_at' => now(), 'updated_at' => now()]); + } + $command = new class extends RunAccountMappingResync { + public bool $edited = false; + protected function runSync(object $job): array + { + if ($this->edited) { + DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)->where('id', $job->id)->update(['status' => 'pending']); + } + return [true, 'OK']; + } + }; + $this->app->make(Kernel::class)->registerCommand($command); + $this->artisan('store-receipt:auto-resync')->assertSuccessful(); + $this->assertSame(['done', 'pending'], DB::table($table)->orderBy('id')->pluck('status')->all()); + $command->edited = true; + $this->artisan('store-receipt:auto-resync')->assertSuccessful(); + $this->assertSame(['done', 'pending'], DB::table($table)->orderBy('id')->pluck('status')->all()); + $this->assertSame([1, 1], DB::table($table)->orderBy('id')->pluck('attempts')->all()); + } +} diff --git a/tests/Unit/EcommerceFinance/AccountMappingResyncDispatcherTest.php b/tests/Unit/EcommerceFinance/AccountMappingResyncDispatcherTest.php index d54f45ef..3c865d21 100644 --- a/tests/Unit/EcommerceFinance/AccountMappingResyncDispatcherTest.php +++ b/tests/Unit/EcommerceFinance/AccountMappingResyncDispatcherTest.php @@ -3,11 +3,75 @@ namespace Tests\Unit\EcommerceFinance; use Modules\EcommerceFinance\Services\AccountMappingResyncDispatcher; +use Modules\EcommerceFinance\Services\AccountMappingStore; +use Illuminate\Database\Schema\Blueprint; +use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Schema; use ReflectionClass; use Tests\TestCase; class AccountMappingResyncDispatcherTest extends TestCase { + public function test_identical_and_display_only_changes_do_not_require_receipt_rebuilds(): void + { + $rule = ['platform' => '京东', 'item' => '测试货款', 'memo' => '收入', 'basis' => '收入', 'feeMemo' => '说明']; + $this->assertFalse(AccountMappingStore::requiresReceiptResync($rule, $rule)); + $this->assertFalse(AccountMappingStore::requiresReceiptResync($rule, array_replace($rule, ['feeMemo' => '更新说明']))); + $this->assertTrue(AccountMappingStore::requiresReceiptResync($rule, array_replace($rule, ['memo' => '退款']))); + $this->assertTrue(AccountMappingStore::requiresReceiptResync($rule, array_replace($rule, ['platform' => '抖音']))); + } + + public function test_only_existing_platform_month_pairs_are_enqueued(): void + { + config(['upstream_pipeline.batch_worker_enabled' => true]); + $schema = Schema::connection('business_mirror'); + $schema->dropIfExists('store_receipt_order_match_rows'); + $schema->create('store_receipt_order_match_rows', function (Blueprint $table): void { + $table->string('platform'); + $table->date('report_month'); + }); + if (! Schema::hasTable(AccountMappingResyncDispatcher::JOBS_TABLE)) { + Schema::create(AccountMappingResyncDispatcher::JOBS_TABLE, function (Blueprint $table): void { + $table->increments('id'); + $table->string('platform'); + $table->string('month'); + $table->string('status'); + $table->text('message')->nullable(); + $table->timestamps(); + }); + } + DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)->delete(); + DB::connection('business_mirror')->table('store_receipt_order_match_rows')->insert([ + ['platform' => '淘系支付宝', 'report_month' => '2026-09-01'], + ['platform' => '淘系微信', 'report_month' => '2026-10-01'], + ]); + $this->assertSame(['jobs' => 2], app(AccountMappingResyncDispatcher::class)->dispatchFor(['淘宝'])); + $this->assertSame(['2026-09'], DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)->where('platform', 'taobao-alipay')->pluck('month')->all()); + $this->assertSame(['2026-10'], DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)->where('platform', 'taobao-wechat')->pluck('month')->all()); + } + + public function test_reimporting_unchanged_rules_does_not_invalidate_history(): void + { + $originalStorage = $this->app->storagePath(); + $directory = sys_get_temp_dir() . '/mapping-resync-' . bin2hex(random_bytes(6)); + \Illuminate\Support\Facades\File::makeDirectory($directory . '/app', 0777, true); + $this->app->useStoragePath($directory); + try { + $store = app(AccountMappingStore::class); + $store->reset(); + $rows = [['platform' => '京东', 'item' => '测试资源优化项目', 'memo' => '收入']]; + $this->assertSame(['京东'], $store->importRows($rows, true)['resync_platforms']); + $same = $store->importRows($rows, true); + $this->assertSame(0, $same['updated']); + $this->assertSame([], $same['resync_platforms']); + $this->assertSame([], $store->importRows([array_replace($rows[0], ['feeMemo' => '更新说明'])], true)['resync_platforms']); + $this->assertSame(['京东'], $store->importRows([array_replace($rows[0], ['basis' => AccountMappingStore::BASIS_IN])], true)['resync_platforms']); + } finally { + $this->app->useStoragePath($originalStorage); + \Illuminate\Support\Facades\File::deleteDirectory($directory); + (new \ReflectionProperty(AccountMappingStore::class, 'allMemo'))->setValue(null); + } + } public function test_production_leaves_queue_processing_to_the_limited_batch_worker(): void { config(['upstream_pipeline.batch_worker_enabled' => true]); diff --git a/tests/Unit/EcommerceFinance/CourierSurchargeSettlementMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/CourierSurchargeSettlementMirrorRepositoryTest.php index 5cade37e..294f95bd 100644 --- a/tests/Unit/EcommerceFinance/CourierSurchargeSettlementMirrorRepositoryTest.php +++ b/tests/Unit/EcommerceFinance/CourierSurchargeSettlementMirrorRepositoryTest.php @@ -345,6 +345,43 @@ class CourierSurchargeSettlementMirrorRepositoryTest extends TestCase $this->app->instance(UpstreamReadOnlyQueryFactory::class, $factory); } + public function test_completed_waybill_index_uses_keyed_lookup_and_preserves_work_order_context(): void + { + $this->createTable(); + DB::connection('business_mirror')->table(CourierSurchargeSettlementMirrorRepository::TABLE)->insert([ + 'rule_id' => 1, 'surcharge_type' => '退回费', 'courier_name' => '测试快递', 'waybill' => 'UNRELATED', + 'settlement_amount' => 0, 'settlement_status' => '已计算', 'pulled_at' => now(), 'created_at' => now(), 'updated_at' => now(), + ]); + $schema = Schema::connection('business_mirror'); + $schema->create('js_jst_work_order_info', function (Blueprint $table): void { + $table->char('_mirror_key', 64)->primary(); + $table->unsignedInteger('Id'); + $table->string('title'); + $table->text('businessFieldList'); + }); + config(['system_database.mode' => 'shared']); + (require base_path('database/migrations/2026_10_19_170000_create_work_order_waybill_lookup.php'))->up(); + $fields = [ + ['fieldId' => 'lIdField', 'value' => '001234'], + ['fieldId' => 'SELECT_CASCADEField1413856620250813172526', 'value' => '催件/未到'], + ['fieldId' => 'soIdField', 'value' => 'SO-TEST'], + ]; + $db = DB::connection('business_mirror'); + $db->table('js_jst_work_order_info')->insert([ + '_mirror_key' => str_repeat('a', 64), 'Id' => 10, 'title' => '查件', 'businessFieldList' => json_encode($fields), + ]); + app(\Modules\EcommerceFinance\Services\WorkOrderWaybillLookup::class)->backfill(500, 1, 0); + $db->enableQueryLog(); + $results = $this->repository->decideImportRows([ + ['waybill' => '001234', 'surcharge_type' => '退回费'], + ]); + $this->assertSame('催件', $results[0]['inspection_type']); + $this->assertStringContainsString('查件', $results[0]['reason']); + $queries = collect($db->getQueryLog())->pluck('query')->implode("\n"); + $this->assertStringContainsString('work_order_waybill_lookup', $queries); + $this->assertStringNotContainsString('businessFieldList" like', $queries); + } + private function createTable(): void { Schema::connection('business_mirror')->create( diff --git a/tests/Unit/EcommerceFinance/JdDailyBillTest.php b/tests/Unit/EcommerceFinance/JdDailyBillTest.php index 6aa66dbc..25ae397b 100644 --- a/tests/Unit/EcommerceFinance/JdDailyBillTest.php +++ b/tests/Unit/EcommerceFinance/JdDailyBillTest.php @@ -90,6 +90,23 @@ class JdDailyBillTest extends TestCase app(JdDailyBillRepository::class)->paginate([], 'occurred_at', 'asc', 1, 10); } + public function test_paging_reuses_summary_and_options_but_reimport_invalidates_them(): void + { + $this->import([$this->row('123', '10.00'), $this->row('124', '2.00')], 2, '12.00', '0.00'); + $repo = app(JdDailyBillRepository::class); + $repo->paginate([], 'occurred_at', 'asc', 1, 1); + $repo->filterOptions(); + $db = DB::connection('business_mirror'); + $db->enableQueryLog(); + $this->assertSame(12.0, $repo->paginate([], 'occurred_at', 'asc', 2, 1)['summary']['income']); + $repo->filterOptions(); + $queries = strtolower(collect($db->getQueryLog())->pluck('query')->implode("\n")); + $this->assertStringNotContainsString('sum(income)', $queries); + $this->assertStringNotContainsString('select distinct', $queries); + $this->import([$this->row('123', '25.00')], 1, '25.00', '0.00'); + $this->assertSame(25.0, $repo->paginate([], 'occurred_at', 'asc', 1, 1)['summary']['income']); + } + public function test_cutoff_validates_rows_and_refuses_to_shorten_existing_coverage(): void { $row = array_replace($this->row('123', '10.00'), ['费用结算时间' => '2026-08-21 23:59:59']); diff --git a/tests/Unit/EcommerceFinance/OrderDataMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/OrderDataMirrorRepositoryTest.php index 7e853c5b..25c52e14 100644 --- a/tests/Unit/EcommerceFinance/OrderDataMirrorRepositoryTest.php +++ b/tests/Unit/EcommerceFinance/OrderDataMirrorRepositoryTest.php @@ -136,6 +136,11 @@ class OrderDataMirrorRepositoryTest extends TestCase $none = $repository->afterSales(array_replace($base, ['shop_names' => ['没有这个店铺']]), 1, 25); $this->assertSame([], $none['rows']); + \Illuminate\Support\Facades\Cache::put(OrderDataMirrorRepository::SHOP_NAMES_CACHE_KEY, ['拼多多-甲店', '京东-乙店', '抖音-丙店'], 300); + $db->enableQueryLog(); + $selected = $repository->afterSales(array_replace($base, ['shop_names' => ['拼多多-甲店'], 'shop_names_exact' => true]), 1, 25); + $this->assertSame(['拼多多-甲店'], $this->sortedShops($selected['rows'])); + $this->assertDoesNotMatchRegularExpression('/shop_name\s+LIKE/i', collect($db->getQueryLog())->pluck('query')->implode("\n")); } public function test_shop_names_lists_distinct_shops_from_the_order_mirror(): void diff --git a/tests/Unit/EcommerceFinance/StoreReportPaidSalesIncrementalTest.php b/tests/Unit/EcommerceFinance/StoreReportPaidSalesIncrementalTest.php new file mode 100644 index 00000000..2c639789 --- /dev/null +++ b/tests/Unit/EcommerceFinance/StoreReportPaidSalesIncrementalTest.php @@ -0,0 +1,125 @@ + ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => '']]); + DB::purge('business_mirror'); + config(['cache.stores.upstream_shared' => ['driver' => 'array'], 'system_database.mode' => 'shared']); + $schema = Schema::connection('business_mirror'); + foreach (['store_report_import_states', 'store_report_paid_sales_rows', 'store_receipt_order_match_rows', StoreReportPaidSalesChanges::TABLE] as $table) $schema->dropIfExists($table); + (require base_path('database/migrations/2026_09_01_000000_create_store_report_mirror_tables.php'))->up(); + $schema->table('store_report_import_states', function (Blueprint $table): void { + $table->string('run_mode')->nullable(); + $table->timestamp('initial_completed_at')->nullable(); + }); + (require base_path('database/migrations/2026_10_16_234000_create_store_report_paid_sales_rows.php'))->up(); + $schema->create('store_receipt_order_match_rows', function (Blueprint $table): void { + $table->increments('id'); + $table->string('platform'); + $table->date('report_month'); + $table->date('occurred_date'); + $table->string('store_name'); + foreach (['income_amount', 'discount_amount', 'offset_amount', 'refund_amount'] as $column) $table->decimal($column, 20, 4)->nullable(); + }); + (require base_path('database/migrations/2026_10_19_171000_track_paid_sales_dirty_dates.php'))->up(); + DB::connection('business_mirror')->table('store_receipt_order_match_rows')->insert([ + ['platform' => 'A', 'report_month' => '2026-10-01', 'occurred_date' => '2026-10-01', 'store_name' => '测试店', 'income_amount' => 100, 'discount_amount' => -5, 'offset_amount' => 0, 'refund_amount' => -10], + ['platform' => 'B', 'report_month' => '2026-10-01', 'occurred_date' => '2026-10-01', 'store_name' => '测试店', 'income_amount' => 20, 'discount_amount' => 0, 'offset_amount' => 0, 'refund_amount' => 0], + ['platform' => 'A', 'report_month' => '2026-09-01', 'occurred_date' => '2026-09-30', 'store_name' => '测试店', 'income_amount' => 40, 'discount_amount' => 0, 'offset_amount' => 0, 'refund_amount' => 0], + ]); + return new StoreReportMirrorSynchronizer(\Mockery::mock(StoreReportUpstreamRepository::class), app('db')); + } + + public function test_amount_change_without_row_count_change_only_rebuilds_affected_days(): void + { + $sync = $this->setupSummary(); + $db = DB::connection('business_mirror'); + $this->assertSame('succeeded', $sync->syncPaidSales()['status']); + $untouchedId = $db->table('store_report_paid_sales_rows')->where('occurred_on', '2026-09-30')->value('id'); + $changes = app(StoreReportPaidSalesChanges::class); + $changes->markPeriod('2026-10', 'A'); + $db->table('store_receipt_order_match_rows')->where('id', 1)->update(['income_amount' => 120]); + $this->assertTrue($sync->syncPaidSales()['changed']); + $this->assertSame(125.0, (float) $db->table('store_report_paid_sales_rows')->where('occurred_on', '2026-10-01')->value('paid_amount')); + $this->assertSame($untouchedId, $db->table('store_report_paid_sales_rows')->where('occurred_on', '2026-09-30')->value('id')); + $db->enableQueryLog(); + $this->assertFalse($sync->syncPaidSales()['changed']); + $this->assertSame([], collect($db->getQueryLog())->filter(fn ($query) => str_contains($query['query'], 'store_receipt_order_match_rows'))->all()); + } + + public function test_date_moves_and_deletions_clear_old_totals_and_include_other_platforms(): void + { + $sync = $this->setupSummary(); + $db = DB::connection('business_mirror'); + $sync->syncPaidSales(); + $changes = app(StoreReportPaidSalesChanges::class); + $changes->markPeriod('2026-10', 'A'); + $db->table('store_receipt_order_match_rows')->where('id', 1)->update(['occurred_date' => '2026-10-03']); + $changes->markPeriod('2026-10', 'A'); + $sync->syncPaidSales(); + $this->assertSame(20.0, (float) $db->table('store_report_paid_sales_rows')->where('occurred_on', '2026-10-01')->value('paid_amount')); + $this->assertSame(85.0, (float) $db->table('store_report_paid_sales_rows')->where('occurred_on', '2026-10-03')->value('paid_amount')); + $changes->markPeriod('2026-10', 'B'); + $db->table('store_receipt_order_match_rows')->where('platform', 'B')->delete(); + $sync->syncPaidSales(); + $this->assertSame(0, $db->table('store_report_paid_sales_rows')->where('occurred_on', '2026-10-01')->count()); + $this->assertSame(125.0, (float) $db->table('store_report_paid_sales_rows')->sum('paid_amount')); + } + + public function test_a_new_revision_is_not_lost_when_an_older_snapshot_is_acknowledged(): void + { + $this->setupSummary(); + $changes = app(StoreReportPaidSalesChanges::class); + $changes->markDates(['2026-10-01']); + $old = $changes->pending(); + $changes->markDates(['2026-10-01']); + $changes->acknowledge($old); + $this->assertCount(1, $changes->pending()); + } + + public function test_active_writer_defers_summary_and_other_writers_without_consuming_changes(): void + { + $sync = $this->setupSummary(); + $sync->syncPaidSales(); + $changes = app(StoreReportPaidSalesChanges::class); + $changes->markDates(['2026-10-01']); + $lock = Cache::store('upstream_shared')->lock('store-report:paid-sales', 3600); + $this->assertTrue($lock->get()); + try { + $this->assertFalse($sync->syncPaidSales()['changed']); + $this->assertCount(1, $changes->pending()); + $this->artisan('store-receipt:order-match:sync', ['month' => '2026-10'])->assertExitCode(75); + } finally { + $lock->release(); + } + $this->assertTrue($sync->syncPaidSales()['changed']); + $this->assertSame([], $changes->pending()); + } + + public function test_aggregation_failure_keeps_previous_totals_and_dirty_dates(): void + { + $sync = $this->setupSummary(); + $db = DB::connection('business_mirror'); + $sync->syncPaidSales(); + app(StoreReportPaidSalesChanges::class)->markDates(['2026-10-01']); + $db->listen(function ($query): void { + if (str_starts_with($query->sql, 'INSERT INTO store_report_paid_sales_rows')) throw new \RuntimeException('test_aggregate_failure'); + }); + $this->assertSame('failed', $sync->syncPaidSales()['status']); + $this->assertSame(105.0, (float) $db->table('store_report_paid_sales_rows')->where('occurred_on', '2026-10-01')->value('paid_amount')); + $this->assertCount(1, app(StoreReportPaidSalesChanges::class)->pending()); + } +} diff --git a/tests/Unit/EcommerceFinance/WorkOrderWaybillLookupTest.php b/tests/Unit/EcommerceFinance/WorkOrderWaybillLookupTest.php new file mode 100644 index 00000000..d9778601 --- /dev/null +++ b/tests/Unit/EcommerceFinance/WorkOrderWaybillLookupTest.php @@ -0,0 +1,81 @@ +dropIfExists($table); + $schema->create(WorkOrderWaybillLookup::SOURCE, function (Blueprint $table): void { + $table->char('_mirror_key', 64)->primary(); + $table->unsignedBigInteger('Id'); + $table->string('title'); + $table->text('businessFieldList'); + }); + config(['system_database.mode' => 'shared']); + (require base_path('database/migrations/2026_10_19_170000_create_work_order_waybill_lookup.php'))->up(); + return app(WorkOrderWaybillLookup::class); + } + + private function row(string $key, int $id, string $waybill, string $title = '查件'): array + { + return ['_mirror_key' => str_pad($key, 64, '0'), 'Id' => $id, 'title' => $title, + 'businessFieldList' => json_encode([['fieldId' => 'lIdField', 'value' => $waybill]])]; + } + + public function test_backfill_resumes_and_does_not_publish_an_incomplete_index(): void + { + $lookup = $this->setupLookup(); + $db = DB::connection('business_mirror'); + $db->table(WorkOrderWaybillLookup::SOURCE)->insert([ + $this->row('a', 1, '001'), $this->row('b', 2, '001'), $this->row('c', 3, '003', '其它'), + ]); + $this->assertFalse($lookup->backfill(2, 1, 0)['complete']); + $this->assertFalse($lookup->ready()); + $this->assertTrue($lookup->backfill(2, 1, 0)['complete']); + $this->assertTrue($lookup->ready()); + $this->assertSame(2, $db->table(WorkOrderWaybillLookup::TABLE)->where('waybill', '001')->count()); + $this->assertSame(0, $lookup->backfill(2, 1, 0)['rows']); + } + + public function test_updates_remove_old_keys_and_preserve_last_field_wins_and_multiple_orders(): void + { + $lookup = $this->setupLookup(); + $db = DB::connection('business_mirror'); + $a = $this->row('a', 1, '001'); + $b = $this->row('b', 2, '001'); + $lookup->storeBatch([$a, $b]); + $a['businessFieldList'] = json_encode([ + ['fieldId' => 'lIdField', 'value' => 'wrong'], ['fieldId' => 'lIdField', 'value' => ' 002 '], + ]); + $lookup->storeBatch([$a]); + $this->assertSame(1, $db->table(WorkOrderWaybillLookup::TABLE)->where('waybill', '001')->count()); + $this->assertSame(1, $db->table(WorkOrderWaybillLookup::TABLE)->where('waybill', '002')->count()); + $b['title'] = '其它'; + $lookup->storeBatch([$b]); + $this->assertSame(0, $db->table(WorkOrderWaybillLookup::TABLE)->where('waybill', '001')->count()); + $a['businessFieldList'] = '{invalid'; + $lookup->storeBatch([$a]); + $this->assertSame(0, $db->table(WorkOrderWaybillLookup::TABLE)->count()); + } + + public function test_full_source_pruning_removes_deleted_records(): void + { + $lookup = $this->setupLookup(); + $db = DB::connection('business_mirror'); + $a = $this->row('a', 1, '001'); + $b = $this->row('b', 2, '002'); + $db->table(WorkOrderWaybillLookup::SOURCE)->insert([$a]); + $lookup->storeBatch([$a, $b]); + $lookup->prune(); + $this->assertSame(['001'], $db->table(WorkOrderWaybillLookup::TABLE)->pluck('waybill')->all()); + } +}