>|null */ private ?array $sourceStatusCache = null; /** @var list|null */ private ?array $jstOrderSourceCache = null; public function __construct(GenericBusinessMirrorRepository $mirror) { $this->connection = $mirror->connection(); } public function ready(): bool { return collect($this->sourceStatuses())->contains( static fn (array $source): bool => (bool) ($source['ready'] ?? false), ); } /** @return array{data:list>,total:int,status:string,last_synced_at:?string,summary:array,source_statuses:array>} */ 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); $total = (int) $this->filteredQuery($filters)->count(); $summary = [ 'count' => 0, 'matched_count' => 0, 'matchable_count' => 0, 'purchase_count' => 0, 'daifa_amount' => 0.0, 'paid_amount' => 0.0, 'price_diff' => 0.0, ]; $pageRows = []; if ($total > 0) { $aggregate = $this->filteredQuery($filters) ->selectRaw( 'COUNT(*) AS row_count, ' . "SUM(CASE WHEN source_type = 'purchase' THEN 1 ELSE 0 END) AS purchase_count, " . "SUM(CASE WHEN source_type = 'ding' THEN 1 ELSE 0 END) AS matchable_count, " . "SUM(CASE WHEN source_type = 'ding' AND product_id IS NOT NULL AND product_id <> '' " . "AND product_name IS NOT NULL AND product_name <> '' THEN 1 ELSE 0 END) AS matched_count, " . "COALESCE(SUM(CASE WHEN source_type = 'ding' THEN daifa_amount ELSE 0 END), 0) AS daifa_amount, " . "COALESCE(SUM(CASE WHEN source_type = 'ding' THEN paid_amount ELSE 0 END), 0) AS paid_amount, " . "COALESCE(SUM(CASE WHEN source_type = 'ding' THEN price_diff ELSE 0 END), 0) AS price_diff" ) ->first(); $summary = [ 'count' => (int) ($aggregate->row_count ?? 0), 'matched_count' => (int) ($aggregate->matched_count ?? 0), 'matchable_count' => (int) ($aggregate->matchable_count ?? 0), 'purchase_count' => (int) ($aggregate->purchase_count ?? 0), 'daifa_amount' => round((float) ($aggregate->daifa_amount ?? 0), 2), 'paid_amount' => round((float) ($aggregate->paid_amount ?? 0), 2), 'price_diff' => round((float) ($aggregate->price_diff ?? 0), 2), ]; $pageRows = $this->filteredQuery($filters) ->orderByDesc('workflow_created_at') ->orderByDesc('process_instance_id') ->orderByDesc('detail_index') ->forPage($page, $perPage) ->get(); } $data = []; foreach ($pageRows as $row) { $data[] = $this->presentRow($row); } $this->attachJstOrderDetails($data); $sources = $this->sourceStatuses(); $readySources = array_filter($sources, static fn (array $source): bool => (bool) ($source['ready'] ?? false)); $synced = array_values(array_filter(array_column($readySources, 'last_synced_at'))); return [ 'data' => $data, 'total' => $total, 'status' => count($readySources) > 1 ? 'available' : (string) (reset($readySources)['status'] ?? 'unknown'), 'last_synced_at' => $synced === [] ? null : max($synced), 'summary' => $summary, 'source_statuses' => $sources, ]; } /** @return \Generator> */ public function exportRows(array $filters): \Generator { if (! $this->ready()) { throw new RuntimeException('business_mirror_dataset_not_ready'); } $rows = $this->filteredQuery($filters) ->orderByDesc('workflow_created_at') ->orderByDesc('process_instance_id') ->orderByDesc('detail_index') ->cursor(); $buffer = []; foreach ($rows as $row) { $buffer[] = $this->presentRow($row); if (count($buffer) >= self::EXPORT_HYDRATE_BATCH) { $this->attachJstOrderDetails($buffer); yield from $buffer; $buffer = []; } } if ($buffer !== []) { $this->attachJstOrderDetails($buffer); yield from $buffer; } } /** * 按线上订单号(钉钉行)或内部单号(采购单行)批量关联聚水潭订单,补充: * - ship_skus:实际发货商品编码(组合装拆开后的子商品编码 × 数量); * - jst_labels:聚水潭订单界面的「标记|多标签」。 * * 只按当前这一批行的单号做等值查询(so_id / o_id 均有索引),不扫整张订单表。 * * @param list> $rows */ private function attachJstOrderDetails(array &$rows): void { foreach ($rows as &$row) { $row['ship_skus'] = ''; $row['ship_sku_items'] = []; $row['jst_labels'] = ''; } unset($row); $soIds = []; $oIds = []; foreach ($rows as $row) { if (($row['source_type'] ?? 'ding') === 'purchase') { $oId = trim((string) ($row['source_order_no'] ?? '')); if ($oId !== '' && ctype_digit($oId)) { $oIds[(int) $oId] = true; } } else { foreach ($this->splitTradeNos((string) ($row['shop_trade_no'] ?? '')) as $soId) { $soIds[$soId] = true; } } } if ($soIds === [] && $oIds === []) { return; } try { [$ordersBySo, $ordersByOid] = $this->loadJstOrders(array_keys($soIds), array_keys($oIds)); } catch (\Throwable $exception) { report($exception); return; } foreach ($rows as &$row) { if (($row['source_type'] ?? 'ding') === 'purchase') { $orders = $ordersByOid[trim((string) ($row['source_order_no'] ?? ''))] ?? []; } else { $orders = []; foreach ($this->splitTradeNos((string) ($row['shop_trade_no'] ?? '')) as $soId) { foreach ($ordersBySo[$soId] ?? [] as $order) { $orders[$order['key']] = $order; } } $orders = array_values($orders); } if ($orders === []) { continue; } $active = array_values(array_filter( $orders, static fn (array $order): bool => ! in_array($order['status'], self::JST_INACTIVE_ORDER_STATUSES, true), )); if ($active === []) { $active = $orders; } $labels = []; $items = []; foreach ($active as $order) { foreach ($order['labels'] as $label) { $labels[$label] = true; } foreach ($order['items'] as $item) { $items[] = $item; } } $shipped = $this->shippedItemsFor($items, (string) ($row['material_no'] ?? '')); $row['ship_sku_items'] = $shipped; $row['ship_skus'] = implode(', ', array_map( fn (array $item): string => $item['sku_id'] . '×' . $this->formatQty($item['qty']), $shipped, )); $row['jst_labels'] = implode(',', array_keys($labels)); } unset($row); } /** * 同一订单里可能有多个组合装或混着普通商品:优先只取由本行「商品编码」 * 拆出来的子商品;本行编码本身就是普通商品时取它自己;都对不上再列整单。 * * @param list $items * @return list */ private function shippedItemsFor(array $items, string $materialNo): array { $materialNo = trim($materialNo); $selected = []; if ($materialNo !== '') { $selected = array_values(array_filter($items, static fn (array $i): bool => $i['combine'] === $materialNo)); if ($selected === []) { $selected = array_values(array_filter( $items, static fn (array $i): bool => $i['combine'] === '' && $i['sku_id'] === $materialNo, )); } } if ($selected === []) { $selected = $items; } $merged = []; foreach ($selected as $item) { if ($item['sku_id'] === '') { continue; } $merged[$item['sku_id']] = ($merged[$item['sku_id']] ?? 0.0) + $item['qty']; } $result = []; foreach ($merged as $skuId => $qty) { if ($qty > 0) { $result[] = ['sku_id' => (string) $skuId, 'qty' => $qty]; } } return $result; } /** * @param list $soIds * @param list $oIds * @return array{0:array>>,1:array>>} */ private function loadJstOrders(array $soIds, array $oIds): array { $bySo = []; $byOid = []; foreach ($this->jstOrderSources() as $source) { $columns = ['o_id', 'so_id']; if ($source['status']) { $columns[] = 'status'; } if ($source['labels']) { $columns[] = 'labels'; } $orders = []; foreach (array_chunk($soIds, self::JST_LOOKUP_CHUNK) as $chunk) { foreach ($this->connection->table($source['orders'])->whereIn('so_id', $chunk)->get($columns) as $order) { $orders[(string) $order->o_id] = $order; } } foreach (array_chunk($oIds, self::JST_LOOKUP_CHUNK) as $chunk) { foreach ($this->connection->table($source['orders'])->whereIn('o_id', $chunk)->get($columns) as $order) { $orders[(string) $order->o_id] = $order; } } if ($orders === []) { continue; } $itemsByOrder = []; if ($source['items'] !== '') { $itemColumns = ['o_id', 'sku_id', 'qty']; if ($source['combine']) { $itemColumns[] = 'src_combine_sku_id'; } foreach (array_chunk(array_map('intval', array_keys($orders)), self::JST_LOOKUP_CHUNK) as $chunk) { foreach ($this->connection->table($source['items'])->whereIn('o_id', $chunk)->get($itemColumns) as $item) { $itemsByOrder[(string) $item->o_id][] = [ 'sku_id' => trim((string) ($item->sku_id ?? '')), 'qty' => (float) ($item->qty ?? 0), 'combine' => trim((string) ($item->src_combine_sku_id ?? '')), ]; } } } foreach ($orders as $oId => $order) { $entry = [ 'key' => $source['orders'] . ':' . $oId, 'status' => trim((string) ($order->status ?? '')), 'labels' => $this->splitLabels((string) ($order->labels ?? '')), 'items' => $itemsByOrder[$oId] ?? [], ]; $soId = trim((string) ($order->so_id ?? '')); if ($soId !== '') { $bySo[$soId][] = $entry; } $byOid[$oId][] = $entry; } } return [$bySo, $byOid]; } /** @return list */ private function jstOrderSources(): array { if ($this->jstOrderSourceCache !== null) { return $this->jstOrderSourceCache; } $schema = Schema::connection($this->connection->getName()); $sources = []; foreach (self::JST_ORDER_SOURCES as [$orders, $items]) { if (! $schema->hasTable($orders) || ! $schema->hasColumns($orders, ['o_id', 'so_id'])) { continue; } $hasItems = $schema->hasTable($items) && $schema->hasColumns($items, ['o_id', 'sku_id', 'qty']); $sources[] = [ 'orders' => $orders, 'items' => $hasItems ? $items : '', 'status' => $schema->hasColumn($orders, 'status'), 'labels' => $schema->hasColumn($orders, 'labels'), 'combine' => $hasItems && $schema->hasColumn($items, 'src_combine_sku_id'), ]; } return $this->jstOrderSourceCache = $sources; } /** @return list */ private function splitTradeNos(string $value): array { $parts = preg_split('/[\s,,;;\/]+/u', trim($value)) ?: []; return array_values(array_unique(array_filter(array_map('trim', $parts), static fn (string $v): bool => $v !== ''))); } /** @return list */ private function splitLabels(string $value): array { $parts = preg_split('/[,,]+/u', trim($value)) ?: []; return array_values(array_unique(array_filter(array_map('trim', $parts), static fn (string $v): bool => $v !== ''))); } private function formatQty(float $qty): string { return abs($qty - round($qty)) < 0.0001 ? (string) (int) round($qty) : rtrim(rtrim(number_format($qty, 4, '.', ''), '0'), '.'); } private function filteredQuery(array $filters): Builder { $query = $this->combinedQuery(); if (($filters['product_id'] ?? '') !== '') { $query->where('product_id', (string) $filters['product_id']); } if (($filters['shop_name'] ?? '') !== '') { $query->where('shop_name', (string) $filters['shop_name']); } if (($filters['supplier_name'] ?? '') !== '') { $query->whereRaw("supplier_name LIKE ? ESCAPE '!'", [$this->contains((string) $filters['supplier_name'])]); } if (($filters['material_no'] ?? '') !== '') { $query->where('material_no', (string) $filters['material_no']); } if (($filters['keyword'] ?? '') !== '') { $keyword = (string) $filters['keyword']; $query->where(function ($q) use ($keyword): void { $q->where('process_instance_id', 'like', '%' . $keyword . '%') ->orWhere('daifa_trade_no', 'like', '%' . $keyword . '%') ->orWhere('shop_trade_no', 'like', '%' . $keyword . '%') ->orWhere('source_order_no', 'like', '%' . $keyword . '%'); }); } if (($filters['status'] ?? '') !== '') { $query->where('status_key', strtoupper((string) $filters['status'])); } if (($filters['created_from'] ?? '') !== '') { $query->where('workflow_created_at', '>=', (string) $filters['created_from']); } if (($filters['created_to'] ?? '') !== '') { $query->where( 'workflow_created_at', '<', Carbon::parse((string) $filters['created_to'])->addDay()->startOfDay()->format('Y-m-d H:i:s'), ); } return $query; } private function combinedQuery(): Builder { $statuses = $this->sourceStatuses(); $queries = []; if ((bool) ($statuses['ding']['ready'] ?? false)) { $queries[] = $this->dingQuery(); } if ((bool) ($statuses['purchase']['ready'] ?? false)) { $queries[] = $this->purchaseQuery(); } if ($queries === []) { throw new RuntimeException('business_mirror_dataset_not_ready'); } $union = array_shift($queries); foreach ($queries as $query) { $union->unionAll($query); } return $this->connection->query()->fromSub($union, 'online_daifa_rows'); } private function dingQuery(): Builder { $status = strtoupper(trim((string) config( 'upstream_mirrors.ding_online_daifa_price.only_status', 'COMPLETED', ))) ?: 'COMPLETED'; return $this->connection->table(self::TARGET_TABLE . ' as d')->select([ DB::raw("'ding' AS source_type"), 'd.process_instance_id', 'd.detail_index', DB::raw('NULL AS purchase_order_id'), DB::raw('NULL AS source_order_no'), 'd.product_id', 'd.product_name', 'd.daifa_trade_no', 'd.shop_trade_no', 'd.shop_name', 'd.payee_name', DB::raw("CASE WHEN d.payee_name = '宁波顾家望业家居有限公司' THEN '刘萧代发' ELSE '1688代发' END AS tag"), 'd.supplier_name', 'd.material_no', 'd.ship_qty', 'd.paid_amount', 'd.daifa_amount', 'd.price_diff', 'd.workflow_payload', 'd.scheme_payload', 'd.task_payload', DB::raw("'" . $status . "' AS status_key"), 'd.workflow_created_at', ]); } private function purchaseQuery(): Builder { return $this->connection->table('product_purchase_price_mirror_rows as p') ->leftJoin('ops_jst_orders as o', 'o.o_id', '=', 'p.source_order_no') ->whereIn('p.document_status', ['Confirmed', 'Finished']) ->select([ DB::raw("'purchase' AS source_type"), 'p.purchase_order_id as process_instance_id', 'p.item_id as detail_index', 'p.purchase_order_id', 'p.source_order_no', DB::raw("'' AS product_id"), DB::raw('NULL AS product_name'), DB::raw("'' AS daifa_trade_no"), 'o.so_id as shop_trade_no', 'o.shop_name', DB::raw("'' AS payee_name"), DB::raw("'采购单管理' AS tag"), 'p.supplier_name', 'p.sku_id as material_no', 'p.quantity as ship_qty', DB::raw('NULL AS paid_amount'), 'p.unit_price as daifa_amount', DB::raw('NULL AS price_diff'), DB::raw('NULL AS workflow_payload'), DB::raw('NULL AS scheme_payload'), DB::raw('NULL AS task_payload'), DB::raw('UPPER(p.document_status) AS status_key'), DB::raw('COALESCE(p.document_created_at, p.purchase_updated_at) AS workflow_created_at'), ]); } /** @return array> */ private function sourceStatuses(): array { if ($this->sourceStatusCache !== null) { return $this->sourceStatusCache; } $schema = Schema::connection($this->connection->getName()); $dingState = null; if ($schema->hasTable(self::TARGET_TABLE) && $schema->hasTable('business_mirror_table_states')) { $dingState = $this->connection->table('business_mirror_table_states') ->where('dataset_key', self::DATASET_KEY) ->first(); } $dingReady = $dingState !== null && $dingState->initial_completed_at !== null; $purchaseState = null; $orderState = null; $requiredPurchaseColumns = [ 'source_order_no', 'supplier_name', 'quantity', 'document_created_at', ]; $purchaseSchemaReady = $schema->hasTable('product_purchase_price_mirror_rows') && $schema->hasTable('product_purchase_price_sync_states') && $schema->hasTable('ops_jst_orders') && $schema->hasColumns('product_purchase_price_mirror_rows', $requiredPurchaseColumns); if ($purchaseSchemaReady) { $purchaseState = $this->connection->table('product_purchase_price_sync_states') ->where('dataset_key', ProductPurchasePriceMirrorRepository::DATASET_KEY) ->first(); $orderState = $schema->hasTable('business_mirror_table_states') ? $this->connection->table('business_mirror_table_states') ->where('dataset_key', self::ORDER_DATASET_KEY) ->first() : null; } $purchaseReady = $purchaseState !== null && $purchaseState->initial_completed_at !== null && (int) ($purchaseState->projection_version ?? 1) >= ProductPurchasePriceMirrorRepository::CURRENT_PROJECTION_VERSION && $orderState !== null && $orderState->initial_completed_at !== null; return $this->sourceStatusCache = [ 'ding' => [ 'label' => '钉钉线上代发审批', 'ready' => $dingReady, 'status' => $dingState === null ? 'pending' : (string) $dingState->status, 'last_synced_at' => $dingState?->last_succeeded_at === null ? null : (string) $dingState->last_succeeded_at, ], 'purchase' => [ 'label' => '聚水潭采购单管理', 'ready' => $purchaseReady, 'status' => $purchaseReady ? (string) $purchaseState->status : ($purchaseSchemaReady ? 'requires_full_sync' : 'pending'), 'last_synced_at' => $purchaseState?->last_succeeded_at === null ? null : (string) $purchaseState->last_succeeded_at, ], ]; } /** @return array */ private function presentRow(object $row): array { $sourceType = (string) ($row->source_type ?? 'ding'); $purchase = $sourceType === 'purchase'; $workflow = $purchase ? [ 'businessId' => (string) ($row->purchase_order_id ?? ''), 'title' => '采购单管理', 'status' => (string) ($row->status_key ?? ''), ] : $this->decode((string) ($row->workflow_payload ?? ''), []); $productId = (string) ($row->product_id ?? ''); $resolvedName = $this->cleanName($row->product_name ?? null); $product = ! $purchase && $productId !== '' && $resolvedName !== null ? ['matched' => true, 'name' => $resolvedName, 'sku_id' => null, 'i_id' => null] : ['matched' => false, 'name' => null, 'sku_id' => null, 'i_id' => null]; $statusKey = $purchase ? strtoupper((string) ($row->status_key ?? '')) : DingPayableWorkflowStatus::key($workflow); return [ 'source_type' => $sourceType, 'process_instance_id' => (string) $row->process_instance_id, 'detail_index' => (int) $row->detail_index, 'purchase_order_id' => $row->purchase_order_id === null ? null : (int) $row->purchase_order_id, 'source_order_no' => $row->source_order_no === null ? null : (string) $row->source_order_no, 'product_id' => $productId, 'daifa_trade_no' => (string) ($row->daifa_trade_no ?? ''), 'shop_trade_no' => (string) ($row->shop_trade_no ?? ''), 'shop_name' => (string) ($row->shop_name ?? ''), 'payee_name' => (string) ($row->payee_name ?? ''), 'tag' => (string) ($row->tag ?? ''), 'supplier_name' => (string) ($row->supplier_name ?? ''), 'material_no' => (string) ($row->material_no ?? ''), 'ship_qty' => $row->ship_qty === null ? null : (float) $row->ship_qty, 'paid_amount' => $row->paid_amount === null ? null : (float) $row->paid_amount, 'daifa_amount' => $row->daifa_amount === null ? null : (float) $row->daifa_amount, 'price_diff' => $row->price_diff === null ? null : (float) $row->price_diff, 'workflow' => $workflow, 'scheme' => $purchase ? [] : $this->decode((string) ($row->scheme_payload ?? ''), []), 'tasks' => $purchase ? [] : $this->decode((string) ($row->task_payload ?? ''), []), 'status_key' => $statusKey, 'status_label' => $purchase ? match ($statusKey) { 'CONFIRMED' => '已确认', 'FINISHED' => '已完成', default => $statusKey, } : DingPayableWorkflowStatus::label($workflow), 'matched_product' => $product, 'order_matched' => ! $purchase || ((string) ($row->shop_trade_no ?? '') !== '' && (string) ($row->shop_name ?? '') !== ''), 'workflow_created_at' => (string) ($row->workflow_created_at ?? ''), ]; } private function contains(string $value): string { return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], trim($value)) . '%'; } private function cleanName(mixed $value): ?string { if ($value === null) { return null; } $name = trim((string) $value); return $name === '' ? null : $name; } private function decode(string $json, mixed $fallback): mixed { if ($json === '') { return $fallback; } $decoded = json_decode($json, true); return is_array($decoded) ? $decoded : $fallback; } }