diff --git a/app/Console/Commands/ImportDingProcessDetails.php b/app/Console/Commands/ImportDingProcessDetails.php index c9e4639c..aaa1cf04 100644 --- a/app/Console/Commands/ImportDingProcessDetails.php +++ b/app/Console/Commands/ImportDingProcessDetails.php @@ -13,7 +13,7 @@ class ImportDingProcessDetails extends Command {--scope=personal_permission : 数据权限范围标识} {--replace : 导入前清空同一范围的数据}'; - protected $description = '导入钉钉流程明细到共享业务镜像'; + protected $description = '兼容导入旧版铺平 CSV；生产请使用 business-mirror:sync-ding-process-details'; public function handle(DingProcessDetailMirrorImporter $importer): int { diff --git a/app/Console/Commands/SyncBusinessMirrors.php b/app/Console/Commands/SyncBusinessMirrors.php index 24445c13..b80fdb55 100644 --- a/app/Console/Commands/SyncBusinessMirrors.php +++ b/app/Console/Commands/SyncBusinessMirrors.php @@ -7,6 +7,7 @@ use Illuminate\Support\Facades\Log; use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository; use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry; use Modules\EcommerceFinance\Services\BusinessMirrorSynchronizer; +use Modules\EcommerceFinance\Services\DingProcessDetailMirrorSynchronizer; use Modules\EcommerceFinance\Services\DingPayableWorkflowMirrorSynchronizer; use Throwable; @@ -26,6 +27,7 @@ class SyncBusinessMirrors extends Command BusinessMirrorSynchronizer $synchronizer, GenericBusinessMirrorRepository $mirror, ?DingPayableWorkflowMirrorSynchronizer $dingPayableWorkflows = null, + ?DingProcessDetailMirrorSynchronizer $dingProcessDetails = null, ): int { if (! (bool) config('upstream_mirrors.refresh_enabled', false) && ! $this->option('full')) { @@ -128,6 +130,24 @@ class SyncBusinessMirrors extends Command Log::error('钉钉线上代发价格流程镜像同步失败。', ['exception' => $exception::class]); $this->error('operations_replica.rpt.ding_online_daifa_price：同步失败，已保留上一版共享数据。'); } + + // 财务流程明细按结构化明细/付款事实同步。首次全量通过专用命令执行， + // 日常只处理最近更新过的审批，避免每分钟重扫整批钉钉表单。 + if ($dingProcessDetails !== null) { + try { + $detailResult = $dingProcessDetails->sync(false); + if ($detailResult['status'] === 'succeeded') { + $this->info(sprintf( + 'operations_replica.rpt.ding_process_details：同步完成，本轮 %d 个审批。', + $detailResult['workflows'], + )); + } + } catch (Throwable $exception) { + $failed = true; + Log::error('钉钉流程明细结构镜像同步失败。', ['exception' => $exception::class]); + $this->error('operations_replica.rpt.ding_process_details：同步失败，已保留上一版共享数据。'); + } + } } return $failed ? self::FAILURE : self::SUCCESS; diff --git a/app/Console/Commands/SyncDingProcessDetailMirror.php b/app/Console/Commands/SyncDingProcessDetailMirror.php new file mode 100644 index 00000000..97af40a6 --- /dev/null +++ b/app/Console/Commands/SyncDingProcessDetailMirror.php @@ -0,0 +1,48 @@ +sync( + (bool) $this->option('full'), + (bool) $this->option('dry-run'), + ); + } catch (Throwable $exception) { + $this->error('同步失败，已保留上一版共享数据：' . $exception->getMessage()); + + return self::FAILURE; + } + + if ($result['status'] === 'requires_full_sync') { + $this->warn('流程明细尚未完成结构化全量同步，请使用 --full。'); + + return self::SUCCESS; + } + + $this->info(sprintf( + '%s：%d 个审批，%d 条明细，%d 笔付款；金额合计 %.2f，银行付款合计 %.2f。', + $result['status'] === 'validated' ? '校验通过' : '同步完成', + $result['workflows'], + $result['details'], + $result['payments'], + $result['amount_total'], + $result['bank_amount_total'], + )); + + return self::SUCCESS; + } +} diff --git a/config/upstream_mirrors.php b/config/upstream_mirrors.php index 2da85ffc..6cf2c275 100644 --- a/config/upstream_mirrors.php +++ b/config/upstream_mirrors.php @@ -29,6 +29,113 @@ return [ 'max_batches_per_run' => (int) env('DING_PAYABLE_WORKFLOW_MIRROR_MAX_BATCHES', 2), ], + // 流程明细页按钉钉结构化表单同步：审批主表、费用明细、付款明细 + // 分别读取并按业务键落镜像，禁止再把明细表与付款表铺平后逐行求和。 + 'ding_process_details' => [ + 'source_key' => 'operations_replica', + 'scope' => env('DING_PROCESS_DETAILS_SCOPE', 'personal_permission'), + 'start_date' => env('DING_PROCESS_DETAILS_START_DATE', '2026-01-01'), + 'batch_size' => (int) env('DING_PROCESS_DETAILS_BATCH_SIZE', 150), + 'sync_interval_minutes' => (int) env('DING_PROCESS_DETAILS_SYNC_MINUTES', 5), + 'incremental_overlap_hours' => (int) env('DING_PROCESS_DETAILS_OVERLAP_HOURS', 48), + 'processes' => [ + 'PROC-B290696F-EC2A-4F76-87BA-E0A12E10D09F' => [ + 'kind' => 'daily', + 'detail_table' => '收款费用明细', + 'detail_amount_fields' => ['收款金额'], + 'amount_total_fields' => ['收款合计'], + 'payment_table' => '付款明细', + 'payment_total_fields' => ['付款合计'], + ], + 'PROC-39DD70DE-DAB2-4A5F-B8F1-0B0B68BD0C88' => [ + 'document_type' => '应付申请流程', + 'detail_table' => '应付明细', + 'detail_amount_fields' => ['付款金额'], + 'payment_table' => '付款明细', + 'payment_total_fields' => ['付款金额合计'], + ], + 'PROC-ACF09A14-B094-46B1-A9E2-8BB1F4E1FDCF' => [ + 'document_type' => '线上代发业务报销', + 'detail_table' => '收款费用明细', + 'detail_amount_fields' => ['代发订单金额'], + 'payment_table' => '付款明细', + ], + 'PROC-C5159B8D-F412-4DF8-981D-64EBB8D03EB6' => [ + 'document_type' => '种菜核销/种菜报销', + 'detail_table' => '收款费用明细', + 'detail_amount_fields' => ['收款金额'], + 'amount_total_fields' => ['收款合计'], + 'payment_table' => '付款明细', + 'payment_total_fields' => ['付款合计'], + ], + 'PROC-BBAC7187-9E59-4942-941A-59C7EAF0CEDA' => [ + 'document_type' => '快递面单应付', + 'detail_table' => '收款费用明细', + 'detail_amount_fields' => ['扣款金额', '收款金额'], + 'amount_total_fields' => ['面单金额合计'], + 'payment_table' => '付款明细', + 'payment_total_fields' => ['付款合计'], + ], + 'PROC-6407E98B-040A-4C96-AA7D-696650FC1A25' => [ + 'document_type' => '固定资产报废或出售', + 'detail_table' => '固定资产报废或出售明细', + 'detail_amount_fields' => ['出售价格'], + ], + 'PROC-0EBE35D8-5A40-4746-B758-BA5EB8DE1BE3' => [ + 'document_type' => '日常物品报销', + 'detail_table' => '收款费用明细', + 'detail_amount_fields' => ['收款金额'], + 'amount_total_fields' => ['收款合计'], + 'payment_table' => '付款明细', + 'payment_total_fields' => ['付款合计'], + ], + 'PROC-17C65606-3F57-4905-8BB5-F5D313CD87D1' => [ + 'document_type' => '样品流程', + 'detail_table' => '样品明细', + 'detail_amount_fields' => ['样品成本', '成本'], + ], + 'PROC-23F2509C-33DC-411A-9FA6-700830A75B37' => [ + 'document_type' => '一次性采买', + 'detail_table' => '采买基础信息', + 'detail_amount_fields' => ['采购金额'], + 'amount_total_fields' => ['采买总金额'], + ], + 'PROC-5AB333E3-4496-4DF3-A5C1-1FBFAB7C72C8' => [ + 'document_type' => '板材流程', + 'detail_table' => '收款费用明细', + 'detail_amount_fields' => ['实际需支付金额', '订单金额'], + 'amount_total_fields' => ['请款金额'], + ], + 'PROC-657E5EEA-DEAA-4DC5-AD8B-A679FE9AD361' => [ + 'document_type' => '板材流程', + 'detail_table' => '收款费用明细', + 'detail_amount_fields' => ['收款金额'], + 'amount_total_fields' => ['收款合计'], + 'payment_total_fields' => ['付款合计'], + ], + 'PROC-C014F61F-F1F8-4F36-93F8-DF2A7EEAC457' => [ + 'document_type' => '外贸发货费用报销', + 'scalar_detail' => true, + 'detail_amount_fields' => ['收款金额'], + 'payment_total_fields' => ['付款合计'], + ], + 'PROC-1E002C79-C087-4630-B9CE-FAB6DF80FAC5' => [ + 'document_type' => '工资流程', + 'detail_table' => '工资明细表', + 'detail_amount_fields' => ['付款金额'], + 'amount_total_fields' => ['收款合计'], + 'payment_total_fields' => ['付款合计'], + ], + 'PROC-2C1EFBA6-2369-461C-8CCD-05A3664CB8B5' => [ + 'document_type' => '种菜充值', + 'detail_table' => '收款费用明细', + 'detail_amount_fields' => ['收款金额'], + 'amount_total_fields' => ['收款合计'], + 'payment_total_fields' => ['付款合计'], + ], + ], + ], + // 线上代发业务报销流程（钉钉 OA 审批）：数据源为 operations_replica 的 // rpt.ding_Workflow 中固定流程编码。明细按“商品ID”关联到共享业务镜像的 // 商品目录（ops_jst_sku_map / ops_jst_sku）做匹配。页面从共享镜像读取， diff --git a/database/migrations/2026_10_26_210000_add_structured_ding_process_payments.php b/database/migrations/2026_10_26_210000_add_structured_ding_process_payments.php new file mode 100644 index 00000000..194201d8 --- /dev/null +++ b/database/migrations/2026_10_26_210000_add_structured_ding_process_payments.php @@ -0,0 +1,79 @@ +environment('testing') + || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') { + return; + } + + $schema = Schema::connection($this->connection()); + + if ($schema->hasTable('ding_process_detail_mirror') + && ! $schema->hasColumn('ding_process_detail_mirror', 'process_instance_id')) { + $schema->table('ding_process_detail_mirror', function (Blueprint $table): void { + $table->string('process_instance_id', 100) + ->nullable() + ->after('business_id') + ->index('ding_process_detail_instance_index'); + }); + } + + if (! $schema->hasTable('ding_process_payment_mirror')) { + $schema->create('ding_process_payment_mirror', function (Blueprint $table): void { + $table->bigIncrements('id'); + $table->string('business_id', 32); + $table->string('process_instance_id', 100); + $table->unsignedInteger('payment_no'); + $table->string('sequence_spec', 255)->nullable(); + $table->decimal('amount', 18, 2); + $table->string('account_code', 64)->nullable(); + $table->string('account_name', 255)->nullable(); + $table->string('payment_method', 50)->nullable(); + $table->string('paid_at', 50)->nullable(); + $table->string('source_scope', 40)->default('personal_permission')->index(); + $table->dateTime('synced_at', 6); + $table->timestamps(); + + $table->unique( + ['source_scope', 'business_id', 'payment_no'], + 'ding_process_payment_scope_business_no_unique', + ); + $table->index( + ['source_scope', 'business_id'], + 'ding_process_payment_scope_business_index', + ); + }); + } + } + + public function down(): void + { + if (app()->environment('testing') + || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') { + return; + } + + $schema = Schema::connection($this->connection()); + $schema->dropIfExists('ding_process_payment_mirror'); + + if ($schema->hasTable('ding_process_detail_mirror') + && $schema->hasColumn('ding_process_detail_mirror', 'process_instance_id')) { + $schema->table('ding_process_detail_mirror', function (Blueprint $table): void { + $table->dropIndex('ding_process_detail_instance_index'); + $table->dropColumn('process_instance_id'); + }); + } + } +}; diff --git a/deploy/activate-incremental-release.sh b/deploy/activate-incremental-release.sh index 4321641b..8e31b999 100644 --- a/deploy/activate-incremental-release.sh +++ b/deploy/activate-incremental-release.sh @@ -227,6 +227,8 @@ CRON_TZ=Asia/Shanghai */2 * * * * root /usr/bin/flock -n /run/caiwu-esign-polling.lock /usr/bin/docker exec -u www-data caiwu_app php artisan esign:poll-signatures --quiet >/dev/null 2>&1 # 财务台账：每天 02:30 同步 AI 库动账并重算流水；店铺台账默认不自动回填。 30 2 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan finance-ledger:sync-rpt --days=7 --merge --quiet >/dev/null 2>&1 +# 财务流程明细每周完整校准一次，修正超过 48 小时重叠窗口才出现的上游改单或晚到付款。 +20 2 * * 0 root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh business-mirror:sync-ding-process-details --full --quiet >/dev/null 2>&1 # 科目映射保存后保留自动重跑队列，每分钟在受限容器内串行消费。 * * * * * root /usr/bin/flock -n /run/caiwu-account-mapping-resync.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh store-receipt:auto-resync --quiet >/dev/null 2>&1 * * * * * root /usr/bin/flock -n /run/caiwu-account-bills-export.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh account-bills:exports:run --quiet >/dev/null 2>&1 diff --git a/deploy/activate-release.sh b/deploy/activate-release.sh index 5a6684e0..d78b7b3f 100644 --- a/deploy/activate-release.sh +++ b/deploy/activate-release.sh @@ -231,6 +231,8 @@ CRON_TZ=Asia/Shanghai */2 * * * * root /usr/bin/flock -n /run/caiwu-esign-polling.lock /usr/bin/docker exec -u www-data caiwu_app php artisan esign:poll-signatures --quiet >/dev/null 2>&1 # 财务台账：每天 02:30 同步 AI 库动账并重算流水；店铺台账默认不自动回填。 30 2 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan finance-ledger:sync-rpt --days=7 --merge --quiet >/dev/null 2>&1 +# 财务流程明细每周完整校准一次，修正超过 48 小时重叠窗口才出现的上游改单或晚到付款。 +20 2 * * 0 root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh business-mirror:sync-ding-process-details --full --quiet >/dev/null 2>&1 # 科目映射保存后保留自动重跑队列，每分钟在受限容器内串行消费。 * * * * * root /usr/bin/flock -n /run/caiwu-account-mapping-resync.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh store-receipt:auto-resync --quiet >/dev/null 2>&1 * * * * * root /usr/bin/flock -n /run/caiwu-account-bills-export.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh account-bills:exports:run --quiet >/dev/null 2>&1 diff --git a/deploy/finance-batch-imports.json b/deploy/finance-batch-imports.json index 310315db..71b2efec 100644 --- a/deploy/finance-batch-imports.json +++ b/deploy/finance-batch-imports.json @@ -1,5 +1,4 @@ [ - "ding:import-process-details", "accessory-reship:import-courier-configs", "account-ledger-remarks:import", "account-mapping:import-files", diff --git a/deploy/run-finance-batch.sh b/deploy/run-finance-batch.sh index cc47643a..48baabdf 100644 --- a/deploy/run-finance-batch.sh +++ b/deploy/run-finance-batch.sh @@ -8,7 +8,7 @@ CAIWU_BATCH_CALLER_DIR="$PWD" export CAIWU_BATCH_CALLER_DIR case "${1:-}" in *:import|*:import-*|*:import:*|product:monthly-weighted-average:base) is_import=1 ;; - jushuitan:sync-shipping|store-receipt:order-match:sync|store-receipt:product-cost:sync|store-receipt:auto-resync|order-data:warm-shop-candidates|business-mirror:sync-tables|business-mirror:index-work-order-waybills|business-mirror:refresh-paid-sales|account-bills:exports:run|finance-ledger:recalculate|product:monthly-weighted-average:auto|product:monthly-weighted-average:compute|product:monthly-weighted-average:report|finance:monthly-weighted-product-cost:auto) ;; + jushuitan:sync-shipping|store-receipt:order-match:sync|store-receipt:product-cost:sync|store-receipt:auto-resync|order-data:warm-shop-candidates|business-mirror:sync-tables|business-mirror:sync-ding-process-details|business-mirror:index-work-order-waybills|business-mirror:refresh-paid-sales|account-bills:exports:run|finance-ledger:recalculate|product:monthly-weighted-average:auto|product:monthly-weighted-average:compute|product:monthly-weighted-average:report|finance:monthly-weighted-product-cost:auto) ;; *) echo 'Unsupported finance batch command' >&2; exit 64 ;; esac cd "$REMOTE_DIR" diff --git a/docs/上游业务镜像清单.md b/docs/上游业务镜像清单.md index b557e1ec..93657917 100644 --- a/docs/上游业务镜像清单.md +++ b/docs/上游业务镜像清单.md @@ -33,6 +33,7 @@ | 已接入待首次同步 | 店铺营销账户余额的京东账户读取 `js_data_middle_platform.dbo.jd_jzt_finance_records` 镜像 `jsm_jd_jzt_finance_records`；按 `id` 稳定断点、`updated_at` 重叠增量，页面不直连上游。 | | 本次接入 | 普通商品资料的工作平台采购价格接入 `jiansun.dbo.pm_priceAdjustment`（镜像 `js_pm_price_adjustment`）：按 `skuCode` 关联 `jst_sku.sku_id`，取 `effectTime` 不晚于当前时间的最新 `purchasePrice`，并在页面和导出中标记“工作平台价格”；源表没有可靠更新时间，按 `Id` 稳定键追加，首次全量完成前只显示普通商品自身采购价。 | | 待有效凭据后首次同步 | 普通商品当前成本价、商品类型和更新时间继续读取 `operations_replica.rpt.jst_sku`；`rpt.fys_skuCostPriceUpdate` 是一 SKU 一行的旧快照，不能充当变价历史。完整成本价变更仍读取聚水潭经典开放平台“商品操作日志查询”接口 `worklog.query`，只镜像操作内容含“采购价”的记录到 `jushuitan_cost_price_history_rows`；经典接口必须在服务器私有配置中使用同一应用的 `JUSHUITAN_PARTNER_ID / JUSHUITAN_PARTNER_KEY / JUSHUITAN_TOKEN`，不得把新开放平台的 App Key / App Secret / Access Token 混用。页面在首次同步完成前明确显示待同步或失败，不把 0 条解释为真实无历史。 | +| 本次接入 | 流程明细读取 `operations_replica/rpt.ding_Workflow + rpt.ding_WorkflowScheme` 的结构化审批主表、费用明细和付款明细，写入 `ding_process_detail_mirror` 与 `ding_process_payment_mirror`。禁止先把明细表和付款表铺平成 CSV 后逐行求和；付款明细必须按“对应序号”关联到代表明细行，但合计始终读取独立付款事实。首次全量先运行 `business-mirror:sync-ding-process-details --full --dry-run` 校验，再使用 `business-mirror:sync-ding-process-details --full` 发布；日常通过 `updateAt/createAt` 48 小时重叠增量同步。当前纳入日常报销、应付、线上代发、种菜、快递面单、固定资产、日常物品、样品、一次性采买、板材、工资、外贸费用和种菜充值等 14 个财务流程；发现新的明细流程编码时全量同步报错并保留上一版。 | 抖音月中校准可在上述导入命令追加 `--through=YYYY-MM-DD`，文件必须包含该月 1 日至截止日（含当天）的全部店铺和流水。系统仅替换这一日期范围，截止日之后继续读取原始账单镜像；省略参数仍表示完整月份。截止日期不得早于同月已校准范围。每天更新时重新导出月初至昨日的完整结果并核对笔数、收支总额，以覆盖迟到及修改记录，不能把单日增量当作月初至截止日的完整结果导入。 diff --git a/modules/EcommerceFinance/Repositories/DingProcessDetailMirrorRepository.php b/modules/EcommerceFinance/Repositories/DingProcessDetailMirrorRepository.php index 9886a8ea..0b366f47 100644 --- a/modules/EcommerceFinance/Repositories/DingProcessDetailMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/DingProcessDetailMirrorRepository.php @@ -88,11 +88,22 @@ class DingProcessDetailMirrorRepository $query = $this->filteredQuery($filters); $total = (clone $query)->count(); + $bankAmountTotal = $this->structuredPaymentsEnabled() + ? (float) ($this->connection->table('ding_process_payment_mirror') + ->where('source_scope', $this->scope()) + ->whereIn('business_id', function ($bankQuery) use ($filters): void { + $this->applyFilters( + $bankQuery->select('business_id')->from('ding_process_detail_mirror'), + $filters, + ); + }) + ->sum('amount') ?? 0) + : (float) ((clone $query)->sum('bank_amount') ?? 0); $summary = [ 'rows' => $total, 'approvals' => (clone $query)->distinct()->count('business_id'), 'amount_total' => (float) ((clone $query)->sum('amount') ?? 0), - 'bank_amount_total' => (float) ((clone $query)->sum('bank_amount') ?? 0), + 'bank_amount_total' => $bankAmountTotal, ]; $data = $query ->orderByDesc('business_month') @@ -174,8 +185,16 @@ class DingProcessDetailMirrorRepository /** @param array $filters */ private function filteredQuery(array $filters): Builder { - $query = $this->connection->table('ding_process_detail_mirror') - ->where('source_scope', $this->scope()); + return $this->applyFilters( + $this->connection->table('ding_process_detail_mirror'), + $filters, + ); + } + + /** @param array $filters */ + private function applyFilters(Builder $query, array $filters): Builder + { + $query->where('source_scope', $this->scope()); $paidDate = trim((string) ($filters['paid_date'] ?? '')); if (preg_match('/^\d{4}-\d{2}-\d{2}$/', $paidDate) === 1) { @@ -205,4 +224,19 @@ class DingProcessDetailMirrorRepository return $query; } + + private function structuredPaymentsEnabled(): bool + { + if (! Schema::connection($this->connection->getName())->hasTable('ding_process_payment_mirror')) { + return false; + } + + $state = $this->state(); + $cursor = $state['cursor'] ?? null; + if (is_string($cursor)) { + $cursor = json_decode($cursor, true); + } + + return is_array($cursor) && ($cursor['source'] ?? '') === 'structured'; + } } diff --git a/modules/EcommerceFinance/Services/DingProcessDetailMirrorImporter.php b/modules/EcommerceFinance/Services/DingProcessDetailMirrorImporter.php index 34e3b434..d6693333 100644 --- a/modules/EcommerceFinance/Services/DingProcessDetailMirrorImporter.php +++ b/modules/EcommerceFinance/Services/DingProcessDetailMirrorImporter.php @@ -118,7 +118,10 @@ class DingProcessDetailMirrorImporter 'status' => 'succeeded', 'run_mode' => 'full', 'generation' => $generation, - 'cursor' => null, + 'cursor' => json_encode([ + 'source' => 'legacy_csv', + 'mirror_details' => $rowCount, + ], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), 'high_watermark' => null, 'window_started_at' => null, 'window_ends_at' => null, diff --git a/modules/EcommerceFinance/Services/DingProcessDetailMirrorSynchronizer.php b/modules/EcommerceFinance/Services/DingProcessDetailMirrorSynchronizer.php new file mode 100644 index 00000000..71219563 --- /dev/null +++ b/modules/EcommerceFinance/Services/DingProcessDetailMirrorSynchronizer.php @@ -0,0 +1,1302 @@ + '已完成', + 'RUNNING' => '审批中', + 'TERMINATED' => '已终止', + 'CANCELED' => '已撤销', + 'CANCELLED' => '已撤销', + 'REVOKED' => '已撤销', + ]; + + private readonly ConnectionInterface $connection; + + public function __construct( + DatabaseManager $database, + private readonly UpstreamReadOnlyQueryFactory $queries, + ) { + $this->connection = $database->connection( + (string) config('upstream_mirrors.connection', 'business_mirror'), + ); + } + + /** + * @return array{ + * status:string, + * run_mode:string, + * workflows:int, + * details:int, + * payments:int, + * amount_total:float, + * bank_amount_total:float + * } + */ + public function sync(bool $forceFull = false, bool $dryRun = false): array + { + $scope = $this->scope(); + $state = $this->state(); + $structuredState = $this->structuredState($state); + $full = $forceFull; + + if (! $full && ! $structuredState) { + return [ + 'status' => 'requires_full_sync', + 'run_mode' => 'full', + 'workflows' => 0, + 'details' => 0, + 'payments' => 0, + 'amount_total' => 0.0, + 'bank_amount_total' => 0.0, + ]; + } + + if (! $full && ! $this->incrementalSyncDue($state)) { + return [ + 'status' => 'succeeded', + 'run_mode' => 'incremental', + 'workflows' => 0, + 'details' => 0, + 'payments' => 0, + 'amount_total' => 0.0, + 'bank_amount_total' => 0.0, + ]; + } + + $now = now(); + if (! $dryRun) { + $this->markAttempted($state, $full, $now); + } + + try { + $extracted = $full + ? $this->extractFullWindow($now) + : $this->extractIncremental($state, $now); + + if ($dryRun) { + return [ + 'status' => 'validated', + 'run_mode' => $full ? 'full' : 'incremental', + 'workflows' => $extracted['metrics']['workflows'], + 'details' => $extracted['metrics']['details'], + 'payments' => $extracted['metrics']['payments'], + 'amount_total' => $extracted['metrics']['amount_total'], + 'bank_amount_total' => $extracted['metrics']['bank_amount_total'], + ]; + } + + $this->publish( + $extracted['details'], + $extracted['payments'], + $extracted['process_instance_ids'], + $full, + $scope, + $extracted['metrics'], + $now, + ); + + return [ + 'status' => 'succeeded', + 'run_mode' => $full ? 'full' : 'incremental', + 'workflows' => $extracted['metrics']['workflows'], + 'details' => $extracted['metrics']['details'], + 'payments' => $extracted['metrics']['payments'], + 'amount_total' => $extracted['metrics']['amount_total'], + 'bank_amount_total' => $extracted['metrics']['bank_amount_total'], + ]; + } catch (Throwable $exception) { + if (! $dryRun) { + $this->markFailed($state, $exception, $now); + } + throw $exception; + } + } + + /** + * @return array{ + * details:list>, + * payments:list>, + * process_instance_ids:list, + * metrics:array + * } + */ + private function extractFullWindow(Carbon $now): array + { + $start = trim((string) config('upstream_mirrors.ding_process_details.start_date', '2026-01-01')); + $end = $now->copy()->addDay()->toDateString(); + $where = 'w.joinAt >= ? AND w.joinAt < ?'; + $bindings = [$start, $end]; + $this->assertConfiguredSourceProcesses($start, $end); + $expected = $this->countWorkflows($where, $bindings); + $result = $this->extractWorkflows( + $where, + $bindings, + 'w.joinAt', + 'joinAt', + $now, + ); + + if ($result['metrics']['workflows'] !== $expected) { + throw new RuntimeException(sprintf( + '流程明细同步不完整：源审批 %d 个，实际处理 %d 个。', + $expected, + $result['metrics']['workflows'], + )); + } + + return $result; + } + + private function assertConfiguredSourceProcesses(string $start, string $end): void + { + $processCodes = array_keys((array) config( + 'upstream_mirrors.ding_process_details.processes', + [], + )); + $detailTables = array_values(array_unique(array_filter(array_map( + static fn (array $process): string => trim((string) ($process['detail_table'] ?? '')), + array_values((array) config('upstream_mirrors.ding_process_details.processes', [])), + )))); + if ($processCodes === [] || $detailTables === []) { + return; + } + + $processPlaceholders = implode(',', array_fill(0, count($processCodes), '?')); + $tablePlaceholders = implode(',', array_fill(0, count($detailTables), '?')); + $financialTitlePatterns = [ + '%报销%', + '%应付%', + '%采买%', + '%工资%', + '%板材%', + '%借支%', + '%销账%', + '%面单%', + '%外贸发货费用%', + '%种菜%', + ]; + $titleSql = implode(' OR ', array_fill(0, count($financialTitlePatterns), 'w.title LIKE ?')); + $query = $this->queries->forSourceWithTimeout( + self::SOURCE, + (int) config('finance_ledger.query_timeout', 120), + ); + $rows = $query->select( + 'SELECT w.processCode, MIN(w.title) AS sample_title FROM rpt.ding_Workflow w ' + . 'JOIN rpt.ding_WorkflowScheme s ON s.processInstanceId = w.processInstanceId ' + . "WHERE w.joinAt >= ? AND w.joinAt < ? AND s.name IN ({$tablePlaceholders}) " + . "AND w.processCode NOT IN ({$processPlaceholders}) " + . "AND ({$titleSql}) " + . 'GROUP BY w.processCode', + [ + $start, + $end, + ...$detailTables, + ...$processCodes, + ...$financialTitlePatterns, + ], + ); + + if ($rows !== []) { + throw new RuntimeException( + '发现未登记的财务审批流程：' + . implode('、', array_map( + static fn (array $row): string => trim((string) ($row['processCode'] ?? '')) + . '（' . trim((string) ($row['sample_title'] ?? '')) . '）', + array_slice($rows, 0, 5), + )), + ); + } + } + + /** + * @param array|null $state + * @return array{ + * details:list>, + * payments:list>, + * process_instance_ids:list, + * metrics:array + * } + */ + private function extractIncremental(?array $state, Carbon $now): array + { + $overlapHours = max( + 1, + (int) config('upstream_mirrors.ding_process_details.incremental_overlap_hours', 48), + ); + $succeededAt = Carbon::parse((string) ($state['last_succeeded_at'] ?? $now)); + $since = $succeededAt->copy()->subHours($overlapHours)->format('Y-m-d H:i:s'); + $until = $now->copy()->addDay()->toDateString(); + $cursorExpression = 'COALESCE(w.updateAt, w.createAt, w.joinAt)'; + $where = $cursorExpression . ' >= ? AND ' . $cursorExpression . ' < ?'; + $bindings = [$since, $until]; + + return $this->extractWorkflows( + $where, + $bindings, + $cursorExpression, + 'sync_cursor', + $now, + ); + } + + /** + * @param array $bindings + * @return array{ + * details:list>, + * payments:list>, + * process_instance_ids:list, + * metrics:array + * } + */ + private function extractWorkflows( + string $where, + array $bindings, + string $cursorExpression, + string $cursorField, + Carbon $now, + ): array { + $processCodes = array_keys((array) config( + 'upstream_mirrors.ding_process_details.processes', + [], + )); + if ($processCodes === []) { + throw new RuntimeException('流程明细未配置任何钉钉审批流程。'); + } + + $query = $this->queries->forSourceWithTimeout( + self::SOURCE, + (int) config('finance_ledger.query_timeout', 120), + ); + $employees = $this->employees($query); + $processPlaceholders = implode(',', array_fill(0, count($processCodes), '?')); + $batchSize = max( + 20, + min((int) config('upstream_mirrors.ding_process_details.batch_size', 150), 300), + ); + $cursor = null; + $details = []; + $payments = []; + $processInstanceIds = []; + $workflowCount = 0; + $issues = []; + + do { + $cursorSql = ''; + $cursorBindings = []; + if (is_array($cursor)) { + $cursorSql = " AND ({$cursorExpression} > ? OR " + . "({$cursorExpression} = ? AND w.processInstanceId > ?))"; + $cursorBindings = [ + $cursor['value'], + $cursor['value'], + $cursor['process_instance_id'], + ]; + } + + $workflows = $query->select( + 'SELECT w.processInstanceId, w.processCode, w.businessId, w.title, w.status, ' + . 'w.originatorUserId, w.originatorDeptName, w.joinAt, w.updateAt, w.finishTime, ' + . "{$cursorExpression} AS sync_cursor " + . 'FROM rpt.ding_Workflow w ' + . "WHERE w.processCode IN ({$processPlaceholders}) AND {$where}" + . $cursorSql + . " ORDER BY {$cursorExpression}, w.processInstanceId" + . ' OFFSET 0 ROWS FETCH NEXT ' . $batchSize . ' ROWS ONLY', + [...$processCodes, ...$bindings, ...$cursorBindings], + ); + + if ($workflows === []) { + break; + } + + $scheme = $this->schemeValues($query, $workflows); + $batch = $this->normalizeWorkflows($workflows, $scheme, $employees, $now, $issues); + array_push($details, ...$batch['details']); + array_push($payments, ...$batch['payments']); + array_push($processInstanceIds, ...$batch['process_instance_ids']); + $workflowCount += count($workflows); + + $last = $workflows[array_key_last($workflows)]; + $cursor = [ + 'value' => (string) ($last[$cursorField] ?? ''), + 'process_instance_id' => (string) ($last['processInstanceId'] ?? ''), + ]; + } while (count($workflows) === $batchSize); + + if ($issues !== []) { + throw new RuntimeException( + '钉钉流程明细校验失败：' . implode('；', array_slice($issues, 0, 8)), + ); + } + + $amountCents = array_sum(array_column($details, 'amount_cents')); + $bankAmountCents = array_sum(array_column($payments, 'amount_cents')); + + return [ + 'details' => array_values($details), + 'payments' => array_values($payments), + 'process_instance_ids' => array_values(array_unique($processInstanceIds)), + 'metrics' => [ + 'workflows' => $workflowCount, + 'details' => count($details), + 'payments' => count($payments), + 'amount_total' => $amountCents / 100, + 'bank_amount_total' => $bankAmountCents / 100, + ], + ]; + } + + /** @param array $bindings */ + private function countWorkflows(string $where, array $bindings): int + { + $processCodes = array_keys((array) config( + 'upstream_mirrors.ding_process_details.processes', + [], + )); + $placeholders = implode(',', array_fill(0, count($processCodes), '?')); + $query = $this->queries->forSourceWithTimeout( + self::SOURCE, + (int) config('finance_ledger.query_timeout', 120), + ); + + return (int) $query->scalar( + 'SELECT COUNT(*) FROM rpt.ding_Workflow w ' + . "WHERE w.processCode IN ({$placeholders}) AND {$where}", + [...$processCodes, ...$bindings], + ); + } + + /** @return array */ + private function employees(UpstreamReadOnlyQuery $query): array + { + $employees = []; + foreach ($query->select('SELECT userid, name FROM rpt.hr_employee') as $row) { + $id = trim((string) ($row['userid'] ?? '')); + $name = trim((string) ($row['name'] ?? '')); + if ($id !== '' && $name !== '') { + $employees[$id] = $name; + } + } + + return $employees; + } + + /** + * @param list> $workflows + * @return array> + */ + private function schemeValues(UpstreamReadOnlyQuery $query, array $workflows): array + { + $ids = array_values(array_filter(array_map( + static fn (array $workflow): string => trim((string) ($workflow['processInstanceId'] ?? '')), + $workflows, + ))); + if ($ids === []) { + return []; + } + + $idPlaceholders = implode(',', array_fill(0, count($ids), '?')); + $names = $this->schemeNames(); + $namePlaceholders = implode(',', array_fill(0, count($names), '?')); + $rows = $query->select( + 'SELECT processInstanceId, name, value FROM rpt.ding_WorkflowScheme ' + . "WHERE processInstanceId IN ({$idPlaceholders}) " + . "AND name IN ({$namePlaceholders}) ORDER BY id", + [...$ids, ...$names], + ); + $result = []; + foreach ($rows as $row) { + $id = (string) ($row['processInstanceId'] ?? ''); + $name = trim((string) ($row['name'] ?? '')); + if ($id !== '' && $name !== '') { + $result[$id][$name] = $row['value'] ?? null; + } + } + + return $result; + } + + /** @return list */ + private function schemeNames(): array + { + $names = [ + '用途', + '科目编码', + '科目名称', + '支付方式', + '付款方式', + '打款时间', + ]; + foreach ((array) config('upstream_mirrors.ding_process_details.processes', []) as $process) { + foreach ([ + $process['detail_table'] ?? '', + $process['payment_table'] ?? '', + ...(array) ($process['detail_amount_fields'] ?? []), + ...(array) ($process['amount_total_fields'] ?? []), + ...(array) ($process['payment_total_fields'] ?? []), + ] as $name) { + $name = trim((string) $name); + if ($name !== '') { + $names[] = $name; + } + } + } + + return array_values(array_unique($names)); + } + + /** + * @param list> $workflows + * @param array> $scheme + * @param array $employees + * @param list $issues + * @return array{ + * details:list>, + * payments:list>, + * process_instance_ids:list + * } + */ + private function normalizeWorkflows( + array $workflows, + array $scheme, + array $employees, + Carbon $now, + array &$issues, + ): array { + $details = []; + $payments = []; + $processInstanceIds = []; + $processes = (array) config('upstream_mirrors.ding_process_details.processes', []); + $scope = $this->scope(); + + foreach ($workflows as $workflow) { + $processCode = trim((string) ($workflow['processCode'] ?? '')); + $process = (array) ($processes[$processCode] ?? []); + $processInstanceId = trim((string) ($workflow['processInstanceId'] ?? '')); + $businessId = trim((string) ($workflow['businessId'] ?? '')); + if ($processInstanceId === '' || $businessId === '' || $process === []) { + $issues[] = '审批主表存在缺少流程或审批编号的数据'; + continue; + } + + $processInstanceIds[] = $processInstanceId; + $fields = (array) ($scheme[$processInstanceId] ?? []); + $title = trim((string) ($workflow['title'] ?? '')); + $originator = $employees[(string) ($workflow['originatorUserId'] ?? '')] + ?? $this->originatorFromTitle($title); + $documentType = $this->documentType($process, $title); + $status = $this->status((string) ($workflow['status'] ?? '')); + $businessMonth = $this->businessMonth($businessId, (string) ($workflow['joinAt'] ?? '')); + + [$workflowDetails, $workflowAmountCents, $detailTableName] = $this->detailRows( + $workflow, + $process, + $fields, + $scope, + $now, + $issues, + $originator, + $documentType, + $status, + $businessMonth, + ); + $this->validateAmountTotal( + $businessId, + $process, + $fields, + $workflowAmountCents, + $issues, + ); + + $workflowPayments = $this->paymentRows( + $workflow, + $process, + $fields, + $scope, + $now, + $issues, + ); + + if ($workflowDetails === []) { + $amount = $this->moneyFromFields($fields, (array) ($process['detail_amount_fields'] ?? [])); + if (($amount !== null && $amount !== 0) || $workflowPayments !== []) { + $workflowDetails[] = $this->detailRecord( + $workflow, + $process, + $scope, + $now, + 1, + $amount, + $title, + '', + '', + $originator, + $documentType, + $status, + $businessMonth, + ); + } + } + + if ($workflowDetails === [] && $workflowPayments === []) { + continue; + } + + $this->attachPaymentsToDetails($workflowDetails, $workflowPayments); + foreach ($workflowDetails as $detail) { + $details[$businessId . '|' . $detail['line_no']] = $detail; + } + foreach ($workflowPayments as $payment) { + $payments[$businessId . '|' . $payment['payment_no']] = $payment; + } + + if ($detailTableName !== '' && ! array_key_exists($detailTableName, $fields) + && $workflowAmountCents > 0 && $title !== '') { + $issues[] = $businessId . ' 缺少“' . $detailTableName . '”'; + } + } + + return [ + 'details' => array_values($details), + 'payments' => array_values($payments), + 'process_instance_ids' => array_values(array_unique($processInstanceIds)), + ]; + } + + /** + * @param array $workflow + * @param array $process + * @param array $fields + * @param list $issues + * @return array{0:list>,1:int,2:string} + */ + private function detailRows( + array $workflow, + array $process, + array $fields, + string $scope, + Carbon $now, + array &$issues, + string $originator, + string $documentType, + string $status, + string $businessMonth, + ): array { + $tableName = trim((string) ($process['detail_table'] ?? '')); + $rows = $tableName !== '' ? $this->tableRows($fields[$tableName] ?? null, $workflow, $tableName, $issues) : []; + if ($tableName === '' && (bool) ($process['scalar_detail'] ?? false)) { + $rows = [$this->scalarDetailRow($fields)]; + } + + $details = []; + $amountCents = 0; + foreach ($rows as $index => $row) { + $amount = $this->moneyFromRow( + $row, + (array) ($process['detail_amount_fields'] ?? []), + ); + if ($amount !== null) { + $amountCents += $amount; + } + $details[] = $this->detailRecord( + $workflow, + $process, + $scope, + $now, + $index + 1, + $amount, + $this->purpose($row) ?: (string) ($workflow['title'] ?? ''), + (string) ($row['科目编码'] ?? ''), + (string) ($row['科目名称'] ?? ''), + $originator, + $documentType, + $status, + $businessMonth, + ); + } + + return [$details, $amountCents, $tableName]; + } + + /** + * @param array $workflow + * @param array $process + * @param array $fields + * @param list $issues + * @return list> + */ + private function paymentRows( + array $workflow, + array $process, + array $fields, + string $scope, + Carbon $now, + array &$issues, + ): array { + $paymentTable = trim((string) ($process['payment_table'] ?? '')); + $rows = $paymentTable !== '' + ? $this->tableRows($fields[$paymentTable] ?? null, $workflow, $paymentTable, $issues) + : []; + $payments = []; + + foreach ($rows as $row) { + $amountCents = $this->moneyFromRow($row, ['付款金额', '付款总金额']); + if ($amountCents === null) { + $issues[] = (string) ($workflow['businessId'] ?? '') . ' 的付款明细存在空金额'; + continue; + } + + $payments[] = [ + 'business_id' => mb_substr(trim((string) ($workflow['businessId'] ?? '')), 0, 32), + 'process_instance_id' => mb_substr( + trim((string) ($workflow['processInstanceId'] ?? '')), + 0, + 100, + ), + 'payment_no' => count($payments) + 1, + 'sequence_spec' => mb_substr(trim((string) ($row['对应序号'] ?? '')), 0, 255), + 'amount' => $this->moneyString($amountCents), + 'amount_cents' => $amountCents, + 'account_code' => mb_substr(trim((string) ($row['科目编码'] ?? '')), 0, 64), + 'account_name' => mb_substr(trim((string) ($row['科目名称'] ?? '')), 0, 255), + 'payment_method' => mb_substr(trim((string) ($row['付款方式'] ?? '')), 0, 50), + 'paid_at' => mb_substr(trim((string) ($row['打款时间'] ?? '')), 0, 50), + 'source_scope' => $scope, + 'synced_at' => $now, + 'created_at' => $now, + 'updated_at' => $now, + ]; + } + + $declaredCents = $this->firstFieldMoneyCents( + $fields, + (array) ($process['payment_total_fields'] ?? []), + ); + $detailPaymentCents = array_sum(array_column($payments, 'amount_cents')); + + if ($declaredCents !== null) { + if ($payments === [] && $declaredCents > 0) { + $payments[] = [ + 'business_id' => mb_substr(trim((string) ($workflow['businessId'] ?? '')), 0, 32), + 'process_instance_id' => mb_substr( + trim((string) ($workflow['processInstanceId'] ?? '')), + 0, + 100, + ), + 'payment_no' => 1, + 'sequence_spec' => '', + 'amount' => $this->moneyString($declaredCents), + 'amount_cents' => $declaredCents, + 'account_code' => '', + 'account_name' => mb_substr( + trim((string) $this->scalarValue($fields['科目名称'] ?? '')), + 0, + 255, + ), + 'payment_method' => mb_substr( + trim((string) $this->scalarValue( + $fields['支付方式'] ?? $fields['付款方式'] ?? '', + )), + 0, + 50, + ), + 'paid_at' => mb_substr( + trim((string) $this->scalarValue($fields['打款时间'] ?? '')), + 0, + 50, + ), + 'source_scope' => $scope, + 'synced_at' => $now, + 'created_at' => $now, + 'updated_at' => $now, + ]; + } elseif ($payments !== [] && abs($detailPaymentCents - $declaredCents) > 1) { + $issues[] = sprintf( + '%s 的付款明细合计 %.2f 与付款合计 %.2f 不一致', + (string) ($workflow['businessId'] ?? ''), + $detailPaymentCents / 100, + $declaredCents / 100, + ); + } + } + + return $payments; + } + + /** + * @param string $businessId + * @param array $process + * @param array $fields + * @param list $issues + */ + private function validateAmountTotal( + string $businessId, + array $process, + array $fields, + int $detailAmountCents, + array &$issues, + ): void { + $declaredCents = $this->firstFieldMoneyCents( + $fields, + (array) ($process['amount_total_fields'] ?? []), + ); + if ($declaredCents !== null && abs($detailAmountCents - $declaredCents) > 1) { + $issues[] = sprintf( + '%s 的明细合计 %.2f 与审批合计 %.2f 不一致', + $businessId, + $detailAmountCents / 100, + $declaredCents / 100, + ); + } + } + + /** + * @param list> $details + * @param list> $payments + */ + private function attachPaymentsToDetails(array &$details, array $payments): void + { + if ($details === []) { + return; + } + + $lineNumbers = array_map( + static fn (array $detail): int => (int) $detail['line_no'], + $details, + ); + $firstLine = min($lineNumbers); + $grouped = []; + foreach ($payments as $payment) { + preg_match_all('/\d+/', (string) ($payment['sequence_spec'] ?? ''), $matches); + $tokens = array_values(array_unique(array_map('intval', $matches[0] ?? []))); + $target = $firstLine; + foreach ($tokens as $token) { + if (in_array($token, $lineNumbers, true)) { + $target = $token; + break; + } + } + $grouped[$target][] = $payment; + } + + foreach ($details as &$detail) { + $group = $grouped[(int) $detail['line_no']] ?? []; + if ($group === []) { + continue; + } + + $detail['bank_amount'] = $this->moneyString((int) array_sum( + array_column($group, 'amount_cents'), + )); + $detail['bank_account_code'] = $this->joinUnique($group, 'account_code', 255); + $detail['bank_account_name'] = $this->joinUnique($group, 'account_name', 255); + $detail['payment_method'] = $this->joinUnique($group, 'payment_method', 50); + $detail['paid_at'] = $this->joinUnique($group, 'paid_at', 50); + } + unset($detail); + } + + /** + * @param array $workflow + * @param array $process + * @return array + */ + private function detailRecord( + array $workflow, + array $process, + string $scope, + Carbon $now, + int $lineNo, + ?int $amountCents, + string $purpose, + string $accountCode, + string $accountName, + string $originator, + string $documentType, + string $status, + string $businessMonth, + ): array { + return [ + 'business_id' => mb_substr(trim((string) ($workflow['businessId'] ?? '')), 0, 32), + 'process_instance_id' => mb_substr( + trim((string) ($workflow['processInstanceId'] ?? '')), + 0, + 100, + ), + 'line_no' => $lineNo, + 'business_month' => $businessMonth, + 'originator_name' => mb_substr($originator, 0, 100), + 'status' => mb_substr($status, 0, 20), + 'document_type' => mb_substr($documentType, 0, 100), + 'purpose' => $purpose, + 'account_code' => mb_substr($accountCode, 0, 64), + 'account_name' => mb_substr($accountName, 0, 255), + 'amount' => $amountCents === null ? null : $this->moneyString($amountCents), + 'amount_cents' => $amountCents ?? 0, + 'bank_account_code' => null, + 'bank_account_name' => null, + 'bank_amount' => null, + 'payment_method' => null, + 'paid_at' => null, + 'source_scope' => $scope, + 'synced_at' => $now, + 'created_at' => $now, + 'updated_at' => $now, + ]; + } + + /** + * @param mixed $value + * @param array $workflow + * @param list $issues + * @return list> + */ + private function tableRows(mixed $value, array $workflow, string $label, array &$issues): array + { + if ($value === null || $value === '' || $value === '[]') { + return []; + } + + if (is_string($value)) { + $value = json_decode($value, true); + } + if (! is_array($value)) { + $issues[] = (string) ($workflow['businessId'] ?? '') . ' 的“' . $label . '”不是有效表格'; + + return []; + } + + $rows = []; + foreach ($value as $rawRow) { + if (! is_array($rawRow)) { + continue; + } + $row = []; + foreach ((array) ($rawRow['rowValue'] ?? []) as $cell) { + if (! is_array($cell)) { + continue; + } + $name = trim((string) ($cell['label'] ?? '')); + if ($name === '') { + continue; + } + $cellValue = $cell['value'] ?? null; + if (($cellValue === null || $cellValue === '')) { + $extend = $cell['extendValue'] ?? null; + if (is_string($extend)) { + $decoded = json_decode($extend, true); + $extend = is_array($decoded) ? $decoded : null; + } + if (is_array($extend)) { + $cellValue = $extend['label'] ?? null; + } + } + $row[$name] = trim((string) ($cellValue ?? '')); + } + $rows[] = $row; + } + + return $rows; + } + + /** @param array $fields */ + private function scalarDetailRow(array $fields): array + { + $row = []; + foreach ($fields as $name => $value) { + $scalar = $this->scalarValue($value); + if ($scalar !== '') { + $row[(string) $name] = $scalar; + } + } + + return $row; + } + + private function scalarValue(mixed $value): string + { + if (is_scalar($value)) { + return trim((string) $value); + } + if (is_array($value) && array_key_exists('label', $value)) { + return trim((string) ($value['label'] ?? '')); + } + + return ''; + } + + /** + * @param array $row + * @param list $fields + */ + private function moneyFromRow(array $row, array $fields): ?int + { + foreach ($fields as $field) { + if (array_key_exists($field, $row) && trim((string) $row[$field]) !== '') { + return $this->parseMoneyCents($row[$field]); + } + } + + return null; + } + + /** + * @param array $fields + * @param list $names + */ + private function moneyFromFields(array $fields, array $names): ?int + { + return $this->firstFieldMoneyCents($fields, $names); + } + + /** + * @param array $fields + * @param list $names + */ + private function firstFieldMoneyCents(array $fields, array $names): ?int + { + foreach ($names as $name) { + if (! array_key_exists($name, $fields)) { + continue; + } + $value = $this->scalarValue($fields[$name]); + if ($value !== '') { + return $this->parseMoneyCents($value); + } + } + + return null; + } + + private function parseMoneyCents(mixed $value): ?int + { + $value = str_replace([',', '，', '¥', '￥'], '', trim((string) $value)); + if ($value === '') { + return null; + } + if (! is_numeric($value)) { + throw new RuntimeException('钉钉审批金额不是有效数字：' . $value); + } + + return (int) round((float) $value * 100); + } + + private function moneyString(int $cents): string + { + return number_format($cents / 100, 2, '.', ''); + } + + /** @param array $row */ + private function purpose(array $row): string + { + foreach (['用途', '商品名称', '固定资产名称', '规格名称', 'SPU', '条形码'] as $field) { + $value = trim((string) ($row[$field] ?? '')); + if ($value !== '') { + return $value; + } + } + + return ''; + } + + /** @param array $process */ + private function documentType(array $process, string $title): string + { + if (($process['kind'] ?? '') === 'daily') { + foreach ([ + '公摊借支-销账' => '公摊借支-销账', + '店铺借支-销账' => '店铺借支-销账', + '公摊日常报销' => '公摊日常报销', + '店铺日常报销' => '店铺日常报销', + ] as $needle => $label) { + if (str_contains($title, $needle)) { + return $label; + } + } + } + + return trim((string) ($process['document_type'] ?? '')) ?: '其他'; + } + + private function status(string $status): string + { + $status = strtoupper(trim($status)); + + return self::STATUS_MAP[$status] ?? ($status !== '' ? $status : '未知'); + } + + private function businessMonth(string $businessId, string $joinAt): string + { + if (preg_match('/^(\d{4})(\d{2})/', $businessId, $matches) === 1) { + $month = (int) $matches[2]; + if ($month >= 1 && $month <= 12) { + return $matches[1] . $matches[2]; + } + } + $timestamp = strtotime($joinAt); + + return $timestamp === false ? '' : date('Ym', $timestamp); + } + + private function originatorFromTitle(string $title): string + { + $title = preg_replace('/^\s*\d{1,2}-\d{1,2}\+?\s*/u', '', $title) ?? $title; + if (preg_match('/^(.+?)(?:在(?:\d{1,2}-\d{1,2})?)?提交(?:的)?/u', $title, $matches) === 1) { + return trim($matches[1]); + } + + return ''; + } + + /** + * @param list> $rows + */ + private function joinUnique(array $rows, string $field, int $limit): string + { + $values = []; + foreach ($rows as $row) { + $value = trim((string) ($row[$field] ?? '')); + if ($value !== '' && ! in_array($value, $values, true)) { + $values[] = $value; + } + } + + return mb_substr(implode(' | ', $values), 0, $limit); + } + + /** + * @param list> $details + * @param list> $payments + * @param list $processInstanceIds + * @param array $metrics + */ + private function publish( + array $details, + array $payments, + array $processInstanceIds, + bool $full, + string $scope, + array $metrics, + Carbon $now, + ): void { + $previous = $this->state(); + $details = array_map(function (array $row): array { + unset($row['amount_cents']); + + return $row; + }, $details); + $payments = array_map(function (array $row): array { + unset($row['amount_cents']); + + return $row; + }, $payments); + + $this->connection->transaction(function () use ( + $details, + $payments, + $processInstanceIds, + $full, + $scope, + $metrics, + $now, + $previous, + ): void { + if ($full) { + $this->connection->table('ding_process_payment_mirror') + ->where('source_scope', $scope) + ->delete(); + $this->connection->table('ding_process_detail_mirror') + ->where('source_scope', $scope) + ->delete(); + } elseif ($processInstanceIds !== []) { + foreach (array_chunk($processInstanceIds, 500) as $chunk) { + $this->connection->table('ding_process_payment_mirror') + ->where('source_scope', $scope) + ->whereIn('process_instance_id', $chunk) + ->delete(); + $this->connection->table('ding_process_detail_mirror') + ->where('source_scope', $scope) + ->whereIn('process_instance_id', $chunk) + ->delete(); + } + } + + foreach (array_chunk($details, 500) as $chunk) { + $this->connection->table('ding_process_detail_mirror')->insert($chunk); + } + foreach (array_chunk($payments, 500) as $chunk) { + $this->connection->table('ding_process_payment_mirror')->insert($chunk); + } + + $detailCount = (int) $this->connection->table('ding_process_detail_mirror') + ->where('source_scope', $scope) + ->count(); + $paymentCount = (int) $this->connection->table('ding_process_payment_mirror') + ->where('source_scope', $scope) + ->count(); + $amountTotal = (float) ($this->connection->table('ding_process_detail_mirror') + ->where('source_scope', $scope) + ->sum('amount') ?? 0); + $bankAmountTotal = (float) ($this->connection->table('ding_process_payment_mirror') + ->where('source_scope', $scope) + ->sum('amount') ?? 0); + + $values = [ + 'source_key' => self::SOURCE, + 'source_object' => 'ding_Workflow + ding_WorkflowScheme structured details', + 'target_table' => 'ding_process_detail_mirror', + 'status' => 'succeeded', + 'run_mode' => $full ? 'full' : 'incremental', + 'generation' => (string) Str::uuid(), + 'cursor' => json_encode([ + 'source' => 'structured', + 'synced_workflows' => $metrics['workflows'], + 'synced_details' => $metrics['details'], + 'synced_payments' => $metrics['payments'], + 'mirror_details' => $detailCount, + 'mirror_payments' => $paymentCount, + 'amount_total' => round($amountTotal, 2), + 'bank_amount_total' => round($bankAmountTotal, 2), + ], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), + 'high_watermark' => null, + 'window_started_at' => null, + 'window_ends_at' => null, + 'initial_completed_at' => $previous['initial_completed_at'] ?? $now, + 'last_full_succeeded_at' => $full + ? $now + : ($previous['last_full_succeeded_at'] ?? null), + 'last_attempted_at' => $now, + 'last_succeeded_at' => $now, + 'last_upserted_rows' => count($details), + 'mirror_row_count' => $detailCount, + 'last_error_code' => null, + 'updated_at' => $now, + ]; + $this->connection->table('business_mirror_table_states')->updateOrInsert( + ['dataset_key' => $this->datasetKey()], + $values + ['created_at' => $now], + ); + }); + } + + /** @param array|null $state */ + private function markAttempted(?array $state, bool $full, Carbon $now): void + { + $this->connection->table('business_mirror_table_states')->updateOrInsert( + ['dataset_key' => $this->datasetKey()], + [ + 'source_key' => self::SOURCE, + 'source_object' => 'ding_Workflow + ding_WorkflowScheme structured details', + 'target_table' => 'ding_process_detail_mirror', + 'status' => 'running', + 'run_mode' => $full ? 'full' : 'incremental', + 'last_attempted_at' => $now, + 'last_error_code' => null, + 'updated_at' => $now, + 'created_at' => $state['created_at'] ?? $now, + ], + ); + } + + /** @param array|null $state */ + private function markFailed(?array $state, Throwable $exception, Carbon $now): void + { + $this->connection->table('business_mirror_table_states')->updateOrInsert( + ['dataset_key' => $this->datasetKey()], + [ + 'source_key' => self::SOURCE, + 'source_object' => 'ding_Workflow + ding_WorkflowScheme structured details', + 'target_table' => 'ding_process_detail_mirror', + 'status' => 'failed', + 'run_mode' => $state['run_mode'] ?? null, + 'initial_completed_at' => $state['initial_completed_at'] ?? null, + 'last_full_succeeded_at' => $state['last_full_succeeded_at'] ?? null, + 'last_attempted_at' => $now, + 'last_succeeded_at' => $state['last_succeeded_at'] ?? null, + 'last_error_code' => mb_substr( + $exception::class . ':' . $exception->getMessage(), + 0, + 120, + ), + 'updated_at' => $now, + 'created_at' => $state['created_at'] ?? $now, + ], + ); + } + + /** @param array|null $state */ + private function incrementalSyncDue(?array $state): bool + { + if (($state['initial_completed_at'] ?? null) === null + || ($state['last_succeeded_at'] ?? null) === null) { + return true; + } + $minutes = max( + 1, + (int) config('upstream_mirrors.ding_process_details.sync_interval_minutes', 5), + ); + + return Carbon::parse((string) $state['last_succeeded_at']) + ->addMinutes($minutes) + ->lessThanOrEqualTo(now()); + } + + /** @return array|null */ + private function state(): ?array + { + $row = $this->connection->table('business_mirror_table_states') + ->where('dataset_key', $this->datasetKey()) + ->first(); + + return $row === null ? null : (array) $row; + } + + /** @param array|null $state */ + private function structuredState(?array $state): bool + { + if (($state['initial_completed_at'] ?? null) === null) { + return false; + } + $cursor = $state['cursor'] ?? null; + if (is_string($cursor)) { + $cursor = json_decode($cursor, true); + } + + return is_array($cursor) && ($cursor['source'] ?? '') === 'structured'; + } + + private function datasetKey(): string + { + return self::DATASET_PREFIX . $this->scope(); + } + + private function scope(): string + { + $scope = trim((string) config( + 'upstream_mirrors.ding_process_details.scope', + 'personal_permission', + )); + + return $scope !== '' ? $scope : 'personal_permission'; + } +} diff --git a/resources/views/business/pages/financial-operations/process-details-content.blade.php b/resources/views/business/pages/financial-operations/process-details-content.blade.php index 47832fa5..f195fdc2 100644 --- a/resources/views/business/pages/financial-operations/process-details-content.blade.php +++ b/resources/views/business/pages/financial-operations/process-details-content.blade.php @@ -564,6 +564,7 @@

+ 金额合计按审批费用明细汇总，银行付款合计按结构化付款明细汇总，同一笔付款不会因关联多条费用而重复计算。 当前展示的是当前账号权限范围内可读取的钉钉审批流程明细；审批中的金额和行项目仍可能随流程变化。 企业级全量数据待领导开通 OA 权限后切换。

diff --git a/tests/Feature/Business/ProcessDetailsPageTest.php b/tests/Feature/Business/ProcessDetailsPageTest.php index a4ba8c27..508a945a 100644 --- a/tests/Feature/Business/ProcessDetailsPageTest.php +++ b/tests/Feature/Business/ProcessDetailsPageTest.php @@ -74,6 +74,7 @@ class ProcessDetailsPageTest extends FeatureTestCase ->assertSee('管理费用_办公费') ->assertSee('100.00') ->assertSee('当前账号权限范围') + ->assertSee('同一笔付款不会因关联多条费用而重复计算') ->assertSee('导出'); } diff --git a/tests/Unit/EcommerceFinance/DingProcessDetailMirrorSynchronizerTest.php b/tests/Unit/EcommerceFinance/DingProcessDetailMirrorSynchronizerTest.php new file mode 100644 index 00000000..195afbb5 --- /dev/null +++ b/tests/Unit/EcommerceFinance/DingProcessDetailMirrorSynchronizerTest.php @@ -0,0 +1,311 @@ + [ + 'driver' => 'sqlite', + 'database' => ':memory:', + 'prefix' => '', + ], + 'upstream_mirrors.connection' => 'business_mirror', + 'upstream_mirrors.ding_process_details' => [ + 'scope' => 'personal_permission', + 'start_date' => '2026-01-01', + 'batch_size' => 20, + 'sync_interval_minutes' => 5, + 'incremental_overlap_hours' => 48, + 'processes' => [ + 'PROC-ACF09A14-B094-46B1-A9E2-8BB1F4E1FDCF' => [ + 'document_type' => '线上代发业务报销', + 'detail_table' => '收款费用明细', + 'detail_amount_fields' => ['代发订单金额'], + 'amount_total_fields' => ['收款合计'], + 'payment_table' => '付款明细', + ], + ], + ], + ]); + DB::purge('business_mirror'); + $this->createMirrorTables(); + + $this->query = new FakeDingProcessDetailQuery(); + $factory = new class($this->query) extends UpstreamReadOnlyQueryFactory { + public function __construct(private readonly UpstreamReadOnlyQuery $query) {} + + public function forSource(string $source): UpstreamReadOnlyQuery + { + return $this->query; + } + + public function forSourceWithTimeout(string $source, int $timeoutSeconds): UpstreamReadOnlyQuery + { + return $this->query; + } + }; + $this->app->instance(UpstreamReadOnlyQueryFactory::class, $factory); + $this->synchronizer = app(DingProcessDetailMirrorSynchronizer::class); + } + + public function test_structured_sync_counts_a_multi_line_payment_once(): void + { + $this->query->workflows = [[ + 'processInstanceId' => 'process-daifa-1', + 'processCode' => 'PROC-ACF09A14-B094-46B1-A9E2-8BB1F4E1FDCF', + 'businessId' => '202605041811000260216', + 'title' => '陈芝豹提交的线上代发业务报销', + 'status' => 'COMPLETED', + 'originatorUserId' => 'user-1', + 'joinAt' => '2026-05-04 18:12:20', + 'updateAt' => '2026-05-08 09:00:00', + ]]; + $this->query->scheme = $this->daifaScheme(); + + $result = $this->synchronizer->sync(true); + + $this->assertSame('succeeded', $result['status']); + $this->assertSame(1, $result['workflows']); + $this->assertSame(14, $result['details']); + $this->assertSame(1, $result['payments']); + $this->assertEqualsWithDelta(1114.57, $result['amount_total'], 0.001); + $this->assertEqualsWithDelta(1114.57, $result['bank_amount_total'], 0.001); + $this->assertSame(14, DB::connection('business_mirror') + ->table('ding_process_detail_mirror')->count()); + $this->assertSame(1, DB::connection('business_mirror') + ->table('ding_process_payment_mirror')->count()); + $this->assertEqualsWithDelta(1114.57, (float) DB::connection('business_mirror') + ->table('ding_process_detail_mirror')->sum('bank_amount'), 0.001); + + $summary = app(DingProcessDetailMirrorRepository::class) + ->paginate(['status' => '已完成'], 1, 50)['summary']; + $this->assertEqualsWithDelta(1114.57, $summary['amount_total'], 0.001); + $this->assertEqualsWithDelta(1114.57, $summary['bank_amount_total'], 0.001); + } + + public function test_validation_failure_keeps_the_previous_published_rows(): void + { + $this->query->workflows = [[ + 'processInstanceId' => 'process-old', + 'processCode' => 'PROC-ACF09A14-B094-46B1-A9E2-8BB1F4E1FDCF', + 'businessId' => '202605041811000260216', + 'title' => '陈芝豹提交的线上代发业务报销', + 'status' => 'COMPLETED', + 'originatorUserId' => 'user-1', + 'joinAt' => '2026-05-04 18:12:20', + 'updateAt' => '2026-05-08 09:00:00', + ]]; + $this->query->scheme = $this->daifaScheme(); + $this->synchronizer->sync(true); + + $this->query->scheme['收款合计'] = 999999; + $this->query->workflows = [[ + 'processInstanceId' => 'process-new', + 'processCode' => 'PROC-ACF09A14-B094-46B1-A9E2-8BB1F4E1FDCF', + 'businessId' => '202605051811000260217', + 'title' => '陈芝豹提交的线上代发业务报销', + 'status' => 'COMPLETED', + 'originatorUserId' => 'user-1', + 'joinAt' => '2026-05-05 18:12:20', + 'updateAt' => '2026-05-09 09:00:00', + ]]; + + try { + $this->synchronizer->sync(true); + $this->fail('明细合计与审批合计不一致时应停止发布。'); + } catch (RuntimeException $exception) { + $this->assertStringContainsString('明细合计', $exception->getMessage()); + } + + $this->assertSame(14, DB::connection('business_mirror') + ->table('ding_process_detail_mirror')->count()); + $this->assertTrue(DB::connection('business_mirror') + ->table('ding_process_detail_mirror') + ->where('business_id', '202605041811000260216') + ->exists()); + $this->assertSame('failed', DB::connection('business_mirror') + ->table('business_mirror_table_states') + ->value('status')); + } + + /** @return array */ + private function daifaScheme(): array + { + $amounts = [157, 153, 56.1, 63.1, 85, 90, 56.1, 63.1, 66.1, 61.67, 66.1, 65.1, 66.1, 66.1]; + $details = array_map(static fn (float $amount): array => [ + 'rowValue' => [ + ['label' => '序号', 'value' => '1'], + ['label' => '商品ID', 'value' => '877714260849'], + ['label' => '代发订单金额', 'value' => (string) $amount], + ['label' => '科目编码', 'value' => '1405.06'], + ['label' => '科目名称', 'value' => '库存商品_线上代发'], + ], + ], $amounts); + $payment = [[ + 'rowValue' => [ + ['label' => '对应序号', 'value' => '1-2-3-4-5-6-7-8-9-10-11-12-13-14'], + ['label' => '付款方式', 'value' => '银行卡'], + ['label' => '付款金额', 'value' => '1114.57'], + ['label' => '科目编码', 'value' => '1002.02.114'], + ['label' => '科目名称', 'value' => '华夏银行-胡燕'], + ['label' => '打款时间', 'value' => '2026-05-08'], + ], + ]]; + + return [ + '收款费用明细' => json_encode($details, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), + '收款合计' => 1114.57, + '付款明细' => json_encode($payment, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), + ]; + } + + private function createMirrorTables(): void + { + Schema::connection('business_mirror')->create( + 'business_mirror_table_states', + function (Blueprint $table): void { + $table->string('dataset_key')->primary(); + $table->string('source_key'); + $table->string('source_object'); + $table->string('target_table'); + $table->string('status')->default('pending'); + $table->string('run_mode')->nullable(); + $table->string('generation')->nullable(); + $table->json('cursor')->nullable(); + $table->json('high_watermark')->nullable(); + $table->dateTime('window_started_at')->nullable(); + $table->dateTime('window_ends_at')->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')->nullable(); + $table->timestamps(); + }, + ); + + Schema::connection('business_mirror')->create( + 'ding_process_detail_mirror', + function (Blueprint $table): void { + $table->bigIncrements('id'); + $table->string('business_id', 32); + $table->string('process_instance_id', 100)->nullable(); + $table->unsignedInteger('line_no'); + $table->string('business_month', 6); + $table->string('originator_name', 100)->nullable(); + $table->string('status', 20)->nullable(); + $table->string('document_type', 100)->nullable(); + $table->text('purpose')->nullable(); + $table->string('account_code', 64)->nullable(); + $table->string('account_name', 255)->nullable(); + $table->decimal('amount', 18, 2)->nullable(); + $table->string('bank_account_code', 255)->nullable(); + $table->string('bank_account_name', 255)->nullable(); + $table->decimal('bank_amount', 18, 2)->nullable(); + $table->string('payment_method', 50)->nullable(); + $table->string('paid_at', 50)->nullable(); + $table->string('source_scope', 40); + $table->dateTime('synced_at', 6); + $table->timestamps(); + $table->unique(['business_id', 'line_no']); + }, + ); + + Schema::connection('business_mirror')->create( + 'ding_process_payment_mirror', + function (Blueprint $table): void { + $table->bigIncrements('id'); + $table->string('business_id', 32); + $table->string('process_instance_id', 100); + $table->unsignedInteger('payment_no'); + $table->string('sequence_spec', 255)->nullable(); + $table->decimal('amount', 18, 2); + $table->string('account_code', 64)->nullable(); + $table->string('account_name', 255)->nullable(); + $table->string('payment_method', 50)->nullable(); + $table->string('paid_at', 50)->nullable(); + $table->string('source_scope', 40); + $table->dateTime('synced_at', 6); + $table->timestamps(); + $table->unique(['source_scope', 'business_id', 'payment_no']); + }, + ); + } +} + +class FakeDingProcessDetailQuery extends UpstreamReadOnlyQuery +{ + /** @var list> */ + public array $workflows = []; + + /** @var array */ + public array $scheme = []; + + /** @var list */ + public array $calls = []; + + public function __construct() + { + parent::__construct('operations_replica'); + } + + public function scalar(string $sql, array $bindings = []): mixed + { + if (str_contains($sql, 'COUNT(*)') && str_contains($sql, 'rpt.ding_Workflow')) { + return count($this->workflows); + } + + return parent::scalar($sql, $bindings); + } + + public function select(string $sql, array $bindings = []): array + { + $this->calls[] = compact('sql', 'bindings'); + + if (str_contains($sql, 'FROM rpt.hr_employee')) { + return [['userid' => 'user-1', 'name' => '陈芝豹']]; + } + if (str_contains($sql, 'JOIN rpt.ding_WorkflowScheme') + && str_contains($sql, 'NOT IN')) { + return []; + } + if (str_contains($sql, 'FROM rpt.ding_Workflow w')) { + return $this->workflows; + } + if (str_contains($sql, 'FROM rpt.ding_WorkflowScheme')) { + $rows = []; + foreach ($this->scheme as $name => $value) { + $rows[] = [ + 'processInstanceId' => (string) ($this->workflows[0]['processInstanceId'] ?? ''), + 'name' => $name, + 'value' => $value, + ]; + } + + return $rows; + } + + return []; + } +}