Merge pull request 'feat: OCR 新增支持波兰(PL,版式 pables_dpd/pables_inpost)' (#10) from inkreach/ocr-pl into main

Reviewed-on: #10

.python-version

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

api/riin/client.py

@@ -246,14 +246,6 @@ 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/label_ocr.py

@@ -29,20 +29,13 @@ _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"}
-    fields = {col: (getattr(r, attr) or "") for attr, col in _OCR_FIELD_MAP}
+    return {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

@@ -1,144 +0,0 @@
-# -*- 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_for_orders
+from flows.shared import fetch_goods_concurrent
 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_for_orders(client, list(order_meta.keys()))
+        goods_map = fetch_goods_concurrent(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_for_orders,
+    fetch_goods_concurrent,
     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_for_orders",
+    "fetch_riin_all_pages", "fetch_sds_all_pages", "fetch_goods_concurrent",
     "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,6 +21,7 @@ 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
@@ -143,58 +144,39 @@ def fetch_sds_all_pages(client: SdsMngClient, params: dict[str, Any]) -> list[An
 
 # ---------- RIIN 商品 ----------
 
-# 批量接口 distribution/order/goods/list:单批 ≤200 个 id,且要求串行调用(分销侧约定)。
+def fetch_goods_for_order(client: RiinDisClient, order_id: str) -> list[Any]:
-# 相比旧逐单接口(软限流:高频返回成功但空 data),调用量从日均 ~1 万次降到 ~50 次。
+    try:
-GOODS_BATCH_SIZE = 200
+        resp = client.get_order_goods(order_id)
-GOODS_BATCH_INTERVAL_MS = 2000  # 相邻两批的串行间隔
+        return list(resp.data)
+    except Exception as err:  # noqa: BLE001
+        print(f"[RIIN sync] 订单 {order_id} 商品拉取失败: {err}")
+        return []
 
 
-def fetch_goods_for_orders(client: RiinDisClient, order_ids: list[str]) -> dict[str, list[Any]]:
+def fetch_goods_concurrent(client: RiinDisClient, order_ids: list[str]) -> dict[str, list[Any]]:
-    """按 goods/list 批量接口串行拉取多订单商品,返回 {order_id: [goods...]}。
+    """以 GOODS_CONCURRENCY 并发拉取多订单商品,返回 {order_id: [goods...]}。"""
+    from collections.abc import Callable
+    from concurrent.futures import ThreadPoolExecutor
 
-    单批失败(网络/业务错误)整批重试最多 MAX_RETRIES 次,仍失败则该批订单记空列表,
+    results: dict[str, list[Any]] = {}
-    由调用方按"无商品订单"计数(与旧逐单路径的容错语义一致)。
-    """
-    results: dict[str, list[Any]] = {oid: [] for oid in order_ids}
     if not order_ids:
         return results
-    total_batches = -(-len(order_ids) // GOODS_BATCH_SIZE)
+    concurrency = min(GOODS_CONCURRENCY, len(order_ids))
-    print(f"[RIIN sync] 商品批量拉取开始: {len(order_ids)} 个订单 / {total_batches} 批"
+    print(f"[RIIN sync] 商品拉取开始: {len(order_ids)} 个订单 (并发={concurrency})")
-          f" (goods/list 串行, 批间 {GOODS_BATCH_INTERVAL_MS}ms)")
+
-    for bi in range(total_batches):
+    def make(oid: str) -> Callable[[], None]:
-        chunk = order_ids[bi * GOODS_BATCH_SIZE : (bi + 1) * GOODS_BATCH_SIZE]
+        def run() -> None:
-        items = _fetch_goods_batch_with_retry(client, chunk, bi + 1)
+            results[oid] = fetch_goods_for_order(client, oid)
-        for g in items:
+        return run
-            if g.order_id in results:
+
-                results[g.order_id].append(g)
+    with ThreadPoolExecutor(max_workers=concurrency) as pool:
-        if bi < total_batches - 1:
+        list(pool.map(lambda fn: fn(), [make(oid) for oid in order_ids]))
-            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,11 +460,9 @@ 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::timestamptz[]) AS t(order_no,created_at) "
+                "SELECT t.order_no, t.created_at FROM unnest(%s::text[], %s::timestamp[]) AS t(order_no,created_at) "
                 "JOIN \"Orders\" o ON o.\"orderNo\"=t.order_no AND o.\"createdAt\"=t.created_at",
                 _unnest_pairs(pairs),
             )
@@ -625,6 +623,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::timestamptz[]) 或 unnest(%s::text[], %s::text[])。
+    配合 SQL:unnest(%s::text[], %s::timestamp[]) 或 unnest(%s::text[], %s::text[])。
     """
     return [p[0] for p in pairs], [p[1] for p in pairs]

prefect.yaml

@@ -21,7 +21,6 @@ deployments:
   schedules:
   - cron: 30 * * * *
     timezone: Asia/Shanghai
-    active: false  # 暂停:每小时重拉近7天全部订单商品,对 RIIN distribution/order/goods 调用量过大(增量/刷新未完结/日报核验已覆盖)
   work_pool:
     name: inkreach-pool
 - name: riin-refresh-pending
@@ -40,7 +39,6 @@ deployments:
   schedules:
   - cron: 0 6 * * *
     timezone: Asia/Shanghai
-    active: false  # 暂停:与 daily-report 18:30 的核验子流程重复,保留日报一处即可
   work_pool:
     name: inkreach-pool
 - name: sds-order-sync
@@ -110,13 +108,6 @@ 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: