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

Reviewed-on: #14

.gitea/workflows/ci.yml

@@ -12,7 +12,10 @@ jobs:
       - name: 检出代码
         run: |
           REF="${{ github.event.pull_request.head.ref || github.ref_name }}"
-          git clone --depth 1 --branch "$REF" "${{ github.server_url }}/${{ github.repository }}.git" .
+          git clone --depth 1 "${{ github.server_url }}/${{ github.repository }}.git" .
+          # PR head 可能是完整 ref(AGit 建的 PR 为 refs/pull/N/head),clone --branch 只认分支名,改 fetch 后 checkout
+          git fetch --depth 1 origin "$REF"
+          git checkout --detach FETCH_HEAD
       - name: 安装 uv
         run: curl -LsSf https://astral.sh/uv/install.sh | sh
       - name: 安装依赖

.python-version

@@ -0,0 +1 @@
+3.12

api/ocr/models.py

@@ -37,14 +37,16 @@ class OcrResult(BaseModel):
 
 
 # OCR 服务已有分类器模型的国家(版面见 api.md §4)
-SUPPORTED_COUNTRIES = {"US", "UK", "DE", "MX", "SA"}
+SUPPORTED_COUNTRIES = {"US", "UK", "DE", "MX", "SA", "CA", "ES", "IT", "PL"}
 
 # ISO 国家代码 / 中文国名 → OCR 国家代码
 _COUNTRY_MAP: dict[str, str] = {
     # 代码
     "GB": "UK", "UK": "UK", "US": "US", "DE": "DE", "MX": "MX", "SA": "SA",
+    "CA": "CA", "ES": "ES", "IT": "IT", "PL": "PL",
     # 中文名
     "英国": "UK", "美国": "US", "德国": "DE", "墨西哥": "MX", "沙特": "SA",
+    "加拿大": "CA", "西班牙": "ES", "意大利": "IT", "波兰": "PL",
 }
 
 

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:

docker-compose.yml

@@ -15,6 +15,9 @@ services:
   prefect-worker:
     build: .
     container_name: prefect-worker
+    # init 进程(tini)回收孤儿/僵尸:pull 步骤的 git 子进程被杀后没人 wait 会变僵尸,
+    # 堆满 cgroup PID 配额后容器内无法再 fork(2026-09-17 事故根因)
+    init: true
     environment:
       PREFECT_API_URL: http://prefect-server:4200/api
       TZ: Asia/Shanghai

docs/sql/2026-09-28-goods-status-allow-pause.sql

@@ -0,0 +1,13 @@
+-- 2026-09-28 Goods.status 约束新增允许 '暂停'
+-- 背景:RIIN 商品级 goodsStatus 复用平台状态枚举,存在 PAUSE(暂停),
+-- 原约束未包含该值,导致映射层只能落 NULL(chk_goods_status 拒绝其余新值)。
+-- 配套:mappers.py 的 RIIN_GOODS_STATUS_MAP 补 COMPLETE/PAUSE。
+-- 执行方式(两步,避免阻塞 ETL 写入;VALIDATE 只锁 SHARE UPDATE EXCLUSIVE):
+ALTER TABLE "Goods" DROP CONSTRAINT chk_goods_status;
+ALTER TABLE "Goods" ADD CONSTRAINT chk_goods_status CHECK (
+    status IS NULL OR status = ANY (ARRAY[
+        '备货中', '已删除', '已取消', '已完成', '待付款', '待确认',
+        '搁置中', '待发货', '已发货', '退款中', '暂停'
+    ])
+) NOT VALID;
+ALTER TABLE "Goods" VALIDATE CONSTRAINT chk_goods_status;

flows/label_ocr.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import os
+import time
 
 import httpx
 from prefect import flow, task
@@ -16,6 +17,8 @@ from api.ocr import OcrClient, country_to_ocr_code
 from flows.shared import SyncDB, build_stats_rows, emit_sync_summary
 
 BATCH_SIZE = 200
