CaiWuXiTong/app/Business/KukaPostingBase/KukaPostingMaterialMapSynchronizer.php
chenxi 8563a3f635 fix: 允许多个货号共用物料编号 (#1126)
增加品牌SKU到报价单物料编号的回退匹配,同一物料编号可被多个货号共用。
2026-11-11 20:52:48 +08:00

397 lines
14 KiB
PHP

<?php
namespace App\Business\KukaPostingBase;
use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\DatabaseManager;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
use RuntimeException;
use Throwable;
class KukaPostingMaterialMapSynchronizer
{
public const SOURCE_KEY = 'furniture3d';
public const DATASET_KEY = 'furniture3d.kuka-material-map';
private const INSERT_CHUNK = 500;
private const SOURCE_ITEM_CHUNK = 200;
public function __construct(
private readonly DatabaseManager $databases,
private readonly UpstreamReadOnlyQueryFactory $queries,
) {
}
/**
* @return array{
* dry_run: bool,
* items: int,
* matched: int,
* unmatched: int,
* ambiguous: int,
* direct: int,
* archive: int,
* brand_sku: int
* }
*/
public function sync(int $companyId = 1, ?string $period = null, bool $dryRun = false): array
{
$connection = $this->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('当前账期没有可用于匹配的货号。');
}
$quotations = $this->activeQuotations($connection, $companyId);
[$direct, $archive, $brandSku] = $this->sourceCandidates($items);
$rows = $this->resolveRows($companyId, $items, $quotations, $direct, $archive, $brandSku);
$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<string> */
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<string,string> */
private function activeQuotations(ConnectionInterface $connection, int $companyId): array
{
return $connection->table('kuka_posting_quotation_rows')
->where('company_id', $companyId)
->where('record_status', '生效')
->get(['material_number', 'material_desc'])
->mapWithKeys(static function (object $row): array {
$materialNumber = trim((string) $row->material_number);
return $materialNumber === ''
? []
: [$materialNumber => trim((string) $row->material_desc)];
})
->all();
}
/**
* @param list<string> $items
* @return array{
* 0:array<string,array<string,true>>,
* 1:array<string,array<string,true>>,
* 2:array<string,array<string,true>>
* }
*/
private function sourceCandidates(array $items): array
{
$direct = [];
$archive = [];
$brandSku = [];
$query = $this->queries->forSourceWithTimeout(self::SOURCE_KEY, 15);
foreach (array_chunk($items, self::SOURCE_ITEM_CHUNK) as $chunk) {
$directRows = $query->select($this->directSql(count($chunk)), $chunk);
$this->collectCandidates(
$direct,
$directRows,
);
$this->collectCandidates($brandSku, $directRows, 'brand_sku');
$this->collectCandidates(
$archive,
$query->select($this->archiveSql(count($chunk)), $chunk),
);
}
return [$direct, $archive, $brandSku];
}
/**
* @param array<string,array<string,true>> $target
* @param list<array<string,mixed>> $rows
*/
private function collectCandidates(
array &$target,
array $rows,
string $materialField = 'material_number',
): void
{
foreach ($rows as $row) {
$itemNumber = trim((string) ($row['item_number'] ?? ''));
$materialNumber = trim((string) ($row[$materialField] ?? ''));
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, '
. 'TRIM(updates->>\'brandSkuCode\') AS brand_sku '
. 'FROM public."BarcodePrintImportRecord" '
. 'WHERE "normalizedSku" IN (' . $placeholders . ') '
. 'AND (COALESCE(updates->>\'materialNo\', \'\') <> \'\' '
. 'OR COALESCE(updates->>\'brandSkuCode\', \'\') <> \'\') '
. 'GROUP BY "normalizedSku", updates->>\'materialNo\', updates->>\'brandSkuCode\' '
. '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<string> $items
* @param array<string,true> $quoteMaterials
* @param array<string,array<string,true>> $direct
* @param array<string,array<string,true>> $archive
* @param array<string,array<string,true>> $brandSku
* @return list<array<string,mixed>>
*/
private function resolveRows(
int $companyId,
array $items,
array $quotations,
array $direct,
array $archive,
array $brandSku,
): array {
$quoteKeys = array_keys($quotations);
$brandMaterials = $this->quoteBrandMaterials($quotations);
$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,
)));
$brandMatches = [];
foreach (array_keys($brandSku[$itemNumber] ?? []) as $brand) {
foreach (array_keys($brandMaterials[$brand] ?? []) as $materialNumber) {
$brandMatches[] = $materialNumber;
}
}
$brandMatches = array_values(array_unique($brandMatches));
$matches = array_values(array_unique(array_merge(
$directMatches,
$archiveMatches,
$brandMatches,
)));
$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';
} elseif ($directMatches === [] && $archiveMatches === [] && count($brandMatches) === 1) {
$materialNumber = $brandMatches[0];
$matchSource = 'brand_sku';
}
$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 array<string,string> $quotations
* @return array<string,array<string,true>>
*/
private function quoteBrandMaterials(array $quotations): array
{
$materials = [];
foreach ($quotations as $materialNumber => $description) {
$brand = trim(explode('~', $description, 2)[0]);
if ($brand !== '') {
$materials[$brand][$materialNumber] = true;
}
}
return $materials;
}
/** @param list<array<string,mixed>> $rows */
private function stats(array $rows): array
{
$counts = ['matched' => 0, 'unmatched' => 0, 'ambiguous' => 0];
$sources = ['barcode_material_no' => 0, 'archive_barcode' => 0, 'brand_sku' => 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'],
'brand_sku' => $sources['brand_sku'],
];
}
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');
}
}