1303 lines
46 KiB
PHP
1303 lines
46 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Services;
|
||
|
||
use Illuminate\Database\ConnectionInterface;
|
||
use Illuminate\Database\DatabaseManager;
|
||
use Illuminate\Support\Carbon;
|
||
use Illuminate\Support\Str;
|
||
use RuntimeException;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 从钉钉审批主表和结构化表单同步流程明细。
|
||
*
|
||
* 明细和付款必须分别解析：审批表单里的“付款明细”可能对应多个费用序号，
|
||
* 如果先把两张表铺平成 CSV 再逐行求和，会把同一笔付款重复计入。
|
||
*/
|
||
class DingProcessDetailMirrorSynchronizer
|
||
{
|
||
private const DATASET_PREFIX = 'ding.process_details.';
|
||
|
||
private const SOURCE = 'operations_replica';
|
||
|
||
private const STATUS_MAP = [
|
||
'COMPLETED' => '已完成',
|
||
'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<array<string,mixed>>,
|
||
* payments:list<array<string,mixed>>,
|
||
* process_instance_ids:list<string>,
|
||
* metrics:array<string,mixed>
|
||
* }
|
||
*/
|
||
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<string,mixed>|null $state
|
||
* @return array{
|
||
* details:list<array<string,mixed>>,
|
||
* payments:list<array<string,mixed>>,
|
||
* process_instance_ids:list<string>,
|
||
* metrics:array<string,mixed>
|
||
* }
|
||
*/
|
||
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<int,mixed> $bindings
|
||
* @return array{
|
||
* details:list<array<string,mixed>>,
|
||
* payments:list<array<string,mixed>>,
|
||
* process_instance_ids:list<string>,
|
||
* metrics:array<string,mixed>
|
||
* }
|
||
*/
|
||
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<int,mixed> $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<string,string> */
|
||
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<array<string,mixed>> $workflows
|
||
* @return array<string,array<string,mixed>>
|
||
*/
|
||
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<string> */
|
||
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<array<string,mixed>> $workflows
|
||
* @param array<string,array<string,mixed>> $scheme
|
||
* @param array<string,string> $employees
|
||
* @param list<string> $issues
|
||
* @return array{
|
||
* details:list<array<string,mixed>>,
|
||
* payments:list<array<string,mixed>>,
|
||
* process_instance_ids:list<string>
|
||
* }
|
||
*/
|
||
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<string,mixed> $workflow
|
||
* @param array<string,mixed> $process
|
||
* @param array<string,mixed> $fields
|
||
* @param list<string> $issues
|
||
* @return array{0:list<array<string,mixed>>,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<string,mixed> $workflow
|
||
* @param array<string,mixed> $process
|
||
* @param array<string,mixed> $fields
|
||
* @param list<string> $issues
|
||
* @return list<array<string,mixed>>
|
||
*/
|
||
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<string,mixed> $process
|
||
* @param array<string,mixed> $fields
|
||
* @param list<string> $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<array<string,mixed>> $details
|
||
* @param list<array<string,mixed>> $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<string,mixed> $workflow
|
||
* @param array<string,mixed> $process
|
||
* @return array<string,mixed>
|
||
*/
|
||
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<string,mixed> $workflow
|
||
* @param list<string> $issues
|
||
* @return list<array<string,string>>
|
||
*/
|
||
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<string,mixed> $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<string,string> $row
|
||
* @param list<string> $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<string,mixed> $fields
|
||
* @param list<string> $names
|
||
*/
|
||
private function moneyFromFields(array $fields, array $names): ?int
|
||
{
|
||
return $this->firstFieldMoneyCents($fields, $names);
|
||
}
|
||
|
||
/**
|
||
* @param array<string,mixed> $fields
|
||
* @param list<string> $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<string,string> $row */
|
||
private function purpose(array $row): string
|
||
{
|
||
foreach (['用途', '商品名称', '固定资产名称', '规格名称', 'SPU', '条形码'] as $field) {
|
||
$value = trim((string) ($row[$field] ?? ''));
|
||
if ($value !== '') {
|
||
return $value;
|
||
}
|
||
}
|
||
|
||
return '';
|
||
}
|
||
|
||
/** @param array<string,mixed> $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<array<string,mixed>> $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<array<string,mixed>> $details
|
||
* @param list<array<string,mixed>> $payments
|
||
* @param list<string> $processInstanceIds
|
||
* @param array<string,mixed> $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<string,mixed>|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<string,mixed>|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<string,mixed>|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<string,mixed>|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<string,mixed>|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';
|
||
}
|
||
}
|