yeuimu
4a7c44a354
CI / test (pull_request) Successful in 1m11s
- api/riin/client.py: 新增 get_order_goods_batch(distribution/order/goods/list) - flows/shared/clients.py: fetch_goods_concurrent → fetch_goods_for_orders, 批量 200/批、串行、批间 2s,批级重试;移除逐单全局限速(340~500ms)机制 - flows/riin/_pipeline.py: 适配新批量拉取 - flows/shared/db.py: 存在性预查 unnest cast timestamp[] → timestamptz[] (入参为带时区偏移的 tz-aware datetime,timestamptz 才能稳定匹配) 背景:旧逐单接口高频被软限流(成功返回空 data),10-28 起日均缺口 13~33%; 批量接口调用量从日均 ~1 万次降到 ~50 次,已实测回补 7689 单/9717 条商品全量成功。
52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
"""RIIN 同步公共管道:映射订单 → 批量拉商品 → upsert 到 PG。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from flows.shared import fetch_goods_for_orders
|
|
from flows.shared.db import SyncDB
|
|
from flows.shared.mappers import map_riin_goods, map_riin_order
|
|
from api import RiinDisClient
|
|
|
|
|
|
def sync_riin_records(raw_orders: list[Any], client: RiinDisClient, db: SyncDB) -> dict[str, int]:
|
|
order_inputs: list[Any] = []
|
|
order_meta: dict[str, tuple[str, Any]] = {} # raw.id -> (orderNo, createdAt)
|
|
goods_failed = 0
|
|
order_map_failed = 0
|
|
|
|
for raw in raw_orders:
|
|
try:
|
|
o = map_riin_order(raw)
|
|
except Exception as err: # noqa: BLE001
|
|
order_map_failed += 1
|
|
print(f"[RIIN] 映射订单失败 {getattr(raw, 'platform_oid', '?')}: {err}")
|
|
continue
|
|
order_inputs.append(o)
|
|
if raw.id:
|
|
order_meta[raw.id] = (o.orderNo, o.createdAt)
|
|
|
|
print(f"[RIIN] 订单映射完成: 成功 {len(order_inputs)} / 失败 {order_map_failed} (共 {len(raw_orders)})")
|
|
|
|
goods_inputs: list[Any] = []
|
|
if order_meta:
|
|
goods_map = fetch_goods_for_orders(client, list(order_meta.keys()))
|
|
for oid, (order_no, created_at) in order_meta.items():
|
|
items = goods_map.get(oid, [])
|
|
if not items:
|
|
goods_failed += 1
|
|
for g in items:
|
|
try:
|
|
goods_inputs.append(map_riin_goods(g, order_no, created_at))
|
|
except Exception as err: # noqa: BLE001
|
|
print(f"[RIIN] 映射商品失败 order={order_no}: {err}")
|
|
|
|
print(f"[RIIN] 商品映射完成: {len(goods_inputs)} 条, 商品拉取失败订单 {goods_failed} 个")
|
|
|
|
stats = db.sync_orders_and_goods(order_inputs, goods_inputs)
|
|
stats["orderCount"] = len(order_inputs)
|
|
stats["goodsCount"] = len(goods_inputs)
|
|
stats["goodsFetchFailed"] = goods_failed
|
|
return stats
|