+BATCH_RETRIES = 3                                  # 批量请求级重试(OCR 服务偶发秒级重启)
+BATCH_RETRY_BACKOFF = [5, 15]                      # 重试间隔(秒)
 
 _OCR_FIELD_MAP = [
     ("layout", "ocrLayout"), ("courier", "ocrCourier"), ("tracking", "ocrTracking"),
@@ -26,13 +29,20 @@ _OCR_FIELD_MAP = [
 ]
 
 
+# 单号不需要 "-" 的承运商(Evri 单号两段式带 "-",下游查询用无 "-" 形式)
+_DASH_STRIP_COURIERS = {"EVRI"}
+
+
 def _result_to_fields(r) -> dict[str, str | None]:
     if not r.ok:
         err = (r.error or "").lower()
         if "reject" in err or "not a recognized layout" in err:
             return {"ocrLayout": "UNSUPPORTED"}
         return {"ocrLayout": "ERROR"}
-    return {col: (getattr(r, attr) or "") for attr, col in _OCR_FIELD_MAP}
+    fields = {col: (getattr(r, attr) or "") for attr, col in _OCR_FIELD_MAP}
+    if (r.courier or "").strip().upper() in _DASH_STRIP_COURIERS:
+        fields["ocrTracking"] = fields["ocrTracking"].replace("-", "")
+    return fields
 
 
 def _has_running_peer() -> bool:
@@ -63,7 +73,7 @@ def _has_running_peer() -> bool:
 def process_all(pending: list) -> dict[str, int]:
     db = SyncDB()
     ocr = OcrClient()
-    stats = {"success": 0, "error": 0, "unsupported": 0, "batches": 0}
+    stats = {"success": 0, "error": 0, "unsupported": 0, "batches": 0, "deferred": 0}
     try:
         # 1. 按国家分组;不支持的国家直接标 UNSUPPORTED
         by_country: dict[str, list[tuple[str, object, str]]] = {}
@@ -89,12 +99,24 @@ def process_all(pending: list) -> dict[str, int]:
                 urls = [{"url": url, "filename": no} for no, _, url in batch]
                 batch_no = i // BATCH_SIZE + 1
                 total_batches = -(-total // BATCH_SIZE)
-                try:
-                    results = ocr.recognize_batch(urls, country_code)
-                except Exception as err:  # noqa: BLE001
-                    print(f"[OCR] {country_code} batch {batch_no}/{total_batches} 请求失败: {str(err)[:100]}")
-                    db.update_ocr_batch([(no, created_map[no], {"ocrLayout": "ERROR"}) for no in created_map])
-                    stats["error"] += len(batch)
+                results = None
+                for attempt in range(BATCH_RETRIES):
+                    try:
+                        results = ocr.recognize_batch(urls, country_code)
+                        break
+                    except Exception as err:  # noqa: BLE001
+                        if attempt < BATCH_RETRIES - 1:
+                            wait = BATCH_RETRY_BACKOFF[min(attempt, len(BATCH_RETRY_BACKOFF) - 1)]
+                            print(f"[OCR] {country_code} batch {batch_no}/{total_batches} 请求失败"
+                                  f"({str(err)[:80]}),{wait}s 后重试({attempt + 2}/{BATCH_RETRIES})")
+                            time.sleep(wait)
+                        else:
+                            # OCR 服务端短暂不可用(容器秒级重启类故障):订单保持未处理,
+                            # 留给下个小时的 run 自动重试,不落 ERROR(那会是永久终态)。
+                            print(f"[OCR] {country_code} batch {batch_no}/{total_batches} "
+                                  f"重试 {BATCH_RETRIES} 次仍失败,留待下次 run: {str(err)[:100]}")
+                if results is None:
+                    stats["deferred"] += len(batch)
                     continue
 
                 updates = []
@@ -122,12 +144,12 @@ def label_ocr(
     courier_source: str | None = None,
     days: int = 7,
 ) -> dict:
-    """面单 OCR enrichment。默认处理英国+美国,近 7 天下单的订单。
+    """面单 OCR enrichment。默认处理全部已支持国家(见 api.ocr.models.SUPPORTED_COUNTRIES),近 7 天下单的订单。
 
-    手动跑其他国家: label_ocr(country_regions=['德国'])
+    手动跑指定国家: label_ocr(country_regions=['德国'])
     """
     if country_regions is None:
-        country_regions = ['英国', '美国', '墨西哥', '德国', '沙特']
+        country_regions = ['英国', '美国', '墨西哥', '德国', '沙特', '加拿大', '西班牙', '意大利', '波兰']
     if _has_running_peer():
         print("已有 label-ocr 在运行,跳过本次")
         return {"skipped": True}
@@ -152,11 +174,13 @@ def label_ocr(
     after = db.count_ocr_status()
     total_processed = stats["success"] + stats["error"] + stats["unsupported"]
     print(f"完成{scope}: 处理 {total_processed} 单 / {stats['batches']} 批 | "
-          f"成功 {stats['success']} 失败 {stats['error']} 不支持 {stats['unsupported']}")
+          f"成功 {stats['success']} 失败 {stats['error']} 不支持 {stats['unsupported']} "
+          f"遗留待重试 {stats['deferred']}")
     print(f"累计: {after}")
     emit_sync_summary(f"面单 OCR{scope}", build_stats_rows("ALL", {}, [
         ("本批处理", total_processed), ("成功", stats["success"]),
         ("失败", stats["error"]), ("不支持", stats["unsupported"]),
+        ("遗留待重试", stats["deferred"]),
         ("累计已完成", after.get("done", 0)), ("累计待处理", after.get("pending", 0)),
     ]))
     return {"processed": total_processed, **stats, **after}

flows/platform_order/platform_order_upper_sync.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8 -*-
+"""平台订单 platform 字段大写同步:把 platform_order_data.platform 统一转为大写。
+
+优化自 platform_order_upper_sync.py,对应 platform-order-upper-sync。
+"""
+
+import time
+
+import psycopg2
+from prefect import flow
+
+# 数据库连接参数
+conn_params = {
+    "host": "192.168.124.92",
+    "port": 5432,
+    "user": "postgres",
+    "password": "Inkreach123456.",
+    "database": "smt",
+}
+
+
+def log_msg(msg, level="info"):
+    ts = time.strftime('%H:%M:%S')
+    tag = {"success": "[OK]", "error": "[ERR]"}.get(level, "")
+    print(f"[{ts}] {tag} {msg}")
+
+
+def update_platform_upper() -> int:
+    conn = None
+    cur = None
+    try:
+        conn = psycopg2.connect(**conn_params)
+        cur = conn.cursor()
+
+        # 仅更新需要变更的行,避免每次都全表写入
+        update_sql = """
+            UPDATE platform_order_data
+            SET platform = UPPER(platform)
+            WHERE platform IS NOT NULL
+              AND platform <> UPPER(platform);
+        """
+        cur.execute(update_sql)
+        affected_rows = cur.rowcount
+        conn.commit()
+        log_msg(f"platform 大写转换完成，共 {affected_rows} 行受影响", "success")
+        return affected_rows
+    except Exception as e:
+        log_msg(f"platform 大写转换失败：{e}", "error")
+        if conn:
+            conn.rollback()
+        raise
+    finally:
+        if cur:
+            cur.close()
+        if conn:
+            conn.close()
+
+
+@flow(name="platform-order-upper-sync", log_prints=True)
+def platform_order_upper_sync() -> int:
+    return update_platform_upper()
+
+
+if __name__ == "__main__":
+    platform_order_upper_sync.serve(name="platform-order-upper-sync", cron="*/2 * * * *")

flows/platform_order/riin.py

@@ -22,7 +22,7 @@ RIIN_LOGIN_URL = "http://tshirt.riin.com/auth/api/auth/login"
 RIIN_ORDER_URL = "https://tshirt.riin.com/distribution/order/page"
 USERNAME_RIIN = "预上网"
 PASSWORD_RIIN = "Aa123456."
-SUB_ID_LIST = ["10368", "10283", "16905", "11502", "13615", "11531"]
+SUB_ID_LIST = ["10368", "10283", "16905", "11502", "13615", "11531","12279"]
 MAX_RETRY = 3; REQUEST_TIMEOUT = 30; PAGE_SIZE_RIIN = 200
 
 LOOKBACK_DAYS = 6  # 回查14天避免漏单
@@ -208,4 +208,5 @@ async def riin_platform_sync():
     log_msg("===== 完成 =====")
 
 if __name__ == "__main__":
-    riin_platform_sync.serve(name="platform-order-riin", cron="15 * * * *")
+    # 直跑一次用于本地调试;勿改回 serve():同名 upsert 会覆盖 server 上挂 work pool 的正式 deployment
+    riin_platform_sync()

flows/platform_order/rouhui.py

@@ -273,4 +273,5 @@ def rouhui_platform_sync():
     log_msg("===== 提取完成 =====")
 
 if __name__ == "__main__":
-    rouhui_platform_sync.serve(name="platform-order-rouhui", cron="*/15 * * * *")
+    # 直跑一次用于本地调试;勿改回 serve():同名 upsert 会覆盖 server 上挂 work pool 的正式 deployment
+    rouhui_platform_sync()

flows/platform_order/sdspod.py

@@ -359,10 +359,18 @@ def sd_format_item(item):
     create_time = item.get("gmtCreated", "")
     parcel_list = item.get("parcelList", [])
     if parcel_list and isinstance(parcel_list, list):
-        qty = sum(p.get("totalQty", 0) for p in parcel_list)
+        def safe_int(val):
+            try:
+                return int(val)
+            except (ValueError, TypeError):
+                return 0
+
+        qty = sum(safe_int(p.get("totalQty", 0)) for p in parcel_list)
     else:
         qty = 0
 
+
+
     return [
         to_chinese_country(country),
         to_chinese_country(country),
@@ -471,4 +479,5 @@ async def sdspod_platform_sync():
 
 
 if __name__ == "__main__":
-    sdspod_platform_sync.serve(name="platform-order-sdspod",  cron="*/15 * * * *")
+    # 直跑一次用于本地调试;勿改回 serve():同名 upsert 会覆盖 server 上挂 work pool 的正式 deployment
+    sdspod_platform_sync()

flows/platform_order/sync_shop_dim.py

@@ -0,0 +1,144 @@
+# -*- coding: utf-8 -*-
+"""shop_dim 同步 flow：teable(源) -> 本地 smt 库(目标)。
+
+最小包装自 sync_shop_dim.py,对应 shop-dim-sync,每 5 分钟调度一次。
+源: PostgreSQL @ table.inkreach.cc:10432  db=teable  schema=bsehWCxpH6HoHUdkjUd
+    物理表 tblUF7OrBxeRKUyWBc2 (teable 里名为 "shop_dim" 的表)
+目标: PostgreSQL @ 192.168.124.92:5432  db=smt  public.shop_dim
+
+本地直接跑:  python flows/platform_order/sync_shop_dim.py
+"""
+
+import time
+from typing import Any
+
+import psycopg2
+from prefect import flow
+
+SRC = dict(host="table.inkreach.cc", port=10432, dbname="teable",
+           user="read_only_role_bsehWCxpH6HoHUdkjUd",
+           password="ygBuh8Au5J-PjuNQASos0", connect_timeout=25)
+SRC_TABLE = '"bsehWCxpH6HoHUdkjUd"."tblUF7OrBxeRKUyWBc2"'
+
+DST = dict(host="192.168.124.92", port=5432, dbname="smt",
+           user="postgres", password="Inkreach123456.", connect_timeout=15)
+
+COL_MAP = [
+    ("Guo_Jia",            "country"),
+    ("Dian_Pu_Ming_Cheng", "shop_name"),
+    ("DBDian_Pu_Ming",     "db_store_name"),
+    ("Ping_Tai",           "platform"),
+    ("Lei_Xing",           "shop_type"),
+    ("Deng_Ji",            "tier"),
+    ("Fu_Ze_Ren",          "owner"),
+    ("Pin_Lei",            "category"),
+    ("Xia_Dian_Shi_Jian",  "opened_at"),
+    ("Mu_Biao_Jian_Shu",   "sep_target_qty"),
+]
+
+
+def log_msg(msg):
+    print(msg)
+
+
+def connect(cfg, tries=5):
+    last = None
+    for i in range(tries):
+        try:
+            return psycopg2.connect(**cfg)
+        except Exception as e:
+            last = e
+            log_msg(f"  连接重试 {i+1}/{tries}: {e}")
+            time.sleep(3)
+    raise ConnectionError(f"连接失败: {last}")
+
+
+def norm(v, col):
+    if v is None:
+        return None
+    if col == "opened_at":
+        return v.date() if hasattr(v, "date") else v
+    if col == "sep_target_qty":
+        try:
+            return int(round(float(v)))
+        except Exception:
+            return None
+    s = str(v).strip()
+    return s if s != "" else None
+
+
+@flow(name="shop-dim-sync", log_prints=True)
+def shop_dim_sync(dry_run: bool = False, prune: bool = True) -> dict[str, Any]:
+    log_msg(f"[1/4] 读取源表 {SRC_TABLE} ...")
+    sc = connect(SRC)
+    scur = sc.cursor()
+    sel = ", ".join('"%s"' % c for c, _ in COL_MAP)
+    scur.execute("SELECT %s FROM %s" % (sel, SRC_TABLE))
+    raw = scur.fetchall()
+    scur.close(); sc.close()
+    src_rows = [
+        dict(zip([d for _, d in COL_MAP], [norm(v, c) for v, (_, c) in zip(row, COL_MAP)]))
+        for row in raw
+    ]
+    log_msg(f"      源记录: {len(src_rows)} 行")
+
+    log_msg("[2/4] 连接目标 smt.shop_dim ...")
+    dc = connect(DST)
+    dcur = dc.cursor()
+    dcur.execute("SELECT country, db_store_name FROM shop_dim WHERE db_store_name IS NOT NULL")
+    existing = {(r[0], r[1]) for r in dcur.fetchall()}
+    log_msg(f"      目标现有: {len(existing)} 个 (country, db_store_name)")
+
+    src_keys = {(r["country"], r["db_store_name"]) for r in src_rows}
+    to_update = src_keys & existing
+    to_insert = src_keys - existing
+    to_delete = existing - src_keys
+    log_msg(f"[3/4] 计划: 更新 {len(to_update)} | 新增 {len(to_insert)} | 待删 {len(to_delete)}")
+
+    if dry_run:
+        log_msg("[dry-run] 不执行写入。")
+        if to_delete:
+            log_msg("        待删示例: " + str(list(to_delete)[:10]))
+        dcur.close(); dc.close()
+        return {"sourceRows": len(src_rows), "dryRun": True}
+
+    cols = [d for _, d in COL_MAP]
+    upd = 0
+    ins = 0
+    for r in src_rows:
+        vals = [r[c] for c in cols]
+        if (r["country"], r["db_store_name"]) in existing:
+            set_clause = ", ".join(f"{c}=%s" for c in cols)
+            dcur.execute(
+                f"UPDATE shop_dim SET {set_clause} WHERE country=%s AND db_store_name=%s",
+                vals + [r["country"], r["db_store_name"]])
+            upd += dcur.rowcount
+        else:
+            ph = ", ".join(["%s"] * len(cols))
+            dcur.execute(f"INSERT INTO shop_dim ({', '.join(cols)}) VALUES ({ph})", vals)
+            ins += dcur.rowcount
+
+    dele = 0
+    if prune and to_delete:
+        for country, dbn in to_delete:
+            dcur.execute("DELETE FROM shop_dim WHERE country=%s AND db_store_name=%s", (country, dbn))
+            dele += dcur.rowcount
+
+    dc.commit()
+    log_msg(f"[4/4] 完成: 更新 {upd} | 新增 {ins} | 删除 {dele}")
+    dcur.execute("SELECT COUNT(*) FROM shop_dim WHERE db_store_name IS NOT NULL")
+    total = dcur.fetchone()[0]
+    log_msg(f"      目标现有总数: {total}")
+    dcur.close(); dc.close()
+
+    return {
+        "sourceRows": len(src_rows),
+        "updated": upd,
+        "inserted": ins,
+        "deleted": dele,
+        "total": total,
+    }
+
+
+if __name__ == "__main__":
+    shop_dim_sync.serve(name="shop-dim-sync", cron="*/5 * * * *")

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

@@ -21,7 +21,6 @@ from flows.shared.mappers import map_sds_channel_type
 
 T = TypeVar("T")
 
-GOODS_CONCURRENCY = 20
 PAGE_DELAY_MS = 500
 MAX_RETRIES = 3
 RETRY_DELAY_MS = 3000
@@ -144,39 +143,58 @@ def fetch_sds_all_pages(client: SdsMngClient, params: dict[str, Any]) -> list[An
 
 # ---------- RIIN 商品 ----------
 
-def fetch_goods_for_order(client: RiinDisClient, order_id: str) -> list[Any]:
-    try:
-        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 []
+# 批量接口 distribution/order/goods/list:单批 ≤200 个 id,且要求串行调用(分销侧约定)。
+# 相比旧逐单接口(软限流:高频返回成功但空 data),调用量从日均 ~1 万次降到 ~50 次。
+GOODS_BATCH_SIZE = 200
+GOODS_BATCH_INTERVAL_MS = 2000  # 相邻两批的串行间隔
 
 
-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
+def fetch_goods_for_orders(client: RiinDisClient, order_ids: list[str]) -> dict[str, list[Any]]:
+    """按 goods/list 批量接口串行拉取多订单商品,返回 {order_id: [goods...]}。
 
-    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] = {}

flows/shared/db.py

@@ -25,8 +25,9 @@ LOOKUP_BATCH = 10000
 # 单锁无环,写事务之间从结构上不可能死锁。1768851058 = "inkr" 的 ASCII 整数表示。
 DB_WRITE_LOCK_KEY = 1768851058
 LOCK_TIMEOUT = "15s"                          # 等锁超时 → LockNotAvailable → 交给重试兜底
-WRITE_RETRIES = 3                             # 死锁/等锁超时的整事务重试次数
-WRITE_RETRY_BACKOFF = [1, 3, 7]               # 重试退避(秒)
+WRITE_RETRIES = 5                             # 死锁/等锁超时的整事务重试次数(外部写者如 17track
+                                              # 物流回写不走 advisory 锁,冲突窗口可达分钟级,重试需够深)
+WRITE_RETRY_BACKOFF = [1, 3, 7, 15, 30]       # 重试退避(秒;末位预留,最后一次失败直接抛出)
 WRITE_RETRYABLE_ERRORS = (pg_errors.DeadlockDetected, pg_errors.LockNotAvailable)
 
 
@@ -459,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),
             )
