limeng
f6497f5afc
- 凌晨同步时发货明细常未到齐，包裹先按自动发货仓/出库单兜底入库(无省份、无商品)；之后发货明细到齐换成发货明细来源键， 旧兜底包裹因"上游未再出现"被保留，造成重复和空白省份(10 月重跑 2,226 个)。现在：旧包裹的快递单号已由本次其它包裹吐出时 视为被取代并删除；无对应的仍保留待核对。 - 商品金额无法安全匹配(或新商品明细缺失)而保留的旧包裹(10 月 28,201 个)，此前整包不更新；现在商品与金额不动， 省市区、快递名称、仓库、店铺、平台、交易单号、付款时间按本次上游刷新。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014nvGZkg27YMEdhRy6EiUwf
2549 lines
112 KiB
PHP
2549 lines
112 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Services\Jushuitan;
|
||
|
||
use Illuminate\Support\Carbon;
|
||
use Illuminate\Support\Facades\DB;
|
||
use InvalidArgumentException;
|
||
use Modules\EcommerceFinance\Services\UpstreamConnection;
|
||
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
|
||
use PDOException;
|
||
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 UPSTREAM_RECONNECT_RETRIES = 5;
|
||
/** 副本库繁忙超时后的重连等待(毫秒)：逐次拉长，白天整月重跑不再因单次超时中断 */
|
||
private const UPSTREAM_RECONNECT_BACKOFF_MS = [2000, 5000, 10000, 20000, 30000];
|
||
|
||
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 OUTBOUND_OFFSET = 2000000000;
|
||
private const PURCHASE_OUT_OFFSET = 3000000000;
|
||
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 REFRESHED_PACKAGE_FIELDS = [
|
||
'trade_no', 'express_name1', 'warehouse_name', 'shop_name', 'platform', 'pay_time',
|
||
'receiver_state', 'receiver_city', 'receiver_district',
|
||
];
|
||
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 const PURCHASE_OUT = 'rpt.jst_purchaseout';
|
||
|
||
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;
|
||
private int $supersededRemoved = 0;
|
||
|
||
/** @var array<int,true> */
|
||
private array $seenSourceIds = [];
|
||
|
||
/** @var array<string,true> 发货明细已占用的窗口内订单号 */
|
||
private array $expressOrderIds = [];
|
||
|
||
/** @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<string,bool> 本轮 express/autoship 已处理(含按标签排除)的出库单 io_id */
|
||
private array $doneIo = [];
|
||
|
||
/** @var array<string,bool> 本轮已处理的快递单号(去@) */
|
||
private array $doneTrack = [];
|
||
|
||
/** 已实际输出的快递单号(去@)，用于判断虚拟/补差单是否借用了别的包裹单号 */
|
||
private array $emittedTrack = [];
|
||
|
||
/** 快递单号(去@) => [交易单号 => true]：同一单号被不同交易单号共用(合包/拼单)时，按交易单号区分是否已处理 */
|
||
private array $doneTrackSo = [];
|
||
|
||
/** @var array<int,true> 本轮按规则排除的包裹 source_id(取消/拆分/合并/线上发货)，窗口内旧记录需清掉 */
|
||
private array $excludedSourceIds = [];
|
||
|
||
/** @var array<string,string> 最近一次 loadViewItems 的 io_id => 平台(视图 shop_site，工作平台口径) */
|
||
private array $ioSite = [];
|
||
|
||
/** @var array<string,array{st?:string,ci?:string,di?:string}> 最近一次 loadViewItems 的 io_id => 视图收货省市区(工作平台口径) */
|
||
private array $ioAddr = [];
|
||
|
||
/** @var array<string,string> io_id => 出库时间 io_date(工作平台发货时间口径) */
|
||
private array $ioDate = [];
|
||
|
||
/** @var array<string,string> io_id => 出库单快递公司(工作平台按实际出库单取，发货明细可能停在作废前的旧分配) */
|
||
private array $ioLc = [];
|
||
|
||
/** @var array<string,string> io_id => 出库单发货仓 wms_co_id */
|
||
private array $ioWms = [];
|
||
|
||
/** @var array<string,array<string,int>> wms_co_id => [仓库名 => 次数]：从快递公司一致的发货明细行学习仓库名 */
|
||
private array $wmsNames = [];
|
||
|
||
/** @var array<int,string> 包裹 source_id => 出库单 wms_co_id(扫描结束后按学到的仓库名统一回填) */
|
||
private array $pendingWarehouse = [];
|
||
|
||
/** @var array<string,bool> io_id => 是否有效出库单(非作废/取消) */
|
||
private array $ioOk = [];
|
||
|
||
/** 换单配对查询超时而跳过的页数(白天副本库繁忙时，不让单页超时中断整轮同步) */
|
||
private int $pairingSkipped = 0;
|
||
|
||
/** 工作平台口径：快递名称为"无需物流"(补差/信誉/运费等虚拟发货)的包裹，仓库统一显示自动发货仓 */
|
||
private const NO_LOGISTICS = '无需物流';
|
||
private const NO_LOGISTICS_WAREHOUSE = '简尚-自动发货仓';
|
||
|
||
/** @var array<string,string> 最近一次 loadViewItems 的 io_id => 视图店铺名(发货明细缺店铺时兜底) */
|
||
private array $ioShop = [];
|
||
|
||
/** @var array<string,string> 最近一次 loadViewItems 的 io_id => 视图快递名称 expressName(工作平台显示的快递名称) */
|
||
private array $ioExpress = [];
|
||
|
||
/** @var array<string,string> 最近一次 loadViewItems 的 io_id => 视图发货仓 wareHoursName(工作平台显示的仓库) */
|
||
private array $ioWare = [];
|
||
/** @var array<string,array{l_id:mixed,send_date:mixed,status:?string,online:bool}> 已查询订单元数据缓存 o_id=>字段 */
|
||
private array $orderMetadata = [];
|
||
|
||
/** @var array<string,array> 窗口内出库单映射(见 emptyIoMap) */
|
||
private array $outboundMap = ['byL' => [], 'byO' => [], 'byLO' => [], 'byOAll' => [], 'ioL' => []];
|
||
|
||
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->supersededRemoved = 0;
|
||
$this->seenSourceIds = [];
|
||
$this->expressOrderIds = [];
|
||
$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;
|
||
$outb = 0;
|
||
$pout = 0;
|
||
$this->doneIo = [];
|
||
$this->doneTrack = [];
|
||
$this->emittedTrack = [];
|
||
$this->doneTrackSo = [];
|
||
$this->excludedSourceIds = [];
|
||
$this->ioDate = [];
|
||
$this->ioLc = [];
|
||
$this->ioWms = [];
|
||
$this->wmsNames = [];
|
||
$this->pendingWarehouse = [];
|
||
$this->ioOk = [];
|
||
$this->pairingSkipped = 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);
|
||
$this->applyPendingWarehouses($progress);
|
||
$this->orderMetadata = []; // 全表扫描后释放(数百万订单)，后续各步不再使用
|
||
gc_collect_cycles();
|
||
$auto = $this->syncAutoShip($fromS, $toS, $now, $progress);
|
||
$outb = $this->syncOutbound($fromS, $toS, $now, $progress);
|
||
$pout = $this->syncPurchaseOut($fromS, $toS, $now, $progress);
|
||
$purged = $this->purgeExcludedPackages($fromS, $toS);
|
||
if ($progress && $purged > 0) {
|
||
$progress('清除已排除包裹(取消/拆分/合并/线上发货)：' . $purged . ' 行');
|
||
}
|
||
$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);
|
||
$this->applyPendingWarehouses($progress);
|
||
}
|
||
|
||
if ($windowed && ! $skipFreight && $this->staleMirrorPackages === 0) {
|
||
$this->markCovered(self::FREIGHT_DATASET_PREFIX, $fromS, $toS);
|
||
}
|
||
|
||
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['outbound'] = $outb;
|
||
$result['purchase_out'] = $pout;
|
||
$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;
|
||
$result['superseded_packages_removed'] = $this->supersededRemoved;
|
||
$result['tracking_pairing_skipped_pages'] = $this->pairingSkipped;
|
||
}
|
||
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->orderMetadata = [];
|
||
$this->outboundMap = ['byL' => [], 'byO' => [], 'byLO' => [], 'byOAll' => [], 'ioL' => []];
|
||
$this->replacementPackages = [];
|
||
$this->replacementHasFreight = [];
|
||
$this->replacementHasItems = [];
|
||
$this->replacementSyntheticSkuIds = [];
|
||
$this->replacementRowCount = 0;
|
||
$this->skipFreight = false;
|
||
$this->replaceWindow = false;
|
||
$this->dwsFailedDays = [];
|
||
$this->seenSourceIds = [];
|
||
$this->expressOrderIds = [];
|
||
gc_collect_cycles();
|
||
}
|
||
}
|
||
|
||
/** @X 与 X 仅在同一订单、同一发货日去重；提供日期窗口时不使用窗口外记录作匹配；剩余 @ 前缀统一去掉。 */
|
||
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();
|
||
}
|
||
// 剩余 @X：工作平台导出单号不带 @，统一去掉前缀(只处理本次窗口)
|
||
$strip = $this->db->table('jst_oms_shipping')->where('express_no', 'like', '@%');
|
||
if ($fromS !== null && $toS !== null) {
|
||
$strip->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS);
|
||
}
|
||
$strip->update(['express_no' => $this->db->raw('SUBSTR(express_no, 2)')]);
|
||
|
||
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';
|
||
|
||
// 窗口模式：只按主键 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 "
|
||
. "WHERE {$where} ORDER BY e.id DESC";
|
||
$rows = $this->selectWithReconnect($sql, $bind);
|
||
if ($rows === []) {
|
||
break;
|
||
}
|
||
if ($b === $maxBatches) {
|
||
throw new RuntimeException('发货扫描达到批次上限，仍有未读取的上游数据。');
|
||
}
|
||
$cnt = count($rows);
|
||
$scanned += $cnt;
|
||
foreach ($rows as $r) {
|
||
$lastId = $r['id'];
|
||
}
|
||
|
||
// 主表分页不联接订单表；每页只按 inner_order_id 分块补齐订单字段，
|
||
// 并复用已经缓存的结果，避免 LEFT JOIN 在大历史窗口上拖垮游标。
|
||
$this->loadOrderMetadataForRows($rows);
|
||
$rows = array_map(fn (array $row): array => $this->mergeOrderMetadata($row), $rows);
|
||
|
||
if ($windowed) {
|
||
$kept = [];
|
||
foreach ($rows as $r) {
|
||
$jd = $r['jo_send'] ?? null;
|
||
$joTs = ($jd !== null && $jd !== '') ? strtotime((string) $jd) : null;
|
||
$oid = (string) ($r['inner_order_id'] ?? '');
|
||
if ($oid !== '' && $joTs !== null && $joTs !== false && $joTs >= $fromTs && $joTs < $toTs) {
|
||
$this->expressOrderIds[$oid] = true;
|
||
}
|
||
$st = $r['send_time'] ?? null;
|
||
$ts = ($st !== null && $st !== '') ? strtotime((string) $st) : null;
|
||
if ($ts === null) {
|
||
$ts = $joTs;
|
||
}
|
||
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));
|
||
$swap = $this->pairChangedTracking($rows, $ioMap);
|
||
$ioIds = [];
|
||
foreach ($rows as $r) {
|
||
$io = isset($swap[(string) ($r['id'] ?? '')])
|
||
? $swap[(string) $r['id']]['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 (in_array((string) ($r['jo_status'] ?? ''), ['Cancelled', 'Split', 'Merged'], true)) {
|
||
// 订单已取消/已拆分/已合并：实际发货在子单或合并单上，工作平台不出此行。
|
||
// 只登记本行自己的单号，不登记订单单号(可能是子单真实单号，需留给出库单兜底)
|
||
$this->markDone('', (string) ($r['logistic_id'] ?? ''));
|
||
$this->markTrackSo((string) ($r['so_id'] ?? ''), (string) ($r['logistic_id'] ?? ''));
|
||
$this->excludeSource($pkgId);
|
||
continue;
|
||
}
|
||
$sw = $swap[(string) $pkgId] ?? null;
|
||
if ($sw !== null) {
|
||
// 换单：快递明细仍是旧单号，出库单已换成新单号 → 用新单号与对应出库单(工作平台口径)
|
||
$r['logistic_id'] = $sw['l_id'];
|
||
}
|
||
$this->markDone(
|
||
$sw !== null ? $sw['io'] : $this->ioFor($ioMap, (string) ($r['logistic_id'] ?? ''), (string) ($r['inner_order_id'] ?? '')),
|
||
(string) ($r['logistic_id'] ?? ''),
|
||
(string) ($r['jo_lid'] ?? ''),
|
||
$sw !== null ? $sw['old'] : ''
|
||
);
|
||
$this->markTrackSo((string) ($r['so_id'] ?? ''), (string) ($r['logistic_id'] ?? ''), (string) ($r['jo_lid'] ?? ''));
|
||
if ($this->excludedByLabel($r['order_labels'] ?? null) || $this->excludedByLabel($r['jo_labels'] ?? null)) {
|
||
$this->excludeSource($pkgId);
|
||
continue; // 线上发货(平台代发/云仓)：工作平台不拉，对齐排除(发货明细标签可能滞后，订单标签一并判断)
|
||
}
|
||
$so = (string) ($r['so_id'] ?? '');
|
||
$oid = (string) ($r['inner_order_id'] ?? '');
|
||
$io = $sw !== null ? $sw['io'] : $this->ioFor($ioMap, (string) ($r['logistic_id'] ?? ''), $oid);
|
||
$items = ($io !== '' && ! empty($viewItems[$io]))
|
||
? $viewItems[$io]
|
||
: $this->itemsFor($byOid, $bySo, $dws, $oid, $so);
|
||
if (trim((string) ($r['logistic_id'] ?? '')) === '') {
|
||
// 发货明细无单号：出库单单号 → 订单单号 → 空(不借运费表里同订单其它包裹的单号)
|
||
$ioLid = ($io !== '') ? (string) ($ioMap['ioL'][$io] ?? '') : '';
|
||
$joLid = trim((string) ($r['jo_lid'] ?? ''));
|
||
$tracking = $ioLid !== '' ? $ioLid : ($joLid !== '' ? $joLid : null);
|
||
} else {
|
||
$tracking = $this->tracking((string) ($r['logistic_id'] ?? ''), (string) ($r['jo_lid'] ?? ''), $freight, $so);
|
||
}
|
||
// 发货时间 = 出库单 io_date(工作平台口径，样本 400/400 一致)，缺失时回落发货明细时间
|
||
$shipTime = ($io !== '' ? $this->ts($this->ioDate[$io] ?? null) : null)
|
||
?? $this->ts($r['send_time'] ?? null) ?? $this->ts($r['jo_send'] ?? null);
|
||
$pkg = $this->pkgFromExpress($r, $tracking, $shipTime);
|
||
$pkg = $this->withOutboundCarrier($io, $r, $pkg, $pkgId);
|
||
$pkg = $this->withViewCarrier($io, $pkg, $pkgId);
|
||
$pkg['platform'] = $this->platformOf($io, $pkg['platform']);
|
||
$pkg = $this->withViewAddr($io, $pkg);
|
||
$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;
|
||
}
|
||
|
||
/** 按当前页订单键补齐订单字段，并复用已缓存的结果。 */
|
||
private function loadOrderMetadataForRows(array $rows): void
|
||
{
|
||
$oids = [];
|
||
foreach ($rows as $row) {
|
||
$oid = (string) ($row['inner_order_id'] ?? '');
|
||
if ($oid !== '' && ! array_key_exists($oid, $this->orderMetadata)) {
|
||
$oids[$oid] = true;
|
||
}
|
||
}
|
||
foreach (array_chunk(array_keys($oids), 800) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
$sql = "SELECT o_id, l_id, send_date, status, labels FROM rpt.jst_Orders om WHERE om.o_id IN ({$ph})";
|
||
foreach ($this->selectWithReconnect($sql, $chunk) as $row) {
|
||
$oid = (string) ($row['o_id'] ?? '');
|
||
if ($oid === '') {
|
||
continue;
|
||
}
|
||
// 全表扫描会缓存数百万订单：只留判断需要的精简字段(状态仅存会被排除的三种，标签只存是否线上发货)
|
||
$st = (string) ($row['status'] ?? '');
|
||
$this->orderMetadata[$oid] = [
|
||
'l_id' => $row['l_id'] ?? null,
|
||
'send_date' => $row['send_date'] ?? null,
|
||
'status' => in_array($st, ['Cancelled', 'Split', 'Merged'], true) ? $st : null,
|
||
'online' => $this->excludedByLabel($row['labels'] ?? null),
|
||
];
|
||
}
|
||
foreach ($chunk as $oid) {
|
||
// 记录无匹配订单，避免后续分页反复查询同一个缺失键。
|
||
$this->orderMetadata[$oid] ??= ['l_id' => null, 'send_date' => null, 'status' => null, 'online' => false];
|
||
}
|
||
}
|
||
}
|
||
|
||
/** 将订单元数据映射到发货明细，保留测试/兼容来源已提供的字段。 */
|
||
private function mergeOrderMetadata(array $row): array
|
||
{
|
||
$oid = (string) ($row['inner_order_id'] ?? '');
|
||
$metadata = $oid !== '' ? ($this->orderMetadata[$oid] ?? []) : [];
|
||
if (($row['jo_lid'] ?? '') === '' && ($metadata['l_id'] ?? '') !== '') {
|
||
$row['jo_lid'] = $metadata['l_id'];
|
||
}
|
||
if (($row['jo_send'] ?? '') === '' && ($metadata['send_date'] ?? '') !== '') {
|
||
$row['jo_send'] = $metadata['send_date'];
|
||
}
|
||
// 订单状态/标签：取消/拆分/合并 与 订单级"线上发货"标签的排除判断用
|
||
if (($row['jo_status'] ?? '') === '' && ($metadata['status'] ?? '') !== '') {
|
||
$row['jo_status'] = $metadata['status'];
|
||
}
|
||
if (($row['jo_labels'] ?? '') === '' && ! empty($metadata['online'])) {
|
||
$row['jo_labels'] = '线上发货';
|
||
}
|
||
|
||
return $row;
|
||
}
|
||
|
||
/** @return list<array<string,mixed>> */
|
||
private function selectWithReconnect(string $sql, array $bindings): array
|
||
{
|
||
for ($retry = 0; ; $retry++) {
|
||
try {
|
||
return $this->up->select($sql, $bindings);
|
||
} catch (PDOException $e) {
|
||
if (! $this->isTransientCursorFailure($e) || $retry >= self::UPSTREAM_RECONNECT_RETRIES) {
|
||
throw $e;
|
||
}
|
||
|
||
// select() returns only after fetchAll(), so no partial result was consumed.
|
||
UpstreamConnection::reset(self::SOURCE);
|
||
if (! app()->runningUnitTests()) {
|
||
usleep(1000 * (self::UPSTREAM_RECONNECT_BACKOFF_MS[$retry] ?? 30000));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private function isTransientCursorFailure(PDOException $e): bool
|
||
{
|
||
$message = strtolower($e->getMessage());
|
||
|
||
return str_contains($message, 'pdo_dblib: dbnextrow() returned fail')
|
||
|| str_contains($message, 'adaptive server connection timed out [20003]');
|
||
}
|
||
|
||
/** 并入自动发货仓：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 < ?";
|
||
$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->selectWithReconnect($sql, $bind);
|
||
if ($rows === []) {
|
||
break;
|
||
}
|
||
if ($b === $maxBatches) {
|
||
throw new RuntimeException('代发仓扫描达到批次上限，仍有未读取的上游数据。');
|
||
}
|
||
|
||
$cnt = count($rows);
|
||
$eligible = [];
|
||
foreach ($rows as $r) {
|
||
$lastRid = $r['rid'];
|
||
$oid = (string) ($r['o_id'] ?? '');
|
||
if ($oid !== '' && ! isset($this->expressOrderIds[$oid])) {
|
||
$eligible[] = $r;
|
||
}
|
||
}
|
||
$rows = $eligible;
|
||
|
||
$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;
|
||
}
|
||
}
|
||
[$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));
|
||
|
||
foreach ($rows as $r) {
|
||
$oid = (string) ($r['o_id'] ?? '');
|
||
if ($oid === '') {
|
||
continue;
|
||
}
|
||
// 单号已属于其它包裹(发货明细已出)：本单是虚拟/补差单，工作平台此类行快递单号为空
|
||
$tracking = $this->tracking((string) ($r['l_id'] ?? ''), '', $freight, (string) ($r['so_id'] ?? ''));
|
||
if ($tracking !== null && isset($this->emittedTrack[ltrim(trim($tracking), '@')])) {
|
||
$tracking = null;
|
||
}
|
||
$this->markDone($this->ioFor($ioMap, (string) ($r['l_id'] ?? ''), $oid), (string) ($r['l_id'] ?? ''));
|
||
$this->markTrackSo((string) ($r['so_id'] ?? ''), (string) ($r['l_id'] ?? ''));
|
||
if ($this->excludedByLabel($r['labels'] ?? null)) {
|
||
$this->excludeSource(self::AUTOSHIP_OFFSET + (int) $oid);
|
||
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);
|
||
$shipTime = ($io !== '' ? $this->ts($this->ioDate[$io] ?? null) : null) ?? $this->ts($r['send_date'] ?? null);
|
||
$pkg = $this->pkgFromOrder($r, $tracking, $shipTime);
|
||
$pkg['platform'] = $this->platformOf($io, $pkg['platform']);
|
||
$pkg = $this->withViewAddr($io, $pkg);
|
||
$pkgId = self::AUTOSHIP_OFFSET + (int) $oid;
|
||
$pkg = $this->withViewCarrier($io, $pkg, $pkgId);
|
||
$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;
|
||
}
|
||
|
||
/**
|
||
* 平台统一为工作平台口径：优先取发货数据视图 shop_site；视图没有时把原始平台名映射到工作平台归类名。
|
||
*/
|
||
private function platformOf(string $io, mixed $raw): ?string
|
||
{
|
||
if ($io !== '' && isset($this->ioSite[$io])) {
|
||
return $this->ioSite[$io];
|
||
}
|
||
$raw = trim((string) ($raw ?? ''));
|
||
if ($raw === '') {
|
||
return null;
|
||
}
|
||
$base = strtoupper(explode(',', $raw)[0]);
|
||
$map = [
|
||
'天猫' => '淘宝天猫', '淘宝' => '淘宝天猫', '京东' => '京东商城', 'MOBILE' => '京东商城', 'PC' => '京东商城',
|
||
'1688' => '阿里巴巴', '抖音' => '头条放心购', 'TOUTIAOFXG' => '头条放心购', 'PINDUODUO' => '拼多多',
|
||
'DEWU' => '得物', '线下' => '线下店铺',
|
||
];
|
||
|
||
return $map[$base] ?? $raw;
|
||
}
|
||
|
||
/** 记下按规则排除的包裹；同步结束后清掉它在本窗口里的旧记录(此前版本可能写入过)。 */
|
||
private function excludeSource(int $sourceId): void
|
||
{
|
||
if ($sourceId > 0) {
|
||
$this->excludedSourceIds[$sourceId] = true;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 清除本窗口内"本轮明确排除"的包裹旧记录，并视为已扫描(不计入未见镜像包裹)。
|
||
* 只删明确排除的 source_id，未出现在上游扫描里的其它旧记录仍按原逻辑保留待核。
|
||
*/
|
||
private function purgeExcludedPackages(string $fromS, string $toS): int
|
||
{
|
||
$this->flushPending();
|
||
$ids = [];
|
||
foreach (array_keys($this->excludedSourceIds) as $id) {
|
||
if (! isset($this->seenSourceIds[$id])) {
|
||
$ids[] = $id;
|
||
$this->seenSourceIds[$id] = true;
|
||
}
|
||
}
|
||
$removed = 0;
|
||
foreach (array_chunk($ids, 500) as $chunk) {
|
||
$removed += (int) $this->db->table('jst_oms_shipping')->whereIn('source_id', $chunk)
|
||
->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS)->delete();
|
||
}
|
||
$this->excludedSourceIds = [];
|
||
|
||
return $removed;
|
||
}
|
||
|
||
/** 登记 单号×交易单号 已处理。 */
|
||
private function markTrackSo(string $so, string ...$tracks): void
|
||
{
|
||
$so = trim($so);
|
||
foreach ($tracks as $t) {
|
||
$t = ltrim(trim($t), '@');
|
||
if ($t !== '') {
|
||
$this->doneTrackSo[$t][$so] = true;
|
||
}
|
||
}
|
||
}
|
||
|
||
/** 登记已处理的出库单/快递单号，供兜底 pass 防重复。 */
|
||
private function markDone(string $io, string ...$tracks): void
|
||
{
|
||
if ($io !== '') {
|
||
$this->doneIo[$io] = true;
|
||
}
|
||
foreach ($tracks as $t) {
|
||
$t = ltrim(trim($t), '@');
|
||
if ($t !== '') {
|
||
$this->doneTrack[$t] = true;
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 兜底：出库单(按 io_date 窗口)里 express/autoship 都没处理过的包裹，按发货数据视图补吐。
|
||
* 典型：工厂代发仓/自有仓出库单无订单行、或同订单其它包裹在发货明细而本包裹不在。
|
||
* 防重复按 io_id + 快递单号(去@)，不按订单号。
|
||
*/
|
||
private function syncOutbound(string $fromS, string $toS, $now, ?callable $progress = null): int
|
||
{
|
||
$mm = $this->selectWithReconnect('SELECT MIN(Id) AS lo, MAX(Id) AS hi FROM ' . self::OUTBOUND . ' WHERE io_date >= ? AND io_date < ?', [$fromS, $toS]);
|
||
$lo = (int) ($mm[0]['lo'] ?? 0);
|
||
$hi = (int) ($mm[0]['hi'] ?? 0);
|
||
if ($lo <= 0 || $hi < $lo) {
|
||
if ($progress) {
|
||
$progress('出库单兜底：窗口内无出库单');
|
||
}
|
||
|
||
return 0;
|
||
}
|
||
$step = max(5000, (int) ceil(($hi - $lo + 1) / 400));
|
||
$cols = 'io_id, o_id, so_id, l_id, io_date, pay_date, shop_name, labels, status, logistics_company, weight, '
|
||
. 'buyer_message, receiver_state, receiver_city, receiver_district, receiver_phone';
|
||
|
||
$total = 0;
|
||
$scanned = 0;
|
||
$emitted = 0;
|
||
$skip = ['done' => 0, 'notrack' => 0, 'status' => 0, 'label' => 0, 'noitem' => 0];
|
||
$virtualEmitted = 0;
|
||
$seen = [];
|
||
$page = 0;
|
||
for ($a = $lo; $a <= $hi; $a += $step) {
|
||
$page++;
|
||
$rows = $this->selectWithReconnect(
|
||
"SELECT {$cols} FROM " . self::OUTBOUND . ' WHERE Id >= ? AND Id < ? AND io_date >= ? AND io_date < ?',
|
||
[$a, $a + $step, $fromS, $toS]
|
||
);
|
||
$scanned += count($rows);
|
||
$pick = [];
|
||
foreach ($rows as $r) {
|
||
$io = (string) ($r['io_id'] ?? '');
|
||
if ($io === '' || isset($seen[$io])) {
|
||
continue;
|
||
}
|
||
$seen[$io] = true;
|
||
$lid = trim((string) ($r['l_id'] ?? ''));
|
||
// 无单号出库单一般跳过；但"无需物流"的虚拟发货(信誉/补差/运费等)工作平台收在自动发货仓、快递单号为空，需补吐
|
||
$virtual = trim((string) ($r['logistics_company'] ?? '')) === self::NO_LOGISTICS;
|
||
if ($lid === '' && ! $virtual) {
|
||
$skip['notrack']++;
|
||
continue;
|
||
}
|
||
if (isset($this->doneIo[$io])) {
|
||
$skip['done']++;
|
||
continue;
|
||
}
|
||
// 单号已处理过：若是别的交易单号共用此单号(合包/拼单)，仍要补吐；交易单号要等视图表头加载后判断
|
||
$r['_trk_done'] = $lid !== '' && isset($this->doneTrack[ltrim($lid, '@')]);
|
||
$st = (string) ($r['status'] ?? '');
|
||
if (stripos($st, 'cancel') !== false || stripos($st, 'delete') !== false) {
|
||
$skip['status']++;
|
||
continue;
|
||
}
|
||
if ($this->excludedByLabel($r['labels'] ?? null)) {
|
||
$skip['label']++;
|
||
continue;
|
||
}
|
||
$pick[$io] = $r;
|
||
}
|
||
if ($pick !== []) {
|
||
$ioIds = array_map('strval', array_keys($pick));
|
||
$viewItems = $this->loadViewItems($ioIds);
|
||
$heads = $this->loadViewHeads($ioIds);
|
||
$soIds = [];
|
||
foreach ($pick as $pio => $r) {
|
||
$so = $this->outboundSo($r, $heads[(string) $pio] ?? []);
|
||
if ($so !== '') {
|
||
$soIds[$so] = true;
|
||
}
|
||
}
|
||
$skus = $this->loadSkus($this->collectSkuIds($viewItems));
|
||
$freight = $this->skipFreight ? [] : $this->loadFreight(array_map('strval', array_keys($soIds)));
|
||
foreach ($pick as $io => $r) {
|
||
$io = (string) $io;
|
||
if (empty($viewItems[$io])) {
|
||
$skip['noitem']++;
|
||
continue;
|
||
}
|
||
$h = $heads[$io] ?? [];
|
||
$so = $this->outboundSo($r, $h);
|
||
$tracking = trim((string) ($r['l_id'] ?? ''));
|
||
if (! empty($r['_trk_done'])) {
|
||
$t0 = ltrim($tracking, '@');
|
||
if ($so === '' || empty($this->doneTrackSo[$t0]) || isset($this->doneTrackSo[$t0][$so])) {
|
||
$skip['done']++;
|
||
continue;
|
||
}
|
||
}
|
||
$pkg = $this->pkgFromOutbound($r, $tracking !== '' ? $tracking : null, $this->ts($r['io_date'] ?? null), $h, $so);
|
||
$pkg = $this->withViewAddr($io, $pkg);
|
||
$pkg = $this->withViewCarrier($io, $pkg, self::OUTBOUND_OFFSET + (int) $io);
|
||
if (trim((string) ($pkg['express_name1'] ?? '')) === self::NO_LOGISTICS) {
|
||
if (empty($this->ioWare[$io])) {
|
||
$pkg['warehouse_name'] = self::NO_LOGISTICS_WAREHOUSE;
|
||
}
|
||
$virtualEmitted++;
|
||
}
|
||
$total += $this->emitPackage(self::OUTBOUND_OFFSET + (int) $io, $pkg, $viewItems[$io], $skus, $freight, $so, $tracking !== '' ? $tracking : null, $now, true);
|
||
$this->markDone($io, $tracking);
|
||
$this->markTrackSo($so, $tracking);
|
||
$emitted++;
|
||
}
|
||
unset($viewItems, $heads, $skus, $freight);
|
||
}
|
||
if ($progress && $page % 40 === 0) {
|
||
$progress(sprintf('出库单兜底：已扫描 %d 行，补吐 %d 包裹', $scanned, $emitted));
|
||
}
|
||
}
|
||
$this->flushPending();
|
||
if ($progress) {
|
||
$progress(sprintf(
|
||
'出库单兜底完成：扫描 %d 行，补吐 %d 包裹 / %d 行(其中无需物流虚拟发货 %d 包裹)；跳过 已处理%d 无单号%d 取消删除%d 标签%d 无商品%d',
|
||
$scanned, $emitted, $total, $virtualEmitted, $skip['done'], $skip['notrack'], $skip['status'], $skip['label'], $skip['noitem']
|
||
));
|
||
}
|
||
|
||
return $total;
|
||
}
|
||
|
||
/**
|
||
* 采购退货出库(退回供应商)：工作平台的发货数据视图收录了其中走物流的那部分，按视图补吐。
|
||
* 无交易单号/店铺/平台；仓库取视图(发货仓)，收货信息取采购退货单。
|
||
*/
|
||
private function syncPurchaseOut(string $fromS, string $toS, $now, ?callable $progress = null): int
|
||
{
|
||
$rows = $this->selectWithReconnect(
|
||
'SELECT io_id, io_date, status, l_id, so_id, logistics_company, receiver_state, receiver_city, receiver_district, '
|
||
. 'receiver_address, receiver_name, receiver_mobile FROM ' . self::PURCHASE_OUT . ' WHERE io_date >= ? AND io_date < ?',
|
||
[$fromS, $toS]
|
||
);
|
||
$pick = [];
|
||
$skip = ['done' => 0, 'cancel' => 0, 'noitem' => 0];
|
||
foreach ($rows as $r) {
|
||
$io = (string) ($r['io_id'] ?? '');
|
||
if ($io === '' || isset($pick[$io])) {
|
||
continue;
|
||
}
|
||
$lid = trim((string) ($r['l_id'] ?? ''));
|
||
if (isset($this->doneIo[$io]) || ($lid !== '' && isset($this->doneTrack[ltrim($lid, '@')]))) {
|
||
$skip['done']++;
|
||
continue;
|
||
}
|
||
if (stripos((string) ($r['status'] ?? ''), 'cancel') !== false) {
|
||
$skip['cancel']++;
|
||
continue;
|
||
}
|
||
$pick[$io] = $r;
|
||
}
|
||
$total = 0;
|
||
$emitted = 0;
|
||
if ($pick !== []) {
|
||
$ioIds = array_map('strval', array_keys($pick));
|
||
$viewItems = $this->loadViewItems($ioIds);
|
||
// 采购退货：每个条码数量取采购退货明细(jst_purchaseoutItem)，体积=单件长宽高×数量(对齐工作平台)
|
||
$poQty = [];
|
||
foreach (array_chunk($ioIds, 800) as $ck) {
|
||
$ph = implode(',', array_fill(0, count($ck), '?'));
|
||
foreach ($this->selectWithReconnect("SELECT io_id, sku_id, qty FROM rpt.jst_purchaseoutItem WHERE io_id IN ({$ph})", $ck) as $pr) {
|
||
$pk = mb_strtolower(trim((string) ($pr['sku_id'] ?? '')));
|
||
$poQty[(string) $pr['io_id']][$pk] = ($poQty[(string) $pr['io_id']][$pk] ?? 0) + (float) ($pr['qty'] ?? 0);
|
||
}
|
||
}
|
||
foreach ($viewItems as $vio => $vitems) {
|
||
foreach ($vitems as $vi => $it) {
|
||
$pq = $poQty[(string) $vio][mb_strtolower((string) $it['sku'])] ?? null;
|
||
if ($pq === null || $pq <= 0) {
|
||
continue;
|
||
}
|
||
$pq = (int) round($pq);
|
||
$viewItems[$vio][$vi]['qty'] = $pq;
|
||
}
|
||
}
|
||
$skus = $this->loadSkus($this->collectSkuIds($viewItems));
|
||
foreach ($pick as $io => $r) {
|
||
$io = (string) $io;
|
||
if (empty($viewItems[$io])) {
|
||
$skip['noitem']++;
|
||
continue;
|
||
}
|
||
$lid = trim((string) ($r['l_id'] ?? ''));
|
||
$tracking = $lid !== '' ? $lid : null;
|
||
$so = trim((string) ($r['so_id'] ?? ''));
|
||
$pkg = [
|
||
'tracking' => $tracking,
|
||
'so_id' => $so !== '' ? $so : null,
|
||
'express_name1' => $r['logistics_company'] ?? null,
|
||
'ship_time' => $this->ts($r['io_date'] ?? null),
|
||
'pay_time' => null,
|
||
'shop_name' => null,
|
||
'platform' => $this->ioSite[$io] ?? 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' => (string) ($r['receiver_address'] ?? ''),
|
||
'r_name' => $r['receiver_name'] ?? null,
|
||
'r_mobile' => $r['receiver_mobile'] ?? null,
|
||
'r_phone' => null,
|
||
'buyer_memo' => null,
|
||
'weight' => null,
|
||
];
|
||
$pkg = $this->withViewAddr($io, $pkg);
|
||
$total += $this->emitPackage(self::PURCHASE_OUT_OFFSET + (int) $io, $pkg, $viewItems[$io], $skus, [], $so, $tracking, $now, true);
|
||
$this->markDone($io, (string) $tracking);
|
||
$emitted++;
|
||
}
|
||
}
|
||
$this->flushPending();
|
||
if ($progress) {
|
||
$progress(sprintf(
|
||
'采购退货出库完成：窗口 %d 单，补吐 %d 包裹 / %d 行；跳过 已处理%d 已取消%d 视图无商品%d',
|
||
count($rows), $emitted, $total, $skip['done'], $skip['cancel'], $skip['noitem']
|
||
));
|
||
}
|
||
|
||
return $total;
|
||
}
|
||
|
||
/** io_id => 视图表头字段(平台/交易单号/省市区/店铺/快递)，出库单表头缺失时兜底用 */
|
||
private function loadViewHeads(array $ioIds): array
|
||
{
|
||
$out = [];
|
||
foreach (array_chunk(array_values(array_unique(array_filter($ioIds))), 300) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($chunk), '?'));
|
||
$sql = 'SELECT io_id, MAX(shop_site) AS site, MAX(so_id) AS so, MAX(receiver_state) AS st, MAX(receiver_city) AS ci, '
|
||
. 'MAX(receiver_district) AS di, MAX(shop_name) AS sn, MAX(expressName) AS ex FROM ' . self::VIEW
|
||
. " WHERE io_id IN ({$ph}) GROUP BY io_id";
|
||
foreach ($this->selectWithReconnect($sql, $chunk) as $r) {
|
||
$out[(string) $r['io_id']] = $r;
|
||
}
|
||
}
|
||
|
||
return $out;
|
||
}
|
||
|
||
/** 出库单表头 → 包裹级字段 */
|
||
private function outboundSo(array $r, array $h): string
|
||
{
|
||
$so = trim((string) ($r['so_id'] ?? ''));
|
||
|
||
return $so !== '' ? $so : trim((string) ($h['so'] ?? ''));
|
||
}
|
||
|
||
private function pkgFromOutbound(array $r, ?string $tracking, ?string $shipTime, array $h, string $so): array
|
||
{
|
||
$pick = fn (string $a, string $b) => (($r[$a] ?? null) !== null && trim((string) $r[$a]) !== '') ? $r[$a] : ($h[$b] ?? null);
|
||
|
||
return [
|
||
'tracking' => $tracking,
|
||
'so_id' => $so !== '' ? $so : null,
|
||
'express_name1' => $pick('logistics_company', 'ex'),
|
||
'ship_time' => $shipTime,
|
||
'pay_time' => $this->ts($r['pay_date'] ?? null),
|
||
'shop_name' => $pick('shop_name', 'sn'),
|
||
'platform' => $h['site'] ?? null,
|
||
'dis_co_name' => null,
|
||
'warehouse_name' => null,
|
||
'r_state' => $pick('receiver_state', 'st'),
|
||
'r_city' => $pick('receiver_city', 'ci'),
|
||
'r_district' => $pick('receiver_district', 'di'),
|
||
'r_address' => '',
|
||
'r_name' => null,
|
||
'r_mobile' => null,
|
||
'r_phone' => $r['receiver_phone'] ?? null,
|
||
'buyer_memo' => $r['buyer_message'] ?? null,
|
||
'weight' => $r['weight'] ?? null,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 输出一个包裹的所有明细行(每件一行；无商品则一行空商品)。
|
||
* @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);
|
||
$tk = ltrim(trim((string) ($tracking ?? '')), '@');
|
||
if ($tk !== '') {
|
||
$this->emittedTrack[$tk] = true;
|
||
}
|
||
$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;
|
||
}
|
||
|
||
/**
|
||
* 导出明细体积(对齐工作平台 OmsDataDay)：默认 数量×长×宽×高；
|
||
* 代发仓(配置)且非标准承运商 → 长×宽×高；特定快递 → 0；无尺寸 → 空。
|
||
*/
|
||
private function detailVolume(mixed $L, mixed $W, mixed $H, float $qty, string $warehouse, string $express): ?float
|
||
{
|
||
$cfg = (array) config('jushuitan.detail_volume', []);
|
||
$express = trim($express);
|
||
if ($express !== '' && in_array($express, (array) ($cfg['zero_volume_expresses'] ?? []), true)) {
|
||
return 0.0;
|
||
}
|
||
$l = $this->nz($L);
|
||
$w = $this->nz($W);
|
||
$h = $this->nz($H);
|
||
if (! $l || ! $w || ! $h) {
|
||
return null;
|
||
}
|
||
$unit = $l * $w * $h;
|
||
$noQty = in_array(trim($warehouse), (array) ($cfg['no_qty_warehouses'] ?? []), true)
|
||
&& ! in_array($express, (array) ($cfg['qty_carriers_in_no_qty_warehouses'] ?? []), true);
|
||
|
||
return round($noQty ? $unit : $unit * max($qty, 1), 4);
|
||
}
|
||
|
||
/** 组装一件明细行。包裹级字段(实称重量/运费)仅首件携带。 */
|
||
/** 文本对齐工作平台：工作平台入库时把 & 替换成 ^(如 ">" 导出为 "^gt;"，样本 22/22)。 */
|
||
private function wpText(mixed $v): ?string
|
||
{
|
||
if ($v === null || $v === '') {
|
||
return $v === '' ? '' : null;
|
||
}
|
||
|
||
return str_replace('&', '^', (string) $v);
|
||
}
|
||
|
||
/** 仓库名：聚水潭占位名"{未指定发货仓}"在工作平台仓库表里没有对应，导出为空。 */
|
||
private function whName(mixed $w): ?string
|
||
{
|
||
$w = trim((string) ($w ?? ''));
|
||
|
||
return ($w === '' || $w === '{未指定发货仓}') ? null : $w;
|
||
}
|
||
|
||
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' => $this->wpText(($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' => $this->whName($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' => $this->detailVolume($L, $W, $H, $qty, (string) ($pkg['warehouse_name'] ?: ($it['warehouse'] ?? '')), (string) ($pkg['express_name1'] ?? '')),
|
||
'm_name' => $this->wpText($it['mname'] ?? null),
|
||
'm_volume_cm3' => isset($it['mvol']) ? round((float) $it['mvol'], 4) : $vol,
|
||
'm_length_cm' => $it['ml'] ?? null,
|
||
'm_width_cm' => $it['mw'] ?? null,
|
||
'm_height_cm' => $it['mh'] ?? null,
|
||
'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,
|
||
]);
|
||
}
|
||
|
||
/**
|
||
* 快递公司/仓库以实际出库单为准(工作平台口径)：发货明细可能停在作废前的旧分配
|
||
* (如先分给工厂代发仓、后改外采由总仓出库；或换单后仍是旧快递公司)。
|
||
* 出库单快递公司与发货明细一致时不改，并借此学习 wms_co_id => 仓库名。
|
||
*/
|
||
private function withOutboundCarrier(string $io, array $r, array $pkg, int $pkgId): array
|
||
{
|
||
$lc = $io !== '' ? ($this->ioLc[$io] ?? '') : '';
|
||
$wms = $io !== '' ? ($this->ioWms[$io] ?? '') : '';
|
||
$eLc = trim((string) ($r['logistic_company_name'] ?? ''));
|
||
$eWh = trim((string) ($r['warehouse_name'] ?? ''));
|
||
if ($lc !== '' && $lc !== $eLc) {
|
||
$pkg['express_name1'] = $lc;
|
||
}
|
||
if (trim((string) ($pkg['express_name1'] ?? '')) === self::NO_LOGISTICS) {
|
||
$pkg['warehouse_name'] = self::NO_LOGISTICS_WAREHOUSE;
|
||
|
||
return $pkg;
|
||
}
|
||
if ($wms === '') {
|
||
return $pkg;
|
||
}
|
||
// 快递公司一致的明细行用来学习 wms_co_id => 仓库名；仓库统一在扫描结束后按出库单发货仓回填
|
||
if ($lc !== '' && $lc === $eLc && $eWh !== '' && $eWh !== '{未指定发货仓}') {
|
||
$this->wmsNames[$wms][$eWh] = ($this->wmsNames[$wms][$eWh] ?? 0) + 1;
|
||
}
|
||
$this->pendingWarehouse[$pkgId] = $wms;
|
||
|
||
return $pkg;
|
||
}
|
||
|
||
/**
|
||
* 工作平台的快递名称/仓库即发货数据视图的 expressName / wareHoursName(抽样 14/14 一致)：
|
||
* 视图有值时以视图为准；视图无该出库单时保留出库单/发货明细推断的值。
|
||
*/
|
||
private function withViewCarrier(string $io, array $pkg, int $pkgId): array
|
||
{
|
||
if ($io === '') {
|
||
return $pkg;
|
||
}
|
||
if (! empty($this->ioExpress[$io])) {
|
||
$pkg['express_name1'] = $this->ioExpress[$io];
|
||
}
|
||
if (! empty($this->ioWare[$io])) {
|
||
$pkg['warehouse_name'] = $this->ioWare[$io];
|
||
unset($this->pendingWarehouse[$pkgId]);
|
||
}
|
||
|
||
return $pkg;
|
||
}
|
||
|
||
private function wmsName(string $wms): ?string
|
||
{
|
||
if (empty($this->wmsNames[$wms])) {
|
||
return null;
|
||
}
|
||
$names = $this->wmsNames[$wms];
|
||
arsort($names);
|
||
$top = (int) reset($names);
|
||
$sum = array_sum($names);
|
||
// 只在仓库名明确(样本>=3 且占比>=90%)时回填，避免一个发货主体对应多个仓库时误改
|
||
if ($top < 3 || $top < 0.9 * $sum) {
|
||
return null;
|
||
}
|
||
|
||
return (string) array_key_first($names);
|
||
}
|
||
|
||
/** 扫描时还没学到仓库名的包裹：扫描结束后按学到的 wms_co_id => 仓库名统一回填。 */
|
||
private function applyPendingWarehouses(?callable $progress = null): void
|
||
{
|
||
$byName = [];
|
||
foreach ($this->pendingWarehouse as $pkgId => $wms) {
|
||
$name = $this->wmsName($wms);
|
||
if ($name !== null) {
|
||
$byName[$name][] = $pkgId;
|
||
}
|
||
}
|
||
$n = 0;
|
||
foreach ($byName as $name => $ids) {
|
||
foreach (array_chunk($ids, 500) as $chunk) {
|
||
$n += $this->db->table('jst_oms_shipping')->whereIn('source_id', $chunk)
|
||
->where(function ($q): void {
|
||
$q->whereNull('express_name1')->orWhere('express_name1', '!=', self::NO_LOGISTICS);
|
||
})
|
||
->where(function ($q) use ($name): void {
|
||
$q->whereNull('warehouse_name')->orWhere('warehouse_name', '!=', $name);
|
||
})
|
||
->update(['warehouse_name' => $name]);
|
||
}
|
||
}
|
||
if ($progress && $this->pendingWarehouse !== []) {
|
||
$progress(sprintf('出库单仓库回填：待补 %d 个包裹，更新 %d 行', count($this->pendingWarehouse), $n));
|
||
}
|
||
$this->pendingWarehouse = [];
|
||
}
|
||
|
||
/** 发货明细表行 → 包裹级字段(省市区从地址拆) */
|
||
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->selectWithReconnect("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->selectWithReconnect("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->selectWithReconnect("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->selectWithReconnect($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
|
||
{
|
||
$sql = 'SELECT l_id, o_id, io_id, io_date, status, logistics_company, wms_co_id FROM ' . self::OUTBOUND . ' WHERE io_date >= ? AND io_date < ?';
|
||
$this->outboundMap = $this->mapOutboundRows($this->selectWithReconnect($sql, [$fromS, $toS]));
|
||
}
|
||
|
||
/** 出库单行 -> 映射(单号/订单/单号+订单 三种定位，作废重出按有效优先、io_date 最新择优)。 */
|
||
private function mapOutboundRows(array $rows): array
|
||
{
|
||
$map = $this->emptyIoMap();
|
||
foreach ($rows as $r) {
|
||
$this->addIoRow($map, $r, true, true);
|
||
}
|
||
unset($map['rank']);
|
||
|
||
return $map;
|
||
}
|
||
|
||
private function ioMapFor(bool $windowed, array $logisticIds, array $oIds): array
|
||
{
|
||
return $windowed ? $this->outboundMap : $this->loadIoMap($logisticIds, $oIds);
|
||
}
|
||
|
||
/**
|
||
* 出库单映射：byL 单号->io、byO 订单->io、byLO 单号|订单->io、byOAll 订单->[io=>单号](仅有效单)、ioL io->单号。
|
||
* 同一单号/订单可能有多张出库单(作废重出)：优先有效(非 Delete/Cancelled)，再取最新 io_date。
|
||
*/
|
||
private function emptyIoMap(): array
|
||
{
|
||
return ['byL' => [], 'byO' => [], 'byLO' => [], 'byOAll' => [], 'ioL' => [], 'rank' => ['L' => [], 'O' => [], 'LO' => []]];
|
||
}
|
||
|
||
private function addIoRow(array &$m, array $r, bool $indexByL, bool $indexByO): void
|
||
{
|
||
$rio = (string) ($r['io_id'] ?? '');
|
||
if ($rio === '') {
|
||
return;
|
||
}
|
||
if (! empty($r['io_date'])) {
|
||
$this->ioDate[$rio] = (string) $r['io_date'];
|
||
}
|
||
$lc = trim((string) ($r['logistics_company'] ?? ''));
|
||
if ($lc !== '') {
|
||
$this->ioLc[$rio] = $lc;
|
||
}
|
||
$wms = trim((string) ($r['wms_co_id'] ?? ''));
|
||
if ($wms !== '' && $wms !== '0') {
|
||
$this->ioWms[$rio] = $wms;
|
||
}
|
||
$l = ltrim(trim((string) ($r['l_id'] ?? '')), '@');
|
||
$o = (string) ($r['o_id'] ?? '');
|
||
$m['ioL'][$rio] = $l;
|
||
$st = strtolower(trim((string) ($r['status'] ?? '')));
|
||
$ok = ! in_array($st, ['delete', 'deleted', 'cancelled', 'cancel', 'canceled'], true);
|
||
$this->ioOk[$rio] = $ok;
|
||
$rk = ($ok ? '1' : '0') . '|' . (string) ($r['io_date'] ?? '') . '|' . str_pad($rio, 12, '0', STR_PAD_LEFT);
|
||
if ($indexByL && $l !== '' && (! isset($m['rank']['L'][$l]) || strcmp($rk, $m['rank']['L'][$l]) > 0)) {
|
||
$m['byL'][$l] = $rio;
|
||
$m['rank']['L'][$l] = $rk;
|
||
}
|
||
// 同一单号多订单共用(拼单/合包)：按 单号+订单 精确定位，避免拿到别的订单的商品
|
||
$lo = $l . '|' . $o;
|
||
if ($indexByL && $l !== '' && $o !== '' && (! isset($m['rank']['LO'][$lo]) || strcmp($rk, $m['rank']['LO'][$lo]) > 0)) {
|
||
$m['byLO'][$lo] = $rio;
|
||
$m['rank']['LO'][$lo] = $rk;
|
||
}
|
||
if ($indexByO && $o !== '') {
|
||
if ($ok) {
|
||
$m['byOAll'][$o][$rio] = $l;
|
||
}
|
||
if (! isset($m['rank']['O'][$o]) || strcmp($rk, $m['rank']['O'][$o]) > 0) {
|
||
$m['byO'][$o] = $rio;
|
||
$m['rank']['O'][$o] = $rk;
|
||
}
|
||
}
|
||
}
|
||
|
||
/** 快递单号/内部单号 -> 出库单 io_id（非窗口诊断模式按需查询；窗口模式用预载映射）。 */
|
||
private function loadIoMap(array $logisticIds, array $oIds): array
|
||
{
|
||
$map = $this->emptyIoMap();
|
||
foreach (array_chunk(array_values(array_unique(array_filter($logisticIds))), 800) as $chunk) {
|
||
if ($chunk === []) {
|
||
continue;
|
||
}
|
||
$q = [];
|
||
foreach ($chunk as $c) {
|
||
$c = ltrim(trim((string) $c), '@');
|
||
if ($c !== '') {
|
||
$q[$c] = true;
|
||
$q['@' . $c] = true;
|
||
}
|
||
}
|
||
$q = array_keys($q);
|
||
$ph = implode(',', array_fill(0, count($q), '?'));
|
||
foreach ($this->selectWithReconnect('SELECT l_id, io_id, io_date, status, o_id, logistics_company, wms_co_id FROM ' . self::OUTBOUND . " WHERE l_id IN ({$ph})", $q) as $r) {
|
||
$this->addIoRow($map, $r, true, false);
|
||
}
|
||
}
|
||
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->selectWithReconnect('SELECT o_id, io_id, io_date, l_id, status, logistics_company, wms_co_id FROM ' . self::OUTBOUND . " WHERE o_id IN ({$ph})", $chunk) as $r) {
|
||
$this->addIoRow($map, $r, false, true);
|
||
}
|
||
}
|
||
unset($map['rank']);
|
||
|
||
return $map;
|
||
}
|
||
|
||
/**
|
||
* 换单配对：快递明细的单号在出库单里找不到时，到同一内部订单下找一张
|
||
* "出库单有单号、但该单号不在任何快递明细里"的出库单，一对一配对(工作平台换单后只保留新单号)。
|
||
*
|
||
* @return array<string,array{io:string,l_id:string,old:string}> 以快递明细 id 为键
|
||
*/
|
||
private function pairChangedTracking(array $rows, array $ioMap): array
|
||
{
|
||
$cands = [];
|
||
$need = [];
|
||
foreach ($rows as $r) {
|
||
$lg = ltrim(trim((string) ($r['logistic_id'] ?? '')), '@');
|
||
$oid = (string) ($r['inner_order_id'] ?? '');
|
||
$byL = $ioMap['byL'][$lg] ?? null;
|
||
// 快递明细单号只对应作废出库单时，视为已换单(工作平台显示有效出库单的新单号)
|
||
$hasValidIo = $byL !== null && ($this->ioOk[(string) $byL] ?? true);
|
||
if ($lg === '' || $oid === '' || $hasValidIo || empty($ioMap['byOAll'][$oid])
|
||
|| in_array((string) ($r['jo_status'] ?? ''), ['Cancelled', 'Split', 'Merged'], true)) {
|
||
continue;
|
||
}
|
||
$need[] = $r;
|
||
foreach ($ioMap['byOAll'][$oid] as $io => $l) {
|
||
if ($l !== '' && ! isset($this->doneIo[(string) $io]) && ! isset($this->doneTrack[$l])) {
|
||
$cands[$l] = true;
|
||
}
|
||
}
|
||
}
|
||
if ($need === [] || $cands === []) {
|
||
return [];
|
||
}
|
||
// 候选新单号若本身出现在快递明细里(同订单拆包)，不能拿来配对
|
||
$inExpress = [];
|
||
foreach (array_chunk(array_keys($cands), 400) as $chunk) {
|
||
$q = [];
|
||
foreach ($chunk as $c) {
|
||
$q[] = (string) $c;
|
||
$q[] = '@' . $c;
|
||
}
|
||
$ph = implode(',', array_fill(0, count($q), '?'));
|
||
try {
|
||
$found = $this->selectWithReconnect("SELECT DISTINCT logistic_id FROM rpt.jst_order_express_details WHERE logistic_id IN ({$ph})", $q);
|
||
} catch (PDOException $e) {
|
||
if (! $this->isTransientCursorFailure($e)) {
|
||
throw $e;
|
||
}
|
||
// logistic_id 无索引，副本库繁忙时会超时：本页跳过换单配对(保留原单号)，不中断整轮
|
||
$this->pairingSkipped++;
|
||
|
||
return [];
|
||
}
|
||
foreach ($found as $x) {
|
||
$inExpress[ltrim(trim((string) ($x['logistic_id'] ?? '')), '@')] = true;
|
||
}
|
||
}
|
||
$used = [];
|
||
$swap = [];
|
||
foreach ($need as $r) {
|
||
$oid = (string) $r['inner_order_id'];
|
||
$ios = $ioMap['byOAll'][$oid];
|
||
ksort($ios, SORT_STRING);
|
||
foreach ($ios as $io => $l) {
|
||
$io = (string) $io;
|
||
if ($l === '' || isset($inExpress[$l]) || isset($used[$io]) || isset($this->doneIo[$io]) || isset($this->doneTrack[$l])) {
|
||
continue;
|
||
}
|
||
$used[$io] = true;
|
||
$swap[(string) $r['id']] = ['io' => $io, 'l_id' => $l, 'old' => (string) $r['logistic_id']];
|
||
break;
|
||
}
|
||
}
|
||
|
||
return $swap;
|
||
}
|
||
|
||
private function ioFor(array $ioMap, string $logisticId, string $oid): string
|
||
{
|
||
$logisticId = ltrim(trim($logisticId), '@');
|
||
if ($logisticId !== '' && $oid !== '' && ! empty($ioMap['byLO'][$logisticId . '|' . $oid])) {
|
||
return (string) $ioMap['byLO'][$logisticId . '|' . $oid];
|
||
}
|
||
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 之和优先。
|
||
* 多条码行(组合件)：见 appendComboParts，对齐工作平台导出明细的拆件口径。
|
||
*/
|
||
private function loadViewItems(array $ioIds): array
|
||
{
|
||
$agg = [];
|
||
$multi = [];
|
||
$this->ioSite = [];
|
||
$this->ioAddr = [];
|
||
$this->ioShop = [];
|
||
$this->ioExpress = [];
|
||
$this->ioWare = [];
|
||
// 发货数据视图是计算视图，IN 列表过长时副本库容易超时：按 300 个出库单一批
|
||
foreach (array_chunk(array_values(array_unique(array_filter($ioIds))), 300) 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, expressName, l, w, h, volume, weight, supplier_name, shop_site, shop_name, receiver_state, receiver_city, receiver_district FROM '
|
||
. self::VIEW . " WHERE io_id IN ({$ph})";
|
||
foreach ($this->selectWithReconnect($sql, $chunk) as $r) {
|
||
$io = (string) ($r['io_id'] ?? '');
|
||
if ($io !== '' && ! isset($this->ioSite[$io]) && trim((string) ($r['shop_site'] ?? '')) !== '') {
|
||
$this->ioSite[$io] = trim((string) $r['shop_site']);
|
||
}
|
||
if ($io !== '' && ! isset($this->ioShop[$io]) && trim((string) ($r['shop_name'] ?? '')) !== '') {
|
||
$this->ioShop[$io] = trim((string) $r['shop_name']);
|
||
}
|
||
if ($io !== '' && ! isset($this->ioExpress[$io]) && trim((string) ($r['expressName'] ?? '')) !== '') {
|
||
$this->ioExpress[$io] = trim((string) $r['expressName']);
|
||
}
|
||
if ($io !== '' && ! isset($this->ioWare[$io]) && trim((string) ($r['wareHoursName'] ?? '')) !== '') {
|
||
$this->ioWare[$io] = trim((string) $r['wareHoursName']);
|
||
}
|
||
if ($io !== '') {
|
||
foreach (['st' => 'receiver_state', 'ci' => 'receiver_city', 'di' => 'receiver_district'] as $ak => $ac) {
|
||
$av = trim((string) ($r[$ac] ?? ''));
|
||
if ($av !== '' && ! isset($this->ioAddr[$io][$ak])) {
|
||
$this->ioAddr[$io][$ak] = $av;
|
||
}
|
||
}
|
||
}
|
||
$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);
|
||
if ($n > 1) {
|
||
$multi[$io][] = [
|
||
't' => (string) ($r['type'] ?? ''), 'qty' => (float) ($r['qty'] ?? 0), 'bcs' => $bcs,
|
||
'name' => $r['sku_name'] ?? null, 'l' => $r['l'] ?? null, 'w' => $r['w'] ?? null, 'h' => $r['h'] ?? null,
|
||
'vol' => $r['volume'] ?? null, 'shopid' => $r['shop_i_id'] ?? null, 'wh' => $r['wareHoursName'] ?? null,
|
||
'supplier' => $r['supplier_name'] ?? null,
|
||
];
|
||
continue;
|
||
}
|
||
$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'],
|
||
];
|
||
}
|
||
|
||
if ($multi !== []) {
|
||
$this->appendComboParts($multi, $out);
|
||
}
|
||
|
||
return $out;
|
||
}
|
||
|
||
/**
|
||
* 组合件拆件(对齐工作平台导出明细)：视图一行逗号拼接多个部件条码时，每个部件一行——
|
||
* 名称/长宽高取 SKU 资料，体积=长×宽×高；
|
||
* 数量：订单商品 → 运费逐件重量÷SKU单重 → 整数均分，前两者须部件合计=整行数量才采用；
|
||
* 同时带整行 名称/体积(按部件数均摊)/长宽高(m_*)，供合并口径使用(工作平台合并取整行值)。
|
||
* 同一出库单有 type=0 行只用 type=0，否则用最小 type，避免 type=3 等重复行重复计数。
|
||
*/
|
||
private function appendComboParts(array $multi, array &$out): void
|
||
{
|
||
$sel = [];
|
||
$bcAll = [];
|
||
foreach ($multi as $io => $rows) {
|
||
$types = array_values(array_unique(array_map(fn ($r) => $r['t'], $rows)));
|
||
$useT = in_array('0', $types, true) ? '0' : min($types);
|
||
foreach ($rows as $r) {
|
||
if ($r['t'] === $useT) {
|
||
$sel[(string) $io][] = $r;
|
||
foreach ($r['bcs'] as $b) {
|
||
$bcAll[$b] = true;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
$ioKeys = array_map('strval', array_keys($sel));
|
||
|
||
$o2io = [];
|
||
foreach (array_chunk($ioKeys, 800) as $ck) {
|
||
$ph = implode(',', array_fill(0, count($ck), '?'));
|
||
foreach ($this->selectWithReconnect('SELECT io_id, o_id FROM ' . self::OUTBOUND . " WHERE io_id IN ({$ph})", $ck) as $r) {
|
||
$o = (string) ($r['o_id'] ?? '');
|
||
if ($o !== '') {
|
||
$o2io[$o][(string) $r['io_id']] = true;
|
||
}
|
||
}
|
||
}
|
||
$ordQty = [];
|
||
foreach (array_chunk(array_map('strval', array_keys($o2io)), 800) as $ck) {
|
||
$ph = implode(',', array_fill(0, count($ck), '?'));
|
||
foreach ($this->selectWithReconnect("SELECT o_id, sku_id, qty FROM rpt.jst_OrderItem WHERE o_id IN ({$ph})", $ck) as $r) {
|
||
$k = mb_strtolower(trim((string) ($r['sku_id'] ?? '')));
|
||
foreach ($o2io[(string) $r['o_id']] ?? [] as $io => $_) {
|
||
$ordQty[(string) $io][$k] = ($ordQty[(string) $io][$k] ?? 0) + (float) ($r['qty'] ?? 0);
|
||
}
|
||
}
|
||
}
|
||
$fysW = [];
|
||
foreach (array_chunk($ioKeys, 800) as $ck) {
|
||
$ph = implode(',', array_fill(0, count($ck), '?'));
|
||
foreach ($this->selectWithReconnect("SELECT io_id, sku_id, weight FROM rpt.fys_jstFreightCostDetail WHERE io_id IN ({$ph})", $ck) as $r) {
|
||
$k = mb_strtolower(trim((string) ($r['sku_id'] ?? '')));
|
||
$fysW[(string) $r['io_id']][$k] = ($fysW[(string) $r['io_id']][$k] ?? 0) + (float) ($r['weight'] ?? 0);
|
||
}
|
||
}
|
||
$skuL = [];
|
||
foreach ($this->loadSkus(array_map('strval', array_keys($bcAll))) as $k => $v) {
|
||
$skuL[mb_strtolower((string) $k)] = $v;
|
||
}
|
||
$raw = fn ($v) => ($v === null || trim((string) $v) === '') ? null : (float) $v;
|
||
|
||
foreach ($sel as $io => $rows) {
|
||
$io = (string) $io;
|
||
$parts = [];
|
||
foreach ($rows as $r) {
|
||
$bcs = $r['bcs'];
|
||
$n = count($bcs);
|
||
$q = $this->partQty($bcs, (int) round($r['qty']), $ordQty[$io] ?? [], $fysW[$io] ?? [], $skuL);
|
||
$rowVol = $raw($r['vol']);
|
||
foreach ($bcs as $i => $bc) {
|
||
$lk = mb_strtolower($bc);
|
||
$s = $skuL[$lk] ?? [];
|
||
$L = $this->nz($s['l'] ?? null);
|
||
$W = $this->nz($s['w'] ?? null);
|
||
$H = $this->nz($s['h'] ?? null);
|
||
if (! isset($parts[$lk])) {
|
||
$parts[$lk] = [
|
||
'id' => null, 'sku' => $bc, 'iid' => $s['iid'] ?? null,
|
||
'shopid' => ($r['shopid'] ?? null) ?: null,
|
||
'name' => ($s['name'] ?? null) ?: $r['name'],
|
||
'qty' => 0, 'amount' => null, 'warehouse' => ($r['wh'] ?? null) ?: null,
|
||
'l' => $L, 'w' => $W, 'h' => $H,
|
||
'vol' => ($L && $W && $H) ? round($L * $W * $H, 4) : null,
|
||
'weight' => $s['weight'] ?? null,
|
||
'supplier' => ($r['supplier'] ?? null) ?: null,
|
||
'mname' => $r['name'], 'mvol' => null,
|
||
'ml' => $raw($r['l']), 'mw' => $raw($r['w']), 'mh' => $raw($r['h']),
|
||
];
|
||
}
|
||
$parts[$lk]['qty'] += $q[$i];
|
||
if ($rowVol !== null) {
|
||
$parts[$lk]['mvol'] = ($parts[$lk]['mvol'] ?? 0) + $rowVol / $n;
|
||
}
|
||
}
|
||
}
|
||
foreach ($parts as $lk => $p) {
|
||
$p['mvol'] = $p['mvol'] !== null ? round($p['mvol'], 4) : null;
|
||
$hit = false;
|
||
foreach ($out[$io] ?? [] as $idx => $it) {
|
||
if (mb_strtolower((string) $it['sku']) === $lk) {
|
||
$out[$io][$idx]['qty'] += $p['qty'];
|
||
$hit = true;
|
||
break;
|
||
}
|
||
}
|
||
if (! $hit) {
|
||
$out[$io][] = $p;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/** 组合件部件数量：订单商品 → 运费逐件重量÷SKU单重 → 整数均分；前两者须合计=整行数量。 */
|
||
private function partQty(array $bcs, int $rowQty, array $ord, array $fys, array $skuL): array
|
||
{
|
||
$n = count($bcs);
|
||
$keys = array_map(fn ($b) => mb_strtolower($b), $bcs);
|
||
$a = [];
|
||
foreach ($keys as $k) {
|
||
if (! isset($ord[$k])) {
|
||
$a = null;
|
||
break;
|
||
}
|
||
$a[] = (int) round($ord[$k]);
|
||
}
|
||
if ($a !== null && array_sum($a) === $rowQty && min($a) > 0) {
|
||
return $a;
|
||
}
|
||
$b = [];
|
||
foreach ($keys as $k) {
|
||
$sw = (float) ($skuL[$k]['weight'] ?? 0);
|
||
if (! isset($fys[$k]) || $sw <= 0) {
|
||
$b = null;
|
||
break;
|
||
}
|
||
$b[] = (int) round($fys[$k] / $sw);
|
||
}
|
||
if ($b !== null && array_sum($b) === $rowQty && min($b) > 0) {
|
||
return $b;
|
||
}
|
||
$total = max($rowQty, 0);
|
||
$base = intdiv($total, $n);
|
||
$rem = $total - $base * $n;
|
||
$c = [];
|
||
foreach ($keys as $i => $_) {
|
||
$c[] = $base + ($i < $rem ? 1 : 0);
|
||
}
|
||
|
||
return $c;
|
||
}
|
||
|
||
private function nz(mixed $v): ?float
|
||
{
|
||
if ($v === null || trim((string) $v) === '') {
|
||
return null;
|
||
}
|
||
$f = (float) $v;
|
||
|
||
return $f == 0.0 ? null : $f;
|
||
}
|
||
|
||
/** 收货省市区统一取发货数据视图(与工作平台一致)，视图无值时保留原值。 */
|
||
private function withViewAddr(string $io, array $pkg): array
|
||
{
|
||
if ($io !== '' && trim((string) ($pkg['shop_name'] ?? '')) === '' && isset($this->ioShop[$io])) {
|
||
$pkg['shop_name'] = $this->ioShop[$io];
|
||
}
|
||
$a = ($io !== '') ? ($this->ioAddr[$io] ?? []) : [];
|
||
foreach (['st' => 'r_state', 'ci' => 'r_city', 'di' => 'r_district'] as $ak => $pk) {
|
||
if (isset($a[$ak]) && $a[$ak] !== '') {
|
||
$pkg[$pk] = $a[$ak];
|
||
}
|
||
}
|
||
|
||
return $pkg;
|
||
}
|
||
|
||
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 = [];
|
||
$refresh = [];
|
||
foreach ($packages as $id => $packageRows) {
|
||
if (! ($hasItems[$id] ?? false) && ($oldHasItems[$id] ?? false)) {
|
||
$this->packagesRetainedForMissingItems++;
|
||
$refresh[$id] = $packageRows[0] ?? [];
|
||
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++;
|
||
$refresh[$id] = $packageRows[0] ?? [];
|
||
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);
|
||
}
|
||
$this->refreshRetainedPackages($refresh);
|
||
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, '差价专用');
|
||
}
|
||
|
||
/**
|
||
* 线上发货标签排除口径(对齐工作平台)：
|
||
* 工作平台只排除"刷单/批发/纯售后"这类无实物的线上发货虚拟单；
|
||
* 云仓/京仓(CLPS/ECLP)真实发货工作平台是收进的，不能排除。
|
||
* 实测(9.1-9.16)：该口径误排 2、误收 0。
|
||
*/
|
||
private function excludedByLabel(?string $labels): bool
|
||
{
|
||
if ($labels === null || $labels === '') {
|
||
return false;
|
||
}
|
||
if (! str_contains($labels, '线上发货')) {
|
||
return false;
|
||
}
|
||
// 云仓/京仓真实发货：工作平台收进，保留
|
||
if (str_contains($labels, '云仓订单') || str_contains($labels, '京仓订单')
|
||
|| str_contains($labels, 'CLPS') || str_contains($labels, 'ECLP')) {
|
||
return false;
|
||
}
|
||
|
||
return true;
|
||
}
|
||
|
||
/** @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->selectWithReconnect("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->selectWithReconnect($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;
|
||
$superseded = [];
|
||
$rows = $this->db->table('jst_oms_shipping')
|
||
->where('ship_time', '>=', $fromS)->where('ship_time', '<', $toS)
|
||
->selectRaw('source_id, MAX(express_no) AS express_no')->groupBy('source_id')->get();
|
||
foreach ($rows as $row) {
|
||
$id = (int) $row->source_id;
|
||
if (isset($this->seenSourceIds[$id])) {
|
||
continue;
|
||
}
|
||
$tk = ltrim(trim((string) ($row->express_no ?? '')), '@');
|
||
if ($tk !== '' && isset($this->emittedTrack[$tk])) {
|
||
$superseded[] = $id;
|
||
continue;
|
||
}
|
||
$unseen++;
|
||
}
|
||
foreach (array_chunk($superseded, 500) as $chunk) {
|
||
$this->db->table('jst_oms_shipping')->whereIn('source_id', $chunk)->delete();
|
||
}
|
||
$this->supersededRemoved = count($superseded);
|
||
|
||
return $unseen;
|
||
}
|
||
|
||
/** 因商品/金额无法安全替换而保留的旧包裹：包裹级字段(省市区、快递、仓库、店铺等)仍按本次上游刷新，商品与金额不动。 */
|
||
private function refreshRetainedPackages(array $refresh): void
|
||
{
|
||
foreach ($refresh as $id => $new) {
|
||
$set = [];
|
||
foreach (self::REFRESHED_PACKAGE_FIELDS as $field) {
|
||
$v = $new[$field] ?? null;
|
||
if ($v !== null && trim((string) $v) !== '') {
|
||
$set[$field] = $v;
|
||
}
|
||
}
|
||
if ($set !== []) {
|
||
$this->db->table('jst_oms_shipping')->where('source_id', (int) $id)->update($set);
|
||
}
|
||
}
|
||
}
|
||
|
||
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(),
|
||
]);
|
||
}
|
||
}
|