397 lines
14 KiB
PHP
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');
|
|
}
|
|
}
|