From 4a7c44a3542abf10209921f8b4bbf9194653bfb8 Mon Sep 17 00:00:00 2001 From: yeuimu <2197651308@qq.com> Date: Sat, 31 Oct 2026 10:01:50 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20RIIN=20=E5=95=86=E5=93=81=E6=8B=89?= =?UTF-8?q?=E5=8F=96=E6=94=B9=E7=94=A8=20goods/list=20=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3(200/=E6=89=B9=E4=B8=B2=E8=A1=8C)=E6=9B=BF?= =?UTF-8?q?=E6=8D=A2=E9=80=90=E5=8D=95=20order/goods?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 条商品全量成功。 --- api/riin/client.py | 8 ++++ flows/riin/_pipeline.py | 6 +-- flows/shared/__init__.py | 4 +- flows/shared/clients.py | 87 +++++++++++++++++++--------------------- flows/shared/db.py | 6 ++- 5 files changed, 59 insertions(+), 52 deletions(-) diff --git a/api/riin/client.py b/api/riin/client.py index f6527eb..1530a4f 100644 --- a/api/riin/client.py +++ b/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: diff --git a/flows/riin/_pipeline.py b/flows/riin/_pipeline.py index c470d6a..53b1220 100644 --- a/flows/riin/_pipeline.py +++ b/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: diff --git a/flows/shared/__init__.py b/flows/shared/__init__.py index 39cbc25..3abb3e6 100644 --- a/flows/shared/__init__.py +++ b/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", diff --git a/flows/shared/clients.py b/flows/shared/clients.py index ab8d673..51dd459 100644 --- a/flows/shared/clients.py +++ b/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 逐单调用对其服务器压力过大, -# 所有线程共享节拍,相邻两次调用间隔 = GOODS_MIN_INTERVAL_MS + 随机抖动(340~500ms,峰值 < 3 次/秒)。 -_goods_pace_lock = threading.Lock() -_goods_last_call_at = 0.0 +# 批量接口 distribution/order/goods/list:单批 ≤200 个 id,且要求串行调用(分销侧约定)。 +# 相比旧逐单接口(软限流:高频返回成功但空 data),调用量从日均 ~1 万次降到 ~50 次。 +GOODS_BATCH_SIZE = 200 +GOODS_BATCH_INTERVAL_MS = 2000 # 相邻两批的串行间隔 -def _pace_goods_call() -> None: - global _goods_last_call_at - 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() +def fetch_goods_for_orders(client: RiinDisClient, order_ids: list[str]) -> dict[str, list[Any]]: + """按 goods/list 批量接口串行拉取多订单商品,返回 {order_id: [goods...]}。 - -def fetch_goods_for_order(client: RiinDisClient, order_id: str) -> list[Any]: - try: - _pace_goods_call() - 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]] = {} + 单批失败(网络/业务错误)整批重试最多 MAX_RETRIES 次,仍失败则该批订单记空列表, + 由调用方按"无商品订单"计数(与旧逐单路径的容错语义一致)。 + """ + results: dict[str, list[Any]] = {oid: [] for oid in order_ids} if not order_ids: return results - concurrency = min(GOODS_CONCURRENCY, len(order_ids)) - print(f"[RIIN sync] 商品拉取开始: {len(order_ids)} 个订单 (并发={concurrency})") - - def make(oid: str) -> Callable[[], None]: - def run() -> None: - results[oid] = fetch_goods_for_order(client, oid) - return run - - with ThreadPoolExecutor(max_workers=concurrency) as pool: - list(pool.map(lambda fn: fn(), [make(oid) for oid in order_ids])) + total_batches = -(-len(order_ids) // GOODS_BATCH_SIZE) + print(f"[RIIN sync] 商品批量拉取开始: {len(order_ids)} 个订单 / {total_batches} 批" + f" (goods/list 串行, 批间 {GOODS_BATCH_INTERVAL_MS}ms)") + for bi in range(total_batches): + chunk = order_ids[bi * GOODS_BATCH_SIZE : (bi + 1) * GOODS_BATCH_SIZE] + items = _fetch_goods_batch_with_retry(client, chunk, bi + 1) + for g in items: + if g.order_id in results: + results[g.order_id].append(g) + 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] = {} diff --git a/flows/shared/db.py b/flows/shared/db.py index 76b1c6d..18ec2a8 100644 --- a/flows/shared/db.py +++ b/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]