Merge pull request 'feat: RIIN 商品拉取改用 goods/list 批量接口(200/批串行)替换逐单 order/goods' (#14) from fix/riin-goods-batch-list into main

Reviewed-on: #14

api/riin/client.py

@@ -246,6 +246,14 @@ class RiinDisClient(TokenHttpClient):
         data = self.post("distribution/order/goods", body={"orderId": order_id})
         return OrderGoodsResponse.model_validate(data)
 
+    def get_order_goods_batch(self, ids: list[str]) -> OrderGoodsResponse:
+        """批量获取订单商品(distribution/order/goods/list)。
+
+        上游要求:单批 ≤200 个 id 且串行调用;高频/并发会被软限流(成功但空 data)。
+        """
+        data = self.post("distribution/order/goods/list", body={"ids": ids})
+        return OrderGoodsResponse.model_validate(data)
+
     def get_order_count(
         self, params: OrderPageRequest | None = None
     ) -> OrderListCountResponse:

flows/riin/_pipeline.py

@@ -1,10 +1,10 @@
-"""RIIN 同步公共管道:映射订单 → 并发拉商品 → upsert 到 PG。"""
+"""RIIN 同步公共管道:映射订单 → 批量拉商品 → upsert 到 PG。"""
 
 from __future__ import annotations
 
 from typing import Any
 
-from flows.shared import fetch_goods_concurrent
+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
@@ -31,7 +31,7 @@ def sync_riin_records(raw_orders: list[Any], client: RiinDisClient, db: SyncDB)
 
     goods_inputs: list[Any] = []
     if order_meta:
-        goods_map = fetch_goods_concurrent(client, list(order_meta.keys()))
+        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:

flows/shared/__init__.py

@@ -5,7 +5,7 @@ from flows.shared.artifacts import build_stats_rows, emit_sync_summary, emit_ver
 from flows.shared.clients import (
     create_riin_client,
     create_sds_client,
-    fetch_goods_concurrent,
+    fetch_goods_for_orders,
     fetch_riin_all_pages,
     fetch_sds_all_pages,
     floor_to_hour,
@@ -23,7 +23,7 @@ __all__ = [
     "clients", "db", "mappers", "SyncDB",
     "OrderUpsertInput", "GoodsUpsertInput",
     "create_riin_client", "create_sds_client",
-    "fetch_riin_all_pages", "fetch_sds_all_pages", "fetch_goods_concurrent",
+    "fetch_riin_all_pages", "fetch_sds_all_pages", "fetch_goods_for_orders",
     "get_channel_type_map", "refresh_channel_type_map", "resolve_courier_source",
     "format_date", "start_of_day", "floor_to_hour",
     "build_stats_rows", "emit_sync_summary", "emit_verify_report",

flows/shared/clients.py

@@ -8,8 +8,6 @@
 from __future__ import annotations
 
 import json
-import random
-import threading
 import time
 from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, TypeVar
@@ -23,9 +21,6 @@ from flows.shared.mappers import map_sds_channel_type
 
 T = TypeVar("T")
 
-GOODS_CONCURRENCY = 5
-GOODS_MIN_INTERVAL_MS = 340  # 相邻两次 order/goods 调用全局最小间隔(峰值 < 3 次/秒)
-GOODS_MAX_JITTER_MS = 160  # 在最小间隔上叠加的随机抖动上限,实际间隔 340~500ms 随机
 PAGE_DELAY_MS = 500
 MAX_RETRIES = 3
 RETRY_DELAY_MS = 3000
@@ -148,56 +143,58 @@ def fetch_sds_all_pages(client: SdsMngClient, params: dict[str, Any]) -> list[An
 
 # ---------- RIIN 商品 ----------
 
-# 全局限速:RIIN 分销侧反馈 distribution/order/goods 逐单调用对其服务器压力过大,
+# 批量接口 distribution/order/goods/list:单批 ≤200 个 id,且要求串行调用(分销侧约定)。
-# 所有线程共享节拍,相邻两次调用间隔 = GOODS_MIN_INTERVAL_MS + 随机抖动(340~500ms,峰值 < 3 次/秒)。
+# 相比旧逐单接口(软限流:高频返回成功但空 data),调用量从日均 ~1 万次降到 ~50 次。
-_goods_pace_lock = threading.Lock()
+GOODS_BATCH_SIZE = 200
-_goods_last_call_at = 0.0
+GOODS_BATCH_INTERVAL_MS = 2000  # 相邻两批的串行间隔
 
 
-def _pace_goods_call() -> None:
+def fetch_goods_for_orders(client: RiinDisClient, order_ids: list[str]) -> dict[str, list[Any]]:
-    global _goods_last_call_at
+    """按 goods/list 批量接口串行拉取多订单商品,返回 {order_id: [goods...]}。
-    with _goods_pace_lock:
-        interval = (GOODS_MIN_INTERVAL_MS + random.uniform(0, GOODS_MAX_JITTER_MS)) / 1000
-        wait = _goods_last_call_at + interval - time.monotonic()
-        if wait > 0:
-            time.sleep(wait)
-        _goods_last_call_at = time.monotonic()
 
-
+    单批失败(网络/业务错误)整批重试最多 MAX_RETRIES 次,仍失败则该批订单记空列表,
-def fetch_goods_for_order(client: RiinDisClient, order_id: str) -> list[Any]:
+    由调用方按"无商品订单"计数(与旧逐单路径的容错语义一致)。
-    try:
+    """
-        _pace_goods_call()
+    results: dict[str, list[Any]] = {oid: [] for oid in order_ids}
-        resp = client.get_order_goods(order_id)
-        return list(resp.data)
-    except Exception as err:  # noqa: BLE001
-        print(f"[RIIN sync] 订单 {order_id} 商品拉取失败: {err}")
-        return []
-
-
-def fetch_goods_concurrent(client: RiinDisClient, order_ids: list[str]) -> dict[str, list[Any]]:
-    """以 GOODS_CONCURRENCY 并发拉取多订单商品,返回 {order_id: [goods...]}。"""
-    from collections.abc import Callable
-    from concurrent.futures import ThreadPoolExecutor
-
-    results: dict[str, list[Any]] = {}
     if not order_ids:
         return results
-    concurrency = min(GOODS_CONCURRENCY, len(order_ids))
+    total_batches = -(-len(order_ids) // GOODS_BATCH_SIZE)
-    print(f"[RIIN sync] 商品拉取开始: {len(order_ids)} 个订单 (并发={concurrency})")
+    print(f"[RIIN sync] 商品批量拉取开始: {len(order_ids)} 个订单 / {total_batches} 批"
-
+          f" (goods/list 串行, 批间 {GOODS_BATCH_INTERVAL_MS}ms)")
-    def make(oid: str) -> Callable[[], None]:
+    for bi in range(total_batches):
-        def run() -> None:
+        chunk = order_ids[bi * GOODS_BATCH_SIZE : (bi + 1) * GOODS_BATCH_SIZE]
-            results[oid] = fetch_goods_for_order(client, oid)
+        items = _fetch_goods_batch_with_retry(client, chunk, bi + 1)
-        return run
+        for g in items:
-
+            if g.order_id in results:
-    with ThreadPoolExecutor(max_workers=concurrency) as pool:
+                results[g.order_id].append(g)
-        list(pool.map(lambda fn: fn(), [make(oid) for oid in order_ids]))
+        if bi < total_batches - 1:
+            time.sleep(GOODS_BATCH_INTERVAL_MS / 1000)
     total_goods = sum(len(v) for v in results.values())
     empty_orders = sum(1 for v in results.values() if not v)
-    print(f"[RIIN sync] 商品拉取完成: 共 {total_goods} 条商品, {empty_orders} 个订单无商品")
+    print(f"[RIIN sync] 商品批量拉取完成: 共 {total_goods} 条商品, {empty_orders} 个订单无商品")
     return results
 
 
+def _fetch_goods_batch_with_retry(client: RiinDisClient, chunk: list[str], batch_no: int) -> list[Any]:
+    last_err: Exception | None = None
+    for attempt in range(1, MAX_RETRIES + 1):
+        try:
+            resp = fetch_with_retry(
+                lambda: client.get_order_goods_batch(chunk), f"RIIN goods 批次 {batch_no}"
+            )
+            if resp.successful is False:
+                raise RuntimeError(f"goods/list 业务失败: {resp.message}")
+            return list(resp.data)
+        except Exception as err:  # noqa: BLE001
+            last_err = err
+            print(f"[RIIN sync] 商品批次 {batch_no} 第 {attempt}/{MAX_RETRIES} 次失败: {str(err)[:120]}")
+            if attempt < MAX_RETRIES:
+                time.sleep(RETRY_DELAY_MS * attempt / 1000)
+    assert last_err is not None
+    print(f"[RIIN sync] 商品批次 {batch_no} 放弃, {len(chunk)} 个订单本轮无商品")
+    return []
+
+
 # ---------- SDS 物流渠道字典 ----------
 
 _channel_cache: dict[str, Any] = {}

flows/shared/db.py

@@ -460,9 +460,11 @@ class SyncDB:
         stats = {"ordersCreated": 0, "ordersUpdated": 0, "goodsCreated": 0, "goodsUpdated": 0, "goodsRemoved": 0}
         with conn.cursor() as cur:
             # 统计已存在 → created/updated
+            # 注:必须 cast 成 timestamptz——入参是带时区偏移的 tz-aware datetime,
+            # cast 成 timestamp 会丢弃偏移,在会话时区非 +08 时匹配不到既有行,导致 created/updated 统计全错。
             pairs = [(o.orderNo, o.createdAt) for o in orders]
             cur.execute(
-                "SELECT t.order_no, t.created_at FROM unnest(%s::text[], %s::timestamp[]) AS t(order_no,created_at) "
+                "SELECT t.order_no, t.created_at FROM unnest(%s::text[], %s::timestamptz[]) AS t(order_no,created_at) "
                 "JOIN \"Orders\" o ON o.\"orderNo\"=t.order_no AND o.\"createdAt\"=t.created_at",
                 _unnest_pairs(pairs),
             )
@@ -623,6 +625,6 @@ class SyncDB:
 def _unnest_pairs(pairs: list[tuple]) -> tuple[list, list]:
     """构造 psycopg2 unnest 双数组参数:返回 (col_a_list, col_b_list)。
 
-    配合 SQL:unnest(%s::text[], %s::timestamp[]) 或 unnest(%s::text[], %s::text[])。
+    配合 SQL:unnest(%s::text[], %s::timestamptz[]) 或 unnest(%s::text[], %s::text[])。
     """
     return [p[0] for p in pairs], [p[1] for p in pairs]