@@ -622,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]

flows/shared/mappers.py

@@ -33,6 +33,10 @@ RIIN_EXPRESS_SOURCE_MAP: dict[int, str] = {
 RIIN_GOODS_STATUS_MAP: dict[str, str] = {
     "NOT_SHIPPED": "待发货", "SHIPPED": "已发货", "CANCEL": "已取消",
     "CANCELLED": "已取消", "REFUNDING": "退款中",
+    # 商品级状态复用平台状态枚举,COMPLETE/PAUSE 缺失曾导致 1/3 商品行 status 为 NULL
+    "COMPLETE": "已完成", "PAUSE": "暂停",
+    # CLOSE 同为商品终态(Goods 约束无"已关闭",取"已取消");UNKNOWN 保留 NULL 并由告警暴露
+    "CLOSE": "已取消",
 }
 
 RIIN_PLATFORM_STATUS_MAP: dict[str, str] = {
@@ -55,6 +59,17 @@ SDS_CHANNEL_TYPE_LABEL_MAP: dict[str, str] = {
 
 # ---------- 日期解析 ----------
 
+# 未知枚举值告警:同值只报一次,避免批量同步刷屏(对齐 TS 版 warnUnknown)
+_warned_values: set[str] = set()
+
+
+def _warn_unknown(map_name: str, value: object) -> None:
+    key = f"{map_name}:{value!r}"
+    if key not in _warned_values:
+        _warned_values.add(key)
+        print(f"[mapper] unknown {map_name} value: {value!r}")
+
+
 def safe_date(value: Any, context: str = "") -> datetime:
     """解析日期字符串(无时区视为 +08:00),返回 tz-aware datetime。"""
     if not value or not isinstance(value, str):
@@ -188,15 +203,20 @@ class GoodsUpsertInput:
 
 # ---------- RIIN 映射 ----------
 
-def map_riin_system_status(order_status: int) -> str | None:
-    return RIIN_SYSTEM_STATUS_MAP.get(order_status)
+def map_riin_system_status(order_status: int | None) -> str | None:
+    if order_status is None:
+        return None
+    v = RIIN_SYSTEM_STATUS_MAP.get(order_status)
+    if v is None:
+        _warn_unknown("RIIN systemStatus", order_status)
+    return v
 
 
 def map_riin_order(raw: Any) -> OrderUpsertInput:
     platform_oid = raw.platform_oid
     ordered_at = _parse(raw.push_time, f"RIIN order {platform_oid} pushTime")
     confirmed_at = _parse(raw.distributor_push_time, f"RIIN order {platform_oid} distributorPushTime")
-    system_status = map_riin_system_status(raw.order_status) if raw.order_status is not None else None
+    system_status = map_riin_system_status(raw.order_status)
     is_terminal = system_status in ("已发货", "已关闭", "已取消")
     completed_at: datetime | None = None
     if is_terminal:
@@ -236,6 +256,15 @@ def map_riin_order(raw: Any) -> OrderUpsertInput:
     )
 
 
+def map_riin_goods_status(goods_status: str | None) -> str | None:
+    if not goods_status:
+        return None
+    v = RIIN_GOODS_STATUS_MAP.get(goods_status)
+    if v is None:
+        _warn_unknown("RIIN goodsStatus", goods_status)
+    return v
+
+
 def map_riin_goods(raw: Any, parent_order_no: str, parent_created_at: datetime) -> GoodsUpsertInput:
     product_code = raw.product_code
     item_name = product_code.split("-")[0] if product_code else None
@@ -250,7 +279,7 @@ def map_riin_goods(raw: Any, parent_order_no: str, parent_created_at: datetime)
         itemSize=raw.size_code,
         itemStyleNo=style_no,
         quantity=raw.num if raw.num is not None else 1,
-        status=RIIN_GOODS_STATUS_MAP.get(raw.goods_status) if raw.goods_status else None,
+        status=map_riin_goods_status(raw.goods_status),
     )
 
 
@@ -262,13 +291,27 @@ def map_sds_channel_type(channel_type: Any) -> str | None:
     return SDS_CHANNEL_TYPE_LABEL_MAP.get(channel_type)
 
 
-def map_sds_system_status(status: int) -> str | None:
-    return SDS_SYSTEM_STATUS_MAP.get(status)
+def map_sds_system_status(status: int | None) -> str | None:
+    if status is None:
+        return None
+    v = SDS_SYSTEM_STATUS_MAP.get(status)
+    if v is None:
+        _warn_unknown("SDS systemStatus", status)
+    return v
+
+
+def map_sds_goods_status(status: int | None) -> str | None:
+    if status is None:
+        return None
+    v = SDS_GOODS_STATUS_MAP.get(status)
+    if v is None:
+        _warn_unknown("SDS goodsStatus", status)
+    return v
 
 
 def map_sds_order(raw: Any, channel_type_map: dict[int, str | None], pre_courier: str | None = None) -> OrderUpsertInput:
     ordered_at = _parse(raw.gmt_pay, f"SDS order {raw.no} gmtPay")
-    system_status = map_sds_system_status(raw.status) if raw.status is not None else None
+    system_status = map_sds_system_status(raw.status)
     is_terminal = system_status in ("已完成", "已取消")
     completed_at: datetime | None = None
     if is_terminal and raw.finish_time and raw.finish_time > 0:
@@ -369,7 +412,7 @@ def map_sds_goods(raw: Any, parent_order_no: str, parent_created_at: datetime) -
         itemSize=raw.product_size,
         itemStyleNo=_parse_sds_style_no(product_name),
         quantity=raw.num if raw.num is not None else 1,
-        status=SDS_GOODS_STATUS_MAP.get(raw.status) if raw.status is not None else None,
+        status=map_sds_goods_status(raw.status),
     )
 
 

