'已完成', '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 { $schemeHighWatermark = $this->schemeHighWatermark(); $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'], )); } $result['metrics']['scheme_high_watermark'] = $schemeHighWatermark; 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 { $schemeHighWatermark = $this->schemeHighWatermark(); $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]; $cursor = $this->stateCursor($state); $lastSchemeId = max(0, (int) ($cursor['scheme_high_watermark'] ?? 0)); $result = $this->extractWorkflows( $where, $bindings, $cursorExpression, 'sync_cursor', $now, ); if ($lastSchemeId > 0) { $changedProcessIds = $this->schemeChangedProcessIds( $lastSchemeId, $schemeHighWatermark, ); if ($changedProcessIds !== []) { foreach (array_chunk($changedProcessIds, 500) as $chunk) { $placeholders = implode(',', array_fill(0, count($chunk), '?')); $schemeResult = $this->extractWorkflows( "w.processInstanceId IN ({$placeholders})", $chunk, 'w.processInstanceId', 'processInstanceId', $now, ); $result = $this->mergeExtracted($result, $schemeResult); } } } $result['metrics']['scheme_high_watermark'] = $schemeHighWatermark; return $result; } /** * New scheme rows can be appended after the workflow itself stopped * changing. Track their numeric ID so late payment/details are discovered * without rescanning every workflow. * * @return list */ private function schemeChangedProcessIds(int $afterId, int $throughId): array { if ($throughId <= $afterId) { return []; } $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), ); $rows = $query->select( 'SELECT DISTINCT s.processInstanceId ' . 'FROM rpt.ding_WorkflowScheme s ' . 'JOIN rpt.ding_Workflow w ON w.processInstanceId = s.processInstanceId ' . 'WHERE TRY_CONVERT(bigint, s.id) > ? ' . 'AND TRY_CONVERT(bigint, s.id) <= ? ' . "AND w.processCode IN ({$placeholders}) " . 'ORDER BY s.processInstanceId', [$afterId, $throughId, ...$processCodes], ); return array_values(array_filter(array_map( static fn (array $row): string => trim((string) ($row['processInstanceId'] ?? '')), $rows, ))); } private function schemeHighWatermark(): int { $query = $this->queries->forSourceWithTimeout( self::SOURCE, (int) config('finance_ledger.query_timeout', 120), ); return max(0, (int) $query->scalar( 'SELECT MAX(TRY_CONVERT(bigint, id)) FROM rpt.ding_WorkflowScheme', )); } /** @param array|null $state @return array */ private function stateCursor(?array $state): array { $cursor = $state['cursor'] ?? null; if (is_string($cursor)) { $cursor = json_decode($cursor, true); } return is_array($cursor) ? $cursor : []; } /** * @param array{details:list>,payments:list>,process_instance_ids:list,metrics:array} $first * @param array{details:list>,payments:list>,process_instance_ids:list,metrics:array} $second * @return array{details:list>,payments:list>,process_instance_ids:list,metrics:array} */ private function mergeExtracted(array $first, array $second): array { if ($second['process_instance_ids'] === []) { return $first; } $details = []; foreach ([...$first['details'], ...$second['details']] as $row) { $key = (string) ($row['business_id'] ?? '') . '|' . (int) ($row['line_no'] ?? 0); $details[$key] = $row; } $payments = []; foreach ([...$first['payments'], ...$second['payments']] as $row) { $key = (string) ($row['business_id'] ?? '') . '|' . (int) ($row['payment_no'] ?? 0); $payments[$key] = $row; } $details = array_values($details); $payments = array_values($payments); $processInstanceIds = array_values(array_unique([ ...$first['process_instance_ids'], ...$second['process_instance_ids'], ])); return [ 'details' => $details, 'payments' => $payments, 'process_instance_ids' => $processInstanceIds, 'metrics' => [ 'workflows' => count($processInstanceIds), 'details' => count($details), 'payments' => count($payments), 'amount_total' => array_sum(array_column($details, 'amount_cents')) / 100, 'bank_amount_total' => array_sum(array_column($payments, 'amount_cents')) / 100, ], ]; } /** * @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'] ?? '')); $paymentSourceState = $this->paymentSourceState($process, $fields); [$workflowDetails, $workflowAmountCents, $detailTableName] = $this->detailRows( $workflow, $process, $fields, $scope, $now, $issues, $originator, $documentType, $status, $businessMonth, $paymentSourceState, ); $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'] ?? [])); $workflowDetails[] = $this->detailRecord( $workflow, $process, $scope, $now, 1, $amount, $title, '', '', $originator, $documentType, $status, $businessMonth, $paymentSourceState, ); } $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, string $paymentSourceState, ): 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, $paymentSourceState, ); } 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 = []; $declaredCents = $this->firstFieldMoneyCents( $fields, (array) ($process['payment_total_fields'] ?? []), ); foreach ($rows as $row) { $amountCents = $this->moneyFromRow($row, ['付款金额', '付款总金额']); if ($amountCents === null) { if (trim((string) ($row['对应序号'] ?? '')) === '' && trim((string) ($row['科目编码'] ?? '')) === '' && trim((string) ($row['科目名称'] ?? '')) === '') { continue; } $issues[] = (string) ($workflow['businessId'] ?? '') . ' 的付款明细存在空金额'; continue; } $accountCode = trim((string) ($row['科目编码'] ?? '')); if ($accountCode !== '' && ! preg_match('/^1(?:001|002|012|013)(?:\.|$)/', $accountCode)) { 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($accountCode, 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, ]; } $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; } /** * Distinguish an empty payment table from a payment form section that the * upstream replica has not delivered yet. Missing sections must never be * presented as "未付款". * * @param array $process * @param array $fields */ private function paymentSourceState(array $process, array $fields): string { $paymentTable = trim((string) ($process['payment_table'] ?? '')); $paymentTotalFields = array_values(array_filter(array_map( static fn (mixed $value): string => trim((string) $value), (array) ($process['payment_total_fields'] ?? []), ))); if ($paymentTable === '' && $paymentTotalFields === []) { return 'not_applicable'; } if ($paymentTable !== '' && array_key_exists($paymentTable, $fields)) { return 'present'; } foreach ($paymentTotalFields as $field) { if (array_key_exists($field, $fields)) { return 'present'; } } return 'missing'; } /** * @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, string $paymentSourceState, ): 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), 'payment_source_state' => mb_substr($paymentSourceState, 0, 24), '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), 'scheme_high_watermark' => (int) ($metrics['scheme_high_watermark'] ?? 0), ], 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'; } }