limeng
b038909cb0
- 聚水潭奇门客户端与发货同步器骨架:QimenClient(签名+POST)、ShippingDataSynchronizer(erp.sctrade.query 列表 + erp.sradesku.query 明细 + upsert + modified 水位) - 发货明细扁平展示表 jst_oms_shipping(对齐工作平台发货数据 47 列/顺序,用于快递运费核算);原始表 jst_sctrade/_items/_skus/_receiver + jst_sync_state - 快递应付新增「发货数据」页:Pages.contentForShippingData + 视图 + business_navigation 导航项(slug=shipping-data) - ShippingOrderRepository 读展示表;artisan jushuitan:sync-shipping + 每10分钟增量调度(JST_SYNC_ENABLED 默认关) - 落库连接 jushuitan_local(本地容器,env 可覆盖);演示 Seeder(6 条家具发货样例);.env.jushuitan.example Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014nvGZkg27YMEdhRy6EiUwf
279 lines
11 KiB
PHP
279 lines
11 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Services\Jushuitan;
|
||
|
||
use Illuminate\Database\ConnectionInterface;
|
||
use Illuminate\Support\Carbon;
|
||
use Illuminate\Support\Facades\DB;
|
||
use Illuminate\Support\Facades\Log;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 聚水潭发货（销售出库单）同步器。
|
||
*
|
||
* 复刻工作平台 ErpTradeData 的两步拉取：
|
||
* 1) erp.sctrade.query 按修改时间窗口分页拉出库单列表 -> 得到每单 id / modified
|
||
* 2) erp.sradesku.query 按 tradeId 拉单据完整明细 -> 主单 + items + skus + receiver
|
||
* 增量以 modified 为水位，带重叠窗口；按 source_id upsert 去重。
|
||
*
|
||
* 注意：奇门返回的 JSON 字段路径以聚水潭实际返回为准，本骨架的取值点已用
|
||
* pluck() 集中，联调时按真实返回调整即可，不影响主流程结构。
|
||
*/
|
||
class ShippingDataSynchronizer
|
||
{
|
||
private const DATASET = 'jst.sctrade';
|
||
|
||
private readonly ConnectionInterface $db;
|
||
|
||
public function __construct(
|
||
private readonly QimenClient $client,
|
||
private readonly array $config,
|
||
) {
|
||
$this->db = DB::connection($config['connection'] ?? 'business_mirror');
|
||
}
|
||
|
||
public static function make(): self
|
||
{
|
||
return new self(QimenClient::fromConfig(), (array) config('jushuitan'));
|
||
}
|
||
|
||
/**
|
||
* 执行一次增量同步。
|
||
*
|
||
* @param ?Carbon $from 起始修改时间；默认取水位 - overlap
|
||
* @param ?Carbon $to 截止修改时间；默认 now
|
||
*/
|
||
public function syncIncremental(?Carbon $from = null, ?Carbon $to = null): array
|
||
{
|
||
$sync = $this->config['sync'] ?? [];
|
||
$overlap = max((int) ($sync['overlap_hours'] ?? 48), 0);
|
||
$limit = max((int) ($sync['page_limit'] ?? 100), 1);
|
||
$maxPages = (int) ($sync['max_pages'] ?? 0);
|
||
|
||
$state = $this->state();
|
||
$from ??= isset($state['watermark']) && $state['watermark']
|
||
? Carbon::parse($state['watermark'])->subHours($overlap)
|
||
: now()->subHours(max((int) ($sync['window_hours'] ?? 24), 1));
|
||
$to ??= now();
|
||
|
||
$this->markRunning();
|
||
$processed = 0;
|
||
$maxModified = $from;
|
||
|
||
try {
|
||
$page = 1;
|
||
do {
|
||
$list = $this->client->call($this->method('sctrade_list'), [
|
||
'updateSt' => $from->format('Y-m-d H:i:s'),
|
||
'updateEt' => $to->format('Y-m-d H:i:s'),
|
||
'limit' => $limit,
|
||
'page' => $page,
|
||
]);
|
||
|
||
$orders = $this->pluckOrders($list);
|
||
foreach ($orders as $order) {
|
||
$id = $this->pluckId($order);
|
||
if ($id === null) {
|
||
continue;
|
||
}
|
||
$detail = $this->client->call($this->method('sctrade_detail'), ['tradeId' => $id]);
|
||
$data = $detail['data'] ?? $detail;
|
||
$this->upsertOrder($data);
|
||
$processed++;
|
||
|
||
$modified = $this->pluckModified($data);
|
||
if ($modified && (! $maxModified || $modified->gt($maxModified))) {
|
||
$maxModified = $modified;
|
||
}
|
||
}
|
||
|
||
$hasNext = count($orders) >= $limit && ($maxPages === 0 || $page < $maxPages);
|
||
$page++;
|
||
} while ($hasNext);
|
||
|
||
$this->markSucceeded($maxModified);
|
||
} catch (Throwable $e) {
|
||
$this->markFailed($e->getMessage());
|
||
report($e);
|
||
throw $e;
|
||
}
|
||
|
||
return ['processed' => $processed, 'watermark' => optional($maxModified)->toDateTimeString()];
|
||
}
|
||
|
||
/* ---------------- upsert（按 source_id 去重） ---------------- */
|
||
|
||
private function upsertOrder(array $data): void
|
||
{
|
||
$sourceId = $this->pluckId($data);
|
||
if ($sourceId === null) {
|
||
return;
|
||
}
|
||
$now = now();
|
||
|
||
$this->db->table('jst_sctrade')->updateOrInsert(
|
||
['source_id' => $sourceId],
|
||
[
|
||
'sourceno' => $data['so_id'] ?? ($data['sourceno'] ?? null),
|
||
'io_id' => $data['io_id'] ?? null,
|
||
'shop_id' => $data['shop_id'] ?? null,
|
||
'shopcode' => $data['shopcode'] ?? null,
|
||
'wms_co_id' => $data['wms_co_id'] ?? null,
|
||
'platform' => $data['platform'] ?? null,
|
||
'status' => $data['status'] ?? null,
|
||
'paytime' => $this->ts($data['pay_date'] ?? ($data['paytime'] ?? null)),
|
||
'sended' => $this->ts($data['sended'] ?? null),
|
||
'created' => $this->ts($data['created'] ?? null),
|
||
'modified' => $this->ts($data['modified'] ?? null),
|
||
'receiverstate' => $data['receiver_state'] ?? ($data['receiverstate'] ?? null),
|
||
'receivercity' => $data['receiver_city'] ?? ($data['receivercity'] ?? null),
|
||
'receiverdistrict' => $data['receiver_district'] ?? ($data['receiverdistrict'] ?? null),
|
||
'receivername' => $data['receiver_name'] ?? ($data['receivername'] ?? null),
|
||
'receivermobile' => $data['receiver_mobile'] ?? ($data['receivermobile'] ?? null),
|
||
'receiverphone' => $data['receiver_phone'] ?? ($data['receiverphone'] ?? null),
|
||
'receiveraddress' => $data['receiver_address'] ?? ($data['receiveraddress'] ?? null),
|
||
'weight' => $data['weight'] ?? null,
|
||
'freight' => $data['freight'] ?? null,
|
||
'raw_json' => json_encode($data, JSON_UNESCAPED_UNICODE),
|
||
'synced_at' => $now,
|
||
]
|
||
);
|
||
|
||
foreach (($data['items'] ?? []) as $item) {
|
||
$iid = $item['id'] ?? null;
|
||
if ($iid === null) {
|
||
continue;
|
||
}
|
||
$this->db->table('jst_sctrade_items')->updateOrInsert(
|
||
['source_id' => $iid],
|
||
[
|
||
'hid' => $sourceId,
|
||
'oid' => $item['oid'] ?? null,
|
||
'shopproid' => $item['shop_i_id'] ?? ($item['shopproid'] ?? null),
|
||
'shoprocode' => $item['sku_id'] ?? ($item['shoprocode'] ?? null),
|
||
'shopskucode' => $item['shopskucode'] ?? null,
|
||
'shoproname' => $item['name'] ?? ($item['shoproname'] ?? null),
|
||
'qty' => $item['qty'] ?? null,
|
||
'raw_json' => json_encode($item, JSON_UNESCAPED_UNICODE),
|
||
'synced_at' => $now,
|
||
]
|
||
);
|
||
}
|
||
|
||
foreach (($data['tradeskus'] ?? ($data['skus'] ?? [])) as $sku) {
|
||
$sid = $sku['id'] ?? null;
|
||
if ($sid === null) {
|
||
continue;
|
||
}
|
||
$this->db->table('jst_sctrade_skus')->updateOrInsert(
|
||
['source_id' => $sid],
|
||
[
|
||
'hid' => $sourceId,
|
||
'procode' => $sku['sku_id'] ?? ($sku['procode'] ?? null),
|
||
'sku_id' => $sku['sku_id'] ?? null,
|
||
'proname' => $sku['name'] ?? ($sku['proname'] ?? null),
|
||
'scmname' => $sku['scmname'] ?? null,
|
||
'postcode' => $sku['postcode'] ?? ($sku['l_id'] ?? null),
|
||
'logistics_company' => $sku['logistics_company'] ?? null,
|
||
'qty' => $sku['qty'] ?? null,
|
||
'sended' => $this->ts($sku['sended'] ?? null),
|
||
'raw_json' => json_encode($sku, JSON_UNESCAPED_UNICODE),
|
||
'synced_at' => $now,
|
||
]
|
||
);
|
||
}
|
||
|
||
$receiver = $data['tradeReceiver'] ?? ($data['receiver'] ?? null);
|
||
if (is_array($receiver)) {
|
||
$this->db->table('jst_sctrade_receiver')->updateOrInsert(
|
||
['hid' => $sourceId],
|
||
[
|
||
'receivername' => $receiver['receivername'] ?? ($receiver['name'] ?? null),
|
||
'receiverstate' => $receiver['receiverstate'] ?? null,
|
||
'receivercity' => $receiver['receivercity'] ?? null,
|
||
'receiverdistrict' => $receiver['receiverdistrict'] ?? null,
|
||
'receiveraddress' => $receiver['receiveraddress'] ?? null,
|
||
'receivermobile' => $receiver['receivermobile'] ?? null,
|
||
'receiverphone' => $receiver['receiverphone'] ?? null,
|
||
'receiverzip' => $receiver['receiverzip'] ?? null,
|
||
'raw_json' => json_encode($receiver, JSON_UNESCAPED_UNICODE),
|
||
'synced_at' => $now,
|
||
]
|
||
);
|
||
}
|
||
}
|
||
|
||
/* ---------------- 返回体取值点（按聚水潭实际返回调整） ---------------- */
|
||
|
||
/** @return array<int,array> */
|
||
private function pluckOrders(array $list): array
|
||
{
|
||
$data = $list['data'] ?? $list;
|
||
return $data['datas'] ?? ($data['orders'] ?? ($data['list'] ?? []));
|
||
}
|
||
|
||
private function pluckId(array $order): ?int
|
||
{
|
||
$id = $order['id'] ?? ($order['io_id'] ?? null);
|
||
return $id === null ? null : (int) $id;
|
||
}
|
||
|
||
private function pluckModified(array $data): ?Carbon
|
||
{
|
||
return $this->ts($data['modified'] ?? null);
|
||
}
|
||
|
||
private function ts(mixed $value): ?Carbon
|
||
{
|
||
if ($value === null || $value === '' ) {
|
||
return null;
|
||
}
|
||
try {
|
||
if (is_numeric($value)) {
|
||
$v = (int) $value;
|
||
return Carbon::createFromTimestamp($v > 9999999999 ? intdiv($v, 1000) : $v);
|
||
}
|
||
return Carbon::parse((string) $value);
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/* ---------------- 同步状态 ---------------- */
|
||
|
||
private function state(): array
|
||
{
|
||
$row = $this->db->table('jst_sync_state')->where('dataset_key', self::DATASET)->first();
|
||
return $row ? (array) $row : [];
|
||
}
|
||
|
||
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(?Carbon $watermark): void
|
||
{
|
||
$update = ['status' => 'succeeded', 'last_succeeded_at' => now(), 'last_error' => null, 'updated_at' => now()];
|
||
if ($watermark) {
|
||
$update['watermark'] = $watermark;
|
||
}
|
||
$this->db->table('jst_sync_state')->where('dataset_key', self::DATASET)->update($update);
|
||
}
|
||
|
||
private function markFailed(string $error): void
|
||
{
|
||
$this->db->table('jst_sync_state')->where('dataset_key', self::DATASET)->update([
|
||
'status' => 'failed', 'last_error' => mb_substr($error, 0, 1000), 'updated_at' => now(),
|
||
]);
|
||
}
|
||
|
||
private function method(string $key): string
|
||
{
|
||
return $this->config['methods'][$key] ?? '';
|
||
}
|
||
}
|