fix: EVRI 面单 OCR 单号入库前去掉 '-'

sceptical-finch(riin-recent-sync 10-10 16:04)失败复盘:17track 物流回写
(宿主机本地进程,不走 advisory 锁)与 sync 写库在 panel 触发器链
(panel.orders/dims/dims_daily)上交叉持锁,冲突窗口持续约 3 分钟,
原 3 次重试(1/3/7s)全部耗尽。加深重试使其可跨过分钟级外部冲突窗口。

.python-version

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

flows/label_ocr.py

@@ -29,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:

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/shared/clients.py

@@ -8,6 +8,8 @@
 from __future__ import annotations
 
 import json
+import random
+import threading
 import time
 from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, TypeVar
@@ -21,7 +23,9 @@ from flows.shared.mappers import map_sds_channel_type
 
 T = TypeVar("T")
 
-GOODS_CONCURRENCY = 20
+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
@@ -144,8 +148,25 @@ 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
+
+
+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_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

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)
 
 

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
@@ -108,6 +110,13 @@ deployments:
     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
 - name: label-ocr
   entrypoint: flows/label_ocr.py:label_ocr
   schedules: