1371 lines
58 KiB
PHP
1371 lines
58 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Services\Jushuitan;
|
||
|
||
use Illuminate\Support\Carbon;
|
||
use Illuminate\Support\Facades\DB;
|
||
use InvalidArgumentException;
|
||
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
|
||
use RuntimeException;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 直读经营分析副本库(operations_replica)聚水潭发货明细，清洗写入 jst_oms_shipping。
|
||
* 对齐工作平台口径：明细=一个发货单×每件商品一行(导出明细)；合并由查询层按包裹归并(导出)。
|
||
*
|
||
* 每行 = 一个发货单(包裹, express_details) × 其一件商品(OrderItem)。
|
||
* - source_id = 发货单 id(包裹)
|
||
* - sku_source_id = 商品明细 id(件)
|
||
* - pkg_count = '1/N'(同包裹件数)
|
||
* - 体积 volume_cm3 = 数量 × 长 × 宽 × 高(工作平台 Svolume 规则)
|
||
* - 包裹级字段(实称重量/运费)只落在首件，其余置空 —— 便于合并 SUM 不重复
|
||
* - 剔除补差虚拟单；并入自动发货仓(jst_Orders 已发货但不在发货明细表)
|
||
*/
|
||
class ShippingReplicaSynchronizer
|
||
{
|
||
private const SOURCE = 'operations_replica';
|
||
private const DATASET = 'jst.sctrade';
|
||
private const FREIGHT_DATASET_PREFIX = 'jst.sctrade.freight.';
|
||
private const ITEMS_DATASET_PREFIX = 'jst.sctrade.items.';
|
||
private const AUTOSHIP_OFFSET = 1000000000;
|
||
private const DEMO_SOURCE_IDS = [9000001, 9000002, 9000003, 9000004, 9000005, 9000006];
|
||
private const FLUSH = 300;
|
||
private const PRESERVED_PACKAGE_FIELDS = [
|
||
'buyer_nick', 'seller_memo', 'package_no', 'download_time', 'freight_rate', 'freight_formula',
|
||
'settle_volume_cm3', 'recon_weight_kg', 'recon_volume_cm3', 'front_no',
|
||
];
|
||
private const PRESERVED_CONTACT_FIELDS = ['receiver_name', 'receiver_mobile', 'receiver_phone'];
|
||
private const PRESERVED_ITEM_FIELDS = ['finished_barcode', 'finished_name', 'finished_supplier'];
|
||
|
||
private const VIEW = 'rpt.Jst_SendData_View_extension';
|
||
private const OUTBOUND = 'rpt.jst_salesOutboundInfo';
|
||
|
||
private int $maxExpressBatches = 5000;
|
||
private int $maxAutoShipBatches = 600;
|
||
private bool $skipFreight = false;
|
||
private bool $replaceWindow = false;
|
||
|
||
/** @var array<int,list<array<string,mixed>>> */
|
||
private array $replacementPackages = [];
|
||
/** @var array<int,bool> */
|
||
private array $replacementHasFreight = [];
|
||
/** @var array<int,bool> */
|
||
private array $replacementHasItems = [];
|
||
/** @var array<int,array<int,true>> */
|
||
private array $replacementSyntheticSkuIds = [];
|
||
private int $replacementRowCount = 0;
|
||
private int $packagesProcessed = 0;
|
||
private int $packagesWithoutItems = 0;
|
||
private int $packagesRetainedForMissingItems = 0;
|
||
private int $packagesRetainedForUnmatchedAmount = 0;
|
||
private int $staleMirrorPackages = 0;
|
||
|
||
/** @var array<int,true> */
|
||
private array $seenSourceIds = [];
|
||
|
||
/** @var list<string> */
|
||
private array $dwsFailedDays = [];
|
||
|
||
private readonly \Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery $up;
|
||
private readonly \Illuminate\Database\ConnectionInterface $db;
|
||
|
||
/** @var array<int,array> */
|
||
private array $buffer = [];
|
||
|
||
/** @var array<string,array> 窗口内 dws 销售全量商品缓存 sourceno=>items */
|
||
private array $dwsCache = [];
|
||
|
||
/** @var array{byL:array<string,string>,byO:array<string,string>} 窗口内出库单映射 */
|
||
private array $outboundMap = ['byL' => [], 'byO' => []];
|
||
|
||
public function __construct(UpstreamReadOnlyQueryFactory $factory)
|
||
{
|
||
$this->up = $factory->forSource(self::SOURCE);
|
||
$this->db = DB::connection((string) config('jushuitan.connection', 'jushuitan_local'));
|
||
$this->db->disableQueryLog();
|
||
}
|
||
|
||
public static function make(): self
|
||
{
|
||
return new self(app(UpstreamReadOnlyQueryFactory::class));
|
||
}
|
||
|
||
public function sync(?Carbon $from = null, ?Carbon $to = null, int $limit = 2000, ?callable $progress = null, bool $skipFreight = false): array
|
||
{
|
||
$this->markRunning();
|
||
try {
|
||
$windowed = $from && $to;
|
||
if ($skipFreight && ! $windowed) {
|
||
throw new InvalidArgumentException('跳过运费仅支持指定起止日期的历史补录。');
|
||
}
|
||
$this->skipFreight = $skipFreight;
|
||
$this->replaceWindow = (bool) $windowed;
|
||
$this->packagesProcessed = 0;
|
||
$this->packagesWithoutItems = 0;
|
||
$this->packagesRetainedForMissingItems = 0;
|
||
$this->packagesRetainedForUnmatchedAmount = 0;
|
||
$this->staleMirrorPackages = 0;
|
||
$this->seenSourceIds = [];
|
||
$this->dwsFailedDays = [];
|
||
$now = now();
|
||
$fromS = $windowed ? $from->format('Y-m-d H:i:s') : null;
|
||
$toS = $windowed ? $to->format('Y-m-d H:i:s') : null;
|
||
if ($skipFreight) {
|
||
$this->markFreightIncomplete($fromS, $toS);
|
||
}
|
||
|
||
$this->db->table('jst_oms_shipping')->whereIn('source_id', self::DEMO_SOURCE_IDS)->delete();
|
||
if ($skipFreight && $progress) {
|
||
$progress('仅补数量：按包裹分批替换，已有运费/结算重量在事务中保留；不预删日期窗口');
|
||
}
|
||
|
||
$total = 0;
|
||
$auto = 0;
|
||
if ($windowed) {
|
||
$this->preloadDws($from, $to);
|
||
$this->preloadOutboundMap($fromS, $toS);
|
||
if ($progress) {
|
||
$progress('dws 预载完成：' . count($this->dwsCache) . ' 个 sourceno');
|
||
$progress(sprintf(
|
||
'出库映射预载完成：%d 个快递单号，%d 个内部单号',
|
||
count($this->outboundMap['byL']),
|
||
count($this->outboundMap['byO']),
|
||
));
|
||
}
|
||
$total = $this->syncExpress(true, $fromS, $toS, $limit, $now, $progress);
|
||
$auto = $this->syncAutoShip($fromS, $toS, $now, $progress);
|
||
$this->staleMirrorPackages = $this->countUnseenMirrorPackages($fromS, $toS);
|
||
if (! $skipFreight && $this->staleMirrorPackages === 0) {
|
||
$removed = $this->dedupAtPrefix($fromS, $toS);
|
||
if ($progress && $removed > 0) {
|
||
$progress('去除@重复单号：' . $removed . ' 条');
|
||
}
|
||
}
|
||
} else {
|
||
$total = $this->syncExpress(false, $fromS, $toS, $limit, $now, $progress);
|
||
}
|
||
|
||
if ($windowed && ($this->dwsFailedDays !== [] || $this->packagesWithoutItems > 0
|
||
|| $this->packagesRetainedForUnmatchedAmount > 0)) {
|
||
$this->markItemsIncomplete($fromS, $toS);
|
||
} elseif ($windowed && ! $skipFreight && $this->staleMirrorPackages === 0) {
|
||
$this->markCovered(self::ITEMS_DATASET_PREFIX, $fromS, $toS);
|
||
}
|
||
|
||
$result = ['processed' => $total, 'autoship' => $auto];
|
||
if ($windowed) {
|
||
$result['packages'] = $this->packagesProcessed;
|
||
$result['unique_packages'] = count($this->seenSourceIds);
|
||
$result['packages_without_items'] = $this->packagesWithoutItems;
|
||
$result['packages_retained_for_missing_items'] = $this->packagesRetainedForMissingItems;
|
||
$result['packages_retained_for_unmatched_amount'] = $this->packagesRetainedForUnmatchedAmount;
|
||
$result['dws_failed_days'] = $this->dwsFailedDays;
|
||
$result['stale_mirror_packages'] = $this->staleMirrorPackages;
|
||
}
|
||
if ($skipFreight) {
|
||
$result['freight_incomplete'] = true;
|
||
}
|
||
|
||
if ($this->staleMirrorPackages > 0) {
|
||
$this->markIncompleteWindow($this->staleMirrorPackages);
|
||
} else {
|
||
$this->markSucceeded();
|
||
}
|
||
return $result;
|
||
} catch (Throwable $e) {
|
||
$this->markFailed($e->getMessage());
|
||
throw $e;
|
||
} finally {
|
||
$this->dwsCache = [];
|
||
$this->outboundMap = ['byL' => [], 'byO' => []];
|
||
$this->replacementPackages = [];
|
||
$this->replacementHasFreight = [];
|
||
$this->replacementHasItems = [];
|
||
$this->replacementSyntheticSkuIds = [];
|
||
$this->replacementRowCount = 0;
|
||
$this->skipFreight = false;
|
||
$this->replaceWindow = false;
|
||
$this->dwsFailedDays = [];
|
||
$this->seenSourceIds = [];
|
||
gc_collect_cycles();
|
||
}
|
||
}
|
||
|
||
/** 仅在同一订单、同一发货日去重；提供日期窗口时不使用窗口外记录作匹配。 */
|
||
public function dedupAtPrefix(?string $fromS = null, ?string $toS = null): int
|
||
{
|
||
$query = $this->db->table('jst_oms_shipping')
|
||
->whereNotNull('express_no')->where('express_no', '!=', '');
|
||
if ($fromS !== null && $toS !== null) {
|
||
$query->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS);
|
||
}
|
||
$clean = [];
|
||
$prefixed = [];
|
||
foreach ($query->distinct()->cursor(['source_id', 'trade_no', 'express_no', 'ship_time']) as $row) {
|
||
$tradeNo = trim((string) $row->trade_no);
|
||
$expressNo = (string) $row->express_no;
|
||
$sourceId = (int) $row->source_id;
|
||
$shipDate = substr((string) $row->ship_time, 0, 10);
|
||
if ($tradeNo === '' || $sourceId <= 0 || $shipDate === '') {
|
||
continue;
|
||
}
|
||
if ($expressNo[0] === '@') {
|
||
$prefixed[$sourceId] = [$tradeNo, $shipDate, substr($expressNo, 1)];
|
||
} else {
|
||
$clean[$tradeNo][$shipDate][$expressNo] = true;
|
||
}
|
||
}
|
||
$toDel = [];
|
||
foreach ($prefixed as $sourceId => [$tradeNo, $shipDate, $expressNo]) {
|
||
if ($expressNo !== '' && isset($clean[$tradeNo][$shipDate][$expressNo])) {
|
||
$toDel[] = $sourceId;
|
||
}
|
||
}
|
||
$removed = 0;
|
||
foreach (array_chunk($toDel, 500) as $chunk) {
|
||
$delete = $this->db->table('jst_oms_shipping')->whereIn('source_id', $chunk);
|
||
if ($fromS !== null && $toS !== null) {
|
||
$delete->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS);
|
||
}
|
||
$removed += (int) $delete->delete();
|
||
}
|
||
|
||
return $removed;
|
||
}
|
||
|
||
/** 主口径：发货单(dlv 镜像) × 每件商品。 */
|
||
private function syncExpress(bool $windowed, ?string $fromS, ?string $toS, int $limit, $now, ?callable $progress = null): int
|
||
{
|
||
$batch = 1500;
|
||
$maxBatches = $this->maxExpressBatches;
|
||
$cols = 'e.id, e.so_id, e.inner_order_id, e.logistic_id, e.logistic_company_name, e.send_time, e.pay_time, '
|
||
. 'e.shop_name, e.platform_name, e.dis_co_name, e.warehouse_name, e.receiver_address, '
|
||
. 'e.weight, e.f_weight, e.buyer_message, e.order_labels, jo.l_id AS jo_lid, jo.send_date AS jo_send';
|
||
|
||
// 窗口模式：只按主键 e.id 翻页(有索引，毫秒级)，send_time 窗口过滤放到 PHP，
|
||
// 避免每页都对无索引的 send_time 做全表扫描(那是"卡死"的真凶)。
|
||
$fromTs = $fromS !== null ? strtotime($fromS) : null;
|
||
$toTs = $toS !== null ? strtotime($toS) : null;
|
||
|
||
$total = 0;
|
||
$scanned = 0;
|
||
$lastId = null;
|
||
for ($b = 0; $b <= $maxBatches; $b++) {
|
||
$where = $windowed ? '1 = 1' : 'e.send_time IS NOT NULL';
|
||
$bind = [];
|
||
if ($lastId !== null) {
|
||
$where .= ' AND e.id < ?';
|
||
$bind[] = $lastId;
|
||
}
|
||
$sql = "SELECT TOP {$batch} {$cols} FROM rpt.jst_order_express_details e "
|
||
. 'LEFT JOIN rpt.jst_Orders jo ON jo.o_id = e.inner_order_id '
|
||
. "WHERE {$where} ORDER BY e.id DESC";
|
||
$rows = $this->up->select($sql, $bind);
|
||
if ($rows === []) {
|
||
break;
|
||
}
|
||
if ($b === $maxBatches) {
|
||
throw new RuntimeException('发货扫描达到批次上限，仍有未读取的上游数据。');
|
||
}
|
||
$cnt = count($rows);
|
||
$scanned += $cnt;
|
||
foreach ($rows as $r) {
|
||
$lastId = $r['id'];
|
||
}
|
||
|
||
if ($windowed) {
|
||
$kept = [];
|
||
foreach ($rows as $r) {
|
||
$st = $r['send_time'] ?? null;
|
||
$ts = ($st !== null && $st !== '') ? strtotime((string) $st) : null;
|
||
if ($ts === null) {
|
||
$jd = $r['jo_send'] ?? null;
|
||
$ts = ($jd !== null && $jd !== '') ? strtotime((string) $jd) : null;
|
||
}
|
||
if ($ts !== null && $ts >= $fromTs && $ts < $toTs) {
|
||
$kept[] = $r;
|
||
}
|
||
}
|
||
$rows = $kept;
|
||
}
|
||
|
||
$oids = [];
|
||
$soIds = [];
|
||
$logs = [];
|
||
foreach ($rows as $r) {
|
||
$so = (string) ($r['so_id'] ?? '');
|
||
if ($so !== '') {
|
||
$soIds[$so] = true;
|
||
}
|
||
$oid = (string) ($r['inner_order_id'] ?? '');
|
||
if ($oid !== '') {
|
||
$oids[$oid] = true;
|
||
}
|
||
$lg = (string) ($r['logistic_id'] ?? '');
|
||
if ($lg !== '') {
|
||
$logs[$lg] = true;
|
||
}
|
||
}
|
||
[$byOid, $bySo] = $this->loadAllItems(array_keys($oids), array_keys($soIds), ! $windowed);
|
||
$dws = [];
|
||
foreach (array_keys($soIds) as $mso) {
|
||
if (empty($bySo[$mso]) && ! empty($this->dwsCache[$mso])) {
|
||
$dws[$mso] = $this->dwsCache[$mso];
|
||
}
|
||
}
|
||
// 主口径：发货数据视图按出库单(io_id)取实发商品
|
||
$ioMap = $this->ioMapFor($windowed, array_keys($logs), array_keys($oids));
|
||
$ioIds = [];
|
||
foreach ($rows as $r) {
|
||
$io = $this->ioFor($ioMap, (string) ($r['logistic_id'] ?? ''), (string) ($r['inner_order_id'] ?? ''));
|
||
if ($io !== '') {
|
||
$ioIds[$io] = true;
|
||
}
|
||
}
|
||
$viewItems = $this->loadViewItems(array_keys($ioIds));
|
||
$skus = $this->loadSkus($this->collectSkuIds($byOid, $bySo, $dws, $viewItems));
|
||
$freight = $this->skipFreight ? [] : $this->loadFreight(array_keys($soIds));
|
||
|
||
foreach ($rows as $r) {
|
||
$pkgId = (int) ($r['id'] ?? 0);
|
||
if ($pkgId === 0) {
|
||
continue;
|
||
}
|
||
if ($this->excludedByLabel($r['order_labels'] ?? null)) {
|
||
continue; // 线上发货(平台代发/云仓)：工作平台不拉，对齐排除
|
||
}
|
||
$so = (string) ($r['so_id'] ?? '');
|
||
$oid = (string) ($r['inner_order_id'] ?? '');
|
||
$io = $this->ioFor($ioMap, (string) ($r['logistic_id'] ?? ''), $oid);
|
||
$items = ($io !== '' && ! empty($viewItems[$io]))
|
||
? $viewItems[$io]
|
||
: $this->itemsFor($byOid, $bySo, $dws, $oid, $so);
|
||
$tracking = $this->tracking((string) ($r['logistic_id'] ?? ''), (string) ($r['jo_lid'] ?? ''), $freight, $so);
|
||
$shipTime = $this->ts($r['send_time'] ?? null) ?? $this->ts($r['jo_send'] ?? null);
|
||
$pkg = $this->pkgFromExpress($r, $tracking, $shipTime);
|
||
$total += $this->emitPackage($pkgId, $pkg, $items, $skus, $freight, $so, $tracking, $now);
|
||
}
|
||
|
||
unset($rows, $byOid, $bySo, $dws, $skus, $freight);
|
||
if ($b % 20 === 0) {
|
||
gc_collect_cycles();
|
||
}
|
||
if ($progress && $b % 20 === 0) {
|
||
$progress(sprintf('发货：已扫描 %d 行，命中窗口 %d 件', $scanned, $total));
|
||
}
|
||
if ($cnt < $batch) {
|
||
break;
|
||
}
|
||
if (! $windowed && $total >= $limit) {
|
||
break;
|
||
}
|
||
}
|
||
if ($progress) {
|
||
$progress(sprintf('发货完成：扫描 %d 行，处理 %d 件', $scanned, $total));
|
||
}
|
||
$this->flushPending();
|
||
|
||
return $total;
|
||
}
|
||
|
||
/** 并入自动发货仓：jst_Orders 已发货但不在发货明细表。 */
|
||
private function syncAutoShip(string $fromS, string $toS, $now, ?callable $progress = null): int
|
||
{
|
||
$batch = 400;
|
||
$maxBatches = $this->maxAutoShipBatches;
|
||
$cols = 'o.Id AS rid, o.o_id, o.so_id, o.l_id, o.logistics_company, o.order_from, o.labels, '
|
||
. 'o.shop_name, o.send_date, o.pay_date, o.weight, o.buyer_message, '
|
||
. 'o.receiver_state, o.receiver_city, o.receiver_district, o.receiver_address, '
|
||
. 'o.receiver_name, o.receiver_mobile, o.receiver_phone';
|
||
|
||
$total = 0;
|
||
$lastRid = null;
|
||
|
||
for ($b = 0; $b <= $maxBatches; $b++) {
|
||
$where = "o.status = 'Sent' AND o.send_date >= ? AND o.send_date < ? "
|
||
. 'AND NOT EXISTS (SELECT 1 FROM rpt.jst_order_express_details e WHERE e.inner_order_id = o.o_id)';
|
||
$bind = [$fromS, $toS];
|
||
if ($lastRid !== null) {
|
||
$where .= ' AND o.Id < ?';
|
||
$bind[] = $lastRid;
|
||
}
|
||
$sql = "SELECT TOP {$batch} {$cols} FROM rpt.jst_Orders o WHERE {$where} ORDER BY o.Id DESC";
|
||
$rows = $this->up->select($sql, $bind);
|
||
if ($rows === []) {
|
||
break;
|
||
}
|
||
if ($b === $maxBatches) {
|
||
throw new RuntimeException('代发仓扫描达到批次上限，仍有未读取的上游数据。');
|
||
}
|
||
|
||
$oids = [];
|
||
$soIds = [];
|
||
$logs = [];
|
||
foreach ($rows as $r) {
|
||
$so = (string) ($r['so_id'] ?? '');
|
||
if ($so !== '') {
|
||
$soIds[$so] = true;
|
||
}
|
||
$oid = (string) ($r['o_id'] ?? '');
|
||
if ($oid !== '') {
|
||
$oids[$oid] = true;
|
||
}
|
||
$lg = (string) ($r['l_id'] ?? '');
|
||
if ($lg !== '') {
|
||
$logs[$lg] = true;
|
||
}
|
||
$lastRid = $r['rid'];
|
||
}
|
||
[$byOid, $bySo] = $this->loadAllItems(array_keys($oids), array_keys($soIds), false);
|
||
$dws = [];
|
||
foreach (array_keys($soIds) as $mso) {
|
||
if (empty($bySo[$mso]) && ! empty($this->dwsCache[$mso])) {
|
||
$dws[$mso] = $this->dwsCache[$mso];
|
||
}
|
||
}
|
||
$ioMap = $this->ioMapFor(true, array_keys($logs), array_keys($oids));
|
||
$ioIds = [];
|
||
foreach ($rows as $r) {
|
||
$io = $this->ioFor($ioMap, (string) ($r['l_id'] ?? ''), (string) ($r['o_id'] ?? ''));
|
||
if ($io !== '') {
|
||
$ioIds[$io] = true;
|
||
}
|
||
}
|
||
$viewItems = $this->loadViewItems(array_keys($ioIds));
|
||
$skus = $this->loadSkus($this->collectSkuIds($byOid, $bySo, $dws, $viewItems));
|
||
$freight = $this->skipFreight ? [] : $this->loadFreight(array_keys($soIds));
|
||
|
||
$cnt = count($rows);
|
||
foreach ($rows as $r) {
|
||
$oid = (string) ($r['o_id'] ?? '');
|
||
if ($oid === '') {
|
||
continue;
|
||
}
|
||
if ($this->excludedByLabel($r['labels'] ?? null)) {
|
||
continue; // 线上发货：对齐排除
|
||
}
|
||
$so = (string) ($r['so_id'] ?? '');
|
||
$io = $this->ioFor($ioMap, (string) ($r['l_id'] ?? ''), $oid);
|
||
$items = ($io !== '' && ! empty($viewItems[$io]))
|
||
? $viewItems[$io]
|
||
: $this->itemsFor($byOid, $bySo, $dws, $oid, $so);
|
||
$tracking = $this->tracking((string) ($r['l_id'] ?? ''), '', $freight, $so);
|
||
$shipTime = $this->ts($r['send_date'] ?? null);
|
||
$pkg = $this->pkgFromOrder($r, $tracking, $shipTime);
|
||
$pkgId = self::AUTOSHIP_OFFSET + (int) $oid;
|
||
$total += $this->emitPackage($pkgId, $pkg, $items, $skus, $freight, $so, $tracking, $now, true);
|
||
}
|
||
unset($rows, $byOid, $bySo, $dws, $skus, $freight);
|
||
if ($b % 30 === 0) {
|
||
gc_collect_cycles();
|
||
}
|
||
if ($cnt < $batch) {
|
||
break;
|
||
}
|
||
}
|
||
if ($progress) {
|
||
$progress(sprintf('代发仓完成：处理 %d 件', $total));
|
||
}
|
||
$this->flushPending();
|
||
|
||
return $total;
|
||
}
|
||
|
||
/**
|
||
* 输出一个包裹的所有明细行(每件一行；无商品则一行空商品)。
|
||
* @return int 组装行数
|
||
*/
|
||
private function emitPackage(int $pkgId, array $pkg, array $items, array $skus, array $freight, string $so, ?string $tracking, $now, bool $auto = false): int
|
||
{
|
||
$items = array_values(array_filter($items, fn ($it) => ! $this->isPricePatch($it['name'] ?? null)));
|
||
$n = count($items);
|
||
$this->seenSourceIds[$pkgId] = true;
|
||
$this->packagesProcessed++;
|
||
if ($n === 0) {
|
||
$this->packagesWithoutItems++;
|
||
}
|
||
$f = $freight[$so . '|' . ($tracking ?? '')] ?? ($freight[$so . '|'] ?? []);
|
||
$hasFreight = isset($f['cost']);
|
||
$packageRows = [];
|
||
|
||
if ($n === 0) {
|
||
$packageRows[] = $this->packageRow($pkgId, $pkgId, $this->buildRow($pkg, [], [], $f, true, '1/1', $now, $auto));
|
||
$this->writePackage($pkgId, $packageRows, $hasFreight, false, []);
|
||
|
||
return 1;
|
||
}
|
||
|
||
$count = 0;
|
||
$syntheticSkuIds = [];
|
||
foreach ($items as $idx => $it) {
|
||
$first = ($idx === 0);
|
||
$sk = (! empty($it['sku']) && isset($skus[(string) $it['sku']])) ? $skus[(string) $it['sku']] : [];
|
||
$skuSid = (int) ($it['id'] ?? ($pkgId * 100 + $idx));
|
||
if (($it['id'] ?? null) === null) {
|
||
$syntheticSkuIds[$skuSid] = true;
|
||
}
|
||
$packageRows[] = $this->packageRow($pkgId, $skuSid, $this->buildRow($pkg, $it, $sk, $f, $first, '1/' . $n, $now, $auto));
|
||
$count++;
|
||
}
|
||
|
||
$this->writePackage($pkgId, $packageRows, $hasFreight, true, $syntheticSkuIds);
|
||
|
||
return $count;
|
||
}
|
||
|
||
/** 组装一件明细行。包裹级字段(实称重量/运费)仅首件携带。 */
|
||
private function buildRow(array $pkg, array $it, array $sk, array $f, bool $first, string $countStr, $now, bool $auto): array
|
||
{
|
||
$qty = (float) ($it['qty'] ?? 0);
|
||
$skuW = (float) (($it['weight'] ?? null) ?: ($sk['weight'] ?? 0));
|
||
$onlineW = $skuW > 0 ? round($skuW * ($qty > 0 ? $qty : 1), 4) : null;
|
||
$L = ($it['l'] ?? null) ?: ($sk['l'] ?? null);
|
||
$W = ($it['w'] ?? null) ?: ($sk['w'] ?? null);
|
||
$H = ($it['h'] ?? null) ?: ($sk['h'] ?? null);
|
||
$vol = ($it['vol'] ?? null) ? round((float) $it['vol'], 4) : (($L && $W && $H) ? round((float) $L * (float) $W * (float) $H, 4) : null);
|
||
|
||
$pkgWeight = (float) ($pkg['weight'] ?? 0) ?: null;
|
||
$freightW = (float) ($f['w'] ?? 0);
|
||
$settleW = $this->skipFreight ? null : ($freightW > 0 ? $freightW : ($pkgWeight ?: $onlineW));
|
||
|
||
return $this->cleanPayload([
|
||
'express_no' => $pkg['tracking'],
|
||
'trade_no' => $pkg['so_id'],
|
||
'item_id' => $it['shopid'] ?? null,
|
||
'barcode' => $it['sku'] ?? null,
|
||
'product_name' => ($it['name'] ?? null) ?: ($sk['name'] ?? null),
|
||
'qty' => $it['qty'] ?? null,
|
||
'pkg_count' => $countStr,
|
||
'product_code' => ($it['iid'] ?? null) ?: ($sk['iid'] ?? null),
|
||
'express_name1' => $pkg['express_name1'],
|
||
'express_name2' => ($f['name'] ?? null) ?: $pkg['express_name1'],
|
||
'ship_time' => $pkg['ship_time'],
|
||
'pay_time' => $pkg['pay_time'],
|
||
'shop_name' => $pkg['shop_name'],
|
||
'platform' => $pkg['platform'],
|
||
'supplier_name' => ($it['supplier'] ?? null) ?: (($sk['supplier'] ?? null) ?: $pkg['dis_co_name']),
|
||
'warehouse_name' => $pkg['warehouse_name'] ?: ($it['warehouse'] ?? null),
|
||
'receiver_state' => $pkg['r_state'],
|
||
'receiver_city' => $pkg['r_city'],
|
||
'receiver_district' => $pkg['r_district'],
|
||
'receiver_address' => $pkg['r_address'],
|
||
'receiver_name' => $pkg['r_name'],
|
||
'receiver_mobile' => $pkg['r_mobile'],
|
||
'receiver_phone' => $pkg['r_phone'],
|
||
'online_weight' => $onlineW,
|
||
'actual_weight' => $first ? $pkgWeight : null,
|
||
'length_cm' => $L,
|
||
'width_cm' => $W,
|
||
'height_cm' => $H,
|
||
'volume_cm3' => $vol,
|
||
'weight_kg' => $first ? $settleW : null,
|
||
'settle_weight_kg' => $first ? $settleW : null,
|
||
'amount' => $it['amount'] ?? null,
|
||
'freight' => $first ? ($f['cost'] ?? null) : null,
|
||
'settle_freight' => $first ? ($f['cost'] ?? null) : null,
|
||
'buyer_memo' => $pkg['buyer_memo'],
|
||
'modified' => $pkg['ship_time'],
|
||
'synced_at' => $now,
|
||
]);
|
||
}
|
||
|
||
/** 发货明细表行 → 包裹级字段(省市区从地址拆) */
|
||
private function pkgFromExpress(array $r, ?string $tracking, ?string $shipTime): array
|
||
{
|
||
$addr = (string) ($r['receiver_address'] ?? '');
|
||
$parts = $addr !== '' ? array_values(array_filter(explode('-', $addr))) : [];
|
||
|
||
return [
|
||
'tracking' => $tracking,
|
||
'so_id' => $r['so_id'] ?? null,
|
||
'express_name1' => $r['logistic_company_name'] ?? null,
|
||
'ship_time' => $shipTime,
|
||
'pay_time' => $this->ts($r['pay_time'] ?? null),
|
||
'shop_name' => $r['shop_name'] ?? null,
|
||
'platform' => $r['platform_name'] ?? null,
|
||
'dis_co_name' => $r['dis_co_name'] ?? null,
|
||
'warehouse_name' => $r['warehouse_name'] ?? null,
|
||
'r_state' => $parts[0] ?? null,
|
||
'r_city' => $parts[1] ?? null,
|
||
'r_district' => $parts[2] ?? null,
|
||
'r_address' => $addr,
|
||
'r_name' => null,
|
||
'r_mobile' => null,
|
||
'r_phone' => null,
|
||
'buyer_memo' => $r['buyer_message'] ?? null,
|
||
'weight' => $r['weight'] ?? null,
|
||
];
|
||
}
|
||
|
||
/** jst_Orders 行(自动发货仓) → 包裹级字段 */
|
||
private function pkgFromOrder(array $r, ?string $tracking, ?string $shipTime): array
|
||
{
|
||
return [
|
||
'tracking' => $tracking,
|
||
'so_id' => $r['so_id'] ?? null,
|
||
'express_name1' => $r['logistics_company'] ?? null,
|
||
'ship_time' => $shipTime,
|
||
'pay_time' => $this->ts($r['pay_date'] ?? null),
|
||
'shop_name' => $r['shop_name'] ?? null,
|
||
'platform' => $r['order_from'] ?? null,
|
||
'dis_co_name' => null,
|
||
'warehouse_name' => null,
|
||
'r_state' => $r['receiver_state'] ?? null,
|
||
'r_city' => $r['receiver_city'] ?? null,
|
||
'r_district' => $r['receiver_district'] ?? null,
|
||
'r_address' => $r['receiver_address'] ?? '',
|
||
'r_name' => $r['receiver_name'] ?? null,
|
||
'r_mobile' => $r['receiver_mobile'] ?? null,
|
||
'r_phone' => $r['receiver_phone'] ?? null,
|
||
'buyer_memo' => $r['buyer_message'] ?? null,
|
||
'weight' => $r['weight'] ?? null,
|
||
];
|
||
}
|
||
|
||
private function tracking(string $logisticId, string $joLid, array $freight, string $so): ?string
|
||
{
|
||
if ($logisticId !== '') {
|
||
return $logisticId;
|
||
}
|
||
if ($joLid !== '') {
|
||
return $joLid;
|
||
}
|
||
$f = $freight[$so . '|'] ?? [];
|
||
$no = (string) ($f['no'] ?? '');
|
||
|
||
return $no !== '' ? $no : null;
|
||
}
|
||
|
||
/** @return array{0:array<string,array>,1:array<string,array>} [byOid, bySo] */
|
||
private function loadAllItems(array $oids, array $soIds, bool $loadBySoId = true): array
|
||
{
|
||
$byOid = [];
|
||
$bySo = [];
|
||
$sel = 'o_id, raw_so_id, sku_id, i_id, shop_i_id, name, qty, amount, send_warehouse, Id';
|
||
|
||
foreach (array_chunk($oids, 800) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
foreach ($this->up->select("SELECT {$sel} FROM rpt.jst_OrderItem WHERE o_id IN ({$ph}) ORDER BY Id", $chunk) as $r) {
|
||
$k = (string) ($r['o_id'] ?? '');
|
||
if ($k !== '') {
|
||
$byOid[$k][] = $this->itemArr($r);
|
||
}
|
||
}
|
||
}
|
||
// raw_so_id lacks a usable index in the production replica. Windowed sync already
|
||
// preloads DWS rows by sourceno, so querying this fallback can turn every batch into
|
||
// a full-table scan and exceed the upstream timeout. Keep it only for the small,
|
||
// non-windowed diagnostic mode where DWS is not preloaded.
|
||
if ($loadBySoId) {
|
||
foreach (array_chunk($soIds, 800) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
foreach ($this->up->select("SELECT {$sel} FROM rpt.jst_OrderItem WHERE raw_so_id IN ({$ph}) ORDER BY Id", $chunk) as $r) {
|
||
$k = (string) ($r['raw_so_id'] ?? '');
|
||
if ($k !== '') {
|
||
$bySo[$k][] = $this->itemArr($r);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
return [$byOid, $bySo];
|
||
}
|
||
|
||
/** 开跑前按天预加载窗口内 dws 销售全量到内存，避免逐批扫表。 */
|
||
private function preloadDws(\Illuminate\Support\Carbon $from, \Illuminate\Support\Carbon $to): void
|
||
{
|
||
$this->dwsCache = [];
|
||
$seen = [];
|
||
$day = $from->copy()->subDay()->startOfDay();
|
||
$end = $to->copy();
|
||
while ($day < $end) {
|
||
$d = $day->format('Y-m-d');
|
||
$day = $day->addDay();
|
||
try {
|
||
$rows = $this->up->select("SELECT sourceno, item_id, skuoid, goods_count, sales_amount FROM rpt.dws_sn_sales_full_info_di WHERE stat_date = ?", [$d]);
|
||
} catch (Throwable) {
|
||
$this->dwsFailedDays[] = $d;
|
||
continue;
|
||
}
|
||
foreach ($rows as $r) {
|
||
$so = (string) ($r['sourceno'] ?? '');
|
||
$bc = (string) ($r['skuoid'] ?? '');
|
||
if ($so === '' || $bc === '') {
|
||
continue;
|
||
}
|
||
$k = $so.'|'.$bc;
|
||
if (isset($seen[$k])) {
|
||
continue;
|
||
}
|
||
$seen[$k] = true;
|
||
$this->dwsCache[$so][] = [
|
||
'id' => null, 'sku' => $bc, 'iid' => null, 'shopid' => $r['item_id'] ?? null,
|
||
'name' => null, 'qty' => $r['goods_count'] ?? null, 'amount' => $r['sales_amount'] ?? null, 'warehouse' => null,
|
||
];
|
||
}
|
||
unset($rows);
|
||
}
|
||
}
|
||
|
||
/** jst_OrderItem 缺失时的兜底：dws 销售全量按 sourceno 取商品(条码/商品ID/数量/金额)。 */
|
||
private function loadDwsItems(array $soIds): array
|
||
{
|
||
$out = [];
|
||
$seen = [];
|
||
foreach (array_chunk($soIds, 500) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
$sql = "SELECT sourceno, item_id, skuoid, goods_count, sales_amount FROM rpt.dws_sn_sales_full_info_di WHERE sourceno IN ({$ph})";
|
||
foreach ($this->up->select($sql, $chunk) as $r) {
|
||
$so = (string) ($r['sourceno'] ?? '');
|
||
$barcode = (string) ($r['skuoid'] ?? '');
|
||
if ($so === '' || $barcode === '') {
|
||
continue;
|
||
}
|
||
$k = $so.'|'.$barcode;
|
||
if (isset($seen[$k])) {
|
||
continue;
|
||
}
|
||
$seen[$k] = true;
|
||
$out[$so][] = [
|
||
'id' => null, 'sku' => $barcode, 'iid' => null, 'shopid' => $r['item_id'] ?? null,
|
||
'name' => null, 'qty' => $r['goods_count'] ?? null, 'amount' => $r['sales_amount'] ?? null, 'warehouse' => null,
|
||
];
|
||
}
|
||
}
|
||
|
||
return $out;
|
||
}
|
||
|
||
private function itemArr(array $r): array
|
||
{
|
||
return [
|
||
'id' => $r['Id'] ?? null, 'sku' => $r['sku_id'] ?? null, 'iid' => $r['i_id'] ?? null,
|
||
'shopid' => $r['shop_i_id'] ?? null, 'name' => $r['name'] ?? null,
|
||
'qty' => $r['qty'] ?? null, 'amount' => $r['amount'] ?? null, 'warehouse' => $r['send_warehouse'] ?? null,
|
||
];
|
||
}
|
||
|
||
/** 该包裹的商品：优先 o_id 精确，取不到回落 so_id。 */
|
||
private function itemsFor(array $byOid, array $bySo, array $dws, string $oid, string $so): array
|
||
{
|
||
if ($oid !== '' && ! empty($byOid[$oid])) {
|
||
return $byOid[$oid];
|
||
}
|
||
if ($so !== '' && ! empty($bySo[$so])) {
|
||
return $bySo[$so];
|
||
}
|
||
if ($so !== '' && ! empty($dws[$so])) {
|
||
return $dws[$so]; // jst_OrderItem 取不到时，用 dws 销售全量兜底
|
||
}
|
||
|
||
return [];
|
||
}
|
||
|
||
/** 窗口同步按 io_date 一次预载，避免对无可用索引的 l_id/o_id 反复执行 IN 查询。 */
|
||
private function preloadOutboundMap(string $fromS, string $toS): void
|
||
{
|
||
$this->outboundMap = ['byL' => [], 'byO' => []];
|
||
$sql = 'SELECT l_id, o_id, io_id FROM ' . self::OUTBOUND . ' WHERE io_date >= ? AND io_date < ?';
|
||
foreach ($this->up->select($sql, [$fromS, $toS]) as $r) {
|
||
$io = (string) ($r['io_id'] ?? '');
|
||
if ($io === '') {
|
||
continue;
|
||
}
|
||
$l = (string) ($r['l_id'] ?? '');
|
||
if ($l !== '' && ! isset($this->outboundMap['byL'][$l])) {
|
||
$this->outboundMap['byL'][$l] = $io;
|
||
}
|
||
$o = (string) ($r['o_id'] ?? '');
|
||
if ($o !== '' && ! isset($this->outboundMap['byO'][$o])) {
|
||
$this->outboundMap['byO'][$o] = $io;
|
||
}
|
||
}
|
||
}
|
||
|
||
private function ioMapFor(bool $windowed, array $logisticIds, array $oIds): array
|
||
{
|
||
return $windowed ? $this->outboundMap : $this->loadIoMap($logisticIds, $oIds);
|
||
}
|
||
|
||
/** 快递单号/内部单号 -> 出库单 io_id（优先快递单号，精确到该次出库）。 */
|
||
private function loadIoMap(array $logisticIds, array $oIds): array
|
||
{
|
||
$byL = [];
|
||
$byO = [];
|
||
foreach (array_chunk(array_values(array_unique(array_filter($logisticIds))), 800) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
foreach ($this->up->select('SELECT l_id, io_id FROM ' . self::OUTBOUND . " WHERE l_id IN ({$ph})", $chunk) as $r) {
|
||
$l = (string) ($r['l_id'] ?? '');
|
||
if ($l !== '' && ! isset($byL[$l])) {
|
||
$byL[$l] = (string) ($r['io_id'] ?? '');
|
||
}
|
||
}
|
||
}
|
||
foreach (array_chunk(array_values(array_unique(array_filter($oIds))), 800) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
foreach ($this->up->select('SELECT o_id, io_id FROM ' . self::OUTBOUND . " WHERE o_id IN ({$ph})", $chunk) as $r) {
|
||
$o = (string) ($r['o_id'] ?? '');
|
||
if ($o !== '' && ! isset($byO[$o])) {
|
||
$byO[$o] = (string) ($r['io_id'] ?? '');
|
||
}
|
||
}
|
||
}
|
||
|
||
return ['byL' => $byL, 'byO' => $byO];
|
||
}
|
||
|
||
private function ioFor(array $ioMap, string $logisticId, string $oid): string
|
||
{
|
||
if ($logisticId !== '' && ! empty($ioMap['byL'][$logisticId])) {
|
||
return (string) $ioMap['byL'][$logisticId];
|
||
}
|
||
if ($oid !== '' && ! empty($ioMap['byO'][$oid])) {
|
||
return (string) $ioMap['byO'][$oid];
|
||
}
|
||
|
||
return '';
|
||
}
|
||
|
||
/** io_id -> 实发商品[itemArr]（发货数据视图，按(io_id,条码)去重，type=0 之和优先）。 */
|
||
private function loadViewItems(array $ioIds): array
|
||
{
|
||
$agg = [];
|
||
foreach (array_chunk(array_values(array_unique(array_filter($ioIds))), 800) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
$sql = 'SELECT io_id, sku_ids, qty, type, sku_name, shop_i_id, wareHoursName, l, w, h, volume, weight, supplier_name FROM '
|
||
. self::VIEW . " WHERE io_id IN ({$ph})";
|
||
foreach ($this->up->select($sql, $chunk) as $r) {
|
||
$io = (string) ($r['io_id'] ?? '');
|
||
$skuIds = (string) ($r['sku_ids'] ?? '');
|
||
if ($io === '' || $skuIds === '') {
|
||
continue;
|
||
}
|
||
// 一行可能逗号拼接多个条码 —— 对齐工作平台导出明细：拆成每条码一件；
|
||
// 数量与体积按条码数均摊，保证合并求和后与整行值一致（工作平台合并体积"整箱算一次"）。
|
||
$bcs = array_values(array_filter(array_map('trim', explode(',', $skuIds)), fn ($x) => $x !== ''));
|
||
if ($bcs === []) {
|
||
continue;
|
||
}
|
||
$n = count($bcs);
|
||
$isT0 = ((string) ($r['type'] ?? '') === '0');
|
||
$perQty = (float) ($r['qty'] ?? 0) / $n;
|
||
$rowVol = ($r['volume'] !== null && $r['volume'] !== '') ? (float) $r['volume'] : null;
|
||
$perVol = ($rowVol !== null) ? $rowVol / $n : null;
|
||
foreach ($bcs as $bc) {
|
||
$k = $io . '|' . mb_strtolower($bc);
|
||
if (! isset($agg[$k])) {
|
||
$agg[$k] = [
|
||
'io' => $io, 'sku' => $bc, 't0' => 0.0, 't0n' => 0, 'any' => 0.0, 'anyn' => 0,
|
||
't0vol' => 0.0, 't0voln' => 0, 'anyvol' => 0.0, 'anyvoln' => 0,
|
||
'shopid' => null, 'name' => null, 'wh' => null,
|
||
'l' => null, 'w' => null, 'h' => null,
|
||
'weight' => null, 'supplier' => null,
|
||
];
|
||
}
|
||
if ($isT0) {
|
||
$agg[$k]['t0'] += $perQty;
|
||
$agg[$k]['t0n']++;
|
||
if ($perVol !== null) {
|
||
$agg[$k]['t0vol'] += $perVol;
|
||
$agg[$k]['t0voln']++;
|
||
}
|
||
}
|
||
$agg[$k]['any'] += $perQty;
|
||
$agg[$k]['anyn']++;
|
||
if ($perVol !== null) {
|
||
$agg[$k]['anyvol'] += $perVol;
|
||
$agg[$k]['anyvoln']++;
|
||
}
|
||
// 名称/长宽高/供应商沿用整行值(共享)：任意行回填一次
|
||
if (empty($agg[$k]['name']) && ! empty($r['sku_name'])) {
|
||
$agg[$k]['name'] = $r['sku_name'];
|
||
}
|
||
if (empty($agg[$k]['shopid']) && ! empty($r['shop_i_id'])) {
|
||
$agg[$k]['shopid'] = $r['shop_i_id'];
|
||
}
|
||
if (empty($agg[$k]['wh']) && ! empty($r['wareHoursName'])) {
|
||
$agg[$k]['wh'] = $r['wareHoursName'];
|
||
}
|
||
foreach (['l', 'w', 'h', 'weight'] as $ff) {
|
||
if (empty($agg[$k][$ff]) && ! empty($r[$ff])) {
|
||
$agg[$k][$ff] = $r[$ff];
|
||
}
|
||
}
|
||
if (empty($agg[$k]['supplier']) && ! empty($r['supplier_name'])) {
|
||
$agg[$k]['supplier'] = $r['supplier_name'];
|
||
}
|
||
}
|
||
}
|
||
}
|
||
$out = [];
|
||
foreach ($agg as $a) {
|
||
$qty = $a['t0n'] > 0 ? $a['t0'] : ($a['anyn'] > 0 ? $a['any'] / $a['anyn'] : 0);
|
||
$qty = (int) round($qty);
|
||
if ($qty < 1) {
|
||
$qty = 1;
|
||
}
|
||
// 体积：type=0 各条码分摊之和优先；否则取任一 type 的均值(去 type 重复)
|
||
$vol = $a['t0voln'] > 0
|
||
? round($a['t0vol'], 4)
|
||
: ($a['anyvoln'] > 0 ? round($a['anyvol'] / $a['anyvoln'], 4) : null);
|
||
$out[$a['io']][] = [
|
||
'id' => null, 'sku' => $a['sku'], 'iid' => null,
|
||
'shopid' => $a['shopid'], 'name' => $a['name'],
|
||
'qty' => $qty, 'amount' => null, 'warehouse' => $a['wh'],
|
||
'l' => $a['l'], 'w' => $a['w'], 'h' => $a['h'], 'vol' => $vol,
|
||
'weight' => $a['weight'], 'supplier' => $a['supplier'],
|
||
];
|
||
}
|
||
|
||
return $out;
|
||
}
|
||
|
||
private function collectSkuIds(array ...$maps): array
|
||
{
|
||
$ids = [];
|
||
foreach ($maps as $map) {
|
||
foreach ($map as $items) {
|
||
foreach ($items as $it) {
|
||
if (! empty($it['sku'])) {
|
||
$ids[(string) $it['sku']] = true;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
return array_keys($ids);
|
||
}
|
||
|
||
private function push(int $sourceId, int $skuSourceId, array $payload): void
|
||
{
|
||
$payload['source_id'] = $sourceId;
|
||
$payload['sku_source_id'] = $skuSourceId;
|
||
$this->buffer[] = $payload;
|
||
if (count($this->buffer) >= self::FLUSH) {
|
||
$this->flush();
|
||
}
|
||
}
|
||
|
||
private function packageRow(int $sourceId, int $skuSourceId, array $payload): array
|
||
{
|
||
$payload['source_id'] = $sourceId;
|
||
$payload['sku_source_id'] = $skuSourceId;
|
||
|
||
return $payload;
|
||
}
|
||
|
||
/** @param list<array<string,mixed>> $rows */
|
||
private function writePackage(int $sourceId, array $rows, bool $hasFreight, bool $hasItems, array $syntheticSkuIds): void
|
||
{
|
||
if (! $this->replaceWindow) {
|
||
foreach ($rows as $row) {
|
||
$this->push($sourceId, (int) $row['sku_source_id'], $row);
|
||
}
|
||
return;
|
||
}
|
||
|
||
$this->replacementRowCount -= count($this->replacementPackages[$sourceId] ?? []);
|
||
$this->replacementPackages[$sourceId] = $rows;
|
||
$this->replacementHasFreight[$sourceId] = $hasFreight;
|
||
$this->replacementHasItems[$sourceId] = $hasItems;
|
||
$this->replacementSyntheticSkuIds[$sourceId] = $syntheticSkuIds;
|
||
$this->replacementRowCount += count($rows);
|
||
if ($this->replacementRowCount >= self::FLUSH) {
|
||
$this->flushReplacements();
|
||
}
|
||
}
|
||
|
||
private function flushPending(): void
|
||
{
|
||
if ($this->replaceWindow) {
|
||
$this->flushReplacements();
|
||
} else {
|
||
$this->flush();
|
||
}
|
||
}
|
||
|
||
private function flushReplacements(): void
|
||
{
|
||
if ($this->replacementPackages === []) {
|
||
return;
|
||
}
|
||
$packages = $this->replacementPackages;
|
||
$hasFreight = $this->replacementHasFreight;
|
||
$hasItems = $this->replacementHasItems;
|
||
$syntheticSkuIds = $this->replacementSyntheticSkuIds;
|
||
$this->db->transaction(function () use ($packages, $hasFreight, $hasItems, $syntheticSkuIds): void {
|
||
$ids = array_keys($packages);
|
||
$previous = [];
|
||
$previousItems = [];
|
||
$oldAmountSkuIds = [];
|
||
$oldHasItems = [];
|
||
$oldRows = $this->db->table('jst_oms_shipping')->whereIn('source_id', $ids)
|
||
->select(array_merge(
|
||
['source_id', 'sku_source_id', 'express_no', 'express_name2', 'freight', 'settle_freight', 'weight_kg', 'settle_weight_kg',
|
||
'item_id', 'barcode', 'product_name', 'product_code', 'qty', 'amount'],
|
||
self::PRESERVED_PACKAGE_FIELDS,
|
||
self::PRESERVED_CONTACT_FIELDS,
|
||
self::PRESERVED_ITEM_FIELDS,
|
||
))
|
||
->get();
|
||
foreach ($oldRows as $oldRow) {
|
||
$id = (int) $oldRow->source_id;
|
||
$row = (array) $oldRow;
|
||
$previousItems[$id][(int) $oldRow->sku_source_id] = $row;
|
||
if ($row['amount'] !== null) {
|
||
$oldAmountSkuIds[$id][(int) $oldRow->sku_source_id] = true;
|
||
}
|
||
foreach (['item_id', 'barcode', 'product_name', 'product_code', 'qty', 'amount',
|
||
'finished_barcode', 'finished_name', 'finished_supplier'] as $field) {
|
||
if ($row[$field] !== null && $row[$field] !== '') {
|
||
$oldHasItems[$id] = true;
|
||
break;
|
||
}
|
||
}
|
||
if (! isset($previous[$id])) {
|
||
$previous[$id] = $row;
|
||
continue;
|
||
}
|
||
foreach (array_merge(
|
||
['freight', 'settle_freight', 'weight_kg', 'settle_weight_kg', 'express_no', 'express_name2'],
|
||
self::PRESERVED_PACKAGE_FIELDS,
|
||
self::PRESERVED_CONTACT_FIELDS,
|
||
) as $field) {
|
||
if (($previous[$id][$field] ?? null) === null || $previous[$id][$field] === '') {
|
||
$previous[$id][$field] = $row[$field];
|
||
}
|
||
}
|
||
}
|
||
|
||
$rows = [];
|
||
$replaceIds = [];
|
||
foreach ($packages as $id => $packageRows) {
|
||
if (! ($hasItems[$id] ?? false) && ($oldHasItems[$id] ?? false)) {
|
||
$this->packagesRetainedForMissingItems++;
|
||
continue;
|
||
}
|
||
$unmatchedOldAmounts = $oldAmountSkuIds[$id] ?? [];
|
||
$unsafeAmount = false;
|
||
foreach ($packageRows as &$packageRow) {
|
||
$skuId = (int) $packageRow['sku_source_id'];
|
||
$oldItem = $previousItems[$id][$skuId] ?? null;
|
||
if ($packageRow['amount'] !== null) {
|
||
unset($unmatchedOldAmounts[$skuId]);
|
||
continue;
|
||
}
|
||
if ($oldItem === null || $oldItem['amount'] === null) {
|
||
continue;
|
||
}
|
||
if (isset($syntheticSkuIds[$id][$skuId])) {
|
||
$oldBarcode = trim((string) $oldItem['barcode']);
|
||
$newBarcode = trim((string) ($packageRow['barcode'] ?? ''));
|
||
if ($oldBarcode === '' || $newBarcode === '' || $oldBarcode !== $newBarcode) {
|
||
$unsafeAmount = true;
|
||
break;
|
||
}
|
||
}
|
||
$packageRow['amount'] = $oldItem['amount'];
|
||
unset($unmatchedOldAmounts[$skuId]);
|
||
}
|
||
unset($packageRow);
|
||
if (! $unsafeAmount && $unmatchedOldAmounts !== []) {
|
||
foreach ($packageRows as $packageRow) {
|
||
if ($packageRow['amount'] === null) {
|
||
$unsafeAmount = true;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
if ($unsafeAmount) {
|
||
$this->packagesRetainedForUnmatchedAmount++;
|
||
continue;
|
||
}
|
||
$replaceIds[] = $id;
|
||
foreach ($packageRows as &$packageRow) {
|
||
foreach (self::PRESERVED_PACKAGE_FIELDS as $field) {
|
||
$packageRow[$field] ??= null;
|
||
}
|
||
foreach (self::PRESERVED_ITEM_FIELDS as $field) {
|
||
$packageRow[$field] = $previousItems[$id][(int) $packageRow['sku_source_id']][$field] ?? null;
|
||
}
|
||
}
|
||
unset($packageRow);
|
||
if (isset($previous[$id])) {
|
||
foreach (self::PRESERVED_PACKAGE_FIELDS as $field) {
|
||
$packageRows[0][$field] = $previous[$id][$field] ?? null;
|
||
}
|
||
foreach (self::PRESERVED_CONTACT_FIELDS as $field) {
|
||
if (($packageRows[0][$field] ?? null) === null || $packageRows[0][$field] === '') {
|
||
$packageRows[0][$field] = $previous[$id][$field] ?? null;
|
||
}
|
||
}
|
||
if ($this->skipFreight) {
|
||
$packageRows[0] = $this->withPreservedFreight($packageRows[0], $previous[$id]);
|
||
} elseif (! ($hasFreight[$id] ?? false)) {
|
||
$packageRows[0] = $this->withPreservedCharges($packageRows[0], $previous[$id], false);
|
||
}
|
||
}
|
||
array_push($rows, ...$packageRows);
|
||
}
|
||
if ($replaceIds === []) {
|
||
return;
|
||
}
|
||
$this->db->table('jst_oms_shipping')->whereIn('source_id', $replaceIds)->delete();
|
||
foreach (array_chunk($rows, self::FLUSH) as $chunk) {
|
||
$this->db->table('jst_oms_shipping')->insert($chunk);
|
||
}
|
||
});
|
||
|
||
$this->replacementPackages = [];
|
||
$this->replacementHasFreight = [];
|
||
$this->replacementHasItems = [];
|
||
$this->replacementSyntheticSkuIds = [];
|
||
$this->replacementRowCount = 0;
|
||
}
|
||
|
||
private function flush(): void
|
||
{
|
||
if ($this->buffer === []) {
|
||
return;
|
||
}
|
||
$this->db->table('jst_oms_shipping')->insertOrIgnore($this->buffer);
|
||
$this->buffer = [];
|
||
}
|
||
|
||
private function isPricePatch(?string $name): bool
|
||
{
|
||
if ($name === null || $name === '') {
|
||
return false;
|
||
}
|
||
|
||
return str_contains($name, '补收差价') || str_contains($name, '差价专用');
|
||
}
|
||
|
||
/** 线上发货(平台代发/云仓)标签：工作平台不拉取，caiwu 对齐排除。 */
|
||
private function excludedByLabel(?string $labels): bool
|
||
{
|
||
if ($labels === null || $labels === '') {
|
||
return false;
|
||
}
|
||
|
||
return str_contains($labels, '线上发货');
|
||
}
|
||
|
||
/** @return array<string,array> sku_id => 尺寸/重量/供应商 */
|
||
private function loadSkus(array $skuIds): array
|
||
{
|
||
$out = [];
|
||
foreach (array_chunk($skuIds, 800) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
foreach ($this->up->select("SELECT sku_id, l, w, h, pack_volume, weight, supplier_name, name, i_id FROM rpt.jst_sku WHERE sku_id IN ({$ph})", $chunk) as $r) {
|
||
$out[(string) $r['sku_id']] = [
|
||
'l' => $r['l'] ?? null, 'w' => $r['w'] ?? null, 'h' => $r['h'] ?? null,
|
||
'vol' => $r['pack_volume'] ?? null, 'weight' => $r['weight'] ?? null,
|
||
'supplier' => $r['supplier_name'] ?? null,
|
||
'name' => $r['name'] ?? null, 'iid' => $r['i_id'] ?? null,
|
||
];
|
||
}
|
||
}
|
||
|
||
return $out;
|
||
}
|
||
|
||
/** @return array<string,array> so_id|expressNo => 运费 (含 so_id| 兜底键) */
|
||
private function loadFreight(array $soIds): array
|
||
{
|
||
$out = [];
|
||
foreach (array_chunk($soIds, 800) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
$sql = 'SELECT so_id, expressNo, SUM(freightCost) AS cost, SUM(weight) AS w, '
|
||
. 'MAX(expressName) AS name, MAX(receiver_state) AS state '
|
||
. "FROM rpt.fys_jstFreightCostDetail WHERE so_id IN ({$ph}) GROUP BY so_id, expressNo";
|
||
foreach ($this->up->select($sql, $chunk) as $r) {
|
||
$expno = (string) ($r['expressNo'] ?? '');
|
||
$val = ['cost' => $r['cost'] ?? null, 'w' => $r['w'] ?? null, 'name' => $r['name'] ?? null, 'state' => $r['state'] ?? null, 'no' => $expno ?: null];
|
||
$out[((string) $r['so_id']) . '|' . $expno] = $val;
|
||
$out[((string) $r['so_id']) . '|'] ??= $val;
|
||
}
|
||
}
|
||
|
||
return $out;
|
||
}
|
||
|
||
private function countUnseenMirrorPackages(string $fromS, string $toS): int
|
||
{
|
||
$unseen = 0;
|
||
$ids = $this->db->table('jst_oms_shipping')
|
||
->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS)
|
||
->distinct()->pluck('source_id');
|
||
foreach ($ids as $id) {
|
||
if (! isset($this->seenSourceIds[(int) $id])) {
|
||
$unseen++;
|
||
}
|
||
}
|
||
|
||
return $unseen;
|
||
}
|
||
|
||
private function withPreservedFreight(array $payload, array $previous): array
|
||
{
|
||
$payload = $this->withPreservedCharges($payload, $previous);
|
||
if (! empty($previous['express_no'])) {
|
||
$payload['express_no'] = $previous['express_no'];
|
||
}
|
||
$payload['express_name2'] = $previous['express_name2'] ?? $payload['express_name2'];
|
||
|
||
return $payload;
|
||
}
|
||
|
||
private function withPreservedCharges(array $payload, array $previous, bool $includeNull = true): array
|
||
{
|
||
foreach (['freight', 'settle_freight', 'weight_kg', 'settle_weight_kg'] as $field) {
|
||
if ($includeNull || $previous[$field] !== null) {
|
||
$payload[$field] = $previous[$field];
|
||
}
|
||
}
|
||
|
||
return $payload;
|
||
}
|
||
|
||
private function cleanPayload(array $data): array
|
||
{
|
||
foreach ($data as $k => $v) {
|
||
if (is_string($v)) {
|
||
$data[$k] = $this->cleanStr($v);
|
||
}
|
||
}
|
||
|
||
return $data;
|
||
}
|
||
|
||
private function cleanStr(string $v): string
|
||
{
|
||
$v = str_replace("\0", '', $v);
|
||
$clean = @iconv('UTF-8', 'UTF-8//IGNORE', $v);
|
||
if ($clean === false) {
|
||
$clean = mb_convert_encoding($v, 'UTF-8', 'UTF-8');
|
||
}
|
||
|
||
return $clean;
|
||
}
|
||
|
||
private function ts(mixed $v): ?string
|
||
{
|
||
if ($v === null || $v === '') {
|
||
return null;
|
||
}
|
||
try {
|
||
return Carbon::parse((string) $v)->format('Y-m-d H:i:s');
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
private function markRunning(): void
|
||
{
|
||
$this->db->table('jst_sync_state')->updateOrInsert(
|
||
['dataset_key' => self::DATASET],
|
||
['status' => 'running', 'last_started_at' => now(), 'updated_at' => now(), 'created_at' => now()]
|
||
);
|
||
}
|
||
|
||
private function markSucceeded(): void
|
||
{
|
||
$this->db->table('jst_sync_state')->where('dataset_key', self::DATASET)->update([
|
||
'status' => 'succeeded', 'last_succeeded_at' => now(), 'last_error' => null, 'watermark' => now(), 'updated_at' => now(),
|
||
]);
|
||
}
|
||
|
||
private function windowDatasetKey(string $prefix, string $fromS, string $toS): string
|
||
{
|
||
return $prefix . Carbon::parse($fromS)->format('YmdHis') . '-' . Carbon::parse($toS)->format('YmdHis');
|
||
}
|
||
|
||
private function markFreightIncomplete(string $fromS, string $toS): void
|
||
{
|
||
$this->db->table('jst_sync_state')->updateOrInsert(
|
||
['dataset_key' => $this->windowDatasetKey(self::FREIGHT_DATASET_PREFIX, $fromS, $toS)],
|
||
[
|
||
'status' => 'pending',
|
||
'last_error' => '本窗口采用数量优先补录；上游运费查询已跳过，新记录的运费和结算重量待补齐，空白不代表 0。',
|
||
'updated_at' => now(),
|
||
'created_at' => now(),
|
||
],
|
||
);
|
||
}
|
||
|
||
private function markItemsIncomplete(string $fromS, string $toS): void
|
||
{
|
||
$detail = sprintf(
|
||
'部分日期窗口的商品或金额待核对：%d 个包裹未匹配商品，%d 个包裹金额无法安全匹配，DWS 预载失败 %d 天。',
|
||
$this->packagesWithoutItems,
|
||
$this->packagesRetainedForUnmatchedAmount,
|
||
count($this->dwsFailedDays),
|
||
);
|
||
$this->db->table('jst_sync_state')->updateOrInsert(
|
||
['dataset_key' => $this->windowDatasetKey(self::ITEMS_DATASET_PREFIX, $fromS, $toS)],
|
||
['status' => 'pending', 'last_error' => $detail, 'updated_at' => now(), 'created_at' => now()],
|
||
);
|
||
}
|
||
|
||
private function markCovered(string $prefix, string $fromS, string $toS): void
|
||
{
|
||
$from = Carbon::parse($fromS)->format('YmdHis');
|
||
$to = Carbon::parse($toS)->format('YmdHis');
|
||
$states = $this->db->table('jst_sync_state')->where('dataset_key', 'like', $prefix . '%')
|
||
->where('status', '!=', 'succeeded')->pluck('dataset_key');
|
||
foreach ($states as $key) {
|
||
$range = substr((string) $key, strlen($prefix));
|
||
if (! preg_match('/^\d{14}-\d{14}$/', $range)) {
|
||
continue;
|
||
}
|
||
[$pendingFrom, $pendingTo] = explode('-', $range, 2);
|
||
if ($from <= $pendingFrom && $pendingTo <= $to) {
|
||
$this->db->table('jst_sync_state')->where('dataset_key', $key)->update([
|
||
'status' => 'succeeded', 'last_error' => null, 'last_succeeded_at' => now(), 'updated_at' => now(),
|
||
]);
|
||
}
|
||
}
|
||
}
|
||
|
||
private function markIncompleteWindow(int $unseen): void
|
||
{
|
||
$this->db->table('jst_sync_state')->where('dataset_key', self::DATASET)->update([
|
||
'status' => 'partial',
|
||
'last_error' => "本窗口尚有 {$unseen} 个镜像包裹未在上游扫描中出现，已保留原记录，需按源键核对。",
|
||
'updated_at' => now(),
|
||
]);
|
||
}
|
||
|
||
private function markFailed(string $err): void
|
||
{
|
||
$this->db->table('jst_sync_state')->where('dataset_key', self::DATASET)->update([
|
||
'status' => 'failed', 'last_error' => mb_substr($err, 0, 1000), 'updated_at' => now(),
|
||
]);
|
||
}
|
||
}
|