prefect.yaml

@@ -21,6 +21,7 @@ deployments:
   schedules:
   - cron: 30 * * * *
     timezone: Asia/Shanghai
+    active: false  # 暂停:每小时重拉近7天全部订单商品,对 RIIN distribution/order/goods 调用量过大(增量/刷新未完结/日报核验已覆盖)
   work_pool:
     name: inkreach-pool
 - name: riin-refresh-pending
@@ -39,6 +40,7 @@ deployments:
   schedules:
   - cron: 0 6 * * *
     timezone: Asia/Shanghai
+    active: false  # 暂停:与 daily-report 18:30 的核验子流程重复,保留日报一处即可
   work_pool:
     name: inkreach-pool
 - name: sds-order-sync
@@ -97,7 +99,21 @@ deployments:
 - name: platform-order-sdspod
   entrypoint: flows/platform_order/sdspod.py:sdspod_platform_sync
   schedules:
-  - cron: '*/30 * * * *'
+  - cron: 0 * * * *
+    timezone: Asia/Shanghai
+  work_pool:
+    name: inkreach-pool
+- name: platform-order-upper-sync
+  entrypoint: flows/platform_order/platform_order_upper_sync.py:platform_order_upper_sync
+  schedules:
+  - cron: '*/2 * * * *'
+    timezone: Asia/Shanghai
+  work_pool:
+    name: inkreach-pool
+- name: shop-dim-sync
+  entrypoint: flows/platform_order/sync_shop_dim.py:shop_dim_sync
+  schedules:
+  - cron: '*/5 * * * *'
     timezone: Asia/Shanghai
   work_pool:
     name: inkreach-pool

