From d87ca965939fbf65eabfcbe75b81e8fd9f2cbb86 Mon Sep 17 00:00:00 2001 From: limeng <1131177064@qq.com> Date: Fri, 21 Aug 2026 14:49:16 +0800 Subject: [PATCH] =?UTF-8?q?=E6=99=AE=E9=80=9A=E5=95=86=E5=93=81=E5=A2=9E?= =?UTF-8?q?=E5=8A=A01688=E4=BB=A3=E5=8F=91=E9=87=87=E8=B4=AD=E5=8D=95?= =?UTF-8?q?=E4=BB=B7=E6=9D=A5=E6=BA=90=20(#74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 保留并标注普通商品采购价，接入已完成1688代发采购单价的只读共享业务镜像。 --- .../SyncProductPurchasePriceMirror.php | 110 +++++++++ app/Http/Controllers/Business/Pages.php | 14 ++ config/upstream_mirrors.php | 13 ++ ...e_product_purchase_price_mirror_tables.php | 81 +++++++ deploy/activate-incremental-release.sh | 2 +- deploy/activate-release.sh | 2 +- docs/上游业务镜像清单.md | 3 + .../ProductCatalogMirrorRepository.php | 110 ++++++++- .../ProductPurchasePriceMirrorRepository.php | 214 ++++++++++++++++++ .../ProductPurchasePriceRepository.php | 48 ++++ ...ProductPurchasePriceMirrorSynchronizer.php | 111 +++++++++ .../normal-products.blade.php | 70 +++++- .../Business/ProductCatalogPageTest.php | 55 +++++ ...ductPurchasePriceMirrorSyncCommandTest.php | 86 +++++++ .../Deployment/BusinessMirrorScheduleTest.php | 5 + .../ProductCatalogMirrorRepositoryTest.php | 89 ++++++++ ...uctPurchasePriceMirrorSynchronizerTest.php | 80 +++++++ .../ProductPurchasePriceRepositoryTest.php | 53 +++++ 18 files changed, 1138 insertions(+), 8 deletions(-) create mode 100644 app/Console/Commands/SyncProductPurchasePriceMirror.php create mode 100644 database/migrations/2026_08_21_120000_create_product_purchase_price_mirror_tables.php create mode 100644 modules/EcommerceFinance/Repositories/ProductPurchasePriceMirrorRepository.php create mode 100644 modules/EcommerceFinance/Repositories/ProductPurchasePriceRepository.php create mode 100644 modules/EcommerceFinance/Services/ProductPurchasePriceMirrorSynchronizer.php create mode 100644 tests/Feature/Console/ProductPurchasePriceMirrorSyncCommandTest.php create mode 100644 tests/Unit/EcommerceFinance/ProductPurchasePriceMirrorSynchronizerTest.php create mode 100644 tests/Unit/EcommerceFinance/ProductPurchasePriceRepositoryTest.php diff --git a/app/Console/Commands/SyncProductPurchasePriceMirror.php b/app/Console/Commands/SyncProductPurchasePriceMirror.php new file mode 100644 index 00000000..36314e84 --- /dev/null +++ b/app/Console/Commands/SyncProductPurchasePriceMirror.php @@ -0,0 +1,110 @@ +option('full')) { + return self::SUCCESS; + } + + try { + $state = $mirror->state(); + $full = $this->shouldRunFull($state); + $result = $synchronizer->sync( + $full, + $this->option('batch') === null ? null : (int) $this->option('batch'), + $this->option('max-batches') === null ? null : (int) $this->option('max-batches'), + ); + } catch (Throwable $exception) { + Log::error('采购单价共享镜像同步失败。', ['exception' => $exception::class]); + $this->error('采购单价同步失败，已保留上一版共享数据。'); + + return self::FAILURE; + } + + if ($result['status'] === 'requires_full_sync') { + $this->warn('采购单价尚未完成首次全量同步，需经授权后显式使用 --full。'); + + return self::SUCCESS; + } + + $this->info(sprintf( + '采购单价%s：%s，本轮 %d 批，累计处理 %d 行。', + $result['mode'] === 'full' ? '全量校准' : '增量同步', + $result['status'] === 'succeeded' ? '完成' : '进行中', + $result['batches'], + $result['rows'], + )); + + return self::SUCCESS; + } + + /** @param array|null $state */ + private function shouldRunFull(?array $state): bool + { + if ((bool) $this->option('full')) { + return true; + } + + if (is_array($state) + && ($state['run_mode'] ?? null) === 'full' + && in_array($state['status'] ?? null, ['running', 'failed'], true)) { + return true; + } + + if (! is_array($state) || ($state['initial_completed_at'] ?? null) === null) { + return false; + } + + $now = now(); + $weekday = min(max((int) config( + 'upstream_mirrors.product_purchase_prices.reconcile_weekday', + 0, + ), 0), 6); + $startHour = min(max((int) config( + 'upstream_mirrors.product_purchase_prices.reconcile_start_hour', + 2, + ), 0), 23); + $windowHours = min(max((int) config( + 'upstream_mirrors.product_purchase_prices.reconcile_window_hours', + 6, + ), 1), 24 - $startHour); + + if ($now->dayOfWeek !== $weekday + || $now->hour < $startHour + || $now->hour >= $startHour + $windowHours) { + return false; + } + + $lastFull = $state['last_full_succeeded_at'] ?? null; + if ($lastFull === null || $lastFull === '') { + return true; + } + + $hours = max((int) config( + 'upstream_mirrors.product_purchase_prices.full_reconcile_hours', + 168, + ), 1); + + return Carbon::parse((string) $lastFull)->addHours($hours)->lte($now); + } +} diff --git a/app/Http/Controllers/Business/Pages.php b/app/Http/Controllers/Business/Pages.php index de08cdf1..0fdb8f45 100644 --- a/app/Http/Controllers/Business/Pages.php +++ b/app/Http/Controllers/Business/Pages.php @@ -672,6 +672,13 @@ class Pages extends Controller 'productCatalogError' => null, 'productCatalogState' => $state, 'productCatalogSyncStatus' => $syncStatus, + 'productCatalogProductPriceSource' => ProductCatalogMirrorRepository::PRODUCT_SOURCE_LABEL, + 'productCatalogPurchasePriceSource' => ProductCatalogMirrorRepository::PURCHASE_SOURCE_LABEL, + 'productCatalogPurchasePriceSyncStatus' => $result['purchase_price_sync_status'] ?? [ + 'code' => 'unavailable', + 'initialized' => false, + 'last_succeeded_at' => null, + ], ]; } catch (Throwable $exception) { report($exception); @@ -710,6 +717,13 @@ class Pages extends Controller 'productCatalogError' => $error, 'productCatalogState' => $state, 'productCatalogSyncStatus' => $syncStatus, + 'productCatalogProductPriceSource' => ProductCatalogMirrorRepository::PRODUCT_SOURCE_LABEL, + 'productCatalogPurchasePriceSource' => ProductCatalogMirrorRepository::PURCHASE_SOURCE_LABEL, + 'productCatalogPurchasePriceSyncStatus' => [ + 'code' => 'unavailable', + 'initialized' => false, + 'last_succeeded_at' => null, + ], ]; } diff --git a/config/upstream_mirrors.php b/config/upstream_mirrors.php index 32e54a7f..d455e74f 100644 --- a/config/upstream_mirrors.php +++ b/config/upstream_mirrors.php @@ -31,6 +31,19 @@ return [ 'reconcile_window_hours' => (int) env('PRODUCT_CATALOG_MIRROR_RECONCILE_WINDOW_HOURS', 6), ], + 'product_purchase_prices' => [ + 'dataset_key' => 'product-inventory.purchase-order-prices', + 'source_key' => 'jiansun', + 'source_object' => 'dbo.jst_purchase + dbo.jst_purchaseItem', + 'batch_size' => (int) env('PRODUCT_PURCHASE_PRICE_MIRROR_BATCH_SIZE', 1000), + 'max_batches_per_run' => (int) env('PRODUCT_PURCHASE_PRICE_MIRROR_MAX_BATCHES', 5), + 'overlap_hours' => (int) env('PRODUCT_PURCHASE_PRICE_MIRROR_OVERLAP_HOURS', 48), + 'full_reconcile_hours' => (int) env('PRODUCT_PURCHASE_PRICE_MIRROR_FULL_RECONCILE_HOURS', 168), + 'reconcile_weekday' => (int) env('PRODUCT_PURCHASE_PRICE_MIRROR_RECONCILE_WEEKDAY', 0), + 'reconcile_start_hour' => (int) env('PRODUCT_PURCHASE_PRICE_MIRROR_RECONCILE_START_HOUR', 2), + 'reconcile_window_hours' => (int) env('PRODUCT_PURCHASE_PRICE_MIRROR_RECONCILE_WINDOW_HOURS', 6), + ], + /* |-------------------------------------------------------------------------- | 上游原表镜像目录 diff --git a/database/migrations/2026_08_21_120000_create_product_purchase_price_mirror_tables.php b/database/migrations/2026_08_21_120000_create_product_purchase_price_mirror_tables.php new file mode 100644 index 00000000..ae3d3164 --- /dev/null +++ b/database/migrations/2026_08_21_120000_create_product_purchase_price_mirror_tables.php @@ -0,0 +1,81 @@ +environment('testing') + || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') { + return; + } + + $schema = Schema::connection($this->connection()); + + if (! $schema->hasTable('product_purchase_price_sync_states')) { + $schema->create('product_purchase_price_sync_states', function (Blueprint $table): void { + $table->string('dataset_key', 120)->primary(); + $table->string('source_key', 64); + $table->string('source_object', 191); + $table->string('status', 24)->default('pending'); + $table->string('run_mode', 24)->nullable(); + $table->uuid('generation')->nullable(); + $table->dateTime('window_started_at', 6)->nullable(); + $table->dateTime('window_ends_at', 6)->nullable(); + $table->string('cursor_updated_at', 33)->nullable(); + $table->unsignedBigInteger('cursor_item_id')->nullable(); + $table->dateTime('initial_completed_at')->nullable(); + $table->dateTime('last_full_succeeded_at')->nullable(); + $table->dateTime('last_attempted_at')->nullable(); + $table->dateTime('last_succeeded_at')->nullable(); + $table->unsignedBigInteger('last_upserted_rows')->default(0); + $table->unsignedBigInteger('mirror_row_count')->default(0); + $table->string('last_error_code', 120)->nullable(); + $table->timestamps(); + }); + } + + if (! $schema->hasTable('product_purchase_price_mirror_rows')) { + $schema->create('product_purchase_price_mirror_rows', function (Blueprint $table): void { + $table->unsignedBigInteger('item_id')->primary(); + $table->unsignedBigInteger('purchase_order_id')->index(); + $table->string('sku_id', 255)->nullable()->index(); + $table->string('style_id', 255)->nullable()->index(); + $table->decimal('unit_price', 20, 4)->nullable(); + $table->string('document_status', 50)->nullable()->index(); + $table->text('labels')->nullable(); + $table->dateTime('finished_at', 6)->nullable()->index(); + $table->dateTime('purchase_updated_at', 6)->index(); + $table->uuid('_mirror_generation')->nullable()->index(); + $table->char('row_hash', 64); + $table->dateTime('_mirror_synced_at', 6)->index(); + $table->timestamps(); + + $table->index( + ['document_status', 'sku_id', 'finished_at'], + 'product_purchase_price_latest_lookup', + ); + }); + } + } + + public function down(): void + { + if (app()->environment('testing') + || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') { + return; + } + + $schema = Schema::connection($this->connection()); + $schema->dropIfExists('product_purchase_price_mirror_rows'); + $schema->dropIfExists('product_purchase_price_sync_states'); + } +}; diff --git a/deploy/activate-incremental-release.sh b/deploy/activate-incremental-release.sh index fdb7cd81..3331e4e6 100644 --- a/deploy/activate-incremental-release.sh +++ b/deploy/activate-incremental-release.sh @@ -129,7 +129,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha" # 账单提醒等既有任务。同一把锁保证快照与业务镜像不会并发施压上游。 # 普通商品优先高频增量，并在配置的每周低峰窗口自动继续全量校准。 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF' -* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1 +* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1 EOF chmod 644 /etc/cron.d/caiwu-jiansun-snapshots diff --git a/deploy/activate-release.sh b/deploy/activate-release.sh index 6c51a60f..9dec209a 100644 --- a/deploy/activate-release.sh +++ b/deploy/activate-release.sh @@ -133,7 +133,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha" # 账单提醒等既有任务。同一把锁保证快照与业务镜像不会并发施压上游。 # 普通商品优先高频增量，并在配置的每周低峰窗口自动继续全量校准。 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF' -* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1 +* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1 EOF chmod 644 /etc/cron.d/caiwu-jiansun-snapshots diff --git a/docs/上游业务镜像清单.md b/docs/上游业务镜像清单.md index 9bfdc4f4..568f536f 100644 --- a/docs/上游业务镜像清单.md +++ b/docs/上游业务镜像清单.md @@ -12,6 +12,7 @@ | --- | --- | | 已完成 | `operations_replica/rpt.Bulk_Price_Limit` 已进入类型化表 `product_price_mirror_rows`，商品价格页面已读取镜像并支持完整分页。 | | 已完成 | `operations_replica/rpt.jst_sku` 已进入 `ops_jst_sku`，普通商品资料页面固定读取 `sku_type = normal` 的共享镜像，并支持稳定分页和业务筛选。 | +| 本次接入 | `jiansun/dbo.jst_purchase + dbo.jst_purchaseItem` 通过固定联表查询进入 `product_purchase_price_mirror_rows`；普通商品资料只读取 `Finished` 且带“1688一键代发”标签的最近采购单价，并与原商品采购价分别标注来源。 | | 本次建立 | `config/upstream_mirrors.php` 中登记全部当前代码引用的原表；通用表级同步框架负责建表、稳定游标、断点续传、增量重叠和全量校准。 | | 已开始逐域切换 | 公司、店铺、商标页面在对应镜像完成首次全量后自动改读完整镜像表；首次全量前保留旧快照作为兼容回滚。其他业务域仍需逐项切换，不得宣称全项目已经完成。 | @@ -39,6 +40,7 @@ | 板材 | `dbo.finance_boardDeduction`、`dbo.pm_plainPlateData`、`dbo.pm_boardPurchasePrice`、`dbo.pm_boardPriceChangeImport`、`dbo.pm_boardApplyDetail` | | 封边条 | `dbo.pm_edgeBandData`、`dbo.pm_edgeBandApplyDetail`、`dbo.pm_edgeBandProfitInfo`、`dbo.finance_edgeBandDeductio`、`dbo.pm_averageUnitPrice` | | 库存与货款对账 | `dbo.pm_boardInventoryDetail_snapshot`、`dbo.fys_supplierGoods_sttlement_purchase`、`dbo.fys_supplierGoods_sttlement_sale`、`dbo.pm_purchaseIn` | +| 采购单价 | `dbo.jst_purchase`、`dbo.jst_purchaseItem`（受控字段联表镜像，不复制其他采购业务字段） | 其中 `fys_supplierGoods_sttlement_sale` 当前约 280 万行，是首次全量同步的最大对象。必须在低峰期小批量、可断点执行，禁止把全量同步塞进发布流程。 @@ -50,6 +52,7 @@ 4. 普通商品在最近完整校准超过 168 小时后，于周日 02:00-08:00 低峰窗口启动全量校准；一旦启动，后续每分钟从稳定游标继续，直至完整结束。首次全量仍必须显式授权，不能由定时任务自行启动。 5. 除专用数据集外，通用任务每轮默认只处理 1 个数据集、每个数据集 2 批，每批 500 行，避免同时压迫两个旧系统。 6. 页面切换必须以“总记录数、第一页、第二页、末页、筛选、汇总、空数据和更新时间”作为验收，不能只看首页有数据。 +7. 普通商品的采购单价同步以采购主表 `updateTime` 加明细 `Id` 作为稳定游标；页面按商品编码选择最近完成的 1688 一键代发价格。同步保存全部状态，使完成单后续取消时也能撤回旧价格；首次全量前只显示普通商品自身采购价。 ## 后续切换顺序 diff --git a/modules/EcommerceFinance/Repositories/ProductCatalogMirrorRepository.php b/modules/EcommerceFinance/Repositories/ProductCatalogMirrorRepository.php index 192e3733..9ccfbfa7 100644 --- a/modules/EcommerceFinance/Repositories/ProductCatalogMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/ProductCatalogMirrorRepository.php @@ -15,6 +15,10 @@ class ProductCatalogMirrorRepository { public const DATASET_KEY = 'operations_replica.rpt.jst_sku'; + public const PRODUCT_SOURCE_LABEL = '商品及库存管理(普通商品资料)'; + + public const PURCHASE_SOURCE_LABEL = '采购单管理-1688代发单价'; + private readonly ConnectionInterface $connection; public function __construct( @@ -78,7 +82,7 @@ class ProductCatalogMirrorRepository return ['code' => $code, 'lag_minutes' => $lagMinutes]; } - /** @return array{data:list,total:int} */ + /** @return array{data:list,total:int,purchase_price_sync_status:array{code:string,initialized:bool,last_succeeded_at:?string}} */ public function paginate(array $filters, int $page, int $perPage): array { if (! $this->ready()) { @@ -151,7 +155,109 @@ class ProductCatalogMirrorRepository ->get() ->all(); - return ['data' => $rows, 'total' => $total]; + $purchasePriceSyncStatus = $this->purchasePriceSyncStatus(); + $this->attachPurchasePrices($rows, $purchasePriceSyncStatus['initialized']); + + return [ + 'data' => $rows, + 'total' => $total, + 'purchase_price_sync_status' => $purchasePriceSyncStatus, + ]; + } + + /** @param list $rows */ + private function attachPurchasePrices(array $rows, bool $initialized): void + { + foreach ($rows as $row) { + $row->product_purchase_price_source = self::PRODUCT_SOURCE_LABEL; + $row->purchase_order_unit_price = null; + $row->purchase_order_finished_at = null; + $row->purchase_order_id = null; + $row->purchase_order_price_source = self::PURCHASE_SOURCE_LABEL; + } + + if ($rows === []) { + return; + } + + if (! $initialized) { + return; + } + + $skuIds = collect($rows) + ->pluck('sku_id') + ->filter(static fn (mixed $value): bool => is_string($value) && trim($value) !== '') + ->map(static fn (string $value): string => trim($value)) + ->unique() + ->values() + ->all(); + if ($skuIds === []) { + return; + } + + $candidates = $this->connection->table('product_purchase_price_mirror_rows') + ->where('document_status', 'Finished') + ->where('labels', 'like', '%1688一键代发%') + ->whereNotNull('unit_price') + ->whereIn('sku_id', $skuIds) + ->orderBy('sku_id') + ->orderByDesc('finished_at') + ->orderByDesc('purchase_order_id') + ->orderByDesc('item_id') + ->get([ + 'sku_id', + 'unit_price', + 'finished_at', + 'purchase_order_id', + ]) + ->unique('sku_id') + ->keyBy('sku_id'); + + foreach ($rows as $row) { + $purchase = $candidates->get(trim((string) ($row->sku_id ?? ''))); + if ($purchase === null) { + continue; + } + + $row->purchase_order_unit_price = $purchase->unit_price; + $row->purchase_order_finished_at = $purchase->finished_at; + $row->purchase_order_id = $purchase->purchase_order_id; + } + } + + /** @return array{code:string,initialized:bool,last_succeeded_at:?string} */ + private function purchasePriceSyncStatus(): array + { + $schema = Schema::connection($this->connection->getName()); + if (! $schema->hasTable('product_purchase_price_sync_states') + || ! $schema->hasTable('product_purchase_price_mirror_rows')) { + return ['code' => 'unavailable', 'initialized' => false, 'last_succeeded_at' => null]; + } + + $state = $this->connection->table('product_purchase_price_sync_states') + ->where('dataset_key', ProductPurchasePriceMirrorRepository::DATASET_KEY) + ->first(); + if ($state === null || $state->initial_completed_at === null) { + $code = match ((string) ($state->status ?? 'pending')) { + 'running' => 'initializing', + 'failed' => 'initial_failed', + default => 'pending', + }; + + return [ + 'code' => $code, + 'initialized' => false, + 'last_succeeded_at' => null, + ]; + } + + return [ + 'code' => 'ready', + 'initialized' => true, + 'last_succeeded_at' => ($state->last_succeeded_at ?? null) === null + ? null + : (string) $state->last_succeeded_at, + ]; } private function dataset(): BusinessMirrorDataset diff --git a/modules/EcommerceFinance/Repositories/ProductPurchasePriceMirrorRepository.php b/modules/EcommerceFinance/Repositories/ProductPurchasePriceMirrorRepository.php new file mode 100644 index 00000000..1c5ff984 --- /dev/null +++ b/modules/EcommerceFinance/Repositories/ProductPurchasePriceMirrorRepository.php @@ -0,0 +1,214 @@ +connection = $database->connection( + (string) config('upstream_mirrors.connection', 'business_mirror') + ); + } + + public function available(): bool + { + $schema = Schema::connection($this->connection->getName()); + + return $schema->hasTable('product_purchase_price_sync_states') + && $schema->hasTable('product_purchase_price_mirror_rows'); + } + + public function initialized(): bool + { + $state = $this->state(); + + return is_array($state) && ($state['initial_completed_at'] ?? null) !== null; + } + + /** @return array|null */ + public function state(): ?array + { + if (! $this->available()) { + return null; + } + + $row = $this->connection->table('product_purchase_price_sync_states') + ->where('dataset_key', self::DATASET_KEY) + ->first(); + + return $row === null ? null : (array) $row; + } + + public function startRun( + string $mode, + string $generation, + Carbon $since, + Carbon $until, + bool $resume, + ): array { + $now = now(); + $values = [ + 'source_key' => (string) config('upstream_mirrors.product_purchase_prices.source_key', 'jiansun'), + 'source_object' => (string) config( + 'upstream_mirrors.product_purchase_prices.source_object', + 'dbo.jst_purchase + dbo.jst_purchaseItem', + ), + 'status' => 'running', + 'run_mode' => $mode, + 'generation' => $generation, + 'last_attempted_at' => $now, + 'last_error_code' => null, + 'updated_at' => $now, + ]; + + if (! $resume) { + $values += [ + 'window_started_at' => $since, + 'window_ends_at' => $until, + 'cursor_updated_at' => null, + 'cursor_item_id' => null, + 'last_upserted_rows' => 0, + ]; + } + + $query = $this->connection->table('product_purchase_price_sync_states') + ->where('dataset_key', self::DATASET_KEY); + + if ($query->exists()) { + $query->update($values); + } else { + $this->connection->table('product_purchase_price_sync_states')->insert( + ['dataset_key' => self::DATASET_KEY, 'created_at' => $now] + $values, + ); + } + + return $this->state() ?? []; + } + + /** @param list> $rows */ + public function storeBatch(array $rows, array $cursor, string $generation, int $cumulativeRows): void + { + $now = now(); + $mapped = array_map(function (array $row) use ($generation, $now): array { + $values = [ + 'item_id' => (int) $row['item_id'], + 'purchase_order_id' => (int) $row['purchase_order_id'], + 'sku_id' => $this->nullableString($row['sku_id'] ?? null), + 'style_id' => $this->nullableString($row['style_id'] ?? null), + 'unit_price' => $row['unit_price'] ?? null, + 'document_status' => $this->nullableString($row['document_status'] ?? null), + 'labels' => $this->nullableString($row['labels'] ?? null), + 'finished_at' => $row['finished_at'] ?? null, + 'purchase_updated_at' => $row['purchase_updated_at'], + ]; + + return $values + [ + '_mirror_generation' => $generation, + 'row_hash' => hash('sha256', json_encode( + $values, + JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION, + )), + '_mirror_synced_at' => $now, + 'created_at' => $now, + 'updated_at' => $now, + ]; + }, $rows); + + $this->connection->transaction(function () use ($mapped, $cursor, $cumulativeRows, $now): void { + $this->connection->table('product_purchase_price_mirror_rows')->upsert( + $mapped, + ['item_id'], + [ + 'purchase_order_id', 'sku_id', 'style_id', 'unit_price', + 'document_status', 'labels', 'finished_at', 'purchase_updated_at', + '_mirror_generation', 'row_hash', '_mirror_synced_at', 'updated_at', + ], + ); + + $this->connection->table('product_purchase_price_sync_states') + ->where('dataset_key', self::DATASET_KEY) + ->update([ + 'cursor_updated_at' => $cursor['updated_at'], + 'cursor_item_id' => $cursor['item_id'], + 'last_upserted_rows' => $cumulativeRows, + 'updated_at' => $now, + ]); + }); + } + + public function complete(string $mode, string $generation, Carbon $until, int $rows): void + { + $now = now(); + + $this->connection->transaction(function () use ($mode, $generation, $until, $rows, $now): void { + if ($mode === 'full') { + $existingRows = $this->connection->table('product_purchase_price_mirror_rows')->count(); + if ($rows === 0 && $existingRows > 0) { + throw new RuntimeException('product_purchase_price_unexpected_empty_source'); + } + + $this->connection->table('product_purchase_price_mirror_rows') + ->where(function ($query) use ($generation): void { + $query->whereNull('_mirror_generation') + ->orWhere('_mirror_generation', '!=', $generation); + }) + ->delete(); + } + + $values = [ + 'status' => 'succeeded', + 'cursor_updated_at' => null, + 'cursor_item_id' => null, + 'last_succeeded_at' => $until, + 'last_upserted_rows' => $rows, + 'mirror_row_count' => $this->connection + ->table('product_purchase_price_mirror_rows') + ->count(), + 'last_error_code' => null, + 'updated_at' => $now, + ]; + + if ($mode === 'full') { + $values['initial_completed_at'] = $now; + $values['last_full_succeeded_at'] = $now; + } + + $this->connection->table('product_purchase_price_sync_states') + ->where('dataset_key', self::DATASET_KEY) + ->update($values); + }); + } + + public function fail(string $errorCode): void + { + $this->connection->table('product_purchase_price_sync_states') + ->where('dataset_key', self::DATASET_KEY) + ->update([ + 'status' => 'failed', + 'last_error_code' => mb_substr($errorCode, 0, 120), + 'updated_at' => now(), + ]); + } + + private function nullableString(mixed $value): ?string + { + if ($value === null) { + return null; + } + + $value = trim((string) $value); + + return $value === '' ? null : $value; + } +} diff --git a/modules/EcommerceFinance/Repositories/ProductPurchasePriceRepository.php b/modules/EcommerceFinance/Repositories/ProductPurchasePriceRepository.php new file mode 100644 index 00000000..713674d6 --- /dev/null +++ b/modules/EcommerceFinance/Repositories/ProductPurchasePriceRepository.php @@ -0,0 +1,48 @@ +> + */ + public function mirrorBatch(?array $cursor, string $since, string $until, int $limit): array + { + $limit = min(max($limit, 1), 5000); + $where = [ + 'p.[updateTime] >= ?', + 'p.[updateTime] < ?', + ]; + $bindings = [$since, $until]; + + if ($cursor !== null) { + $where[] = '(p.[updateTime] > ? OR (p.[updateTime] = ? AND i.[Id] > ?))'; + $bindings[] = $cursor['updated_at']; + $bindings[] = $cursor['updated_at']; + $bindings[] = $cursor['item_id']; + } + + $sql = 'SELECT TOP (' . $limit . ') ' + . 'i.[Id] AS item_id, p.[po_id] AS purchase_order_id, ' + . 'i.[sku_id] AS sku_id, i.[i_id] AS style_id, i.[price] AS unit_price, ' + . 'p.[status] AS document_status, p.[labels] AS labels, ' + . 'CONVERT(varchar(27), p.[finish_time], 126) AS finished_at, ' + . 'CONVERT(varchar(27), p.[updateTime], 126) AS purchase_updated_at ' + . 'FROM [dbo].[jst_purchase] p ' + . 'INNER JOIN [dbo].[jst_purchaseItem] i ON i.[po_id] = p.[po_id] ' + . 'WHERE (' . implode(') AND (', $where) . ') ' + . 'ORDER BY p.[updateTime], i.[Id]'; + + return $this->queries->forSource(self::SOURCE)->select($sql, $bindings); + } +} diff --git a/modules/EcommerceFinance/Services/ProductPurchasePriceMirrorSynchronizer.php b/modules/EcommerceFinance/Services/ProductPurchasePriceMirrorSynchronizer.php new file mode 100644 index 00000000..0690d2aa --- /dev/null +++ b/modules/EcommerceFinance/Services/ProductPurchasePriceMirrorSynchronizer.php @@ -0,0 +1,111 @@ +mirror->available()) { + throw new RuntimeException('product_purchase_price_mirror_schema_missing'); + } + + $state = $this->mirror->state(); + if (! $full && ($state['initial_completed_at'] ?? null) === null) { + return ['status' => 'requires_full_sync', 'mode' => 'incremental', 'batches' => 0, 'rows' => 0]; + } + + $mode = $full ? 'full' : 'incremental'; + $resume = is_array($state) + && ($state['run_mode'] ?? null) === $mode + && in_array($state['status'] ?? null, ['running', 'failed'], true) + && ($state['window_started_at'] ?? null) !== null + && ($state['window_ends_at'] ?? null) !== null; + $generation = $resume && is_string($state['generation'] ?? null) + ? (string) $state['generation'] + : (string) Str::uuid(); + + if ($resume) { + $since = Carbon::parse((string) $state['window_started_at']); + $until = Carbon::parse((string) $state['window_ends_at']); + } elseif ($full) { + $since = Carbon::create(1900, 1, 1, 0, 0, 0, config('app.timezone')); + $until = now()->addSecond(); + } else { + $overlap = max((int) config('upstream_mirrors.product_purchase_prices.overlap_hours', 48), 1); + $since = Carbon::parse((string) $state['last_succeeded_at'])->subHours($overlap); + $until = now()->addSecond(); + } + + $state = $this->mirror->startRun($mode, $generation, $since, $until, $resume); + $cursor = ($state['cursor_updated_at'] ?? null) === null + ? null + : [ + 'updated_at' => (string) $state['cursor_updated_at'], + 'item_id' => (int) ($state['cursor_item_id'] ?? 0), + ]; + $batchSize = min(max( + $batchSize ?? (int) config('upstream_mirrors.product_purchase_prices.batch_size', 1000), + 1, + ), 5000); + $maxBatches = max( + $maxBatches ?? (int) config('upstream_mirrors.product_purchase_prices.max_batches_per_run', 5), + 0, + ); + $batches = 0; + $rows = (int) ($state['last_upserted_rows'] ?? 0); + + try { + while (true) { + $batch = $this->upstream->mirrorBatch( + $cursor, + $since->format('Y-m-d\TH:i:s.v'), + $until->format('Y-m-d\TH:i:s.v'), + $batchSize, + ); + + if ($batch === []) { + $this->mirror->complete($mode, $generation, $until, $rows); + + return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows]; + } + + $last = $batch[array_key_last($batch)]; + $cursor = [ + 'updated_at' => (string) $last['purchase_updated_at'], + 'item_id' => (int) $last['item_id'], + ]; + $rows += count($batch); + $batches++; + $this->mirror->storeBatch($batch, $cursor, $generation, $rows); + + if (count($batch) < $batchSize) { + $this->mirror->complete($mode, $generation, $until, $rows); + + return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows]; + } + + if ($maxBatches > 0 && $batches >= $maxBatches) { + return ['status' => 'running', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows]; + } + } + } catch (Throwable $exception) { + $this->mirror->fail($exception::class); + throw $exception; + } + } +} diff --git a/resources/views/business/pages/product-inventory/normal-products.blade.php b/resources/views/business/pages/product-inventory/normal-products.blade.php index 6cf30bc8..a53b16fb 100644 --- a/resources/views/business/pages/product-inventory/normal-products.blade.php +++ b/resources/views/business/pages/product-inventory/normal-products.blade.php @@ -4,6 +4,13 @@ $productCatalogError = $content['productCatalogError'] ?? null; $productCatalogState = $content['productCatalogState'] ?? null; $productCatalogSyncStatus = $content['productCatalogSyncStatus'] ?? ['code' => 'unavailable', 'lag_minutes' => null]; + $productCatalogProductPriceSource = $content['productCatalogProductPriceSource'] ?? '商品及库存管理(普通商品资料)'; + $productCatalogPurchasePriceSource = $content['productCatalogPurchasePriceSource'] ?? '采购单管理-1688代发单价'; + $productCatalogPurchasePriceSyncStatus = $content['productCatalogPurchasePriceSyncStatus'] ?? [ + 'code' => 'unavailable', + 'initialized' => false, + 'last_succeeded_at' => null, + ]; $hasFilters = collect($productCatalogFilters)->contains( static fn ($value) => $value !== null && $value !== '' ); @@ -193,6 +200,19 @@ font-size: 0.8125rem; } + .product-catalog-notice { + display: flex; + align-items: center; + gap: 8px; + margin-bottom: 12px; + padding: 10px 12px; + border: 1px solid rgba(147, 95, 7, 0.35); + border-radius: 6px; + background: rgba(147, 95, 7, 0.06); + color: var(--vip-warning, #935f07); + font-size: 0.8125rem; + } + .product-catalog-table-wrap { overflow-x: auto; border: 1px solid var(--vip-border, #d8d2c7); @@ -202,7 +222,7 @@ .product-catalog-table { width: 100%; - min-width: 1940px; + min-width: 2120px; border-collapse: collapse; font-size: 0.78125rem; } @@ -240,6 +260,19 @@ white-space: nowrap; } + .product-catalog-price-cell { + min-width: 168px; + } + + .product-catalog-price-source { + display: block; + margin-top: 3px; + color: var(--catalog-muted); + font-size: 0.6875rem; + line-height: 1.25; + white-space: nowrap; + } + .product-catalog-table .is-code { font-variant-numeric: tabular-nums; white-space: nowrap; @@ -454,6 +487,25 @@ @endif + @if (!($productCatalogPurchasePriceSyncStatus['initialized'] ?? false) && !$productCatalogError) +
+ + + @if (($productCatalogPurchasePriceSyncStatus['code'] ?? '') === 'initializing') + 1688 代发单价正在执行首次同步，完成前当前仅显示普通商品采购价。 + @elseif (($productCatalogPurchasePriceSyncStatus['code'] ?? '') === 'initial_failed') + 1688 代发单价首次同步失败，当前仅显示普通商品采购价。 + @else + 1688 代发单价尚未完成首次同步，当前仅显示普通商品采购价。 + @endif + +
+ @endif +
@@ -469,7 +521,8 @@ - + + @@ -499,9 +552,18 @@ - @foreach (['sale_price', 'cost_price', 'purchase_price', 'market_price'] as $priceField) + @foreach (['sale_price', 'cost_price'] as $priceField) @endforeach + + + @empty -
供应商 销售价 成本价采购价普通商品采购价1688 代发单价 市场价 商品状态 库存同步{{ $row->brand ?? '-' }} {{ $row->category ?? '-' }} {{ $row->supplier_name ?? '-' }}{{ isset($row->{$priceField}) && $row->{$priceField} !== null ? number_format((float) $row->{$priceField}, 2) : '-' }} + {{ isset($row->purchase_price) && $row->purchase_price !== null ? number_format((float) $row->purchase_price, 2) : '-' }} + {{ $row->product_purchase_price_source ?? $productCatalogProductPriceSource }} + + {{ isset($row->purchase_order_unit_price) && $row->purchase_order_unit_price !== null ? number_format((float) $row->purchase_order_unit_price, 2) : '-' }} + {{ $row->purchase_order_price_source ?? $productCatalogPurchasePriceSource }} + {{ isset($row->market_price) && $row->market_price !== null ? number_format((float) $row->market_price, 2) : '-' }} {{ $enabledLabel }} @@ -516,7 +578,7 @@
+ @if ($productCatalogError) 数据暂时不可用。 diff --git a/tests/Feature/Business/ProductCatalogPageTest.php b/tests/Feature/Business/ProductCatalogPageTest.php index ca747512..015d64d7 100644 --- a/tests/Feature/Business/ProductCatalogPageTest.php +++ b/tests/Feature/Business/ProductCatalogPageTest.php @@ -58,6 +58,11 @@ class ProductCatalogPageTest extends FeatureTestCase 'sale_price' => 899, 'cost_price' => 500, 'purchase_price' => 480, + 'product_purchase_price_source' => '商品及库存管理(普通商品资料)', + 'purchase_order_unit_price' => 65, + 'purchase_order_finished_at' => '2026-08-20 11:03:05', + 'purchase_order_id' => 827380, + 'purchase_order_price_source' => '采购单管理-1688代发单价', 'market_price' => 1099, 'brand' => '北木', 'category' => '住宅家具', @@ -67,6 +72,11 @@ class ProductCatalogPageTest extends FeatureTestCase 'modified' => '2026-08-15 10:00:00', ]], 'total' => 1, + 'purchase_price_sync_status' => [ + 'code' => 'ready', + 'initialized' => true, + 'last_succeeded_at' => '2026-08-21 12:00:00', + ], ]); $this->app->instance(ProductCatalogMirrorRepository::class, $repository); @@ -95,10 +105,55 @@ class ProductCatalogPageTest extends FeatureTestCase ->assertSeeText('SKU-001') ->assertSeeText('岩板餐桌') ->assertSeeText('899.00') + ->assertSeeText('普通商品采购价') + ->assertSeeText('1688 代发单价') + ->assertSeeText('商品及库存管理(普通商品资料)') + ->assertSeeText('采购单管理-1688代发单价') + ->assertSeeText('65.00') + ->assertDontSeeText('1688 代发单价尚未完成首次同步') ->assertSeeText('正常同步') ->assertDontSee('pic_url'); } + public function test_normal_products_page_marks_purchase_prices_as_not_yet_synchronized(): void + { + $this->grantAdminPermission(); + + $state = [ + 'status' => 'succeeded', + 'initial_completed_at' => '2026-08-14 21:15:36', + 'last_succeeded_at' => '2026-08-21 12:00:00', + 'mirror_row_count' => 547217, + ]; + $repository = \Mockery::mock(ProductCatalogMirrorRepository::class); + $repository->shouldReceive('available')->once()->andReturnTrue(); + $repository->shouldReceive('state')->once()->andReturn($state); + $repository->shouldReceive('syncStatus')->once()->with($state)->andReturn([ + 'code' => 'healthy', + 'lag_minutes' => 1, + ]); + $repository->shouldReceive('ready')->once()->andReturnTrue(); + $repository->shouldReceive('paginate')->once()->andReturn([ + 'data' => [], + 'total' => 0, + 'purchase_price_sync_status' => [ + 'code' => 'pending', + 'initialized' => false, + 'last_succeeded_at' => null, + ], + ]); + $this->app->instance(ProductCatalogMirrorRepository::class, $repository); + + $this->loginAs()->get(route('business.pages.show', [ + 'section' => 'product-inventory', + 'page' => 'normal-products', + ])) + ->assertOk() + ->assertSee('data-purchase-price-sync-state="pending"', false) + ->assertSeeText('1688 代发单价尚未完成首次同步') + ->assertSeeText('当前仅显示普通商品采购价'); + } + public function test_normal_products_page_does_not_treat_initial_sync_as_empty_data(): void { $this->grantAdminPermission(); diff --git a/tests/Feature/Console/ProductPurchasePriceMirrorSyncCommandTest.php b/tests/Feature/Console/ProductPurchasePriceMirrorSyncCommandTest.php new file mode 100644 index 00000000..3a6bbdb2 --- /dev/null +++ b/tests/Feature/Console/ProductPurchasePriceMirrorSyncCommandTest.php @@ -0,0 +1,86 @@ + true, + 'upstream_mirrors.product_purchase_prices' => [ + 'batch_size' => 1000, + 'max_batches_per_run' => 5, + 'full_reconcile_hours' => 168, + 'reconcile_weekday' => 0, + 'reconcile_start_hour' => 2, + 'reconcile_window_hours' => 6, + ], + ]); + } + + protected function tearDown(): void + { + Carbon::setTestNow(); + parent::tearDown(); + } + + public function test_it_runs_incremental_sync_after_initialization(): void + { + Carbon::setTestNow('2026-08-21 10:00:00'); + $mirror = Mockery::mock(ProductPurchasePriceMirrorRepository::class); + $mirror->shouldReceive('state')->once()->andReturn([ + 'status' => 'succeeded', + 'run_mode' => 'incremental', + 'initial_completed_at' => '2026-08-14 10:00:00', + 'last_full_succeeded_at' => '2026-08-17 02:00:00', + ]); + $synchronizer = Mockery::mock(ProductPurchasePriceMirrorSynchronizer::class); + $synchronizer->shouldReceive('sync') + ->once() + ->with(false, null, null) + ->andReturn([ + 'status' => 'succeeded', + 'mode' => 'incremental', + 'batches' => 1, + 'rows' => 12, + ]); + $this->app->instance(ProductPurchasePriceMirrorRepository::class, $mirror); + $this->app->instance(ProductPurchasePriceMirrorSynchronizer::class, $synchronizer); + + $this->artisan('business-mirror:sync-product-purchase-prices') + ->expectsOutputToContain('采购单价增量同步：完成') + ->assertSuccessful(); + } + + public function test_it_does_not_start_the_first_full_sync_automatically(): void + { + Carbon::setTestNow('2026-08-23 02:30:00'); + $mirror = Mockery::mock(ProductPurchasePriceMirrorRepository::class); + $mirror->shouldReceive('state')->once()->andReturn(null); + $synchronizer = Mockery::mock(ProductPurchasePriceMirrorSynchronizer::class); + $synchronizer->shouldReceive('sync') + ->once() + ->with(false, null, null) + ->andReturn([ + 'status' => 'requires_full_sync', + 'mode' => 'incremental', + 'batches' => 0, + 'rows' => 0, + ]); + $this->app->instance(ProductPurchasePriceMirrorRepository::class, $mirror); + $this->app->instance(ProductPurchasePriceMirrorSynchronizer::class, $synchronizer); + + $this->artisan('business-mirror:sync-product-purchase-prices') + ->expectsOutputToContain('需经授权后显式使用 --full') + ->assertSuccessful(); + } +} diff --git a/tests/Unit/Deployment/BusinessMirrorScheduleTest.php b/tests/Unit/Deployment/BusinessMirrorScheduleTest.php index f2867862..171aac4b 100644 --- a/tests/Unit/Deployment/BusinessMirrorScheduleTest.php +++ b/tests/Unit/Deployment/BusinessMirrorScheduleTest.php @@ -11,18 +11,23 @@ class BusinessMirrorScheduleTest extends TestCase foreach (['activate-release.sh', 'activate-incremental-release.sh'] as $script) { $contents = file_get_contents(dirname(__DIR__, 3) . '/deploy/' . $script); $catalogPosition = strpos($contents, 'business-mirror:sync-product-catalog --quiet'); + $purchasePricePosition = strpos($contents, 'business-mirror:sync-product-purchase-prices --quiet'); $snapshotPosition = strpos($contents, 'upstream:snapshots:refresh-all --quiet'); $pricePosition = strpos($contents, 'business-mirror:sync-product-prices --quiet'); $genericPosition = strpos($contents, 'business-mirror:sync-tables --quiet'); $this->assertNotFalse($catalogPosition, $script); + $this->assertNotFalse($purchasePricePosition, $script); $this->assertNotFalse($snapshotPosition, $script); $this->assertNotFalse($pricePosition, $script); $this->assertNotFalse($genericPosition, $script); $this->assertLessThan($snapshotPosition, $catalogPosition, $script); + $this->assertLessThan($snapshotPosition, $purchasePricePosition, $script); + $this->assertLessThan($purchasePricePosition, $catalogPosition, $script); $this->assertLessThan($pricePosition, $catalogPosition, $script); $this->assertLessThan($genericPosition, $catalogPosition, $script); $this->assertSame(1, substr_count($contents, 'business-mirror:sync-product-catalog --quiet'), $script); + $this->assertSame(1, substr_count($contents, 'business-mirror:sync-product-purchase-prices --quiet'), $script); $this->assertStringContainsString('docker exec -u www-data caiwu_app', $contents, $script); $this->assertStringContainsString('/run/caiwu-upstream-sync.lock', $contents, $script); } diff --git a/tests/Unit/EcommerceFinance/ProductCatalogMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/ProductCatalogMirrorRepositoryTest.php index b5b92801..7c501ddc 100644 --- a/tests/Unit/EcommerceFinance/ProductCatalogMirrorRepositoryTest.php +++ b/tests/Unit/EcommerceFinance/ProductCatalogMirrorRepositoryTest.php @@ -117,6 +117,61 @@ class ProductCatalogMirrorRepositoryTest extends TestCase $this->assertObjectNotHasProperty('pic_url', $result['data'][0]); } + public function test_it_attaches_the_latest_finished_1688_drop_shipping_price_without_overwriting_product_price(): void + { + $this->markReady(1); + $this->insertProduct(1, ['sku_id' => 'SKU-1688', 'purchase_price' => 55]); + $this->createPurchasePriceTables(true); + $this->insertPurchasePrice(10, [ + 'sku_id' => 'SKU-1688', + 'unit_price' => 60, + 'finished_at' => '2026-08-18 10:00:00', + ]); + $this->insertPurchasePrice(11, [ + 'sku_id' => 'SKU-1688', + 'unit_price' => 65, + 'finished_at' => '2026-08-20 11:03:05', + ]); + $this->insertPurchasePrice(12, [ + 'sku_id' => 'SKU-1688', + 'unit_price' => 99, + 'document_status' => 'Cancelled', + 'finished_at' => '2026-08-21 10:00:00', + ]); + $this->insertPurchasePrice(13, [ + 'sku_id' => 'SKU-1688', + 'unit_price' => 88, + 'labels' => '普通采购', + 'finished_at' => '2026-08-21 11:00:00', + ]); + + $result = $this->repository->paginate($this->filters(), 1, 25); + $row = $result['data'][0]; + + $this->assertSame('55', (string) $row->purchase_price); + $this->assertSame('65', (string) $row->purchase_order_unit_price); + $this->assertSame(1011, (int) $row->purchase_order_id); + $this->assertSame('采购单管理-1688代发单价', $row->purchase_order_price_source); + $this->assertSame('商品及库存管理(普通商品资料)', $row->product_purchase_price_source); + $this->assertTrue($result['purchase_price_sync_status']['initialized']); + } + + public function test_it_keeps_products_visible_when_purchase_price_mirror_is_not_initialized(): void + { + $this->markReady(1); + $this->insertProduct(1, ['sku_id' => 'SKU-PENDING']); + $this->createPurchasePriceTables(false); + + $result = $this->repository->paginate($this->filters(), 1, 25); + $row = $result['data'][0]; + + $this->assertSame('SKU-PENDING', $row->sku_id); + $this->assertNull($row->purchase_order_unit_price); + $this->assertSame('采购单管理-1688代发单价', $row->purchase_order_price_source); + $this->assertFalse($result['purchase_price_sync_status']['initialized']); + $this->assertSame('pending', $result['purchase_price_sync_status']['code']); + } + public function test_it_refuses_queries_before_the_initial_sync_completes(): void { DB::connection('business_mirror')->table('business_mirror_table_states')->insert([ @@ -228,6 +283,40 @@ class ProductCatalogMirrorRepositoryTest extends TestCase }); } + private function createPurchasePriceTables(bool $initialized): void + { + Schema::connection('business_mirror')->create('product_purchase_price_sync_states', function (Blueprint $table): void { + $table->string('dataset_key')->primary(); + $table->dateTime('initial_completed_at')->nullable(); + }); + Schema::connection('business_mirror')->create('product_purchase_price_mirror_rows', function (Blueprint $table): void { + $table->unsignedBigInteger('item_id')->primary(); + $table->unsignedBigInteger('purchase_order_id'); + $table->string('sku_id')->nullable(); + $table->decimal('unit_price')->nullable(); + $table->string('document_status')->nullable(); + $table->text('labels')->nullable(); + $table->dateTime('finished_at')->nullable(); + }); + DB::connection('business_mirror')->table('product_purchase_price_sync_states')->insert([ + 'dataset_key' => 'product-inventory.purchase-order-prices', + 'initial_completed_at' => $initialized ? '2026-08-21 12:00:00' : null, + ]); + } + + private function insertPurchasePrice(int $itemId, array $overrides = []): void + { + DB::connection('business_mirror')->table('product_purchase_price_mirror_rows')->insert(array_merge([ + 'item_id' => $itemId, + 'purchase_order_id' => 1000 + $itemId, + 'sku_id' => 'SKU-' . $itemId, + 'unit_price' => 50, + 'document_status' => 'Finished', + 'labels' => '1688密文下单,1688一键代发,1688费用更新', + 'finished_at' => '2026-08-20 10:00:00', + ], $overrides)); + } + private function markReady(int $rows): void { DB::connection('business_mirror')->table('business_mirror_table_states')->insert([ diff --git a/tests/Unit/EcommerceFinance/ProductPurchasePriceMirrorSynchronizerTest.php b/tests/Unit/EcommerceFinance/ProductPurchasePriceMirrorSynchronizerTest.php new file mode 100644 index 00000000..f6517004 --- /dev/null +++ b/tests/Unit/EcommerceFinance/ProductPurchasePriceMirrorSynchronizerTest.php @@ -0,0 +1,80 @@ +createMock(ProductPurchasePriceRepository::class); + $upstream->expects($this->never())->method('mirrorBatch'); + $mirror = $this->createMock(ProductPurchasePriceMirrorRepository::class); + $mirror->method('available')->willReturn(true); + $mirror->method('state')->willReturn(null); + + $result = (new ProductPurchasePriceMirrorSynchronizer($upstream, $mirror))->sync(); + + $this->assertSame('requires_full_sync', $result['status']); + } + + public function test_full_sync_uses_stable_update_time_and_item_id_cursor(): void + { + Carbon::setTestNow('2026-08-21 12:00:00'); + $rows = [[ + 'item_id' => 100, + 'purchase_order_id' => 827380, + 'sku_id' => 'DF-A1DNZ001203', + 'purchase_updated_at' => '2026-08-20T11:03:32.000', + ], [ + 'item_id' => 101, + 'purchase_order_id' => 827381, + 'sku_id' => 'DF-G1DNZ002401', + 'purchase_updated_at' => '2026-08-20T11:03:32.000', + ]]; + $upstream = $this->createMock(ProductPurchasePriceRepository::class); + $upstream->expects($this->exactly(2))->method('mirrorBatch') + ->willReturnOnConsecutiveCalls($rows, []); + $mirror = $this->createMock(ProductPurchasePriceMirrorRepository::class); + $mirror->method('available')->willReturn(true); + $mirror->method('state')->willReturn(null); + $mirror->expects($this->once())->method('startRun') + ->with( + 'full', + $this->isType('string'), + $this->isInstanceOf(Carbon::class), + $this->isInstanceOf(Carbon::class), + false, + ) + ->willReturn(['last_upserted_rows' => 0]); + $mirror->expects($this->once())->method('storeBatch') + ->with( + $rows, + ['updated_at' => '2026-08-20T11:03:32.000', 'item_id' => 101], + $this->isType('string'), + 2, + ); + $mirror->expects($this->once())->method('complete') + ->with('full', $this->isType('string'), $this->isInstanceOf(Carbon::class), 2); + + $result = (new ProductPurchasePriceMirrorSynchronizer($upstream, $mirror))->sync(true, 2, 0); + + $this->assertSame([ + 'status' => 'succeeded', + 'mode' => 'full', + 'batches' => 1, + 'rows' => 2, + ], $result); + } +} diff --git a/tests/Unit/EcommerceFinance/ProductPurchasePriceRepositoryTest.php b/tests/Unit/EcommerceFinance/ProductPurchasePriceRepositoryTest.php new file mode 100644 index 00000000..5d73d4b9 --- /dev/null +++ b/tests/Unit/EcommerceFinance/ProductPurchasePriceRepositoryTest.php @@ -0,0 +1,53 @@ +sql = $sql; + $this->bindings = $bindings; + + return []; + } + }; + $factory = new class($query) extends UpstreamReadOnlyQueryFactory { + public function __construct(private readonly UpstreamReadOnlyQuery $query) {} + public function forSource(string $source): UpstreamReadOnlyQuery + { + TestCase::assertSame('jiansun', $source); + + return $this->query; + } + }; + + (new ProductPurchasePriceRepository($factory))->mirrorBatch( + ['updated_at' => '2026-08-20T10:00:00.000', 'item_id' => 100], + '2026-08-19T00:00:00.000', + '2026-08-21T00:00:00.000', + 1000, + ); + + $this->assertStringContainsString('SELECT TOP (1000)', $query->sql); + $this->assertStringContainsString('[dbo].[jst_purchase] p', $query->sql); + $this->assertStringContainsString('[dbo].[jst_purchaseItem] i', $query->sql); + $this->assertStringContainsString('i.[po_id] = p.[po_id]', $query->sql); + $this->assertStringContainsString('p.[updateTime] >= ?', $query->sql); + $this->assertStringContainsString('p.[updateTime] = ? AND i.[Id] > ?', $query->sql); + $this->assertStringContainsString('ORDER BY p.[updateTime], i.[Id]', $query->sql); + $this->assertStringNotContainsString("p.[status] = 'Finished'", $query->sql); + $this->assertSame(100, $query->bindings[4]); + } +}