diff --git a/app/Console/Commands/DedupAt.php b/app/Console/Commands/DedupAt.php index 6d2d6a11..94e292d6 100644 --- a/app/Console/Commands/DedupAt.php +++ b/app/Console/Commands/DedupAt.php @@ -6,14 +6,14 @@ use Illuminate\Console\Command; use Modules\EcommerceFinance\Services\Jushuitan\ShippingReplicaSynchronizer; /** - * 一次性：删除现有发货数据里以 @ 开头、去@后已存在的重复单号。 + * 一次性：全表扫描，仅删除同订单、同发货日有干净单号对应的 @ 包裹。 * php artisan jst:dedup-at */ class DedupAt extends Command { protected $signature = 'jst:dedup-at'; - protected $description = '去除 @ 开头的重复单号（去@后已存在则删）'; + protected $description = '去除同订单、同发货日且有干净单号对应的 @ 包裹'; public function handle(): int { diff --git a/modules/EcommerceFinance/Console/SyncJushuitanShipping.php b/modules/EcommerceFinance/Console/SyncJushuitanShipping.php index 021124d4..a27f74ea 100644 --- a/modules/EcommerceFinance/Console/SyncJushuitanShipping.php +++ b/modules/EcommerceFinance/Console/SyncJushuitanShipping.php @@ -17,7 +17,8 @@ use Modules\EcommerceFinance\Services\Jushuitan\ShippingReplicaSynchronizer; */ class SyncJushuitanShipping extends Command { - protected $signature = 'jushuitan:sync-shipping {--from=} {--to=} {--days=} {--limit=2000}'; + protected $signature = 'jushuitan:sync-shipping {--from=} {--to=} {--days=} {--limit=2000} + {--skip-freight : 仅补发货数量；保留已有运费，新记录运费留空并标记部分完成}'; protected $description = '从经营分析副本库直读聚水潭发货明细，清洗写入 caiwu 发货数据表'; @@ -35,6 +36,12 @@ class SyncJushuitanShipping extends Command $to = now()->startOfDay()->addDay(); } + $skipFreight = (bool) $this->option('skip-freight'); + if ($skipFreight && (! $from || ! $to)) { + $this->error('--skip-freight 需要指定 --from 和 --to，或使用 --days。'); + return self::FAILURE; + } + $limit = (int) ($this->option('limit') ?: 2000); $range = ($from && $to) ? $from->toDateString() . ' ~ ' . $to->toDateString() : '最近 ' . $limit . ' 条'; @@ -46,13 +53,40 @@ class SyncJushuitanShipping extends Command } }; try { - $result = ShippingReplicaSynchronizer::make()->sync($from, $to, $limit, $progress); + $result = ShippingReplicaSynchronizer::make()->sync($from, $to, $limit, $progress, $skipFreight); } catch (\Throwable $e) { $this->error('同步失败：' . $e->getMessage()); return self::FAILURE; } - $this->info('完成：写入 ' . ($result['processed'] ?? 0) . ' 条发货明细。'); + $expressRows = (int) ($result['processed'] ?? 0); + $autoRows = (int) ($result['autoship'] ?? 0); + $summary = sprintf('发货明细 %d 条，代发仓 %d 条，合计处理 %d 条明细', $expressRows, $autoRows, $expressRows + $autoRows); + $packageSummary = isset($result['unique_packages']) + ? sprintf('唯一包裹源键 %d 个，包裹记录处理 %d 次；', $result['unique_packages'], $result['packages'] ?? 0) + : ''; + if ($result['freight_incomplete'] ?? false) { + $this->warn('本次窗口' . $packageSummary . $summary . '；新记录运费与结算重量待补齐，空白不代表 0。'); + } else { + $this->info('完成：' . $packageSummary . $summary . '。'); + } + if (($result['packages_without_items'] ?? 0) > 0 || ($result['dws_failed_days'] ?? []) !== []) { + $this->warn(sprintf( + '商品或金额待核对：%d 个包裹未匹配商品，DWS 预载失败 %d 天；页面包裹数不等于商品明细完整性。', + $result['packages_without_items'] ?? 0, + count($result['dws_failed_days'] ?? []), + )); + } + if (($result['packages_retained_for_missing_items'] ?? 0) > 0) { + $this->warn('其中 ' . $result['packages_retained_for_missing_items'] . ' 个包裹因新商品明细缺失而保留原有商品及金额，本次未完成这些包裹的更新。'); + } + if (($result['packages_retained_for_unmatched_amount'] ?? 0) > 0) { + $this->warn('其中 ' . $result['packages_retained_for_unmatched_amount'] . ' 个包裹因商品金额无法安全匹配而保留原记录，金额待核对。'); + } + if (($result['stale_mirror_packages'] ?? 0) > 0) { + $this->error('本窗口仍有 ' . $result['stale_mirror_packages'] . ' 个镜像包裹未在上游扫描中出现；已保留原记录，需按源键核对后才能认定数量完整。'); + return self::FAILURE; + } return self::SUCCESS; } } diff --git a/modules/EcommerceFinance/Repositories/ShippingOrderRepository.php b/modules/EcommerceFinance/Repositories/ShippingOrderRepository.php index e6bfc714..af89f9ad 100644 --- a/modules/EcommerceFinance/Repositories/ShippingOrderRepository.php +++ b/modules/EcommerceFinance/Repositories/ShippingOrderRepository.php @@ -260,9 +260,21 @@ class ShippingOrderRepository return ['code' => 'unavailable', 'last_succeeded_at' => null]; } + $freightPending = $this->db->table('jst_sync_state') + ->where('dataset_key', 'like', 'jst.sctrade.freight.%')->where('status', '!=', 'succeeded')->count(); + $itemsPending = $this->db->table('jst_sync_state') + ->where('dataset_key', 'like', 'jst.sctrade.items.%')->where('status', '!=', 'succeeded')->count(); + $code = (string) ($row->status ?? 'unavailable'); + return [ - 'code' => (string) ($row->status ?? 'unavailable'), + 'code' => ($freightPending || $itemsPending) && $code === 'succeeded' ? 'partial' : $code, 'last_succeeded_at' => $row->last_succeeded_at ?? null, + 'freight_pending' => (bool) $freightPending, + 'warning' => implode(' ', array_filter([ + $code === 'partial' ? '部分包裹尚需与上游源键核对，原有镜像记录已保留。' : null, + $freightPending ? "{$freightPending} 个日期窗口的运费和结算重量待补，空白不代表 0。" : null, + $itemsPending ? "{$itemsPending} 个日期窗口的商品或金额待核对。" : null, + ])), ]; } catch (\Throwable) { return ['code' => 'unavailable', 'last_succeeded_at' => null]; diff --git a/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php b/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php index 82851015..09e57edc 100644 --- a/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php +++ b/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php @@ -4,7 +4,9 @@ namespace Modules\EcommerceFinance\Services\Jushuitan; use Illuminate\Support\Carbon; use Illuminate\Support\Facades\DB; +use InvalidArgumentException; use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory; +use RuntimeException; use Throwable; /** @@ -23,13 +25,47 @@ class ShippingReplicaSynchronizer { private const SOURCE = 'operations_replica'; private const DATASET = 'jst.sctrade'; + private const FREIGHT_DATASET_PREFIX = 'jst.sctrade.freight.'; + private const ITEMS_DATASET_PREFIX = 'jst.sctrade.items.'; private const AUTOSHIP_OFFSET = 1000000000; private const DEMO_SOURCE_IDS = [9000001, 9000002, 9000003, 9000004, 9000005, 9000006]; private const FLUSH = 300; + private const PRESERVED_PACKAGE_FIELDS = [ + 'buyer_nick', 'seller_memo', 'package_no', 'download_time', 'freight_rate', 'freight_formula', + 'settle_volume_cm3', 'recon_weight_kg', 'recon_volume_cm3', 'front_no', + ]; + private const PRESERVED_CONTACT_FIELDS = ['receiver_name', 'receiver_mobile', 'receiver_phone']; + private const PRESERVED_ITEM_FIELDS = ['finished_barcode', 'finished_name', 'finished_supplier']; private const VIEW = 'rpt.Jst_SendData_View_extension'; private const OUTBOUND = 'rpt.jst_salesOutboundInfo'; + private int $maxExpressBatches = 5000; + private int $maxAutoShipBatches = 600; + private bool $skipFreight = false; + private bool $replaceWindow = false; + + /** @var array>> */ + private array $replacementPackages = []; + /** @var array */ + private array $replacementHasFreight = []; + /** @var array */ + private array $replacementHasItems = []; + /** @var array> */ + private array $replacementSyntheticSkuIds = []; + private int $replacementRowCount = 0; + private int $packagesProcessed = 0; + private int $packagesWithoutItems = 0; + private int $packagesRetainedForMissingItems = 0; + private int $packagesRetainedForUnmatchedAmount = 0; + private int $staleMirrorPackages = 0; + + /** @var array */ + private array $seenSourceIds = []; + + /** @var list */ + private array $dwsFailedDays = []; + private readonly \Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery $up; private readonly \Illuminate\Database\ConnectionInterface $db; @@ -54,19 +90,33 @@ class ShippingReplicaSynchronizer return new self(app(UpstreamReadOnlyQueryFactory::class)); } - public function sync(?Carbon $from = null, ?Carbon $to = null, int $limit = 2000, ?callable $progress = null): array + public function sync(?Carbon $from = null, ?Carbon $to = null, int $limit = 2000, ?callable $progress = null, bool $skipFreight = false): array { $this->markRunning(); try { $windowed = $from && $to; + if ($skipFreight && ! $windowed) { + throw new InvalidArgumentException('跳过运费仅支持指定起止日期的历史补录。'); + } + $this->skipFreight = $skipFreight; + $this->replaceWindow = (bool) $windowed; + $this->packagesProcessed = 0; + $this->packagesWithoutItems = 0; + $this->packagesRetainedForMissingItems = 0; + $this->packagesRetainedForUnmatchedAmount = 0; + $this->staleMirrorPackages = 0; + $this->seenSourceIds = []; + $this->dwsFailedDays = []; $now = now(); $fromS = $windowed ? $from->format('Y-m-d H:i:s') : null; $toS = $windowed ? $to->format('Y-m-d H:i:s') : null; + if ($skipFreight) { + $this->markFreightIncomplete($fromS, $toS); + } $this->db->table('jst_oms_shipping')->whereIn('source_id', self::DEMO_SOURCE_IDS)->delete(); - if ($windowed) { - $this->db->table('jst_oms_shipping') - ->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS)->delete(); + if ($skipFreight && $progress) { + $progress('仅补数量：按包裹分批替换，已有运费/结算重量在事务中保留；不预删日期窗口'); } $total = 0; @@ -84,49 +134,100 @@ class ShippingReplicaSynchronizer } $total = $this->syncExpress(true, $fromS, $toS, $limit, $now, $progress); $auto = $this->syncAutoShip($fromS, $toS, $now, $progress); - $removed = $this->dedupAtPrefix(); - if ($progress && $removed > 0) { - $progress('去除@重复单号：' . $removed . ' 条'); + $this->staleMirrorPackages = $this->countUnseenMirrorPackages($fromS, $toS); + if (! $skipFreight && $this->staleMirrorPackages === 0) { + $removed = $this->dedupAtPrefix($fromS, $toS); + if ($progress && $removed > 0) { + $progress('去除@重复单号：' . $removed . ' 条'); + } } } else { $total = $this->syncExpress(false, $fromS, $toS, $limit, $now, $progress); } - $this->markSucceeded(); - return ['processed' => $total, 'autoship' => $auto]; + if ($windowed && ($this->dwsFailedDays !== [] || $this->packagesWithoutItems > 0 + || $this->packagesRetainedForUnmatchedAmount > 0)) { + $this->markItemsIncomplete($fromS, $toS); + } elseif ($windowed && ! $skipFreight && $this->staleMirrorPackages === 0) { + $this->markCovered(self::ITEMS_DATASET_PREFIX, $fromS, $toS); + } + + $result = ['processed' => $total, 'autoship' => $auto]; + if ($windowed) { + $result['packages'] = $this->packagesProcessed; + $result['unique_packages'] = count($this->seenSourceIds); + $result['packages_without_items'] = $this->packagesWithoutItems; + $result['packages_retained_for_missing_items'] = $this->packagesRetainedForMissingItems; + $result['packages_retained_for_unmatched_amount'] = $this->packagesRetainedForUnmatchedAmount; + $result['dws_failed_days'] = $this->dwsFailedDays; + $result['stale_mirror_packages'] = $this->staleMirrorPackages; + } + if ($skipFreight) { + $result['freight_incomplete'] = true; + } + + if ($this->staleMirrorPackages > 0) { + $this->markIncompleteWindow($this->staleMirrorPackages); + } else { + $this->markSucceeded(); + } + return $result; } catch (Throwable $e) { $this->markFailed($e->getMessage()); throw $e; } finally { $this->dwsCache = []; $this->outboundMap = ['byL' => [], 'byO' => []]; + $this->replacementPackages = []; + $this->replacementHasFreight = []; + $this->replacementHasItems = []; + $this->replacementSyntheticSkuIds = []; + $this->replacementRowCount = 0; + $this->skipFreight = false; + $this->replaceWindow = false; + $this->dwsFailedDays = []; + $this->seenSourceIds = []; gc_collect_cycles(); } } - /** 去除以 @ 开头、去@后已存在的重复单号（换单/重拉产生；工作平台只保留干净版）。 */ - public function dedupAtPrefix(): int + /** 仅在同一订单、同一发货日去重；提供日期窗口时不使用窗口外记录作匹配。 */ + public function dedupAtPrefix(?string $fromS = null, ?string $toS = null): int { - $all = $this->db->table('jst_oms_shipping') - ->whereNotNull('express_no')->where('express_no', '!=', '') - ->distinct()->pluck('express_no')->all(); - $set = []; - foreach ($all as $e) { - $set[(string) $e] = true; + $query = $this->db->table('jst_oms_shipping') + ->whereNotNull('express_no')->where('express_no', '!=', ''); + if ($fromS !== null && $toS !== null) { + $query->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS); + } + $clean = []; + $prefixed = []; + foreach ($query->distinct()->cursor(['source_id', 'trade_no', 'express_no', 'ship_time']) as $row) { + $tradeNo = trim((string) $row->trade_no); + $expressNo = (string) $row->express_no; + $sourceId = (int) $row->source_id; + $shipDate = substr((string) $row->ship_time, 0, 10); + if ($tradeNo === '' || $sourceId <= 0 || $shipDate === '') { + continue; + } + if ($expressNo[0] === '@') { + $prefixed[$sourceId] = [$tradeNo, $shipDate, substr($expressNo, 1)]; + } else { + $clean[$tradeNo][$shipDate][$expressNo] = true; + } } $toDel = []; - foreach (array_keys($set) as $e) { - $e = (string) $e; - if ($e !== '' && $e[0] === '@' && isset($set[substr($e, 1)])) { - $toDel[] = $e; + foreach ($prefixed as $sourceId => [$tradeNo, $shipDate, $expressNo]) { + if ($expressNo !== '' && isset($clean[$tradeNo][$shipDate][$expressNo])) { + $toDel[] = $sourceId; } } $removed = 0; foreach (array_chunk($toDel, 500) as $chunk) { - if ($chunk === []) { - continue; + $delete = $this->db->table('jst_oms_shipping')->whereIn('source_id', $chunk); + if ($fromS !== null && $toS !== null) { + $delete->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS); } - $removed += (int) $this->db->table('jst_oms_shipping')->whereIn('express_no', $chunk)->delete(); + $removed += (int) $delete->delete(); } return $removed; @@ -136,7 +237,7 @@ class ShippingReplicaSynchronizer private function syncExpress(bool $windowed, ?string $fromS, ?string $toS, int $limit, $now, ?callable $progress = null): int { $batch = 1500; - $maxBatches = 5000; + $maxBatches = $this->maxExpressBatches; $cols = 'e.id, e.so_id, e.inner_order_id, e.logistic_id, e.logistic_company_name, e.send_time, e.pay_time, ' . 'e.shop_name, e.platform_name, e.dis_co_name, e.warehouse_name, e.receiver_address, ' . 'e.weight, e.f_weight, e.buyer_message, e.order_labels, jo.l_id AS jo_lid, jo.send_date AS jo_send'; @@ -149,9 +250,7 @@ class ShippingReplicaSynchronizer $total = 0; $scanned = 0; $lastId = null; - $entered = false; - - for ($b = 0; $b < $maxBatches; $b++) { + for ($b = 0; $b <= $maxBatches; $b++) { $where = $windowed ? '1 = 1' : 'e.send_time IS NOT NULL'; $bind = []; if ($lastId !== null) { @@ -165,14 +264,15 @@ class ShippingReplicaSynchronizer if ($rows === []) { break; } + if ($b === $maxBatches) { + throw new RuntimeException('发货扫描达到批次上限，仍有未读取的上游数据。'); + } $cnt = count($rows); $scanned += $cnt; foreach ($rows as $r) { $lastId = $r['id']; } - $pageNewest = null; - $pageHit = false; if ($windowed) { $kept = []; foreach ($rows as $r) { @@ -182,13 +282,8 @@ class ShippingReplicaSynchronizer $jd = $r['jo_send'] ?? null; $ts = ($jd !== null && $jd !== '') ? strtotime((string) $jd) : null; } - if ($ts !== null && ($pageNewest === null || $ts > $pageNewest)) { - $pageNewest = $ts; - } if ($ts !== null && $ts >= $fromTs && $ts < $toTs) { $kept[] = $r; - $pageHit = true; - $entered = true; } } $rows = $kept; @@ -229,7 +324,7 @@ class ShippingReplicaSynchronizer } $viewItems = $this->loadViewItems(array_keys($ioIds)); $skus = $this->loadSkus($this->collectSkuIds($byOid, $bySo, $dws, $viewItems)); - $freight = $this->loadFreight(array_keys($soIds)); + $freight = $this->skipFreight ? [] : $this->loadFreight(array_keys($soIds)); foreach ($rows as $r) { $pkgId = (int) ($r['id'] ?? 0); @@ -258,10 +353,6 @@ class ShippingReplicaSynchronizer if ($progress && $b % 20 === 0) { $progress(sprintf('发货：已扫描 %d 行，命中窗口 %d 件', $scanned, $total)); } - // 表远大于窗口时提前结束：已进入窗口，且本页整页都早于窗口(id≈时间倒序)。 - if ($windowed && $entered && ! $pageHit && $pageNewest !== null && $fromTs !== null && $pageNewest < $fromTs) { - break; - } if ($cnt < $batch) { break; } @@ -270,9 +361,9 @@ class ShippingReplicaSynchronizer } } if ($progress) { - $progress(sprintf('发货完成：扫描 %d 行，写入 %d 件', $scanned, $total)); + $progress(sprintf('发货完成：扫描 %d 行，处理 %d 件', $scanned, $total)); } - $this->flush(); + $this->flushPending(); return $total; } @@ -281,7 +372,7 @@ class ShippingReplicaSynchronizer private function syncAutoShip(string $fromS, string $toS, $now, ?callable $progress = null): int { $batch = 400; - $maxBatches = 600; + $maxBatches = $this->maxAutoShipBatches; $cols = 'o.Id AS rid, o.o_id, o.so_id, o.l_id, o.logistics_company, o.order_from, o.labels, ' . 'o.shop_name, o.send_date, o.pay_date, o.weight, o.buyer_message, ' . 'o.receiver_state, o.receiver_city, o.receiver_district, o.receiver_address, ' @@ -290,7 +381,7 @@ class ShippingReplicaSynchronizer $total = 0; $lastRid = null; - for ($b = 0; $b < $maxBatches; $b++) { + for ($b = 0; $b <= $maxBatches; $b++) { $where = "o.status = 'Sent' AND o.send_date >= ? AND o.send_date < ? " . 'AND NOT EXISTS (SELECT 1 FROM rpt.jst_order_express_details e WHERE e.inner_order_id = o.o_id)'; $bind = [$fromS, $toS]; @@ -303,6 +394,9 @@ class ShippingReplicaSynchronizer if ($rows === []) { break; } + if ($b === $maxBatches) { + throw new RuntimeException('代发仓扫描达到批次上限，仍有未读取的上游数据。'); + } $oids = []; $soIds = []; @@ -339,7 +433,7 @@ class ShippingReplicaSynchronizer } $viewItems = $this->loadViewItems(array_keys($ioIds)); $skus = $this->loadSkus($this->collectSkuIds($byOid, $bySo, $dws, $viewItems)); - $freight = $this->loadFreight(array_keys($soIds)); + $freight = $this->skipFreight ? [] : $this->loadFreight(array_keys($soIds)); $cnt = count($rows); foreach ($rows as $r) { @@ -370,38 +464,52 @@ class ShippingReplicaSynchronizer } } if ($progress) { - $progress(sprintf('代发仓完成：写入 %d 件', $total)); + $progress(sprintf('代发仓完成：处理 %d 件', $total)); } - $this->flush(); + $this->flushPending(); return $total; } /** * 输出一个包裹的所有明细行(每件一行；无商品则一行空商品)。 - * @return int 写入行数 + * @return int 组装行数 */ private function emitPackage(int $pkgId, array $pkg, array $items, array $skus, array $freight, string $so, ?string $tracking, $now, bool $auto = false): int { $items = array_values(array_filter($items, fn ($it) => ! $this->isPricePatch($it['name'] ?? null))); $n = count($items); + $this->seenSourceIds[$pkgId] = true; + $this->packagesProcessed++; + if ($n === 0) { + $this->packagesWithoutItems++; + } $f = $freight[$so . '|' . ($tracking ?? '')] ?? ($freight[$so . '|'] ?? []); + $hasFreight = isset($f['cost']); + $packageRows = []; if ($n === 0) { - $this->push($pkgId, $pkgId, $this->buildRow($pkg, [], [], $f, true, '1/1', $now, $auto)); + $packageRows[] = $this->packageRow($pkgId, $pkgId, $this->buildRow($pkg, [], [], $f, true, '1/1', $now, $auto)); + $this->writePackage($pkgId, $packageRows, $hasFreight, false, []); return 1; } $count = 0; + $syntheticSkuIds = []; foreach ($items as $idx => $it) { $first = ($idx === 0); $sk = (! empty($it['sku']) && isset($skus[(string) $it['sku']])) ? $skus[(string) $it['sku']] : []; $skuSid = (int) ($it['id'] ?? ($pkgId * 100 + $idx)); - $this->push($pkgId, $skuSid, $this->buildRow($pkg, $it, $sk, $f, $first, '1/' . $n, $now, $auto)); + if (($it['id'] ?? null) === null) { + $syntheticSkuIds[$skuSid] = true; + } + $packageRows[] = $this->packageRow($pkgId, $skuSid, $this->buildRow($pkg, $it, $sk, $f, $first, '1/' . $n, $now, $auto)); $count++; } + $this->writePackage($pkgId, $packageRows, $hasFreight, true, $syntheticSkuIds); + return $count; } @@ -418,7 +526,7 @@ class ShippingReplicaSynchronizer $pkgWeight = (float) ($pkg['weight'] ?? 0) ?: null; $freightW = (float) ($f['w'] ?? 0); - $settleW = $freightW > 0 ? $freightW : ($pkgWeight ?: $onlineW); + $settleW = $this->skipFreight ? null : ($freightW > 0 ? $freightW : ($pkgWeight ?: $onlineW)); return $this->cleanPayload([ 'express_no' => $pkg['tracking'], @@ -582,6 +690,7 @@ class ShippingReplicaSynchronizer try { $rows = $this->up->select("SELECT sourceno, item_id, skuoid, goods_count, sales_amount FROM rpt.dws_sn_sales_full_info_di WHERE stat_date = ?", [$d]); } catch (Throwable) { + $this->dwsFailedDays[] = $d; continue; } foreach ($rows as $r) { @@ -855,6 +964,183 @@ class ShippingReplicaSynchronizer } } + private function packageRow(int $sourceId, int $skuSourceId, array $payload): array + { + $payload['source_id'] = $sourceId; + $payload['sku_source_id'] = $skuSourceId; + + return $payload; + } + + /** @param list> $rows */ + private function writePackage(int $sourceId, array $rows, bool $hasFreight, bool $hasItems, array $syntheticSkuIds): void + { + if (! $this->replaceWindow) { + foreach ($rows as $row) { + $this->push($sourceId, (int) $row['sku_source_id'], $row); + } + return; + } + + $this->replacementRowCount -= count($this->replacementPackages[$sourceId] ?? []); + $this->replacementPackages[$sourceId] = $rows; + $this->replacementHasFreight[$sourceId] = $hasFreight; + $this->replacementHasItems[$sourceId] = $hasItems; + $this->replacementSyntheticSkuIds[$sourceId] = $syntheticSkuIds; + $this->replacementRowCount += count($rows); + if ($this->replacementRowCount >= self::FLUSH) { + $this->flushReplacements(); + } + } + + private function flushPending(): void + { + if ($this->replaceWindow) { + $this->flushReplacements(); + } else { + $this->flush(); + } + } + + private function flushReplacements(): void + { + if ($this->replacementPackages === []) { + return; + } + $packages = $this->replacementPackages; + $hasFreight = $this->replacementHasFreight; + $hasItems = $this->replacementHasItems; + $syntheticSkuIds = $this->replacementSyntheticSkuIds; + $this->db->transaction(function () use ($packages, $hasFreight, $hasItems, $syntheticSkuIds): void { + $ids = array_keys($packages); + $previous = []; + $previousItems = []; + $oldAmountSkuIds = []; + $oldHasItems = []; + $oldRows = $this->db->table('jst_oms_shipping')->whereIn('source_id', $ids) + ->select(array_merge( + ['source_id', 'sku_source_id', 'express_no', 'express_name2', 'freight', 'settle_freight', 'weight_kg', 'settle_weight_kg', + 'item_id', 'barcode', 'product_name', 'product_code', 'qty', 'amount'], + self::PRESERVED_PACKAGE_FIELDS, + self::PRESERVED_CONTACT_FIELDS, + self::PRESERVED_ITEM_FIELDS, + )) + ->get(); + foreach ($oldRows as $oldRow) { + $id = (int) $oldRow->source_id; + $row = (array) $oldRow; + $previousItems[$id][(int) $oldRow->sku_source_id] = $row; + if ($row['amount'] !== null) { + $oldAmountSkuIds[$id][(int) $oldRow->sku_source_id] = true; + } + foreach (['item_id', 'barcode', 'product_name', 'product_code', 'qty', 'amount', + 'finished_barcode', 'finished_name', 'finished_supplier'] as $field) { + if ($row[$field] !== null && $row[$field] !== '') { + $oldHasItems[$id] = true; + break; + } + } + if (! isset($previous[$id])) { + $previous[$id] = $row; + continue; + } + foreach (array_merge( + ['freight', 'settle_freight', 'weight_kg', 'settle_weight_kg', 'express_no', 'express_name2'], + self::PRESERVED_PACKAGE_FIELDS, + self::PRESERVED_CONTACT_FIELDS, + ) as $field) { + if (($previous[$id][$field] ?? null) === null || $previous[$id][$field] === '') { + $previous[$id][$field] = $row[$field]; + } + } + } + + $rows = []; + $replaceIds = []; + foreach ($packages as $id => $packageRows) { + if (! ($hasItems[$id] ?? false) && ($oldHasItems[$id] ?? false)) { + $this->packagesRetainedForMissingItems++; + continue; + } + $unmatchedOldAmounts = $oldAmountSkuIds[$id] ?? []; + $unsafeAmount = false; + foreach ($packageRows as &$packageRow) { + $skuId = (int) $packageRow['sku_source_id']; + $oldItem = $previousItems[$id][$skuId] ?? null; + if ($packageRow['amount'] !== null) { + unset($unmatchedOldAmounts[$skuId]); + continue; + } + if ($oldItem === null || $oldItem['amount'] === null) { + continue; + } + if (isset($syntheticSkuIds[$id][$skuId])) { + $oldBarcode = trim((string) $oldItem['barcode']); + $newBarcode = trim((string) ($packageRow['barcode'] ?? '')); + if ($oldBarcode === '' || $newBarcode === '' || $oldBarcode !== $newBarcode) { + $unsafeAmount = true; + break; + } + } + $packageRow['amount'] = $oldItem['amount']; + unset($unmatchedOldAmounts[$skuId]); + } + unset($packageRow); + if (! $unsafeAmount && $unmatchedOldAmounts !== []) { + foreach ($packageRows as $packageRow) { + if ($packageRow['amount'] === null) { + $unsafeAmount = true; + break; + } + } + } + if ($unsafeAmount) { + $this->packagesRetainedForUnmatchedAmount++; + continue; + } + $replaceIds[] = $id; + foreach ($packageRows as &$packageRow) { + foreach (self::PRESERVED_PACKAGE_FIELDS as $field) { + $packageRow[$field] ??= null; + } + foreach (self::PRESERVED_ITEM_FIELDS as $field) { + $packageRow[$field] = $previousItems[$id][(int) $packageRow['sku_source_id']][$field] ?? null; + } + } + unset($packageRow); + if (isset($previous[$id])) { + foreach (self::PRESERVED_PACKAGE_FIELDS as $field) { + $packageRows[0][$field] = $previous[$id][$field] ?? null; + } + foreach (self::PRESERVED_CONTACT_FIELDS as $field) { + if (($packageRows[0][$field] ?? null) === null || $packageRows[0][$field] === '') { + $packageRows[0][$field] = $previous[$id][$field] ?? null; + } + } + if ($this->skipFreight) { + $packageRows[0] = $this->withPreservedFreight($packageRows[0], $previous[$id]); + } elseif (! ($hasFreight[$id] ?? false)) { + $packageRows[0] = $this->withPreservedCharges($packageRows[0], $previous[$id], false); + } + } + array_push($rows, ...$packageRows); + } + if ($replaceIds === []) { + return; + } + $this->db->table('jst_oms_shipping')->whereIn('source_id', $replaceIds)->delete(); + foreach (array_chunk($rows, self::FLUSH) as $chunk) { + $this->db->table('jst_oms_shipping')->insert($chunk); + } + }); + + $this->replacementPackages = []; + $this->replacementHasFreight = []; + $this->replacementHasItems = []; + $this->replacementSyntheticSkuIds = []; + $this->replacementRowCount = 0; + } + private function flush(): void { if ($this->buffer === []) { @@ -928,6 +1214,43 @@ class ShippingReplicaSynchronizer return $out; } + private function countUnseenMirrorPackages(string $fromS, string $toS): int + { + $unseen = 0; + $ids = $this->db->table('jst_oms_shipping') + ->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS) + ->distinct()->pluck('source_id'); + foreach ($ids as $id) { + if (! isset($this->seenSourceIds[(int) $id])) { + $unseen++; + } + } + + return $unseen; + } + + private function withPreservedFreight(array $payload, array $previous): array + { + $payload = $this->withPreservedCharges($payload, $previous); + if (! empty($previous['express_no'])) { + $payload['express_no'] = $previous['express_no']; + } + $payload['express_name2'] = $previous['express_name2'] ?? $payload['express_name2']; + + return $payload; + } + + private function withPreservedCharges(array $payload, array $previous, bool $includeNull = true): array + { + foreach (['freight', 'settle_freight', 'weight_kg', 'settle_weight_kg'] as $field) { + if ($includeNull || $previous[$field] !== null) { + $payload[$field] = $previous[$field]; + } + } + + return $payload; + } + private function cleanPayload(array $data): array { foreach ($data as $k => $v) { @@ -977,6 +1300,67 @@ class ShippingReplicaSynchronizer ]); } + private function windowDatasetKey(string $prefix, string $fromS, string $toS): string + { + return $prefix . Carbon::parse($fromS)->format('YmdHis') . '-' . Carbon::parse($toS)->format('YmdHis'); + } + + private function markFreightIncomplete(string $fromS, string $toS): void + { + $this->db->table('jst_sync_state')->updateOrInsert( + ['dataset_key' => $this->windowDatasetKey(self::FREIGHT_DATASET_PREFIX, $fromS, $toS)], + [ + 'status' => 'pending', + 'last_error' => '本窗口采用数量优先补录；上游运费查询已跳过，新记录的运费和结算重量待补齐，空白不代表 0。', + 'updated_at' => now(), + 'created_at' => now(), + ], + ); + } + + private function markItemsIncomplete(string $fromS, string $toS): void + { + $detail = sprintf( + '部分日期窗口的商品或金额待核对：%d 个包裹未匹配商品，%d 个包裹金额无法安全匹配，DWS 预载失败 %d 天。', + $this->packagesWithoutItems, + $this->packagesRetainedForUnmatchedAmount, + count($this->dwsFailedDays), + ); + $this->db->table('jst_sync_state')->updateOrInsert( + ['dataset_key' => $this->windowDatasetKey(self::ITEMS_DATASET_PREFIX, $fromS, $toS)], + ['status' => 'pending', 'last_error' => $detail, 'updated_at' => now(), 'created_at' => now()], + ); + } + + private function markCovered(string $prefix, string $fromS, string $toS): void + { + $from = Carbon::parse($fromS)->format('YmdHis'); + $to = Carbon::parse($toS)->format('YmdHis'); + $states = $this->db->table('jst_sync_state')->where('dataset_key', 'like', $prefix . '%') + ->where('status', '!=', 'succeeded')->pluck('dataset_key'); + foreach ($states as $key) { + $range = substr((string) $key, strlen($prefix)); + if (! preg_match('/^\d{14}-\d{14}$/', $range)) { + continue; + } + [$pendingFrom, $pendingTo] = explode('-', $range, 2); + if ($from <= $pendingFrom && $pendingTo <= $to) { + $this->db->table('jst_sync_state')->where('dataset_key', $key)->update([ + 'status' => 'succeeded', 'last_error' => null, 'last_succeeded_at' => now(), 'updated_at' => now(), + ]); + } + } + } + + private function markIncompleteWindow(int $unseen): void + { + $this->db->table('jst_sync_state')->where('dataset_key', self::DATASET)->update([ + 'status' => 'partial', + 'last_error' => "本窗口尚有 {$unseen} 个镜像包裹未在上游扫描中出现，已保留原记录，需按源键核对。", + 'updated_at' => now(), + ]); + } + private function markFailed(string $err): void { $this->db->table('jst_sync_state')->where('dataset_key', self::DATASET)->update([ diff --git a/resources/views/business/pages/courier-payables/shipping-data.blade.php b/resources/views/business/pages/courier-payables/shipping-data.blade.php index 3f1f6f6e..052e948f 100644 --- a/resources/views/business/pages/courier-payables/shipping-data.blade.php +++ b/resources/views/business/pages/courier-payables/shipping-data.blade.php @@ -1,7 +1,7 @@ @php $filters = $shippingFilters ?? []; $sync = $shippingSync ?? ['code' => 'unavailable', 'last_succeeded_at' => null]; - $statusLabels = ['succeeded' => '数据新鲜', 'running' => '同步中', 'pending' => '等待同步', 'failed' => '同步失败', 'unavailable' => '尚未同步']; + $statusLabels = ['succeeded' => '数据新鲜', 'partial' => !empty($sync['freight_pending']) ? '部分数据已同步 · 运费待补' : '部分数据已同步 · 商品或金额待核对', 'running' => '同步中', 'pending' => '等待同步', 'failed' => '同步失败', 'unavailable' => '尚未同步']; // 按工作平台发货数据界面顺序的 47 列： [列key, 表头, 对齐, 类型] $columns = [ @@ -120,10 +120,14 @@
{{ $statusLabels[$sync['code'] ?? 'unavailable'] ?? '状态未知' }} - @if (!empty($sync['last_succeeded_at'])) 最近同步 {{ $sync['last_succeeded_at'] }} @endif + @if (!empty($sync['last_succeeded_at'])) {{ ($sync['code'] ?? '') === 'partial' ? '最近成功同步' : '最近同步' }} {{ $sync['last_succeeded_at'] }} @endif
+ @if (!empty($sync['warning'])) +
{{ $sync['warning'] }}
+ @endif + @if (!empty($shippingError))
{{ $shippingError }}
@endif diff --git a/tests/Feature/Business/ShippingDataPageTest.php b/tests/Feature/Business/ShippingDataPageTest.php index 11bf8619..bc35f636 100644 --- a/tests/Feature/Business/ShippingDataPageTest.php +++ b/tests/Feature/Business/ShippingDataPageTest.php @@ -59,6 +59,78 @@ class ShippingDataPageTest extends FeatureTestCase ->assertSeeText('导出明细'); } + public function test_page_shows_partial_backfill_and_pending_freight_warning(): void + { + $this->grantPermission(); + + $repository = Mockery::mock(ShippingOrderRepository::class); + $repository->shouldReceive('paginate')->once()->andReturn(['data' => [], 'total' => 0]); + $repository->shouldReceive('syncStatus')->once()->andReturn([ + 'code' => 'partial', + 'last_succeeded_at' => '2026-09-24 09:00:00', + 'freight_pending' => true, + 'warning' => '2 个日期窗口的运费和结算重量待补，空白不代表 0。', + ]); + $this->app->instance(ShippingOrderRepository::class, $repository); + + $this->loginAs()->get(route('business.pages.show', [ + 'section' => 'courier-payables', + 'page' => 'shipping-data', + 'ship_from' => '2026-09-01', + 'ship_to' => '2026-09-24', + ])) + ->assertOk() + ->assertSeeText('部分数据已同步 · 运费待补') + ->assertSeeText('最近成功同步 2026-09-24 09:00:00') + ->assertSeeText('2 个日期窗口的运费和结算重量待补，空白不代表 0。') + ->assertDontSeeText('数据新鲜'); + } + + public function test_page_distinguishes_pending_items_from_pending_freight(): void + { + $this->grantPermission(); + + $repository = Mockery::mock(ShippingOrderRepository::class); + $repository->shouldReceive('paginate')->once()->andReturn(['data' => [], 'total' => 0]); + $repository->shouldReceive('syncStatus')->once()->andReturn([ + 'code' => 'partial', + 'last_succeeded_at' => null, + 'freight_pending' => false, + 'warning' => '1 个日期窗口的商品或金额待核对。', + ]); + $this->app->instance(ShippingOrderRepository::class, $repository); + + $this->loginAs()->get(route('business.pages.show', [ + 'section' => 'courier-payables', + 'page' => 'shipping-data', + ])) + ->assertOk() + ->assertSeeText('部分数据已同步 · 商品或金额待核对') + ->assertSeeText('1 个日期窗口的商品或金额待核对。') + ->assertDontSeeText('运费待补'); + } + + public function test_page_hides_repository_sql_exception_details(): void + { + $this->grantPermission(); + + $repository = Mockery::mock(ShippingOrderRepository::class); + $repository->shouldReceive('paginate')->once()->andThrow(new \RuntimeException( + 'SQLSTATE[42S02]: table SECRET_MIRROR_TABLE does not exist', + )); + $repository->shouldNotReceive('syncStatus'); + $this->app->instance(ShippingOrderRepository::class, $repository); + + $this->loginAs()->get(route('business.pages.show', [ + 'section' => 'courier-payables', + 'page' => 'shipping-data', + ])) + ->assertOk() + ->assertSeeText('发货数据暂时无法读取（可能尚未运行迁移或同步）。') + ->assertDontSee('SQLSTATE') + ->assertDontSee('SECRET_MIRROR_TABLE'); + } + public function test_export_uses_current_filters_and_keeps_identifiers_unprefixed(): void { $this->grantPermission(); diff --git a/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php b/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php index 84160cdc..b9b2407c 100644 --- a/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php +++ b/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php @@ -2,10 +2,13 @@ namespace Tests\Unit\EcommerceFinance; +use Illuminate\Support\Carbon; use Illuminate\Support\Facades\DB; use Modules\EcommerceFinance\Services\Jushuitan\ShippingReplicaSynchronizer; use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery; use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory; +use Modules\EcommerceFinance\Repositories\ShippingOrderRepository; +use RuntimeException; use Tests\TestCase; class ShippingReplicaSynchronizerTest extends TestCase @@ -78,6 +81,769 @@ class ShippingReplicaSynchronizerTest extends TestCase $this->assertSame([], $byTrade); } + public function test_windowed_scan_continues_past_an_old_page_to_find_later_shipments(): void + { + $pages = [ + array_map(fn (int $id): array => ['id' => $id, 'send_time' => $id === 4500 ? '2026-09-02 12:00:00' : null], range(4500, 3001)), + array_map(fn (int $id): array => ['id' => $id, 'send_time' => '2026-08-31 12:00:00'], range(3000, 1501)), + [['id' => 1500, 'send_time' => '2026-09-03 12:00:00']], + ]; + $scanCalls = 0; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls, $pages): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return $pages[$scanCalls++] ?? []; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), + ); + + $this->assertSame(3, $scanCalls); + $this->assertSame(2, $result['processed']); + $this->assertSame(2, $result['unique_packages']); + $this->assertSame([1500, 4500], DB::connection('business_mirror')->table('jst_oms_shipping') + ->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all()); + } + + public function test_window_result_distinguishes_duplicate_package_rows_from_unique_source_ids(): void + { + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + $row = ['id' => 5000, 'send_time' => '2026-09-02 12:00:00']; + return [$row, $row]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true, + ); + + $this->assertSame(2, $result['packages']); + $this->assertSame(1, $result['unique_packages']); + $this->assertSame(1, DB::connection('business_mirror')->table('jst_oms_shipping') + ->distinct()->count('source_id')); + } + + public function test_express_scan_fails_when_more_rows_remain_after_batch_limit(): void + { + DB::connection('business_mirror')->table('jst_oms_shipping')->insert([ + 'source_id' => 3000, 'sku_source_id' => 3000, 'ship_time' => '2026-09-02 12:00:00', + 'express_no' => 'old-tracking', 'freight' => 18.75, + ]); + $pages = [ + array_map(fn (int $id): array => ['id' => $id, 'send_time' => $id === 3000 ? '2026-09-02 12:00:00' : null], range(3000, 1501)), + [['id' => 1500]], + ]; + $scanCalls = 0; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls, $pages): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return $pages[$scanCalls++] ?? []; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + $synchronizer = new ShippingReplicaSynchronizer($factory); + $this->setScanBatchLimit($synchronizer, 'maxExpressBatches', 1); + + try { + $synchronizer->sync(Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true); + $this->fail('Expected an incomplete express scan to fail.'); + } catch (RuntimeException $e) { + $this->assertStringContainsString('发货扫描达到批次上限', $e->getMessage()); + } + + $this->assertSame(2, $scanCalls); + $this->assertSame('failed', DB::connection('business_mirror')->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade')->value('status')); + $this->assertSame('pending', DB::connection('business_mirror')->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade.freight.20260901000000-20260905000000')->value('status')); + $original = DB::connection('business_mirror')->table('jst_oms_shipping')->where('source_id', 3000)->first(); + $this->assertSame('old-tracking', $original->express_no); + $this->assertEquals(18.75, $original->freight); + } + + public function test_full_last_batch_succeeds_when_no_more_express_rows_exist(): void + { + $scanCalls = 0; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return $scanCalls++ === 0 + ? array_map(fn (int $id): array => ['id' => $id], range(1500, 1)) + : []; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + $synchronizer = new ShippingReplicaSynchronizer($factory); + $this->setScanBatchLimit($synchronizer, 'maxExpressBatches', 1); + + $result = $synchronizer->sync(Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05')); + + $this->assertSame(2, $scanCalls); + $this->assertSame(0, $result['processed']); + $this->assertSame('succeeded', DB::connection('business_mirror')->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade')->value('status')); + } + + public function test_auto_ship_scan_fails_when_more_rows_remain_after_batch_limit(): void + { + $pages = [ + array_map(fn (int $id): array => ['rid' => $id, 'o_id' => ''], range(1000, 601)), + [['rid' => 600, 'o_id' => '']], + ]; + $scanCalls = 0; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls, $pages): array { + if (str_contains($sql, 'FROM rpt.jst_Orders o WHERE')) { + return $pages[$scanCalls++] ?? []; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + $synchronizer = new ShippingReplicaSynchronizer($factory); + $this->setScanBatchLimit($synchronizer, 'maxAutoShipBatches', 1); + + try { + $synchronizer->sync(Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05')); + $this->fail('Expected an incomplete auto-ship scan to fail.'); + } catch (RuntimeException $e) { + $this->assertStringContainsString('代发仓扫描达到批次上限', $e->getMessage()); + } + + $this->assertSame(2, $scanCalls); + $this->assertSame('failed', DB::connection('business_mirror')->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade')->value('status')); + } + + public function test_skip_freight_preserves_existing_amounts_and_marks_new_amounts_incomplete(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + ['source_id' => 5000, 'sku_source_id' => 5000, 'ship_time' => '2026-09-02 12:00:00', + 'express_no' => 'existing-tracking', 'express_name2' => 'existing-express', + 'freight' => 12.34, 'settle_freight' => 13.45, 'weight_kg' => 1.2, 'settle_weight_kg' => 1.3], + ['source_id' => 6000, 'sku_source_id' => 6000, 'ship_time' => '2026-09-03 12:00:00', + 'express_no' => 'weight-only-tracking', 'express_name2' => null, + 'freight' => null, 'settle_freight' => null, 'weight_kg' => null, 'settle_weight_kg' => 4.5], + ]); + + $scanCalls = 0; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array { + if (str_contains($sql, 'fys_jstFreightCostDetail')) { + $this->fail('Skip-freight mode must not query the freight view.'); + } + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return $scanCalls++ === 0 ? [ + ['id' => 6000, 'send_time' => '2026-09-03 12:00:00', 'weight' => 2, 'logistic_id' => 'new-weight-tracking'], + ['id' => 5000, 'send_time' => '2026-09-02 12:00:00', 'weight' => 2, 'logistic_id' => 'new-tracking'], + ['id' => 4000, 'send_time' => '2026-09-03 12:00:00', 'weight' => 2], + ] : []; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true, + ); + + $this->assertSame(3, $result['processed']); + $this->assertSame(3, $result['packages']); + $this->assertSame(3, $result['packages_without_items']); + $this->assertTrue($result['freight_incomplete']); + $this->assertSame('succeeded', $db->table('jst_sync_state')->where('dataset_key', 'jst.sctrade')->value('status')); + $this->assertSame('pending', $db->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade.freight.20260901000000-20260905000000')->value('status')); + $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']); + $existing = $db->table('jst_oms_shipping')->where('source_id', 5000)->first(); + $this->assertSame('existing-tracking', $existing->express_no); + $this->assertSame('existing-express', $existing->express_name2); + $this->assertEquals(12.34, $existing->freight); + $this->assertEquals(13.45, $existing->settle_freight); + $this->assertEquals(1.2, $existing->weight_kg); + $this->assertEquals(1.3, $existing->settle_weight_kg); + $weightOnly = $db->table('jst_oms_shipping')->where('source_id', 6000)->first(); + $this->assertSame('weight-only-tracking', $weightOnly->express_no); + $this->assertNull($weightOnly->freight); + $this->assertEquals(4.5, $weightOnly->settle_weight_kg); + $new = $db->table('jst_oms_shipping')->where('source_id', 4000)->first(); + $this->assertNull($new->freight); + $this->assertNull($new->settle_freight); + $this->assertNull($new->weight_kg); + $this->assertNull($new->settle_weight_kg); + $this->assertEquals(2, $new->actual_weight); + + (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-04'), Carbon::parse('2026-09-05'), + ); + $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']); + + (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), + ); + $this->assertSame('pending', $db->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade.freight.20260901000000-20260905000000')->value('status')); + $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']); + } + + public function test_normal_window_preserves_old_charges_when_freight_is_missing_but_uses_new_cost(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + ['source_id' => 5000, 'sku_source_id' => 5000, 'ship_time' => '2026-09-02 12:00:00', + 'express_no' => 'old-5000', 'freight' => 12.34, 'settle_freight' => 13.45, + 'weight_kg' => 1.2, 'settle_weight_kg' => 1.3], + ['source_id' => 6000, 'sku_source_id' => 6000, 'ship_time' => '2026-09-03 12:00:00', + 'express_no' => 'old-6000', 'freight' => 15.00, 'settle_freight' => 15.00, + 'weight_kg' => 1.5, 'settle_weight_kg' => 1.5], + ]); + $db->table('jst_sync_state')->insert([ + 'dataset_key' => 'jst.sctrade.freight.20260901000000-20260905000000', + 'status' => 'pending', + ]); + + $scanCalls = 0; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return $scanCalls++ === 0 ? [ + ['id' => 6000, 'so_id' => 'so-6000', 'logistic_id' => 'new-6000', + 'send_time' => '2026-09-03 12:00:00', 'weight' => 2], + ['id' => 5000, 'so_id' => 'so-5000', 'logistic_id' => 'new-5000', + 'send_time' => '2026-09-02 12:00:00', 'weight' => 2], + ] : []; + } + if (str_contains($sql, 'fys_jstFreightCostDetail')) { + return [['so_id' => 'so-6000', 'expressNo' => 'new-6000', 'cost' => 7.50, + 'w' => 2.5, 'name' => '快递公司']]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), + ); + + $this->assertSame(0, $result['stale_mirror_packages']); + $missing = $db->table('jst_oms_shipping')->where('source_id', 5000)->first(); + $this->assertSame('new-5000', $missing->express_no); + $this->assertEquals(12.34, $missing->freight); + $this->assertEquals(13.45, $missing->settle_freight); + $this->assertEquals(1.2, $missing->weight_kg); + $this->assertEquals(1.3, $missing->settle_weight_kg); + $updated = $db->table('jst_oms_shipping')->where('source_id', 6000)->first(); + $this->assertEquals(7.50, $updated->freight); + $this->assertEquals(7.50, $updated->settle_freight); + $this->assertEquals(2.5, $updated->settle_weight_kg); + $this->assertSame('pending', $db->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade.freight.20260901000000-20260905000000')->value('status')); + } + + public function test_window_replacement_preserves_unreconstructed_package_and_matching_item_fields(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + 'source_id' => 5000, 'sku_source_id' => 101, 'ship_time' => '2026-09-02 12:00:00', + 'package_no' => 'old-package', 'download_time' => '2026-09-02 13:00:00', + 'freight_rate' => 1.25, 'freight_formula' => 'legacy-formula', + 'settle_volume_cm3' => 100, 'recon_weight_kg' => 1.5, 'recon_volume_cm3' => 105, + 'finished_barcode' => 'finished-101', 'finished_name' => '成品甲', + 'finished_supplier' => '供应商甲', + ]); + $db->table('jst_oms_shipping')->insert([ + 'source_id' => 5000, 'sku_source_id' => 102, 'ship_time' => '2026-09-02 12:00:00', + 'buyer_nick' => 'old-buyer', 'seller_memo' => 'old-seller-memo', 'front_no' => 'old-front-no', + 'receiver_name' => 'old-recipient', 'receiver_mobile' => 'old-mobile', 'receiver_phone' => 'old-phone', + 'finished_barcode' => 'finished-102', 'finished_name' => '成品乙', + 'finished_supplier' => '供应商乙', + ]); + $scanCalls = 0; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return $scanCalls++ === 0 ? [[ + 'id' => 5000, 'inner_order_id' => 'order-1', 'send_time' => '2026-09-02 12:00:00', + ]] : []; + } + if (str_contains($sql, 'FROM rpt.jst_OrderItem WHERE o_id IN')) { + return [ + ['o_id' => 'order-1', 'Id' => 101, 'name' => '新商品甲', 'qty' => 1], + ['o_id' => 'order-1', 'Id' => 103, 'name' => '新商品丙', 'qty' => 1], + ]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), + ); + + $this->assertSame([101, 103], $db->table('jst_oms_shipping')->where('source_id', 5000) + ->orderBy('sku_source_id')->pluck('sku_source_id')->map(fn ($id) => (int) $id)->all()); + $first = $db->table('jst_oms_shipping')->where('source_id', 5000)->where('sku_source_id', 101)->first(); + $this->assertSame('old-buyer', $first->buyer_nick); + $this->assertSame('old-seller-memo', $first->seller_memo); + $this->assertSame('old-recipient', $first->receiver_name); + $this->assertSame('old-mobile', $first->receiver_mobile); + $this->assertSame('old-phone', $first->receiver_phone); + $this->assertSame('old-package', $first->package_no); + $this->assertSame('2026-09-02 13:00:00', $first->download_time); + $this->assertEquals(1.25, $first->freight_rate); + $this->assertSame('legacy-formula', $first->freight_formula); + $this->assertEquals(100, $first->settle_volume_cm3); + $this->assertEquals(1.5, $first->recon_weight_kg); + $this->assertEquals(105, $first->recon_volume_cm3); + $this->assertSame('old-front-no', $first->front_no); + $this->assertSame('finished-101', $first->finished_barcode); + $this->assertSame('成品甲', $first->finished_name); + $this->assertSame('供应商甲', $first->finished_supplier); + $second = $db->table('jst_oms_shipping')->where('source_id', 5000)->where('sku_source_id', 103)->first(); + $this->assertNull($second->seller_memo); + $this->assertNull($second->settle_volume_cm3); + $this->assertNull($second->finished_barcode); + $this->assertNull($second->finished_name); + $this->assertNull($second->finished_supplier); + } + + public function test_missing_new_items_preserves_old_product_rows_and_marks_items_pending(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + ['source_id' => 5000, 'sku_source_id' => 101, 'ship_time' => '2026-09-02 12:00:00', + 'barcode' => 'sku-101', 'product_name' => '旧商品甲', 'qty' => 1, 'amount' => 10.50, + 'freight' => 5.25], + ['source_id' => 5000, 'sku_source_id' => 102, 'ship_time' => '2026-09-02 12:00:00', + 'barcode' => 'sku-102', 'product_name' => '旧商品乙', 'qty' => 2, 'amount' => 20.75, + 'freight' => null], + ]); + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql, array $bindings): array { + if (str_contains($sql, 'rpt.dws_sn_sales_full_info_di') && $bindings === ['2026-09-02']) { + throw new RuntimeException('DWS read timed out'); + } + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return [['id' => 5000, 'inner_order_id' => 'order-1', + 'send_time' => '2026-09-02 12:00:00']]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true, + ); + + $this->assertSame(1, $result['packages_without_items']); + $this->assertSame(1, $result['packages_retained_for_missing_items']); + $this->assertContains('2026-09-02', $result['dws_failed_days']); + $this->assertSame('pending', $db->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade.items.20260901000000-20260905000000')->value('status')); + $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']); + $rows = $db->table('jst_oms_shipping')->where('source_id', 5000)->orderBy('sku_source_id')->get(); + $this->assertCount(2, $rows); + $this->assertSame(['sku-101', 'sku-102'], $rows->pluck('barcode')->all()); + $this->assertSame(['旧商品甲', '旧商品乙'], $rows->pluck('product_name')->all()); + $this->assertEquals(10.50, $rows[0]->amount); + $this->assertEquals(20.75, $rows[1]->amount); + $this->assertEquals(5.25, $rows[0]->freight); + } + + public function test_view_item_with_null_amount_preserves_matching_synthetic_sku_amount(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + 'source_id' => 5000, 'sku_source_id' => 500000, 'ship_time' => '2026-09-02 12:00:00', + 'barcode' => 'sku-101', 'product_name' => '旧商品', 'amount' => 10.50, + ]); + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return [['id' => 5000, 'inner_order_id' => 'order-1', + 'send_time' => '2026-09-02 12:00:00']]; + } + if (str_contains($sql, 'FROM rpt.jst_salesOutboundInfo WHERE io_date')) { + return [['o_id' => 'order-1', 'io_id' => 'outbound-1']]; + } + if (str_contains($sql, 'FROM rpt.Jst_SendData_View_extension')) { + return [['io_id' => 'outbound-1', 'sku_ids' => 'sku-101', 'qty' => 1, + 'type' => 0, 'sku_name' => '实发商品', 'volume' => null]]; + } + if (str_contains($sql, 'FROM rpt.jst_OrderItem WHERE o_id IN')) { + return [['o_id' => 'order-1', 'Id' => 101, 'sku_id' => 'other-sku', + 'name' => '订单商品', 'qty' => 1, 'amount' => 999]]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), + ); + + $this->assertSame(0, $result['packages_retained_for_unmatched_amount']); + $row = $db->table('jst_oms_shipping')->where('source_id', 5000)->sole(); + $this->assertSame(500000, (int) $row->sku_source_id); + $this->assertSame('实发商品', $row->product_name); + $this->assertEquals(10.50, $row->amount); + } + + public function test_view_item_with_unmatched_synthetic_sku_retains_old_package_and_marks_items_pending(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + ['source_id' => 5000, 'sku_source_id' => 500000, 'ship_time' => '2026-09-02 12:00:00', + 'barcode' => null, 'product_name' => '旧商品甲', 'amount' => 10.50], + ['source_id' => 6000, 'sku_source_id' => 600000, 'ship_time' => '2026-09-02 13:00:00', + 'barcode' => 'old-sku', 'product_name' => '旧商品乙', 'amount' => 20.75], + ]); + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return [ + ['id' => 5000, 'inner_order_id' => 'order-1', 'send_time' => '2026-09-02 12:00:00'], + ['id' => 6000, 'inner_order_id' => 'order-2', 'send_time' => '2026-09-02 13:00:00'], + ]; + } + if (str_contains($sql, 'FROM rpt.jst_salesOutboundInfo WHERE io_date')) { + return [ + ['o_id' => 'order-1', 'io_id' => 'outbound-1'], + ['o_id' => 'order-2', 'io_id' => 'outbound-2'], + ]; + } + if (str_contains($sql, 'FROM rpt.Jst_SendData_View_extension')) { + return [ + ['io_id' => 'outbound-1', 'sku_ids' => 'new-sku', 'qty' => 1, 'type' => 0, + 'sku_name' => '新商品甲', 'volume' => null], + ['io_id' => 'outbound-2', 'sku_ids' => 'new-sku', 'qty' => 1, 'type' => 0, + 'sku_name' => '新商品乙', 'volume' => null], + ]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), + ); + + $this->assertSame(2, $result['packages_retained_for_unmatched_amount']); + $this->assertSame('pending', $db->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade.items.20260901000000-20260905000000')->value('status')); + $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']); + $rows = $db->table('jst_oms_shipping')->orderBy('source_id')->get(); + $this->assertSame(['旧商品甲', '旧商品乙'], $rows->pluck('product_name')->all()); + $this->assertEquals(10.50, $rows[0]->amount); + $this->assertEquals(20.75, $rows[1]->amount); + } + + public function test_explicit_zero_item_amount_replaces_old_amount(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + 'source_id' => 5000, 'sku_source_id' => 101, 'ship_time' => '2026-09-02 12:00:00', + 'barcode' => 'sku-101', 'amount' => 10.50, + ]); + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return [['id' => 5000, 'inner_order_id' => 'order-1', + 'send_time' => '2026-09-02 12:00:00']]; + } + if (str_contains($sql, 'FROM rpt.jst_OrderItem WHERE o_id IN')) { + return [['o_id' => 'order-1', 'Id' => 101, 'sku_id' => 'sku-101', + 'name' => '新商品', 'qty' => 1, 'amount' => 0]]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), + ); + + $this->assertSame(0, $result['packages_retained_for_unmatched_amount']); + $row = $db->table('jst_oms_shipping')->where('source_id', 5000)->sole(); + $this->assertSame('新商品', $row->product_name); + $this->assertEquals(0, $row->amount); + } + + public function test_shipping_command_reports_express_and_autoship_counts_separately(): void + { + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return [['id' => 5000, 'send_time' => '2026-09-02 12:00:00']]; + } + if (str_contains($sql, 'FROM rpt.jst_Orders o WHERE')) { + return [['rid' => 100, 'o_id' => '123', 'send_date' => '2026-09-02 13:00:00']]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + $this->app->instance(UpstreamReadOnlyQueryFactory::class, $factory); + + $this->artisan('jushuitan:sync-shipping', [ + '--from' => '2026-09-01', '--to' => '2026-09-05', '--skip-freight' => true, + ])->expectsOutputToContain('唯一包裹源键 2 个，包裹记录处理 2 次；发货明细 1 条，代发仓 1 条，合计处理 2 条明细') + ->assertExitCode(0); + } + + public function test_normal_window_failure_preserves_unprocessed_old_packages(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + ['source_id' => 3000, 'sku_source_id' => 3000, 'ship_time' => '2026-09-02 12:00:00', + 'express_no' => 'processed-old', 'freight' => 11.11], + ['source_id' => 1000, 'sku_source_id' => 1000, 'ship_time' => '2026-09-02 12:00:00', + 'express_no' => 'unprocessed-old', 'freight' => 22.22], + ]); + + $scanCalls = 0; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + if ($scanCalls++ > 0) { + throw new RuntimeException('upstream page failed'); + } + return array_map(fn (int $id): array => [ + 'id' => $id, + 'send_time' => $id >= 2701 ? '2026-09-02 12:00:00' : '2026-08-31 12:00:00', + ], range(3000, 1501)); + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + try { + (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), + ); + $this->fail('Expected the second upstream page to fail.'); + } catch (RuntimeException $e) { + $this->assertSame('upstream page failed', $e->getMessage()); + } + + $this->assertSame(2, $scanCalls); + $this->assertSame('failed', $db->table('jst_sync_state')->where('dataset_key', 'jst.sctrade')->value('status')); + $processed = $db->table('jst_oms_shipping')->where('source_id', 3000)->first(); + $this->assertEquals(11.11, $processed->freight); + $unprocessed = $db->table('jst_oms_shipping')->where('source_id', 1000)->first(); + $this->assertSame('unprocessed-old', $unprocessed->express_no); + $this->assertEquals(22.22, $unprocessed->freight); + $this->assertSame(301, $db->table('jst_oms_shipping')->count()); + } + + public function test_at_prefix_dedup_requires_same_trade_and_same_window(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + ['source_id' => 1, 'sku_source_id' => 1, 'ship_time' => '2026-09-02 12:00:00', + 'trade_no' => 'same-trade', 'express_no' => 'clean-1'], + ['source_id' => 2, 'sku_source_id' => 2, 'ship_time' => '2026-09-02 12:00:00', + 'trade_no' => 'same-trade', 'express_no' => '@clean-1'], + ['source_id' => 3, 'sku_source_id' => 3, 'ship_time' => '2026-09-02 12:00:00', + 'trade_no' => 'other-trade', 'express_no' => '@clean-1'], + ['source_id' => 4, 'sku_source_id' => 4, 'ship_time' => '2026-09-08 12:00:00', + 'trade_no' => 'outside-trade', 'express_no' => 'clean-2'], + ['source_id' => 5, 'sku_source_id' => 5, 'ship_time' => '2026-09-02 12:00:00', + 'trade_no' => 'outside-trade', 'express_no' => '@clean-2'], + ['source_id' => 6, 'sku_source_id' => 6, 'ship_time' => '2026-09-03 12:00:00', + 'trade_no' => 'same-trade', 'express_no' => '@clean-1'], + ]); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->willReturn($this->createMock(UpstreamReadOnlyQuery::class)); + + $removed = (new ShippingReplicaSynchronizer($factory))->dedupAtPrefix( + '2026-09-01 00:00:00', '2026-09-05 00:00:00', + ); + + $this->assertSame(1, $removed); + $this->assertSame([1, 3, 4, 5, 6], $db->table('jst_oms_shipping') + ->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all()); + $this->assertSame(0, (new ShippingReplicaSynchronizer($factory))->dedupAtPrefix()); + } + + public function test_non_window_fast_sync_still_ignores_existing_package(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + 'source_id' => 5000, 'sku_source_id' => 5000, + 'ship_time' => '2026-09-02 12:00:00', 'express_no' => 'existing-tracking', 'freight' => 12.34, + ]); + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return [[ + 'id' => 5000, + 'send_time' => '2026-09-02 12:00:00', + 'logistic_id' => 'new-tracking', + ]]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync(); + + $this->assertSame(1, $result['processed']); + $this->assertSame(1, $db->table('jst_oms_shipping')->count()); + $existing = $db->table('jst_oms_shipping')->where('source_id', 5000)->first(); + $this->assertSame('existing-tracking', $existing->express_no); + $this->assertEquals(12.34, $existing->freight); + } + + public function test_failed_dws_preload_is_reported_separately_from_package_count(): void + { + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql, array $bindings): array { + if (str_contains($sql, 'rpt.dws_sn_sales_full_info_di') && $bindings === ['2026-08-31']) { + throw new RuntimeException('read timed out'); + } + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return [['id' => 1234, 'send_time' => '2026-09-02 12:00:00']]; + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), + ); + + $this->assertSame(1, $result['packages']); + $this->assertSame(1, $result['packages_without_items']); + $this->assertSame(['2026-08-31'], $result['dws_failed_days']); + $this->assertSame('pending', DB::connection('business_mirror')->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade.items.20260901000000-20260905000000')->value('status')); + } + + public function test_skip_freight_rolls_back_old_package_when_replacement_insert_conflicts(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + 'source_id' => 5000, 'sku_source_id' => 5000, + 'ship_time' => '2026-09-02 12:00:00', 'express_no' => 'old-tracking', 'freight' => 12.34, + ]); + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->willReturn($upstream); + $synchronizer = new ShippingReplicaSynchronizer($factory); + + $skipFreight = new \ReflectionProperty($synchronizer, 'skipFreight'); + $skipFreight->setAccessible(true); + $skipFreight->setValue($synchronizer, true); + $replacements = new \ReflectionProperty($synchronizer, 'replacementPackages'); + $replacements->setAccessible(true); + $newRow = [ + 'source_id' => 5000, 'sku_source_id' => 123, + 'ship_time' => '2026-09-02 12:00:00', 'express_no' => 'new-tracking', 'express_name2' => null, + 'freight' => null, 'settle_freight' => null, 'weight_kg' => null, 'settle_weight_kg' => null, + 'amount' => null, + 'receiver_name' => null, 'receiver_mobile' => null, 'receiver_phone' => null, + ]; + $replacements->setValue($synchronizer, [5000 => [$newRow, $newRow]]); + $flush = new \ReflectionMethod($synchronizer, 'flushReplacements'); + $flush->setAccessible(true); + + try { + $flush->invoke($synchronizer); + $this->fail('Expected a duplicate replacement row to roll back the package transaction.'); + } catch (\Illuminate\Database\QueryException $e) { + $this->assertStringContainsString('UNIQUE constraint failed', $e->getMessage()); + } + + $rows = $db->table('jst_oms_shipping')->where('source_id', 5000)->get(); + $this->assertCount(1, $rows); + $this->assertSame(5000, (int) $rows[0]->sku_source_id); + $this->assertSame('old-tracking', $rows[0]->express_no); + $this->assertEquals(12.34, $rows[0]->freight); + } + + public function test_skip_freight_keeps_unseen_mirror_package_and_marks_window_partial(): void + { + $db = DB::connection('business_mirror'); + $db->table('jst_oms_shipping')->insert([ + 'source_id' => 7000, 'sku_source_id' => 7000, + 'ship_time' => '2026-09-02 12:00:00', 'express_no' => 'unseen-tracking', 'freight' => 22.50, + ]); + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturn([]); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true, + ); + + $this->assertSame(1, $result['stale_mirror_packages']); + $this->assertSame('partial', $db->table('jst_sync_state')->where('dataset_key', 'jst.sctrade')->value('status')); + $this->assertStringContainsString('源键核对', $db->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade')->value('last_error')); + $old = $db->table('jst_oms_shipping')->where('source_id', 7000)->first(); + $this->assertSame('unseen-tracking', $old->express_no); + $this->assertEquals(22.50, $old->freight); + } + + public function test_partial_sync_status_uses_business_warning_instead_of_raw_error(): void + { + DB::connection('business_mirror')->table('jst_sync_state')->insert([ + 'dataset_key' => 'jst.sctrade', + 'status' => 'partial', + 'last_error' => 'SQLSTATE[42S02]: SECRET_MIRROR_TABLE does not exist', + ]); + + $status = (new ShippingOrderRepository())->syncStatus(); + + $this->assertSame('partial', $status['code']); + $this->assertStringContainsString('源键核对', $status['warning']); + $this->assertStringNotContainsString('SQLSTATE', $status['warning']); + $this->assertStringNotContainsString('SECRET_MIRROR_TABLE', $status['warning']); + } + public function test_sync_removes_only_the_known_demo_source_ids(): void { $db = DB::connection('business_mirror'); @@ -104,4 +870,11 @@ class ShippingReplicaSynchronizerTest extends TestCase $db->table('jst_oms_shipping')->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all(), ); } + + private function setScanBatchLimit(ShippingReplicaSynchronizer $synchronizer, string $property, int $limit): void + { + $reflection = new \ReflectionProperty($synchronizer, $property); + $reflection->setAccessible(true); + $reflection->setValue($synchronizer, $limit); + } }