> */ private array $approvalSnapshots = []; public function __construct( DatabaseManager $database, private readonly UpstreamReadOnlyQueryFactory $queries, private readonly DingTalkProcessInstanceClient $approvalClient, ) { $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 { $this->approvalApiAttempts = 0; $this->approvalApiAvailable = true; $this->approvalSnapshots = []; $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; } } /** * Rebuild selected approvals from the source replica and fill missing * payment components from DingTalk's current process-instance API. * * @param list $businessIds * @return array{ * status:string, * run_mode:string, * workflows:int, * details:int, * payments:int, * amount_total:float, * bank_amount_total:float * } */ public function syncBusinessIds(array $businessIds, bool $dryRun = false): array { $businessIds = array_values(array_unique(array_filter(array_map( static fn (mixed $value): string => trim((string) $value), $businessIds, )))); if ($businessIds === []) { throw new RuntimeException('至少需要一个审批编号。'); } if (count($businessIds) > self::BUSINESS_ID_SYNC_LIMIT) { throw new RuntimeException('单次最多按审批编号补取 50 个流程。'); } $this->approvalApiAttempts = 0; $this->approvalApiAvailable = true; $this->approvalSnapshots = []; $scope = $this->scope(); $state = $this->state(); if (! $this->structuredState($state)) { return [ 'status' => 'requires_full_sync', 'run_mode' => 'full', 'workflows' => 0, 'details' => 0, 'payments' => 0, 'amount_total' => 0.0, 'bank_amount_total' => 0.0, ]; } $now = now(); if (! $dryRun) { $this->markAttempted($state, false, $now); } try { $placeholders = implode(',', array_fill(0, count($businessIds), '?')); $extracted = $this->extractWorkflows( "w.businessId IN ({$placeholders})", $businessIds, 'w.businessId', 'businessId', $now, ); $cursor = $this->stateCursor($state); $extracted['metrics']['approval_api_cursor'] = (string) ( $cursor['approval_api_cursor'] ?? '' ); $extracted['metrics']['scheme_high_watermark'] = (int) ( $cursor['scheme_high_watermark'] ?? 0 ); if ($dryRun) { return [ 'status' => 'validated', 'run_mode' => 'business_ids', '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'], false, $scope, $extracted['metrics'], $now, ); return [ 'status' => 'succeeded', 'run_mode' => 'business_ids', '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)); $apiCursor = (string) ($cursor['approval_api_cursor'] ?? ''); // 先消化历史待核实流程,避免本轮变更流程耗尽接口额度后补查游标永远不前进。 $backlogIds = $this->approvalApiFallbackCandidates( $apiCursor, [], self::APPROVAL_API_BACKLOG_BATCH_LIMIT, ); $result = $backlogIds === [] ? null : $this->extractApprovalApiWorkflows($backlogIds, $now); if ($backlogIds !== []) { $apiCursor = (string) $backlogIds[array_key_last($backlogIds)]; } $incremental = $this->extractWorkflows( $where, $bindings, $cursorExpression, 'sync_cursor', $now, ); $result = $result === null ? $incremental : $this->mergeExtracted($result, $incremental); 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); } } } $fallbackIds = $this->approvalApiFallbackCandidates( $apiCursor, $result['process_instance_ids'], ); if ($fallbackIds !== []) { $result = $this->mergeExtracted( $result, $this->extractApprovalApiWorkflows($fallbackIds, $now), ); $apiCursor = (string) $fallbackIds[array_key_last($fallbackIds)]; } $result['metrics']['scheme_high_watermark'] = $schemeHighWatermark; $result['metrics']['approval_api_cursor'] = $apiCursor; return $result; } /** * @param list $processInstanceIds * @return array{ * details:list>, * payments:list>, * process_instance_ids:list, * metrics:array * } */ private function extractApprovalApiWorkflows(array $processInstanceIds, Carbon $now): array { $result = [ 'details' => [], 'payments' => [], 'process_instance_ids' => [], 'metrics' => [ 'workflows' => 0, 'details' => 0, 'payments' => 0, 'amount_total' => 0.0, 'bank_amount_total' => 0.0, ], ]; foreach (array_chunk($processInstanceIds, 25) as $chunk) { $placeholders = implode(',', array_fill(0, count($chunk), '?')); $batch = $this->extractWorkflows( "w.processInstanceId IN ({$placeholders})", $chunk, 'w.processInstanceId', 'processInstanceId', $now, ); $result = $this->mergeExtracted($result, $batch); } return $result; } /** * Rotate through approvals whose payment component is absent from the * replica. The cursor prevents one repeatedly failing approval from * blocking the rest of the backlog. * * @param list $excludedProcessInstanceIds * @param int|null $maxCandidates * @return list */ private function approvalApiFallbackCandidates( string $afterProcessInstanceId, array $excludedProcessInstanceIds, ?int $maxCandidates = null, ): array { if (! $this->approvalClient->isConfigured() || ! Schema::connection($this->connection->getName())->hasColumn( 'ding_process_detail_mirror', 'payment_source_state', )) { return []; } $cashierRefreshEnabled = Schema::connection($this->connection->getName())->hasColumn( 'ding_process_detail_mirror', 'cashier_checked_at', ); $approverRefreshEnabled = Schema::connection($this->connection->getName())->hasColumn( 'ding_process_detail_mirror', 'current_approver_name', ); $remainingAttempts = self::APPROVAL_API_DRAIN_LIMIT - $this->approvalApiAttempts; if (! $this->approvalApiAvailable || $remainingAttempts <= 0) { return []; } $candidateLimit = $maxCandidates === null ? $remainingAttempts : min($remainingAttempts, max(0, $maxCandidates)); if ($candidateLimit <= 0) { return []; } $cashierStaleBefore = now()->subMinutes(15); $base = $this->connection->table('ding_process_detail_mirror') ->where('source_scope', $this->scope()) ->whereIn('status', ['审批中', '已完成']) ->where(function ($query) use ( $cashierRefreshEnabled, $approverRefreshEnabled, $cashierStaleBefore, ): void { $query->where('payment_source_state', 'missing'); if ($cashierRefreshEnabled) { $query->orWhere(function ($cashier) use ($cashierStaleBefore): void { $cashier->where('status', '审批中') ->where(function ($checked) use ($cashierStaleBefore): void { $checked->whereNull('cashier_checked_at') ->orWhere('cashier_checked_at', '<', $cashierStaleBefore); }); }); } if ($approverRefreshEnabled) { $query->orWhere(function ($approver): void { $approver->where('status', '审批中') ->where(function ($name): void { $name->whereNull('current_approver_name') ->orWhere('current_approver_name', ''); }); }); } }) ->whereNotNull('process_instance_id') ->where('process_instance_id', '!=', ''); $excluded = array_fill_keys($excludedProcessInstanceIds, true); $ids = (clone $base) ->when( $afterProcessInstanceId !== '', fn ($query) => $query->where('process_instance_id', '>', $afterProcessInstanceId), ) ->distinct() ->orderBy('process_instance_id') ->limit($candidateLimit * 2) ->pluck('process_instance_id') ->map(static fn (mixed $value): string => trim((string) $value)) ->filter() ->values() ->all(); $ids = array_values(array_filter( $ids, static fn (string $id): bool => ! isset($excluded[$id]), )); if ($ids === [] && $afterProcessInstanceId !== '') { $ids = (clone $base) ->distinct() ->orderBy('process_instance_id') ->limit($candidateLimit * 2) ->pluck('process_instance_id') ->map(static fn (mixed $value): string => trim((string) $value)) ->filter() ->values() ->all(); $ids = array_values(array_filter( $ids, static fn (string $id): bool => ! isset($excluded[$id]), )); } return array_slice($ids, 0, $candidateLimit); } /** * 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'], ]; } $orderBy = $cursorExpression === 'w.processInstanceId' ? 'w.processInstanceId' : "{$cursorExpression}, w.processInstanceId"; $workflows = $query->select( 'SELECT w.processInstanceId, w.processCode, w.businessId, w.title, w.status, w.result, ' . '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 {$orderBy}" . ' 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($workflow); $businessMonth = $this->businessMonth($businessId, (string) ($workflow['joinAt'] ?? '')); $paymentSourceState = $this->paymentSourceState($process, $fields); $needsPaymentApi = $this->needsApprovalApiFallback( $process, $fields, $paymentSourceState, ); $needsCashierApi = $this->needsCashierApi($process, $workflow); $approvalSnapshot = null; if ($needsPaymentApi || $needsCashierApi || $status === '审批中') { $approvalSnapshot = $this->approvalApiSnapshot($workflow); } if ($needsPaymentApi && $approvalSnapshot !== null) { $apiFields = (array) ($approvalSnapshot['form_values'] ?? []); if ($apiFields !== []) { foreach ($apiFields as $fieldName => $fieldValue) { if (! array_key_exists($fieldName, $fields) || trim((string) ($fields[$fieldName] ?? '')) === '') { $fields[$fieldName] = $fieldValue; } } $paymentSourceState = $this->hasApprovalPaymentSource($process, $apiFields) ? 'api' : $this->paymentSourceState($process, $fields); } } $cashierStatus = $this->cashierStatus($process, $workflow, $approvalSnapshot); $cashierCheckedAt = $cashierStatus === 'pending_sync' ? null : $now; $taskState = $this->taskState($approvalSnapshot, $employees); [$workflowDetails, $workflowAmountCents, $detailTableName] = $this->detailRows( $workflow, $process, $fields, $scope, $now, $issues, $originator, $documentType, $status, $businessMonth, $paymentSourceState, $cashierStatus, $cashierCheckedAt, $taskState, ); $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, $cashierStatus, $cashierCheckedAt, $taskState, ); } $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, string $cashierStatus, ?Carbon $cashierCheckedAt, array $taskState, ): 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, $cashierStatus, $cashierCheckedAt, $taskState, ); } 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 array $process * @param array $fields */ private function needsApprovalApiFallback( array $process, array $fields, string $paymentSourceState, ): bool { if ($paymentSourceState === 'missing') { return true; } if ($paymentSourceState !== 'present') { return false; } foreach ($this->paymentSourceFieldNames($process) as $fieldName) { if (trim((string) ($fields[$fieldName] ?? '')) === '') { return true; } } return false; } /** * @param array $workflow * @return array */ private function approvalApiSnapshot(array $workflow): ?array { $processInstanceId = trim((string) ($workflow['processInstanceId'] ?? '')); if ($processInstanceId === '') { return null; } if (isset($this->approvalSnapshots[$processInstanceId])) { return $this->approvalSnapshots[$processInstanceId]; } if (! $this->approvalClient->isConfigured() || ! $this->approvalApiAvailable || $this->approvalApiAttempts >= self::APPROVAL_API_DRAIN_LIMIT) { return null; } $this->approvalApiAttempts++; try { return $this->approvalSnapshots[$processInstanceId] = $this->approvalClient->processSnapshot($processInstanceId); } catch (Throwable $exception) { if ($exception instanceof DingTalkProcessInstanceException && $exception->stopsRun()) { $this->approvalApiAvailable = false; } Log::warning('钉钉官方审批接口读取流程状态失败。', [ 'process_instance_id' => $processInstanceId, 'exception' => $exception::class, ]); return null; } } /** * @param array $process * @param array $workflow */ private function needsCashierApi(array $process, array $workflow): bool { return $this->paymentSourceFieldNames($process) !== [] && strtoupper(trim((string) ($workflow['status'] ?? ''))) === 'RUNNING'; } /** * @param array $process * @param array $workflow * @param array|null $snapshot */ private function cashierStatus(array $process, array $workflow, ?array $snapshot): string { if ($this->paymentSourceFieldNames($process) === []) { return 'not_applicable'; } $workflowStatus = strtoupper(trim((string) ($workflow['status'] ?? ''))); $workflowResult = strtoupper(trim((string) ($workflow['result'] ?? ''))); if (in_array($workflowStatus, ['TERMINATED', 'CANCELED', 'CANCELLED', 'REVOKED'], true) || in_array($workflowResult, ['REFUSE', 'REJECT', 'REJECTED'], true)) { return 'terminated'; } if ($workflowStatus === 'COMPLETED') { return 'completed'; } if ($workflowStatus !== 'RUNNING' || $snapshot === null) { return 'pending_sync'; } $events = []; foreach ((array) ($snapshot['tasks'] ?? []) as $task) { if (! is_array($task) || ! $this->isCashierLabel((string) ($task['taskGroupName'] ?? ''))) { continue; } $status = strtoupper(trim((string) ($task['status'] ?? ''))); $result = strtoupper(trim((string) ($task['result'] ?? ''))); $time = trim((string) ($task['finishTime'] ?? $task['createTime'] ?? '')); if ($status === 'COMPLETED' && $result === 'AGREE') { $events[] = ['time' => $time, 'state' => 'completed', 'priority' => 2]; } elseif ($status === 'RUNNING') { $events[] = ['time' => $time, 'state' => 'pending', 'priority' => 1]; } elseif (in_array($result, ['REDIRECTED', 'REFUSE', 'REJECT'], true) || $status === 'CANCELED') { $events[] = ['time' => $time, 'state' => 'returned', 'priority' => 3]; } } foreach ((array) ($snapshot['operation_records'] ?? []) as $operation) { if (! is_array($operation) || ! $this->isCashierLabel((string) ($operation['showName'] ?? ''))) { continue; } $type = strtoupper(trim((string) ($operation['type'] ?? ''))); $result = strtoupper(trim((string) ($operation['result'] ?? ''))); $time = trim((string) ($operation['date'] ?? '')); if (str_starts_with($type, 'REDIRECT') || in_array($result, ['REDIRECTED', 'REFUSE', 'REJECT'], true)) { $events[] = ['time' => $time, 'state' => 'returned', 'priority' => 3]; } elseif ($type === 'EXECUTE_TASK_NORMAL' && $result === 'AGREE') { $events[] = ['time' => $time, 'state' => 'completed', 'priority' => 2]; } } if ($events === []) { return 'not_started'; } usort($events, static function (array $left, array $right): int { return [$left['time'], $left['priority']] <=> [$right['time'], $right['priority']]; }); return (string) ($events[array_key_last($events)]['state'] ?? 'not_started'); } /** * @param array|null $snapshot * @return array{ * current_task_name:?string, * current_approver_name:?string, * current_task_started_at:?Carbon, * cashier_task_name:?string, * cashier_task_started_at:?Carbon * } */ private function taskState(?array $snapshot, array $employees): array { $state = [ 'current_task_name' => null, 'current_approver_name' => null, 'current_task_started_at' => null, 'cashier_task_name' => null, 'cashier_task_started_at' => null, ]; if ($snapshot === null) { return $state; } $runningNames = []; $runningApprovers = []; $runningStartedAt = null; $cashierTasks = []; foreach ((array) ($snapshot['tasks'] ?? []) as $task) { if (! is_array($task)) { continue; } $name = trim((string) ($task['taskGroupName'] ?? $task['name'] ?? '')); $status = strtoupper(trim((string) ($task['status'] ?? ''))); $startedAt = $this->taskDateTime($task['createTime'] ?? null); if ($status === 'RUNNING' && $name !== '') { $runningNames[$name] = true; if ($startedAt !== null && ($runningStartedAt === null || $startedAt->lt($runningStartedAt))) { $runningStartedAt = $startedAt; } $approverName = $this->taskApproverName($task, $employees); if ($approverName !== '') { $runningApprovers[$approverName] = true; } } if ($this->isCashierLabel($name)) { $cashierTasks[] = [ 'name' => $name, 'started_at' => $startedAt, 'sort_time' => trim((string) ($task['createTime'] ?? '')), ]; } } foreach ((array) ($snapshot['operation_records'] ?? []) as $operation) { if (! is_array($operation)) { continue; } $name = trim((string) ($operation['showName'] ?? '')); if ($this->isCashierLabel($name)) { $cashierTasks[] = [ 'name' => $name, 'started_at' => $this->taskDateTime($operation['date'] ?? null), 'sort_time' => trim((string) ($operation['date'] ?? '')), ]; } } $cashierTask = null; foreach ($cashierTasks as $candidate) { if ($cashierTask === null || strcmp($candidate['sort_time'], $cashierTask['sort_time']) > 0) { $cashierTask = $candidate; } } return [ 'current_task_name' => $runningNames === [] ? null : mb_substr(implode(' / ', array_keys($runningNames)), 0, 255), 'current_approver_name' => $runningApprovers === [] ? null : mb_substr(implode('、', array_keys($runningApprovers)), 0, 255), 'current_task_started_at' => $runningStartedAt, 'cashier_task_name' => $cashierTask === null ? null : mb_substr((string) $cashierTask['name'], 0, 100), 'cashier_task_started_at' => $cashierTask['started_at'] ?? null, ]; } /** * @param array $task * @param array $employees */ private function taskApproverName(array $task, array $employees): string { foreach (['userId', 'userid', 'staffId'] as $field) { $userId = trim((string) ($task[$field] ?? '')); if ($userId !== '' && trim((string) ($employees[$userId] ?? '')) !== '') { return trim((string) $employees[$userId]); } } foreach (['staffName', 'userName', 'employeeName'] as $field) { $name = trim((string) ($task[$field] ?? '')); if ($name !== '') { return $name; } } return ''; } private function taskDateTime(mixed $value): ?Carbon { $text = trim((string) $value); if ($text === '') { return null; } try { return Carbon::parse(str_replace('T', ' ', preg_replace('/Z$/', '', $text))); } catch (Throwable) { return null; } } private function isCashierLabel(string $label): bool { return str_contains($label, '出纳') && ( str_contains($label, '打款') || str_contains($label, '付款') ); } /** * @param array $process * @param array $apiFields */ private function hasApprovalPaymentSource(array $process, array $apiFields): bool { foreach ($this->paymentSourceFieldNames($process) as $fieldName) { if (array_key_exists($fieldName, $apiFields) && trim((string) ($apiFields[$fieldName] ?? '')) !== '') { return true; } } return false; } /** * @param array $process * @return list */ private function paymentSourceFieldNames(array $process): array { return array_values(array_unique(array_filter([ trim((string) ($process['payment_table'] ?? '')), ...array_map( static fn (mixed $value): string => trim((string) $value), (array) ($process['payment_total_fields'] ?? []), ), ]))); } /** * @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, string $cashierStatus, ?Carbon $cashierCheckedAt, array $taskState, ): 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), 'cashier_status' => mb_substr($cashierStatus, 0, 32), 'cashier_checked_at' => $cashierCheckedAt, 'current_approver_name' => $taskState['current_approver_name'] ?? null, 'current_task_name' => $taskState['current_task_name'] ?? null, 'current_task_started_at' => $taskState['current_task_started_at'] ?? null, 'cashier_task_name' => $taskState['cashier_task_name'] ?? null, 'cashier_task_started_at' => $taskState['cashier_task_started_at'] ?? null, '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'] ?? '')) ?: '其他'; } /** @param array $workflow */ private function status(array $workflow): string { $key = DingPayableWorkflowStatus::key($workflow); return match ($key) { 'RUNNING' => '审批中', 'COMPLETED' => '已完成', 'REJECTED' => '已拒绝', 'WITHDRAWN' => '已撤销', 'FAILED' => '失败', 'EXPIRED' => '已过期', 'DELETED' => '已删除', default => $key !== '' && $key !== 'UNKNOWN' ? $key : '未知', }; } 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' => $this->approvalApiAttempts > 0 ? 'ding_Workflow + ding_WorkflowScheme + DingTalk processInstance API fallback' : '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), 'approval_api_cursor' => (string) ($metrics['approval_api_cursor'] ?? ''), 'approval_api_attempts' => $this->approvalApiAttempts, ], 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'; } }