scripts/backfill_riin_goods_status.py

@@ -0,0 +1,72 @@
+"""一次性回填:RIIN Goods.status 为 NULL 的历史行(终态"已发货"订单)。
+
+背景:RIIN 商品级 goods_status 复用平台状态枚举,COMPLETE/PAUSE 此前无映射,
+约 1/3 商品行 status 被写为 NULL(mappers.py 已补映射)。本脚本按订单终态
+将近似值回填;"已关闭"与非终态订单不在此列,走 riin-by-orders 重拉取真值。
+幂等:只更新 status IS NULL 的行。分批提交,避免长事务。
+
+运行:uv run python scripts/backfill_riin_goods_status.py [--dry-run] [--batch N]
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+
+import psycopg2
+from dotenv import load_dotenv
+
+load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))
+
+DEFAULT_BATCH = 50_000
+
+SQL_COUNT = """
+SELECT count(*) FROM "Goods" g
+JOIN "Orders" o ON o."orderNo" = g."orderNo" AND o."createdAt" = g."createdAt"
+WHERE o."systemName" = 'RIIN' AND o."systemStatus" = '已发货' AND g."status" IS NULL
+"""
+
+# ctid 分批:PostgreSQL 的 UPDATE 不支持 LIMIT,用子查询圈定每批行
+SQL_BACKFILL_BATCH = """
+UPDATE "Goods" g SET "status" = '已发货'
+WHERE g.ctid IN (
+    SELECT g2.ctid FROM "Goods" g2
+    JOIN "Orders" o ON o."orderNo" = g2."orderNo" AND o."createdAt" = g2."createdAt"
+    WHERE o."systemName" = 'RIIN' AND o."systemStatus" = '已发货' AND g2."status" IS NULL
+    LIMIT %s
+)
+"""
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="回填 RIIN Goods.status NULL(已发货订单)")
+    parser.add_argument("--dry-run", action="store_true", help="只统计待回填行数,不写入")
+    parser.add_argument("--batch", type=int, default=DEFAULT_BATCH, help="每事务回填行数")
+    args = parser.parse_args()
+
+    conn = psycopg2.connect(os.environ["DB_DSN"])
+    try:
+        with conn.cursor() as cur:
+            cur.execute(SQL_COUNT)
+            (pending,) = cur.fetchone()
+        print(f"[backfill] 待回填(status IS NULL 且订单已发货): {pending} 行")
+        if args.dry_run or pending == 0:
+            return
+
+        total = 0
+        while True:
+            with conn.cursor() as cur:
+                cur.execute(SQL_BACKFILL_BATCH, (args.batch,))
+                updated = cur.rowcount
+            conn.commit()
+            total += updated
+            print(f"[backfill] 本批 {updated} 行,累计 {total}")
+            if updated < args.batch:
+                break
+        print(f"[backfill] 完成,共回填 {total} 行")
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()

tests/test_mappers.py

@@ -0,0 +1,126 @@
+"""mappers 状态映射单测。
+
+运行:uv run python -m unittest discover -s tests -v
+"""
+
+from __future__ import annotations
+
+import io
+import unittest
+from contextlib import redirect_stdout
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+from flows.shared import mappers
+from flows.shared.mappers import (
+    map_riin_goods,
+    map_riin_goods_status,
+    map_riin_system_status,
+    map_sds_goods_status,
+    map_sds_system_status,
+)
+
+TZ_CN = timezone(timedelta(hours=8))
+
+
+class WarnUnknownTest(unittest.TestCase):
+    def setUp(self):
+        mappers._warned_values.clear()
+
+    def _capture(self):
+        return io.StringIO()
+
+    def test_未知值告警且同值只报一次(self):
+        buf = self._capture()
+        with redirect_stdout(buf):
+            self.assertIsNone(map_riin_goods_status("SOMETHING_NEW"))
+            self.assertIsNone(map_riin_goods_status("SOMETHING_NEW"))  # 第二次不再报
+        lines = [ln for ln in buf.getvalue().splitlines() if "SOMETHING_NEW" in ln]
+        self.assertEqual(len(lines), 1)
+        self.assertIn("[mapper] unknown RIIN goodsStatus", lines[0])
+
+
+class RiinGoodsStatusTest(unittest.TestCase):
+    def setUp(self):
+        mappers._warned_values.clear()
+
+    def test_全部已知枚举可映射(self):
+        cases = {
+            "NOT_SHIPPED": "待发货",
+            "SHIPPED": "已发货",
+            "CANCEL": "已取消",
+            "CANCELLED": "已取消",
+            "REFUNDING": "退款中",
+            "COMPLETE": "已完成",  # 本次补齐(曾导致 NULL)
+            "PAUSE": "暂停",       # 本次补齐(曾导致 NULL)
+            "CLOSE": "已取消",     # 回填重拉时告警发现的新枚举
+        }
+        for raw, want in cases.items():
+            with self.subTest(raw=raw):
+                self.assertEqual(map_riin_goods_status(raw), want)
+
+    def test_空值返回None(self):
+        self.assertIsNone(map_riin_goods_status(None))
+        self.assertIsNone(map_riin_goods_status(""))
+
+    def test_map_riin_goods_接线_PAUSE与COMPLETE不再落NULL(self):
+        # 复现线上订单 GSU1S555400MFLL:4 行商品 goods_status=PAUSE/COMPLETE
+        created = datetime(2026, 9, 14, 9, 39, 24, tzinfo=TZ_CN)
+        base = dict(
+            platform_oid="GSU1S555400MFLL", product_code="DG001-BK",
+            color_code="BK", size_code="L", style_code=None, num=1,
+        )
+        g1 = map_riin_goods(SimpleNamespace(id="P1", goods_status="PAUSE", **base),
+                            "GSU1S555400MFLL", created)
+        g2 = map_riin_goods(SimpleNamespace(id="P2", goods_status="COMPLETE", **base),
+                            "GSU1S555400MFLL", created)
+        self.assertEqual(g1.status, "暂停")
+        self.assertEqual(g2.status, "已完成")
+        self.assertEqual(g1.productionNo, "P1")
+        self.assertEqual(g1.quantity, 1)
+        self.assertEqual(g1.itemName, "DG001")
+
+
+class RiinSystemStatusTest(unittest.TestCase):
+    def setUp(self):
+        mappers._warned_values.clear()
+
+    def test_已知与空值(self):
+        self.assertEqual(map_riin_system_status(12), "已发货")
+        self.assertEqual(map_riin_system_status(-4), "待推送")
+        self.assertIsNone(map_riin_system_status(None))
+
+    def test_未知值返回None并告警(self):
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            self.assertIsNone(map_riin_system_status(999))
+        self.assertIn("unknown RIIN systemStatus", buf.getvalue())
+
+
+class SdsStatusTest(unittest.TestCase):
+    def setUp(self):
+        mappers._warned_values.clear()
+
+    def test_商品状态含已删除(self):
+        self.assertEqual(map_sds_goods_status(99), "已删除")
+        self.assertEqual(map_sds_goods_status(98), "已取消")
+        self.assertEqual(map_sds_goods_status(4), "已完成")
+        self.assertIsNone(map_sds_goods_status(None))
+
+    def test_未知商品状态返回None并告警(self):
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            self.assertIsNone(map_sds_goods_status(6))
+        self.assertIn("unknown SDS goodsStatus", buf.getvalue())
+
+    def test_订单状态与未知告警(self):
+        self.assertEqual(map_sds_system_status(3), "备货中")
+        self.assertIsNone(map_sds_system_status(None))
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            self.assertIsNone(map_sds_system_status(42))
+        self.assertIn("unknown SDS systemStatus", buf.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()