diff --git a/app/Console/Commands/SyncBusinessMirrors.php b/app/Console/Commands/SyncBusinessMirrors.php index 64098240..46772a7c 100644 --- a/app/Console/Commands/SyncBusinessMirrors.php +++ b/app/Console/Commands/SyncBusinessMirrors.php @@ -7,6 +7,7 @@ use Illuminate\Support\Facades\Log; use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository; use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry; use Modules\EcommerceFinance\Services\BusinessMirrorSynchronizer; +use Modules\EcommerceFinance\Services\DingPayableWorkflowMirrorSynchronizer; use Throwable; class SyncBusinessMirrors extends Command @@ -24,6 +25,7 @@ class SyncBusinessMirrors extends Command BusinessMirrorDatasetRegistry $datasets, BusinessMirrorSynchronizer $synchronizer, GenericBusinessMirrorRepository $mirror, + ?DingPayableWorkflowMirrorSynchronizer $dingPayableWorkflows = null, ): int { if (! (bool) config('upstream_mirrors.refresh_enabled', false) && ! $this->option('full')) { @@ -76,6 +78,30 @@ class SyncBusinessMirrors extends Command )); } + // 钉钉应付流程采用按流程聚合的专用镜像；接入同一入口后由现有 + // 生产 cron 自动断点续传，避免额外维护一套调度脚本。 + if ($requested === null && $dingPayableWorkflows !== null && ! app()->environment('testing')) { + try { + $result = $dingPayableWorkflows->sync( + (bool) $this->option('full'), + $this->option('batch') === null ? null : (int) $this->option('batch'), + $this->option('max-batches') === null ? null : (int) $this->option('max-batches'), + ); + if ($result['status'] !== 'requires_full_sync') { + $this->info(sprintf( + 'operations_replica.rpt.ding_payable_workflows：%s，本轮 %d 批，累计处理 %d 个流程。', + $result['status'] === 'succeeded' ? '同步完成' : '同步进行中', + $result['batches'], + $result['rows'], + )); + } + } catch (Throwable $exception) { + $failed = true; + Log::error('钉钉应付流程镜像同步失败。', ['exception' => $exception::class]); + $this->error('operations_replica.rpt.ding_payable_workflows：同步失败，已保留上一版共享数据。'); + } + } + return $failed ? self::FAILURE : self::SUCCESS; } diff --git a/app/Console/Commands/SyncDingPayableWorkflowMirror.php b/app/Console/Commands/SyncDingPayableWorkflowMirror.php new file mode 100644 index 00000000..5b6c91c6 --- /dev/null +++ b/app/Console/Commands/SyncDingPayableWorkflowMirror.php @@ -0,0 +1,47 @@ +option('full')) { + return self::SUCCESS; + } + + try { + $result = $synchronizer->sync( + (bool) $this->option('full'), + $this->option('batch') === null ? null : (int) $this->option('batch'), + $this->option('max-batches') === null ? null : (int) $this->option('max-batches'), + ); + } catch (Throwable $exception) { + Log::error('钉钉应付流程镜像同步失败。', ['exception' => $exception::class]); + $this->error('钉钉应付流程同步失败，已保留上一版共享数据。'); + + return self::FAILURE; + } + + $this->info(sprintf( + '%s：本轮 %d 批，累计处理 %d 个流程。', + $result['status'] === 'succeeded' ? '同步完成' : '同步进行中', + $result['batches'], + $result['rows'], + )); + + return self::SUCCESS; + } +} diff --git a/app/Http/Controllers/Business/Pages.php b/app/Http/Controllers/Business/Pages.php index 2bdd83a5..9b705f9f 100644 --- a/app/Http/Controllers/Business/Pages.php +++ b/app/Http/Controllers/Business/Pages.php @@ -23,6 +23,7 @@ use Modules\EcommerceFinance\Repositories\PinduoduoSpreadBillRepository; use Modules\EcommerceFinance\Repositories\ProductCatalogMirrorRepository; use Modules\EcommerceFinance\Repositories\ProductPriceMirrorRepository; use Modules\EcommerceFinance\Repositories\RawBoardPayablesRepository; +use Modules\EcommerceFinance\Repositories\DingPayableWorkflowMirrorRepository; use Modules\EcommerceFinance\Repositories\SupplierPayableReportRepository; use Modules\EcommerceFinance\Repositories\TmallTaobaoSpreadBillRepository; use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory; @@ -45,6 +46,7 @@ class Pages extends Controller private readonly ProductCatalogMirrorRepository $productCatalog, private readonly ProductPriceMirrorRepository $productPrices, private readonly RawBoardPayablesRepository $rawBoardPayables, + private readonly DingPayableWorkflowMirrorRepository $dingPayableWorkflows, private readonly SupplierPayableReportRepository $supplierPayableReports, private readonly UpstreamQueryCacheFactory $upstreamCaches, ) { @@ -245,10 +247,53 @@ class Pages extends Controller 'product-prices' => $this->contentForProductPrices(), 'board-deductions' => $this->contentForBoardDeductions(), 'supplier-payables-report' => $this->contentForSupplierPayablesReport(), + 'payable-cloud-warehouse-vehicle-fee' => $this->contentForDingPayableWorkflows(), default => null, }; } + protected function contentForDingPayableWorkflows(): array + { + $filters = [ + 'process_instance_id' => mb_substr(trim((string) request('process_instance_id', '')), 0, 100), + ]; + $page = max((int) request('page', 1), 1); + $limit = min(max((int) setting('default.list_limit', 25), 1), 100); + + try { + $result = $this->dingPayableWorkflows->paginate($filters, $page, $limit); + $rows = new LengthAwarePaginator( + Collection::make($result['data'] ?? []), + (int) ($result['total'] ?? 0), + $limit, + $page, + ['path' => request()->url(), 'query' => request()->except('page')], + ); + + return [ + 'view' => 'business.pages.courier-payables.payable-cloud-warehouse-vehicle-fee', + 'dingPayableWorkflowRows' => $rows, + 'dingPayableWorkflowFilters' => $filters, + 'dingPayableWorkflowStatus' => $result['status'] ?? 'unknown', + 'dingPayableWorkflowSyncedAt' => $result['last_synced_at'] ?? null, + 'dingPayableWorkflowError' => null, + ]; + } catch (Throwable $exception) { + report($exception); + + return [ + 'view' => 'business.pages.courier-payables.payable-cloud-warehouse-vehicle-fee', + 'dingPayableWorkflowRows' => new LengthAwarePaginator([], 0, $limit, $page, [ + 'path' => request()->url(), 'query' => request()->except('page'), + ]), + 'dingPayableWorkflowFilters' => $filters, + 'dingPayableWorkflowStatus' => 'unavailable', + 'dingPayableWorkflowSyncedAt' => null, + 'dingPayableWorkflowError' => '钉钉应付流程镜像尚未完成同步，请稍后重试。', + ]; + } + } + protected function contentForNonJingxiRefundData(): array { $request = request(); diff --git a/config/business_navigation.php b/config/business_navigation.php index 196401e3..35bc03f9 100644 --- a/config/business_navigation.php +++ b/config/business_navigation.php @@ -424,6 +424,7 @@ return [ ['slug' => 'compensation-details', 'title' => '赔偿明细'], ['slug' => 'advance-details', 'title' => '借支明细'], ['slug' => 'other-adjustments', 'title' => '其他扣款补款'], + ['slug' => 'payable-cloud-warehouse-vehicle-fee', 'title' => '应付云仓车费'], ['slug' => 'courier-payables-report', 'title' => '快递应付表'], ], ], diff --git a/config/upstream_mirrors.php b/config/upstream_mirrors.php index ae90c8e7..ddd826c1 100644 --- a/config/upstream_mirrors.php +++ b/config/upstream_mirrors.php @@ -3,6 +3,17 @@ return [ 'connection' => env('BUSINESS_MIRROR_DB_CONNECTION', 'business_mirror'), + // 钉钉应付申请流程只同步“应付申请流程”这一固定流程编码， + // 页面从共享业务镜像读取，不在 HTTP 请求中直连上游。 + 'ding_payable_workflows' => [ + 'dataset_key' => 'operations_replica.rpt.ding_payable_workflows', + 'source_key' => 'operations_replica', + 'process_code' => 'PROC-39DD70DE-DAB2-4A5F-B8F1-0B0B68BD0C88', + 'target_table' => 'ding_payable_workflow_mirror', + 'batch_size' => (int) env('DING_PAYABLE_WORKFLOW_MIRROR_BATCH_SIZE', 100), + 'max_batches_per_run' => (int) env('DING_PAYABLE_WORKFLOW_MIRROR_MAX_BATCHES', 2), + ], + // 只有生产刷新节点写镜像；开发环境和网页请求只读共享结果。 'refresh_enabled' => env( 'BUSINESS_MIRROR_REFRESH_ENABLED', diff --git a/database/migrations/2026_08_28_000000_create_ding_payable_workflow_mirror_tables.php b/database/migrations/2026_08_28_000000_create_ding_payable_workflow_mirror_tables.php new file mode 100644 index 00000000..4b062873 --- /dev/null +++ b/database/migrations/2026_08_28_000000_create_ding_payable_workflow_mirror_tables.php @@ -0,0 +1,45 @@ +environment('testing') + || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') { + return; + } + + $schema = Schema::connection($this->connection()); + + if (! $schema->hasTable('ding_payable_workflow_mirror')) { + $schema->create('ding_payable_workflow_mirror', function (Blueprint $table): void { + $table->string('process_instance_id', 100)->primary(); + $table->json('workflow_payload'); + $table->json('scheme_payload'); + $table->json('task_payload')->nullable(); + $table->string('workflow_created_at', 32)->nullable()->index(); + $table->uuid('sync_generation')->nullable()->index(); + $table->dateTime('synced_at', 6); + }); + } + } + + public function down(): void + { + if (app()->environment('testing') + || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') { + return; + } + + Schema::connection($this->connection())->dropIfExists('ding_payable_workflow_mirror'); + } +}; diff --git a/modules/EcommerceFinance/Repositories/DingPayableWorkflowMirrorRepository.php b/modules/EcommerceFinance/Repositories/DingPayableWorkflowMirrorRepository.php new file mode 100644 index 00000000..c999bffb --- /dev/null +++ b/modules/EcommerceFinance/Repositories/DingPayableWorkflowMirrorRepository.php @@ -0,0 +1,155 @@ +connection = $mirror->connection(); + } + + public function ready(): bool + { + $schema = Schema::connection($this->connection->getName()); + if (! $schema->hasTable('ding_payable_workflow_mirror') || ! $schema->hasTable('business_mirror_table_states')) { + return false; + } + + $state = $this->connection->table('business_mirror_table_states') + ->where('dataset_key', self::DATASET_KEY)->first(); + + return $state !== null && $state->initial_completed_at !== null; + } + + /** @return array{data:list>,total:int,status:string,last_synced_at:?string} */ + public function paginate(array $filters, int $page, int $perPage): array + { + if (! $this->ready()) { + throw new RuntimeException('business_mirror_dataset_not_ready'); + } + + $page = max($page, 1); + $perPage = min(max($perPage, 1), 100); + $state = $this->connection->table('business_mirror_table_states')->where('dataset_key', self::DATASET_KEY)->first(); + // 先在镜像库按固定业务关键词缩小候选流程，再解析 JSON；避免为每次页面请求 + // 解码全部历史表单。关键词来自本任务的固定业务筛选，不是用户拼接 SQL。 + // “仓”覆盖江西仓、重庆仓等实际用途写法；后续仍在解析明细时 + // 同时校验类型和车费，避免仅因其他字段出现“仓”而误收流程。 + $records = $this->connection->table('ding_payable_workflow_mirror') + ->where('scheme_payload', 'like', '%快递运费%') + ->where('scheme_payload', 'like', '%仓%') + ->where('scheme_payload', 'like', '%车费%') + ->orderByDesc('workflow_created_at')->orderByDesc('process_instance_id')->get(); + $filtered = []; + + foreach ($records as $record) { + $workflow = $this->decode((string) $record->workflow_payload, []); + if ((string) ($workflow['processCode'] ?? '') !== self::PROCESS_CODE) { + continue; + } + if (($filters['process_instance_id'] ?? '') !== '' + && ! str_contains((string) ($workflow['processInstanceId'] ?? ''), (string) $filters['process_instance_id'])) { + continue; + } + + $scheme = $this->decode((string) $record->scheme_payload, []); + $forms = $this->formFields($scheme); + foreach ($this->detailRows($forms['应付明细'] ?? []) as $detail) { + $type = trim((string) ($detail['类型'] ?? '')); + $purpose = trim((string) ($detail['用途'] ?? '')); + if ($type !== '快递运费' || ! str_contains($purpose, '仓') || ! str_contains($purpose, '车费')) { + continue; + } + + $filtered[] = [ + 'process_instance_id' => (string) ($workflow['processInstanceId'] ?? $record->process_instance_id), + 'workflow' => $workflow, + 'fields' => $forms, + 'detail' => $detail, + 'tasks' => $this->decode((string) ($record->task_payload ?? '[]'), []), + ]; + } + } + + $total = count($filtered); + $data = array_slice($filtered, ($page - 1) * $perPage, $perPage); + + return [ + 'data' => $data, + 'total' => $total, + 'status' => (string) ($state->status ?? 'unknown'), + 'last_synced_at' => $state->last_succeeded_at ? (string) $state->last_succeeded_at : null, + ]; + } + + /** @return array */ + private function formFields(array $rows): array + { + $fields = []; + foreach ($rows as $row) { + if (! is_array($row)) { + continue; + } + $name = trim((string) ($row['name'] ?? '')); + if ($name === '') { + continue; + } + $value = $this->decode((string) ($row['value'] ?? ''), $row['value'] ?? null); + $fields[$name] = $value; + } + return $fields; + } + + /** @return list> */ + private function detailRows(mixed $value): array + { + if (! is_array($value)) { + return []; + } + $details = []; + foreach ($value as $row) { + if (! is_array($row) || ! isset($row['rowValue']) || ! is_array($row['rowValue'])) { + continue; + } + $detail = []; + foreach ($row['rowValue'] as $field) { + if (! is_array($field) || ! isset($field['label'])) { + continue; + } + $detail[(string) $field['label']] = $this->scalarValue($field['value'] ?? null); + } + if ($detail !== []) { + $details[] = $detail; + } + } + return $details; + } + + private function scalarValue(mixed $value): mixed + { + if (is_scalar($value) || $value === null) { + return $value; + } + return json_encode($value, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES); + } + + private function decode(string $json, mixed $fallback): mixed + { + if ($json === '') { + return $fallback; + } + $decoded = json_decode($json, true); + return json_last_error() === JSON_ERROR_NONE ? $decoded : $fallback; + } +} diff --git a/modules/EcommerceFinance/Services/DingPayableWorkflowMirrorSynchronizer.php b/modules/EcommerceFinance/Services/DingPayableWorkflowMirrorSynchronizer.php new file mode 100644 index 00000000..da55a1d2 --- /dev/null +++ b/modules/EcommerceFinance/Services/DingPayableWorkflowMirrorSynchronizer.php @@ -0,0 +1,277 @@ +connection = app('db')->connection( + (string) config('upstream_mirrors.connection', 'business_mirror') + ); + } + + /** @return array{status:string,mode:string,batches:int,rows:int,dataset:string} */ + public function sync(bool $full = false, ?int $batchSize = null, ?int $maxBatches = null): array + { + $config = (array) config('upstream_mirrors.ding_payable_workflows', []); + $dataset = (string) ($config['dataset_key'] ?? 'operations_replica.rpt.ding_payable_workflows'); + $target = (string) ($config['target_table'] ?? 'ding_payable_workflow_mirror'); + $source = (string) ($config['source_key'] ?? 'operations_replica'); + $processCode = (string) ($config['process_code'] ?? ''); + + if ($processCode === '') { + throw new RuntimeException('ding_payable_workflow_mirror_not_initialized'); + } + + $this->ensureMirrorTable($target); + + UpstreamOdbcConnection::reset($source); + $state = $this->state($dataset); + + if (! $full && ($state['initial_completed_at'] ?? null) === null) { + return ['status' => 'requires_full_sync', 'mode' => 'incremental', 'batches' => 0, 'rows' => 0, 'dataset' => $dataset]; + } + + $mode = $full ? 'full' : 'incremental'; + $resume = is_array($state) + && ($state['run_mode'] ?? null) === $mode + && in_array($state['status'] ?? null, ['running', 'failed'], true) + && is_string($state['generation'] ?? null) + && is_array($state['cursor'] ?? null); + $generation = $resume ? (string) $state['generation'] : (string) Str::uuid(); + $cursor = $resume + ? (string) ($state['cursor']['processInstanceId'] ?? '') + : (string) ($this->decodeJson($state['high_watermark'] ?? null)['processInstanceId'] ?? ''); + $rows = $resume ? (int) ($state['last_upserted_rows'] ?? 0) : 0; + $batchSize = min(max($batchSize ?? (int) ($config['batch_size'] ?? 100), 1), 500); + $maxBatches = max($maxBatches ?? (int) ($config['max_batches_per_run'] ?? 2), 0); + + $this->start($dataset, $target, $source, $mode, $generation, $cursor, $resume); + $batches = 0; + + try { + while (true) { + $workflows = $this->readWorkflows($source, $processCode, $cursor, $batchSize); + if ($workflows === []) { + $this->complete($dataset, $target, $mode, $generation, $rows, $cursor); + + return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset]; + } + + $ids = array_values(array_filter(array_map( + static fn (array $row): string => (string) ($row['processInstanceId'] ?? ''), + $workflows, + ))); + $schemes = $this->readRelated($source, 'ding_WorkflowScheme', ['id', 'processInstanceId', 'componentType', 'name', 'value', 'extValue'], $ids, 'id'); + $tasks = $this->readRelated($source, 'ding_WorkflowTask', ['processInstanceId', 'result', 'activityId', 'finishTime', 'createTime', 'userId', 'taskId', 'status', 'id'], $ids, 'id'); + + $schemeById = $this->groupBy($schemes, 'processInstanceId'); + $taskById = $this->groupBy($tasks, 'processInstanceId'); + $now = now(); + $payloads = []; + + foreach ($workflows as $workflow) { + $id = (string) ($workflow['processInstanceId'] ?? ''); + if ($id === '') { + continue; + } + $payloads[] = [ + 'process_instance_id' => $id, + 'workflow_payload' => json_encode($workflow, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR), + 'scheme_payload' => json_encode($schemeById[$id] ?? [], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR), + 'task_payload' => json_encode($taskById[$id] ?? [], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR), + 'workflow_created_at' => isset($workflow['createAt']) ? (string) $workflow['createAt'] : null, + 'sync_generation' => $generation, + 'synced_at' => $now, + ]; + } + + $cursor = (string) ($workflows[array_key_last($workflows)]['processInstanceId'] ?? $cursor); + $rows += count($payloads); + $batches++; + $this->connection->transaction(function () use ($dataset, $target, $payloads, $cursor, $generation, $rows, $now): void { + if ($payloads !== []) { + $this->connection->table($target)->upsert( + $payloads, + ['process_instance_id'], + ['workflow_payload', 'scheme_payload', 'task_payload', 'workflow_created_at', 'sync_generation', 'synced_at'], + ); + } + $this->connection->table('business_mirror_table_states') + ->where('dataset_key', $dataset) + ->update([ + 'cursor' => json_encode(['processInstanceId' => $cursor], JSON_UNESCAPED_UNICODE), + 'last_upserted_rows' => $rows, + 'updated_at' => $now, + ]); + }); + + if (count($workflows) < $batchSize) { + $this->complete($dataset, $target, $mode, $generation, $rows, $cursor); + + return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset]; + } + + if ($maxBatches > 0 && $batches >= $maxBatches) { + return ['status' => 'running', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset]; + } + } + } catch (Throwable $exception) { + $this->connection->table('business_mirror_table_states') + ->where('dataset_key', $dataset) + ->update(['status' => 'failed', 'last_error_code' => mb_substr($exception::class, 0, 120), 'updated_at' => now()]); + throw $exception; + } + } + + private function ensureMirrorTable(string $target): void + { + $schema = Schema::connection($this->connection->getName()); + if ($schema->hasTable($target)) { + return; + } + $schema->create($target, function (Blueprint $table): void { + $table->string('process_instance_id', 100)->primary(); + $table->json('workflow_payload'); + $table->json('scheme_payload'); + $table->json('task_payload')->nullable(); + $table->string('workflow_created_at', 32)->nullable()->index(); + $table->uuid('sync_generation')->nullable()->index(); + $table->dateTime('synced_at', 6); + }); + } + + /** @return array|null */ + private function state(string $dataset): ?array + { + $row = $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset)->first(); + if ($row === null) { + return null; + } + $state = (array) $row; + $state['cursor'] = is_string($state['cursor'] ?? null) ? (json_decode($state['cursor'], true) ?: []) : ($state['cursor'] ?? []); + + return $state; + } + + private function start(string $dataset, string $target, string $source, string $mode, string $generation, string $cursor, bool $resume): void + { + $now = now(); + $values = [ + 'source_key' => $source, + 'source_object' => 'rpt.ding_Workflow + rpt.ding_WorkflowScheme + rpt.ding_WorkflowTask', + 'target_table' => $target, + 'status' => 'running', + 'run_mode' => $mode, + 'generation' => $generation, + 'last_attempted_at' => $now, + 'last_error_code' => null, + 'updated_at' => $now, + ]; + if (! $resume) { + $values += ['cursor' => null, 'window_started_at' => null, 'window_ends_at' => null, 'last_upserted_rows' => 0]; + } + $query = $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset); + $query->exists() + ? $query->update($values) + : $this->connection->table('business_mirror_table_states')->insert(['dataset_key' => $dataset, 'created_at' => $now] + $values); + } + + private function complete(string $dataset, string $target, string $mode, string $generation, int $rows, string $cursor): void + { + $now = now(); + $this->connection->transaction(function () use ($dataset, $target, $mode, $generation, $rows, $cursor, $now): void { + if ($mode === 'full') { + $this->connection->table($target)->where(function ($query) use ($generation): void { + $query->whereNull('sync_generation')->orWhere('sync_generation', '!=', $generation); + })->delete(); + } + $values = [ + 'status' => 'succeeded', + 'cursor' => null, + 'high_watermark' => json_encode(['processInstanceId' => $cursor], JSON_UNESCAPED_UNICODE), + 'last_succeeded_at' => $now, + 'last_upserted_rows' => $rows, + 'mirror_row_count' => $this->connection->table($target)->count(), + 'last_error_code' => null, + 'updated_at' => $now, + ]; + if ($mode === 'full') { + $values['initial_completed_at'] = $now; + $values['last_full_succeeded_at'] = $now; + } + $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset)->update($values); + }); + } + + /** @return array */ + private function decodeJson(mixed $value): array + { + if (is_array($value)) { + return $value; + } + if (! is_string($value) || $value === '') { + return []; + } + $decoded = json_decode($value, true); + return is_array($decoded) ? $decoded : []; + } + + /** @return list> */ + private function readWorkflows(string $source, string $processCode, string $cursor, int $limit): array + { + $where = '[processCode] = ?'; + $bindings = [$processCode]; + if ($cursor !== '') { + $where .= ' AND [processInstanceId] > ?'; + $bindings[] = $cursor; + } + return $this->queries->forSource($source)->select( + 'SELECT TOP (' . $limit . ') [finishTime],[businessId],[title],[result],[originatorUserId],[originatorDeptId],[originatorDeptName],[status],[bizAction],[processInstanceId],[joinAt],[updateAt],[createAt],[processCode],[flowType] FROM [rpt].[ding_Workflow] WHERE ' . $where . ' ORDER BY [processInstanceId]', + $bindings, + ); + } + + /** @param list $ids @param list $columns @return list> */ + private function readRelated(string $source, string $table, array $columns, array $ids, string $order): array + { + if ($ids === []) { + return []; + } + $placeholders = implode(',', array_fill(0, count($ids), '?')); + return $this->queries->forSource($source)->select( + 'SELECT ' . implode(', ', array_map(static fn (string $column): string => '[' . $column . ']', $columns)) + . ' FROM [rpt].[' . $table . '] WHERE [processInstanceId] IN (' . $placeholders . ') ORDER BY [' . $order . ']', + $ids, + ); + } + + /** @param list> $rows @return array>> */ + private function groupBy(array $rows, string $column): array + { + $grouped = []; + foreach ($rows as $row) { + $key = (string) ($row[$column] ?? ''); + if ($key !== '') { + $grouped[$key][] = $row; + } + } + return $grouped; + } +} diff --git a/resources/views/business/pages/courier-payables/payable-cloud-warehouse-vehicle-fee.blade.php b/resources/views/business/pages/courier-payables/payable-cloud-warehouse-vehicle-fee.blade.php new file mode 100644 index 00000000..f1eeefcf --- /dev/null +++ b/resources/views/business/pages/courier-payables/payable-cloud-warehouse-vehicle-fee.blade.php @@ -0,0 +1,113 @@ +@php + $rows = $dingPayableWorkflowRows ?? null; + $filters = $dingPayableWorkflowFilters ?? []; + $status = (string) ($dingPayableWorkflowStatus ?? 'unknown'); + $syncedAt = $dingPayableWorkflowSyncedAt ?? null; + $error = $dingPayableWorkflowError ?? null; + $json = static fn (mixed $value): string => is_array($value) + ? (json_encode($value, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_PRETTY_PRINT) ?: '-') + : (string) ($value ?? '-'); + $value = static function (array $detail, string $key): string { + $v = $detail[$key] ?? null; + return $v === null || $v === '' ? '-' : (string) $v; + }; +@endphp + +
+ + +
+
+ + +
+
+ + 重置 +
+
+ + @if ($error) + + @endif + +
+ 固定流程：应付申请流程 + 筛选：类型=快递运费，且用途同时包含“仓”“车费” + 镜像状态：{{ $status }} + @if ($syncedAt) 最近同步：{{ $syncedAt }} @endif + 共 {{ $rows?->total() ?? 0 }} 条明细 +
+ +
+ + + + + + + + @forelse ($rows ?? [] as $row) + @php($workflow = $row['workflow'] ?? []) + @php($detail = $row['detail'] ?? []) + @php($fields = $row['fields'] ?? []) + + + + + + + + + + + + + + + + + + + + + + + @empty + + @endforelse + +
流程实例 ID标题状态申请人申请部门创建时间完成时间类型用途付款金额货款月份快递支付方式收款人全称开户银行银行账号全部表单字段付款明细附件审批任务
{{ $row['process_instance_id'] ?? '-' }}{{ $workflow['title'] ?? '-' }}{{ $workflow['status'] ?? '-' }}{{ $workflow['originatorUserId'] ?? '-' }}{{ $workflow['originatorDeptName'] ?? '-' }}{{ $workflow['createAt'] ?? '-' }}{{ $workflow['finishTime'] ?? '-' }}{{ $value($detail, '类型') }}{{ $value($detail, '用途') }}{{ $value($detail, '付款金额') }}{{ $value($detail, '货款月份') }}{{ $value($detail, '快递') }}{{ $value($detail, '支付方式') }}{{ $value($detail, '收款人全称') }}{{ $value($detail, '开户银行') }}{{ $value($detail, '银行账号') }}
展开字段（{{ count($fields) }} 项）
{{ $json($fields) }}
展开付款明细
{{ $json($fields['付款明细'] ?? []) }}
展开附件
{{ $json($fields['附件'] ?? ($fields['未签字版合同附件或其他明细附件'] ?? [])) }}
展开审批任务
{{ $json($row['tasks'] ?? []) }}
{{ $error ? '镜像读取失败，请稍后重试。' : '暂无符合条件的应付云仓车费流程；同步完成后会自动显示。' }}
+
+ 当前页 {{ $rows?->count() ?? 0 }} 条 · 第 {{ $rows?->currentPage() ?? 1 }} 页 + + @if ($rows && !$rows->onFirstPage())上一页@endif + @if ($rows && $rows->hasMorePages())下一页@endif + +
+
+
diff --git a/tests/Feature/Business/DingPayableWorkflowPageTest.php b/tests/Feature/Business/DingPayableWorkflowPageTest.php new file mode 100644 index 00000000..c6532951 --- /dev/null +++ b/tests/Feature/Business/DingPayableWorkflowPageTest.php @@ -0,0 +1,63 @@ +user->attachPermission(Permission::where('name', 'read-admin-panel')->firstOrFail()); + $repository = Mockery::mock(DingPayableWorkflowMirrorRepository::class); + $repository->shouldReceive('paginate')->once()->andReturn([ + 'data' => [[ + 'process_instance_id' => 'PROC-TEST-001', + 'workflow' => [ + 'title' => '应付申请流程', + 'status' => 'COMPLETED', + 'originatorUserId' => 'user-1', + 'originatorDeptName' => '财务部', + 'createAt' => '2026-08-28 10:00:00', + 'finishTime' => '2026-08-28 11:00:00', + ], + 'detail' => ['类型' => '快递运费', '用途' => '拉到江西仓的车费', '付款金额' => '1200'], + 'fields' => ['应付明细' => [], '付款明细' => [['付款金额' => '1200']], '附件' => []], + 'tasks' => [['status' => 'COMPLETED']], + ]], + 'total' => 1, + 'status' => 'succeeded', + 'last_synced_at' => '2026-08-28 11:05:00', + ]); + $this->app->instance(DingPayableWorkflowMirrorRepository::class, $repository); + + $this->loginAs()->get(route('business.pages.show', [ + 'section' => 'courier-payables', + 'page' => 'payable-cloud-warehouse-vehicle-fee', + ]))->assertOk() + ->assertSee('应付云仓车费') + ->assertSee('拉到江西仓的车费') + ->assertSee('用途同时包含“仓”“车费”') + ->assertSee('全部表单字段') + ->assertSee('审批任务') + ->assertSee('镜像状态：succeeded'); + } + + public function test_cloud_warehouse_vehicle_fee_page_keeps_an_unavailable_mirror_readable(): void + { + $this->user->attachPermission(Permission::where('name', 'read-admin-panel')->firstOrFail()); + $repository = Mockery::mock(DingPayableWorkflowMirrorRepository::class); + $repository->shouldReceive('paginate')->andThrow(new \RuntimeException('not ready')); + $this->app->instance(DingPayableWorkflowMirrorRepository::class, $repository); + + $this->loginAs()->get(route('business.pages.show', [ + 'section' => 'courier-payables', + 'page' => 'payable-cloud-warehouse-vehicle-fee', + ]))->assertOk() + ->assertSee('钉钉应付流程镜像尚未完成同步，请稍后重试。') + ->assertSee('镜像读取失败，请稍后重试。'); + } +} diff --git a/tests/Feature/Console/DingPayableWorkflowMirrorSyncTest.php b/tests/Feature/Console/DingPayableWorkflowMirrorSyncTest.php new file mode 100644 index 00000000..8dfa7f6a --- /dev/null +++ b/tests/Feature/Console/DingPayableWorkflowMirrorSyncTest.php @@ -0,0 +1,28 @@ + false]); + $synchronizer = Mockery::mock(DingPayableWorkflowMirrorSynchronizer::class); + $synchronizer->shouldReceive('sync')->once()->with(true, 25, 1)->andReturn([ + 'status' => 'succeeded', 'mode' => 'full', 'batches' => 1, 'rows' => 2, + 'dataset' => 'operations_replica.rpt.ding_payable_workflows', + ]); + $this->app->instance(DingPayableWorkflowMirrorSynchronizer::class, $synchronizer); + + $this->artisan('business-mirror:sync-ding-payable-workflows', [ + '--full' => true, + '--batch' => 25, + '--max-batches' => 1, + ])->expectsOutputToContain('同步完成') + ->assertSuccessful(); + } +}