connection(); if (! Schema::connection($this->connectionName())->hasTable('kuka_posting_material_map_rows') || ! Schema::connection($this->connectionName())->hasTable('kuka_posting_base_rows') || ! Schema::connection($this->connectionName())->hasTable('kuka_posting_quotation_rows')) { throw new RuntimeException('顾家物料匹配所需的数据表尚未建立，请先执行数据库迁移。'); } $started = microtime(true); $this->markState('running', 0, null, $started); try { $items = $this->itemNumbers($connection, $companyId, $period); if ($items === []) { throw new RuntimeException('当前账期没有可用于匹配的货号。'); } $quoteMaterials = $this->activeQuoteMaterials($connection, $companyId); [$direct, $archive] = $this->sourceCandidates($items); $rows = $this->resolveRows($companyId, $items, $quoteMaterials, $direct, $archive); $stats = $this->stats($rows); if (! $dryRun) { $connection->transaction(function () use ($connection, $companyId, $rows): void { $connection->table('kuka_posting_material_map_rows') ->where('company_id', $companyId) ->delete(); foreach (array_chunk($rows, self::INSERT_CHUNK) as $chunk) { $connection->table('kuka_posting_material_map_rows')->insert($chunk); } }); $this->markState('succeeded', count($rows), null, $started); } return ['dry_run' => $dryRun] + $stats; } catch (Throwable $exception) { if (! $dryRun) { $this->markState('failed', 0, $exception::class, $started); } throw $exception; } } /** @return list */ private function itemNumbers(ConnectionInterface $connection, int $companyId, ?string $period): array { $items = []; $query = $connection->table('kuka_posting_base_rows') ->where('company_id', $companyId) ->when($period !== null, fn ($builder) => $builder->where('period', $period)); $query->select(['id', 'source_values']) ->orderBy('id') ->chunkById(1000, function ($rows) use (&$items): void { foreach ($rows as $row) { $values = json_decode((string) $row->source_values, true); if (! is_array($values)) { continue; } $itemNumber = trim((string) ($values['货号'] ?? '')); if ($itemNumber !== '') { $items[$itemNumber] = true; } } }); $items = array_keys($items); sort($items, SORT_STRING); return $items; } /** @return array */ private function activeQuoteMaterials(ConnectionInterface $connection, int $companyId): array { return $connection->table('kuka_posting_quotation_rows') ->where('company_id', $companyId) ->where('record_status', '生效') ->pluck('material_number') ->map(static fn (mixed $value): string => trim((string) $value)) ->filter() ->unique() ->mapWithKeys(static fn (string $value): array => [$value => true]) ->all(); } /** * @param list $items * @return array{0:array>,1:array>} */ private function sourceCandidates(array $items): array { $direct = []; $archive = []; $query = $this->queries->forSourceWithTimeout(self::SOURCE_KEY, 15); foreach (array_chunk($items, self::SOURCE_ITEM_CHUNK) as $chunk) { $this->collectCandidates( $direct, $query->select($this->directSql(count($chunk)), $chunk), ); $this->collectCandidates( $archive, $query->select($this->archiveSql(count($chunk)), $chunk), ); } return [$direct, $archive]; } /** * @param array> $target * @param list> $rows */ private function collectCandidates(array &$target, array $rows): void { foreach ($rows as $row) { $itemNumber = trim((string) ($row['item_number'] ?? '')); $materialNumber = trim((string) ($row['material_number'] ?? '')); if ($itemNumber === '' || $materialNumber === '') { continue; } $target[$itemNumber][$materialNumber] = true; } } private function directSql(int $count): string { $placeholders = implode(', ', array_fill(0, $count, '?')); return 'SELECT "normalizedSku" AS item_number, ' . 'TRIM(updates->>\'materialNo\') AS material_number ' . 'FROM public."BarcodePrintImportRecord" ' . 'WHERE "normalizedSku" IN (' . $placeholders . ') ' . 'AND COALESCE(updates->>\'materialNo\', \'\') <> \'\' ' . 'GROUP BY "normalizedSku", updates->>\'materialNo\' ' . 'LIMIT 5000'; } private function archiveSql(int $count): string { $placeholders = implode(', ', array_fill(0, $count, '?')); return 'SELECT TRIM(sku) AS item_number, TRIM(barcode) AS material_number ' . 'FROM public."SupplyChainProductArchive" ' . 'WHERE sku IN (' . $placeholders . ') ' . 'AND COALESCE(TRIM(barcode), \'\') <> \'\' ' . 'GROUP BY sku, barcode ' . 'LIMIT 5000'; } /** * @param list $items * @param array $quoteMaterials * @param array> $direct * @param array> $archive * @return list> */ private function resolveRows( int $companyId, array $items, array $quoteMaterials, array $direct, array $archive, ): array { $quoteKeys = array_keys($quoteMaterials); $rows = []; $now = now(); foreach ($items as $itemNumber) { $directMatches = array_values(array_unique(array_intersect( array_keys($direct[$itemNumber] ?? []), $quoteKeys, ))); $archiveMatches = array_values(array_unique(array_intersect( array_keys($archive[$itemNumber] ?? []), $quoteKeys, ))); $matches = array_values(array_unique(array_merge($directMatches, $archiveMatches))); $materialNumber = null; $matchSource = null; if (count($directMatches) === 1) { $materialNumber = $directMatches[0]; $matchSource = 'barcode_material_no'; } elseif ($directMatches === [] && count($archiveMatches) === 1) { $materialNumber = $archiveMatches[0]; $matchSource = 'archive_barcode'; } $status = match (true) { $materialNumber !== null => 'matched', count($matches) > 1 => 'ambiguous', default => 'unmatched', }; $rows[] = [ 'company_id' => $companyId, 'item_number' => $itemNumber, 'material_number' => $materialNumber, 'match_status' => $status, 'match_source' => $matchSource, 'candidate_count' => count($matches), 'candidate_materials' => json_encode( $matches, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR, ), 'synced_at' => $now, ]; } return $rows; } /** @param list> $rows */ private function stats(array $rows): array { $counts = ['matched' => 0, 'unmatched' => 0, 'ambiguous' => 0]; $sources = ['barcode_material_no' => 0, 'archive_barcode' => 0]; foreach ($rows as $row) { $counts[$row['match_status']]++; if ($row['match_source'] !== null) { $sources[$row['match_source']]++; } } return [ 'items' => count($rows), 'matched' => $counts['matched'], 'unmatched' => $counts['unmatched'], 'ambiguous' => $counts['ambiguous'], 'direct' => $sources['barcode_material_no'], 'archive' => $sources['archive_barcode'], ]; } private function markState( string $status, int $rowCount, ?string $errorCode, float $started, ): void { if (! Schema::connection($this->connectionName())->hasTable('business_mirror_table_states')) { return; } $connection = $this->connection(); $now = now(); $existing = $connection->table('business_mirror_table_states') ->where('dataset_key', self::DATASET_KEY) ->first(); $values = [ 'source_key' => self::SOURCE_KEY, 'source_object' => 'public.SupplyChainProductArchive + public.BarcodePrintImportRecord', 'target_table' => 'kuka_posting_material_map_rows', 'status' => $status, 'run_mode' => 'targeted', 'last_attempted_at' => $now, 'last_error_code' => $errorCode === null ? null : mb_substr($errorCode, 0, 120), 'updated_at' => $now, ]; if ($status === 'succeeded') { $values += [ 'initial_completed_at' => $existing?->initial_completed_at ?? $now, 'last_succeeded_at' => $now, 'last_upserted_rows' => $rowCount, 'mirror_row_count' => $rowCount, ]; if (Schema::connection($this->connectionName()) ->hasColumn('business_mirror_table_states', 'mirror_row_count_is_exact')) { $values['mirror_row_count_is_exact'] = true; } } if ($existing !== null) { $connection->table('business_mirror_table_states') ->where('dataset_key', self::DATASET_KEY) ->update($values); return; } $connection->table('business_mirror_table_states')->insert($values + [ 'dataset_key' => self::DATASET_KEY, 'created_at' => $now, ]); } private function connection(): ConnectionInterface { return $this->databases->connection($this->connectionName()); } private function connectionName(): string { return (string) config('upstream_mirrors.connection', 'business_mirror'); } }