From 89031dfeb4838ab5559af06f75f3621cf164461c Mon Sep 17 00:00:00 2001 From: yueyang <2495544247@qq.com> Date: Fri, 11 Sep 2026 17:56:27 +0800 Subject: [PATCH 1/2] chore: publish pinduoduo receiving bill task scope -- 2.47.2 From 813e215e1f08820ed0928776040ac4b65561fed0 Mon Sep 17 00:00:00 2001 From: yueyang <2495544247@qq.com> Date: Sat, 12 Sep 2026 11:06:38 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20=E6=8E=A5=E5=85=A5=E6=8B=BC?= =?UTF-8?q?=E5=A4=9A=E5=A4=9A=E8=B4=A7=E6=AC=BE=E8=B4=A6=E6=88=B7=E8=B4=A6?= =?UTF-8?q?=E5=8D=95=E9=95=9C=E5=83=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/business_data.php | 3 + config/upstream_mirrors.php | 26 ++- docs/上游业务镜像清单.md | 3 + .../Contracts/BusinessMirrorDataset.php | 3 + .../Http/Controllers/AccountBills.php | 56 +++-- ...oreReceivingAlipayBillMirrorRepository.php | 95 +++++++-- ...ReceivingPinduoduoBillMirrorRepository.php | 49 +++++ .../BusinessMirrorDatasetRegistry.php | 41 +++- .../Services/BusinessMirrorSchemaManager.php | 3 +- .../Services/BusinessMirrorSynchronizer.php | 5 + .../js/components/business/AccountBills.vue | 7 +- .../Business/PinduoduoReceivingBillsTest.php | 142 +++++++++++++ .../BusinessMirrorSynchronizerTest.php | 55 +++++ .../ColumnarBusinessMirrorRepositoryTest.php | 1 + .../GenericUpstreamPipelineTest.php | 2 + ...ivingPinduoduoBillMirrorRepositoryTest.php | 198 ++++++++++++++++++ tests/Unit/View/AccountBillsTableTest.php | 24 +++ 17 files changed, 668 insertions(+), 45 deletions(-) create mode 100644 modules/EcommerceFinance/Repositories/StoreReceivingPinduoduoBillMirrorRepository.php create mode 100644 tests/Feature/Business/PinduoduoReceivingBillsTest.php create mode 100644 tests/Unit/EcommerceFinance/StoreReceivingPinduoduoBillMirrorRepositoryTest.php create mode 100644 tests/Unit/View/AccountBillsTableTest.php diff --git a/config/business_data.php b/config/business_data.php index 5548c0ea..8a73151d 100644 --- a/config/business_data.php +++ b/config/business_data.php @@ -94,6 +94,9 @@ return [ 'jiansun.dbo.jst_small_payment_order_map' => ['role' => 'operational', 'target_store' => 'mariadb', 'consumer' => '小额打款工单'], // Account bills and supplier pages are bounded serving datasets today. + 'jiansun.dbo.jst_originalBillInfo.pinduoduo_receiving' => [ + 'role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '店铺收款账单/拼多多账户', + ], 'jiansun.dbo.alipay_Bill_Detail' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '店铺收款账单/淘系支付宝账户'], 'jiansun.dbo.tmall_spread_bills' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '淘宝推广账单'], 'jiansun.dbo.pdd_spread_bills' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '拼多多推广账单'], diff --git a/config/upstream_mirrors.php b/config/upstream_mirrors.php index b976aa20..a477c8eb 100644 --- a/config/upstream_mirrors.php +++ b/config/upstream_mirrors.php @@ -370,7 +370,31 @@ return [ ], ], - // JianSun 旧业务库：营销账单、供应商应付、板材、封边条和库存。 + // JianSun 旧业务库：拼多多货款、营销账单、供应商应付、板材、封边条和库存。 + // 拼多多货款账单只复制平台原始账单中的固定渠道切片，避免为该页面 + // 重复同步上游 1400 万行原表，也不依赖慢一天的经营分析副本。 + 'jiansun.dbo.jst_originalBillInfo.pinduoduo_receiving' => [ + 'source_key' => 'jiansun', + 'schema' => 'dbo', + 'object' => 'jst_originalBillInfo', + 'target' => 'js_jst_pdd_receiving_bill_rows', + 'keys' => ['Id'], + 'watermark' => 'updateTime', + 'fixed_filters' => [ + 'channel' => 'PinduoduoImport', + ], + 'columns' => [ + 'Id', 'shop_id', 'channel', 'fee_time', 'del_flag', 'primary_key', + 'createTime', 'updateTime', 'auto_id', 'fee_name', 'inout_type', + 'oc_amount', 'p_auto_id', 'shop_i_id', 'shop_sku_id', 'sku_id', + 'so_id', 'combine_type', 'amount', + ], + 'indexes' => [ + ['Id'], + ['fee_time', 'Id', 'inout_type', 'amount'], + ['shop_id', 'fee_time', 'Id', 'inout_type', 'amount', 'fee_name'], + ], + ], 'jiansun.dbo.alipay_Bill_Detail' => [ 'source_key' => 'jiansun', 'schema' => 'dbo', diff --git a/docs/上游业务镜像清单.md b/docs/上游业务镜像清单.md index 1a6fb885..b95baa54 100644 --- a/docs/上游业务镜像清单.md +++ b/docs/上游业务镜像清单.md @@ -22,6 +22,7 @@ | 中央同步接入 | 店铺维度报表从 `jiansun.dbo.jst_originalBillInfo` 和 `dbo.fys_shopInfo_flow` 自动、可断点同步；不再依赖开发者电脑、SQLite 或人工 XLSX。 | | 已完成 | `operations_replica/rpt.jst_originalBillInfo` 已进入通用原始账单镜像 `ops_jst_original_bill_info`；保留当前 19 个业务字段，按 `Id` 断点全量、按 `updateTime` 重叠增量。 | | 页面已切换 | 店铺收款账单的淘系支付宝账户读取 `operations_replica.rpt.jst_originalBillInfo`，并关联 `rpt.jst_shopInfo` 展示店铺；分页、筛选、汇总和导出统一读取共享业务镜像。 | +| 已接入待首次同步 | 店铺收款账单的拼多多账户读取 `jiansun.dbo.jst_originalBillInfo` 中固定 `channel = PinduoduoImport` 的独立切片镜像 `js_jst_pdd_receiving_bill_rows`；不使用慢一天的 `operations_replica` 副本或营销账户 `pdd_spread_bills`。 | | 本次接入 | 普通商品资料的工作平台采购价格接入 `jiansun.dbo.pm_priceAdjustment`（镜像 `js_pm_price_adjustment`）：按 `skuCode` 关联 `jst_sku.sku_id`，取 `effectTime` 不晚于当前时间的最新 `purchasePrice`，并在页面和导出中标记“工作平台价格”；源表没有可靠更新时间，按 `Id` 稳定键追加，首次全量完成前只显示普通商品自身采购价。 | ## 经营分析副本库（operations_replica） @@ -65,6 +66,7 @@ | 工作平台采购价格 | `dbo.pm_priceAdjustment`（`Id` 稳定键；按 `skuCode` 匹配 `jst_sku.sku_id`，页面取当前已生效的最新 `purchasePrice` / `effectTime`） | | 快递资料 | `dbo.wp_express`、`dbo.wp_express_expNet`、`dbo.wp_express_account`、`dbo.wp_express_extension` | | 聚水潭小额打款工单 | `dbo.jst_workOrderInfo`（稳定键 `Id`、增量字段 `updateTime`）与 `dbo.jst_small_payment_order_map`（稳定键 `woNo`、增量字段 `updated_at`） | +| 拼多多货款账户账单 | `dbo.jst_originalBillInfo`（固定 `channel = PinduoduoImport`，镜像 `js_jst_pdd_receiving_bill_rows`） | 其中 `fys_supplierGoods_sttlement_sale` 当前约 280 万行，是首次全量同步的最大对象。必须在低峰期小批量、可断点执行，禁止把全量同步塞进发布流程。 @@ -80,6 +82,7 @@ 8. 聚水潭官方接口尚未接入时，普通商品历史残留排除由负责人提供的最新完整 XLSX 导出刷新。只排除“导出生成前已创建、但导出中缺失”的镜像行，导出生成后的新增商品保持可见。导入使用新同步代次，只有完整解析成功才原子替换排除表；中断或错误保留上一版。后续接入聚水潭官方“普通商品资料查询”及删除/恢复记录后，由官方接口取代人工导出校准。 9. 店铺维度报表是已批准的中央专用数据集：生产节点按 `Id` 断点推进并回看最近 50000 行；原始账单同步后增量维护按店铺/日期汇总的 `store_report_daily_rows`，页面不得现场扫描千万级原始账单。开发机不运行刷新 worker。 10. `operations_replica.rpt.jst_originalBillInfo` 是供多个业务功能复用的原始账单镜像。首次同步由 Codex 备份后自主安排在生产同步节点低峰期分批续传，无需另行授权；页面请求和普通开发机不启动刷新 worker。同步完成后，开发者通过固定 Repository 查询 `ops_jst_original_bill_info`，不得各自创建整表快照。 淘系支付宝账户账单当前仍读取此 MariaDB 镜像，统一调度须持续增量，不因 ClickHouse 尚未启用而暂停。账单冲减则复用“订单数据 → 原始数据 → 订单售后费用明细”，不按该表费用名称推断 500112。 +11. 拼多多货款账户账单使用 `jiansun.dbo.jst_originalBillInfo.pinduoduo_receiving` 数据集。同步器必须以参数绑定固定 `channel = PinduoduoImport`，只复制该渠道切片；页面 Repository 只读 `js_jst_pdd_receiving_bill_rows`。首次全量按 `Id` 断点推进，后续按 `updateTime` 保留重叠窗口，不得切回落后一天的 `operations_replica` 副本。 工作平台采购价格首次全量需在备份和负责人授权后单独执行： diff --git a/modules/EcommerceFinance/Contracts/BusinessMirrorDataset.php b/modules/EcommerceFinance/Contracts/BusinessMirrorDataset.php index 750c0bdd..77ca082c 100644 --- a/modules/EcommerceFinance/Contracts/BusinessMirrorDataset.php +++ b/modules/EcommerceFinance/Contracts/BusinessMirrorDataset.php @@ -19,6 +19,9 @@ interface BusinessMirrorDataset public function watermarkColumn(): ?string; + /** @return array */ + public function fixedFilters(): array; + /** @return list> */ public function indexDefinitions(): array; } diff --git a/modules/EcommerceFinance/Http/Controllers/AccountBills.php b/modules/EcommerceFinance/Http/Controllers/AccountBills.php index 4fe2a186..d7067d5f 100644 --- a/modules/EcommerceFinance/Http/Controllers/AccountBills.php +++ b/modules/EcommerceFinance/Http/Controllers/AccountBills.php @@ -8,6 +8,7 @@ use Illuminate\Routing\Controller; use Modules\EcommerceFinance\AccountBillColumns; use Modules\EcommerceFinance\Models\AccountBill; use Modules\EcommerceFinance\Repositories\StoreReceivingDouyinBillMirrorRepository; +use Modules\EcommerceFinance\Repositories\StoreReceivingPinduoduoBillMirrorRepository; use RuntimeException; class AccountBills extends Controller @@ -27,10 +28,10 @@ class AccountBills extends Controller abort(404, '未知的账户类型。'); } - if ($billType === 'douyin') { + if (in_array($billType, ['douyin', 'pinduoduo'], true)) { return $request->boolean('export') ? $this->export($request, $billType) - : $this->douyinIndex($request); + : $this->mirrorIndex($request, $billType); } $request->validate([ @@ -116,8 +117,8 @@ class AccountBills extends Controller abort_unless(AccountBillColumns::has($bill_type), 404, '未知的账户类型。'); - if ($bill_type === 'douyin') { - return $this->exportDouyin($request); + if (in_array($bill_type, ['douyin', 'pinduoduo'], true)) { + return $this->exportMirror($request, $bill_type); } $filters = $request->validate([ @@ -140,11 +141,13 @@ class AccountBills extends Controller { $this->authorizeStoreReceivingBills(); - abort_unless($bill_type === 'douyin', 404, '该账户类型没有可用的筛选选项。'); + abort_unless(in_array($bill_type, ['douyin', 'pinduoduo'], true), 404, '该账户类型没有可用的筛选选项。'); + + $label = $this->mirrorLabel($bill_type); try { return response()->json( - app(StoreReceivingDouyinBillMirrorRepository::class)->filterOptions() + $this->mirrorRepository($bill_type)->filterOptions() ); } catch (RuntimeException $exception) { if ($exception->getMessage() !== 'store_receiving_alipay_bill_mirror_not_ready') { @@ -152,7 +155,7 @@ class AccountBills extends Controller } return response()->json([ - 'message' => '抖音账户账单共享业务镜像正在首次同步，筛选选项稍后可用。', + 'message' => $label . '账户账单共享业务镜像正在首次同步，筛选选项稍后可用。', 'shops' => [], 'bill_items' => [], ], 503); @@ -249,16 +252,17 @@ class AccountBills extends Controller return $query; } - private function douyinIndex(Request $request) + private function mirrorIndex(Request $request, string $billType) { - $filters = $this->douyinFilters($request); + $filters = $this->mirrorFilters($request); $sort = $this->sort($request); $direction = $this->direction($request); $page = max((int) $request->get('page', 1), 1); $perPage = min(max((int) $request->get('limit', 10), 1), 100); + $label = $this->mirrorLabel($billType); try { - $data = app(StoreReceivingDouyinBillMirrorRepository::class)->paginate( + $data = $this->mirrorRepository($billType)->paginate( $filters, $sort, $direction, @@ -271,26 +275,26 @@ class AccountBills extends Controller } return response()->json([ - 'message' => '抖音账户账单共享业务镜像正在首次同步，完成后将自动显示。', + 'message' => $label . '账户账单共享业务镜像正在首次同步，完成后将自动显示。', 'data' => [], 'total' => 0, 'current_page' => 1, 'last_page' => 1, 'per_page' => $perPage, 'summary' => ['income' => 0, 'expense' => 0, 'balance' => 0, 'count' => 0], - 'columns' => AccountBillColumns::for('douyin'), + 'columns' => AccountBillColumns::for($billType), ], 503); } - $data['columns'] = AccountBillColumns::for('douyin'); + $data['columns'] = AccountBillColumns::for($billType); return response()->json($data); } - private function exportDouyin(Request $request) + private function exportMirror(Request $request, string $billType) { - $filters = $this->douyinFilters($request); - $definition = AccountBillColumns::for('douyin'); + $filters = $this->mirrorFilters($request); + $definition = AccountBillColumns::for($billType); $columns = collect($definition['columns'] ?? []); $requested = $request->query('columns', []); if (is_string($requested)) { @@ -314,14 +318,14 @@ class AccountBills extends Controller $sort = $this->sort($request); $direction = $this->direction($request); try { - $rows = app(StoreReceivingDouyinBillMirrorRepository::class) + $rows = $this->mirrorRepository($billType) ->exportRows($filters, $sort, $direction); } catch (RuntimeException $exception) { if ($exception->getMessage() !== 'store_receiving_alipay_bill_mirror_not_ready') { throw $exception; } - return response()->json(['message' => '抖音账户账单共享业务镜像正在首次同步，完成后将自动导出。'], 503); + return response()->json(['message' => $this->mirrorLabel($billType) . '账户账单共享业务镜像正在首次同步，完成后将自动导出。'], 503); } $headers = $selected->pluck('label')->all(); @@ -336,14 +340,14 @@ class AccountBills extends Controller )->all()); } fclose($csv); - }, 'douyin_account_bills.csv', [ + }, $billType . '_account_bills.csv', [ 'Content-Type' => 'text/csv; charset=UTF-8', 'Cache-Control' => 'private, no-store', ]); } /** @return array */ - private function douyinFilters(Request $request): array + private function mirrorFilters(Request $request): array { $filters = $request->validate([ 'store_name' => ['nullable', 'string', 'max:255'], @@ -356,6 +360,18 @@ class AccountBills extends Controller return array_map(static fn (mixed $value): string => trim((string) $value), $filters); } + private function mirrorRepository(string $billType): object + { + return $billType === 'pinduoduo' + ? app(StoreReceivingPinduoduoBillMirrorRepository::class) + : app(StoreReceivingDouyinBillMirrorRepository::class); + } + + private function mirrorLabel(string $billType): string + { + return $billType === 'pinduoduo' ? '拼多多' : '抖音'; + } + private function sort(Request $request): string { return in_array($request->get('sort'), ['occurred_at', 'income', 'expense', 'balance', 'record_code'], true) diff --git a/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php b/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php index a7659c0d..7fc4a61f 100644 --- a/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php @@ -39,9 +39,10 @@ class StoreReceivingAlipayBillMirrorRepository int $perPage, ?string $channel, string $mapping = 'taobao', + string $datasetKey = self::DATASET, ): array { - $dataset = $this->datasets->get(self::DATASET); + $dataset = $this->datasets->get($datasetKey); if (! $this->mirror->ready($dataset) || ! $this->mirror->ready($this->datasets->get(self::SHOP_DATASET))) { throw new RuntimeException('store_receiving_alipay_bill_mirror_not_ready'); @@ -51,14 +52,15 @@ class StoreReceivingAlipayBillMirrorRepository $perPage = min(max($perPage, 1), 100); $direction = $direction === 'asc' ? 'asc' : 'desc'; $sortColumn = [ - 'record_code' => 'b.Id', + 'record_code' => $mapping === 'pinduoduo' ? 'b.auto_id' : 'b.Id', 'occurred_at' => 'b.fee_time', 'income' => 'b.amount', 'expense' => 'b.amount', 'balance' => 'b.amount', ][$sort] ?? 'b.fee_time'; - $query = $this->billQuery(); + $targetTable = $dataset->targetTable(); + $query = $this->billQuery($targetTable); $this->applyChannel($query, $channel); $this->applyFilters($query, $filters); $state = $this->mirror->state($dataset) ?? []; @@ -66,14 +68,14 @@ class StoreReceivingAlipayBillMirrorRepository // that count exact in their upsert transaction, including inserts. $hasFilters = collect($filters)->contains(static fn (mixed $value): bool => $value !== null && trim((string) $value) !== ''); $recordedCount = $state['mirror_row_count'] ?? null; - $useRecordedCount = $channel === null && ! $hasFilters + $useRecordedCount = $this->channelCoversDataset($channel, $dataset->fixedFilters()) && ! $hasFilters && (($state['status'] ?? null) === 'succeeded' || (bool) ($state['mirror_row_count_is_exact'] ?? false)) && $recordedCount !== null && ctype_digit((string) $recordedCount); $total = $useRecordedCount ? (int) $recordedCount : (clone $query)->count(); $summary = $this->summary($query, $filters, $channel); $billRows = $this->pageRows( - $query, $sortColumn, $direction, $page, $perPage, $filters, $channel, + $query, $sortColumn, $direction, $page, $perPage, $filters, $channel, $targetTable, ); $shopNames = $this->shopNames($billRows); $rows = collect($billRows) @@ -114,40 +116,42 @@ class StoreReceivingAlipayBillMirrorRepository string $direction = 'asc', ?string $channel = null, string $mapping = 'taobao', + string $datasetKey = self::DATASET, ): iterable { - $dataset = $this->datasets->get(self::DATASET); + $dataset = $this->datasets->get($datasetKey); if (! $this->mirror->ready($dataset) || ! $this->mirror->ready($this->datasets->get(self::SHOP_DATASET))) { throw new RuntimeException('store_receiving_alipay_bill_mirror_not_ready'); } - $query = $this->billQuery(); + $targetTable = $dataset->targetTable(); + $query = $this->billQuery($targetTable); $this->applyChannel($query, $channel); $this->applyFilters($query, $filters); $shopNames = $this->allShopNames(); $direction = $direction === 'desc' ? 'desc' : 'asc'; $sortColumn = [ - 'record_code' => 'b.Id', + 'record_code' => $mapping === 'pinduoduo' ? 'b.auto_id' : 'b.Id', 'occurred_at' => 'b.fee_time', 'income' => 'b.amount', 'expense' => 'b.amount', 'balance' => 'b.amount', ][$sort] ?? 'b.fee_time'; - return $this->exportBatches($query, $sortColumn, $direction, $shopNames, $mapping); + return $this->exportBatches($query, $sortColumn, $direction, $shopNames, $mapping, $targetTable); } /** @return array{shops:array,bill_items:array} */ - public function filterOptionsForChannel(string $channel): array + public function filterOptionsForChannel(string $channel, string $datasetKey = self::DATASET): array { - $dataset = $this->datasets->get(self::DATASET); + $dataset = $this->datasets->get($datasetKey); if (! $this->mirror->ready($dataset) || ! $this->mirror->ready($this->datasets->get(self::SHOP_DATASET))) { throw new RuntimeException('store_receiving_alipay_bill_mirror_not_ready'); } - $channelQuery = $this->billQuery(); + $channelQuery = $this->billQuery($dataset->targetTable()); $this->applyChannel($channelQuery, $channel); $shopIds = (clone $channelQuery) @@ -187,12 +191,12 @@ class StoreReceivingAlipayBillMirrorRepository } /** Read bounded pages so a full export does not sort/buffer millions of rows at once. */ - private function exportBatches(Builder $query, string $sortColumn, string $direction, array $shopNames, string $mapping): iterable + private function exportBatches(Builder $query, string $sortColumn, string $direction, array $shopNames, string $mapping, string $targetTable): iterable { $operator = $direction === 'asc' ? '>' : '<'; $sortField = substr($sortColumn, 2); $last = null; - $maximumId = $this->billQuery()->max('b.Id'); + $maximumId = $this->billQuery($targetTable)->max('b.Id'); if ($maximumId === null) return; do { @@ -227,9 +231,9 @@ class StoreReceivingAlipayBillMirrorRepository } while ($rows->count() === 1000); } - private function billQuery(): Builder + private function billQuery(string $targetTable): Builder { - return $this->mirror->connection()->table('ops_jst_original_bill_info as b'); + return $this->mirror->connection()->table($targetTable . ' as b'); } /** @return array */ @@ -241,6 +245,7 @@ class StoreReceivingAlipayBillMirrorRepository int $perPage, array $filters, ?string $channel, + string $targetTable, ): array { if ($channel === null || ($filters['bill_item'] ?? '') === '') { return $query->select('b.*') @@ -252,7 +257,7 @@ class StoreReceivingAlipayBillMirrorRepository } $idQuery = clone $query; - if ($idQuery->getConnection()->getDriverName() === 'mysql') { + if ($targetTable === 'ops_jst_original_bill_info' && $idQuery->getConnection()->getDriverName() === 'mysql') { $idQuery->forceIndex('idx_original_bill_channel_shop'); } $ids = $idQuery->select('b.Id') @@ -265,7 +270,7 @@ class StoreReceivingAlipayBillMirrorRepository return []; } - $rowsById = $this->billQuery()->whereIn('b.Id', $ids)->select('b.*')->get() + $rowsById = $this->billQuery($targetTable)->whereIn('b.Id', $ids)->select('b.*')->get() ->keyBy(static fn (object $row): string => (string) $row->Id); return collect($ids) @@ -372,7 +377,7 @@ class StoreReceivingAlipayBillMirrorRepository if (($filters['keyword'] ?? '') !== '') { $value = '%' . $this->escapeLike((string) $filters['keyword']) . '%'; $query->where(function (Builder $query) use ($value): void { - foreach (['b.fee_name', 'b.primary_key', 'b.p_auto_id', 'b.so_id', 'b.sku_id', 'b.shop_sku_id', 'b.shop_i_id', 'b.channel'] as $column) { + foreach (['b.Id', 'b.auto_id', 'b.fee_name', 'b.primary_key', 'b.p_auto_id', 'b.so_id', 'b.sku_id', 'b.shop_sku_id', 'b.shop_i_id', 'b.channel'] as $column) { $query->orWhere($column, 'like', $value); } }); @@ -386,6 +391,22 @@ class StoreReceivingAlipayBillMirrorRepository } } + /** @param array $fixedFilters */ + private function channelCoversDataset(?string $channel, array $fixedFilters): bool + { + if ($channel === null || $channel === '') { + return true; + } + + foreach ($fixedFilters as $column => $value) { + if (strcasecmp($column, 'channel') === 0 && (string) $value === $channel) { + return true; + } + } + + return false; + } + /** @param array $rows @return array */ private function shopNames(array $rows): array { @@ -448,6 +469,10 @@ class StoreReceivingAlipayBillMirrorRepository /** @return array */ private function mapRowFor(object $row, ?string $shopName, string $mapping): array { + if ($mapping === 'pinduoduo') { + return $this->mapPinduoduoRow($row, $shopName); + } + if ($mapping !== 'douyin') { return $this->mapRow($row, $shopName); } @@ -493,6 +518,38 @@ class StoreReceivingAlipayBillMirrorRepository ]; } + /** @return array */ + private function mapPinduoduoRow(object $row, ?string $shopName): array + { + $amount = abs((float) ($row->amount ?? 0)); + $income = (string) ($row->inout_type ?? '') === '1' ? $amount : 0.0; + $expense = (string) ($row->inout_type ?? '') === '-1' ? $amount : 0.0; + + return [ + 'id' => (string) $row->Id, + 'record_code' => (string) ($row->auto_id ?: $row->Id), + 'store_name' => $shopName, + 'finance_remark' => null, + 'order_no' => $row->so_id ?: $row->p_auto_id, + 'amount' => $amount, + 'custom_remark' => null, + 'bill_item' => $row->fee_name, + 'occurred_at' => $this->dateTime($row->fee_time), + 'income' => $income, + 'expense' => $expense, + 'balance' => $income - $expense, + 'remark' => null, + 'updated_at' => $this->dateTime($row->updateTime), + 'modifier' => 'RPA', + 'payload' => [ + 'merchant_order_no' => $row->so_id ?: $row->p_auto_id, + 'online_order_no' => $row->so_id, + 'finance_type' => $row->fee_name, + 'type' => '拼多多货款账户', + ], + ]; + } + private function directionCode(string $value): string { return match ($value) { diff --git a/modules/EcommerceFinance/Repositories/StoreReceivingPinduoduoBillMirrorRepository.php b/modules/EcommerceFinance/Repositories/StoreReceivingPinduoduoBillMirrorRepository.php new file mode 100644 index 00000000..cc947e8a --- /dev/null +++ b/modules/EcommerceFinance/Repositories/StoreReceivingPinduoduoBillMirrorRepository.php @@ -0,0 +1,49 @@ + */ + public function paginate(array $filters, string $sort, string $direction, int $page, int $perPage): array + { + return $this->source->paginateChannel( + $filters, + $sort, + $direction, + $page, + $perPage, + self::SOURCE_CHANNEL, + 'pinduoduo', + self::DATASET, + ); + } + + /** @return iterable> */ + public function exportRows(array $filters, string $sort = 'occurred_at', string $direction = 'asc'): iterable + { + return $this->source->exportChannelRows( + $filters, + $sort, + $direction, + self::SOURCE_CHANNEL, + 'pinduoduo', + self::DATASET, + ); + } + + /** @return array{shops:array,bill_items:array} */ + public function filterOptions(): array + { + return $this->source->filterOptionsForChannel(self::SOURCE_CHANNEL, self::DATASET); + } +} diff --git a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php index 911af9a2..ec9f9c8e 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php @@ -71,6 +71,19 @@ class BusinessMirrorDatasetRegistry return $this->definitions()[$key]['columns'] ?? null; } + /** @return array */ + public function fixedFilters(string $key): array + { + $definitions = $this->definitions(); + $definition = (array) ($definitions[$key] ?? []); + + if ($definition === []) { + throw new InvalidArgumentException('未知的业务镜像数据集：' . $key); + } + + return $this->normalizeFixedFilters($key, (array) ($definition['fixed_filters'] ?? [])); + } + private function make(string $key, array $definition): BusinessMirrorDataset { $source = trim((string) ($definition['source_key'] ?? '')); @@ -84,6 +97,7 @@ class BusinessMirrorDatasetRegistry (array) ($definition['indexes'] ?? []), )); $indexColumns = array_merge(...($indexes ?: [[]])); + $fixedFilters = $this->normalizeFixedFilters($key, (array) ($definition['fixed_filters'] ?? [])); foreach ([$source, $target] as $identifier) { if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $identifier) !== 1) { @@ -105,7 +119,7 @@ class BusinessMirrorDatasetRegistry throw new InvalidArgumentException('业务镜像数据集的增量字段不安全：' . $key); } - return new readonly class($key, $source, $schema, $object, $target, $keys, $watermark ?: null, $indexes) implements BusinessMirrorDataset { + return new readonly class($key, $source, $schema, $object, $target, $keys, $watermark ?: null, $fixedFilters, $indexes) implements BusinessMirrorDataset { public function __construct( private string $key, private string $source, @@ -114,6 +128,7 @@ class BusinessMirrorDatasetRegistry private string $target, private array $keys, private ?string $watermark, + private array $fixedFilters, private array $indexes, ) { } @@ -125,7 +140,31 @@ class BusinessMirrorDatasetRegistry public function targetTable(): string { return $this->target; } public function keyColumns(): array { return $this->keys; } public function watermarkColumn(): ?string { return $this->watermark; } + public function fixedFilters(): array { return $this->fixedFilters; } public function indexDefinitions(): array { return $this->indexes; } }; } + + /** @return array */ + private function normalizeFixedFilters(string $key, array $filters): array + { + $normalized = []; + $seen = []; + + foreach ($filters as $column => $value) { + $column = is_string($column) ? trim($column) : ''; + $canonical = strtolower($column); + if ($column === '' || mb_strlen($column) > 128 || str_contains($column, "\0") || isset($seen[$canonical])) { + throw new InvalidArgumentException('业务镜像数据集包含无效的固定过滤字段：' . $key); + } + if (! is_scalar($value)) { + throw new InvalidArgumentException('业务镜像数据集包含无效的固定过滤值：' . $key); + } + + $seen[$canonical] = true; + $normalized[$column] = $value; + } + + return $normalized; + } } diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php b/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php index 601b3748..8d4c71e7 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php @@ -122,7 +122,8 @@ class BusinessMirrorSchemaManager } } - foreach ([...$dataset->keyColumns(), $dataset->watermarkColumn()] as $required) { + $fixedFilterColumns = array_keys($dataset->fixedFilters()); + foreach ([...$dataset->keyColumns(), $dataset->watermarkColumn(), ...$fixedFilterColumns] as $required) { if ($required !== null && ! in_array(strtolower($required), $names, true)) { throw new RuntimeException('business_mirror_required_column_missing:' . $dataset->key()); } diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php index 673340dd..bb1a0441 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php @@ -120,6 +120,11 @@ class BusinessMirrorSynchronizer $where = []; $bindings = []; + foreach ($dataset->fixedFilters() as $column => $value) { + $where[] = $this->quote($column) . ' = ?'; + $bindings[] = $value; + } + if ($mode === 'incremental' && $dataset->watermarkColumn() !== null) { $watermark = $this->quote($dataset->watermarkColumn()); $where[] = $watermark . ' >= ? AND ' . $watermark . ' < ?'; diff --git a/resources/assets/js/components/business/AccountBills.vue b/resources/assets/js/components/business/AccountBills.vue index 8b1481b6..b94daeac 100644 --- a/resources/assets/js/components/business/AccountBills.vue +++ b/resources/assets/js/components/business/AccountBills.vue @@ -5,14 +5,14 @@