From 51dbde453861ad3e810e2f7ca61db04156531d39 Mon Sep 17 00:00:00 2001
From: yeuimu <2197651308@qq.com>
Date: Thu, 23 Jul 2026 16:15:24 +0000
Subject: [PATCH] =?UTF-8?q?feat:=20OCR=20=E9=9D=A2=E5=8D=95=20enrichment?=
 =?UTF-8?q?=20flow=20(api/ocr/=20+=20flows/sds/label=5Focr.py=20+=20db=20O?=
 =?UTF-8?q?CR=20=E6=96=B9=E6=B3=95)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 api/ocr/__init__.py    |  9 ++++
 api/ocr/client.py      | 43 ++++++++++++++++++
 api/ocr/models.py      | 56 ++++++++++++++++++++++++
 flows/sds/label_ocr.py | 98 ++++++++++++++++++++++++++++++++++++++++++
 flows/shared/db.py     | 53 +++++++++++++++++++++++
 prefect.yaml           |  7 +++
 6 files changed, 266 insertions(+)
 create mode 100644 api/ocr/__init__.py
 create mode 100644 api/ocr/client.py
 create mode 100644 api/ocr/models.py
 create mode 100644 flows/sds/label_ocr.py

diff --git a/api/ocr/__init__.py b/api/ocr/__init__.py
new file mode 100644
index 0000000..893a04d
--- /dev/null
+++ b/api/ocr/__init__.py
@@ -0,0 +1,9 @@
+"""OCR 面单识别服务客户端(无鉴权内部服务)。
+
+Inference API: http://192.168.124.92:9801
+"""
+
+from api.ocr.client import OcrClient
+from api.ocr.models import OcrResult, SUPPORTED_COUNTRIES, country_to_ocr_code
+
+__all__ = ["OcrClient", "OcrResult", "SUPPORTED_COUNTRIES", "country_to_ocr_code"]
diff --git a/api/ocr/client.py b/api/ocr/client.py
new file mode 100644
index 0000000..4c91487
--- /dev/null
+++ b/api/ocr/client.py
@@ -0,0 +1,43 @@
+"""OCR 面单识别 Inference API 客户端(httpx,无鉴权)。
+
+端点: POST /recognize_url  JSON {url, country}
+"""
+
+from __future__ import annotations
+
+import httpx
+
+from api.ocr.models import OcrResult
+
+DEFAULT_BASE_URL = "http://192.168.124.92:9801"
+DEFAULT_TIMEOUT = 60.0
+
+
+class OcrClient:
+    """OCR 推理服务客户端。"""
+
+    def __init__(self, base_url: str = DEFAULT_BASE_URL, timeout: float = DEFAULT_TIMEOUT) -> None:
+        self._http = httpx.Client(base_url=base_url, timeout=timeout)
+
+    def recognize_url(self, url: str, country: str = "US") -> OcrResult:
+        """通过 PDF URL 识别面单,返回 OcrResult。"""
+        resp = self._http.post("/recognize_url", json={"url": url, "country": country})
+        resp.raise_for_status()
+        return OcrResult.model_validate(resp.json())
+
+    def health(self) -> bool:
+        """健康检查。"""
+        try:
+            resp = self._http.get("/health")
+            return resp.status_code == 200
+        except Exception:  # noqa: BLE001
+            return False
+
+    def close(self) -> None:
+        self._http.close()
+
+    def __enter__(self) -> "OcrClient":
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        self.close()
diff --git a/api/ocr/models.py b/api/ocr/models.py
new file mode 100644
index 0000000..c217d69
--- /dev/null
+++ b/api/ocr/models.py
@@ -0,0 +1,56 @@
+"""OCR 面单识别响应模型 + 国家代码映射。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict
+
+
+class OcrResult(BaseModel):
+    """单张面单 OCR 识别结果。"""
+
+    model_config = ConfigDict(extra="allow")
+
+    file: str | None = None
+    layout: str | None = None
+    courier: str | None = None
+    courier_name: str | None = None
+    country: str | None = None
+    error: str | None = None
+    tracking: str | None = None
+    s_name: str | None = None
+    s_street: str | None = None
+    s_city: str | None = None
+    s_state: str | None = None
+    s_zip: str | None = None
+    r_name: str | None = None
+    r_street: str | None = None
+    r_city: str | None = None
+    r_state: str | None = None
+    r_zip: str | None = None
+    marker: str | None = None
+
+    @property
+    def ok(self) -> bool:
+        return not self.error
+
+
+# OCR 服务已有分类器模型的国家(版面见 api.md §4)
+SUPPORTED_COUNTRIES = {"US", "UK", "DE", "MX", "PL"}
+
+# ISO 国家代码 / 中文国名 → OCR 国家代码
+_COUNTRY_MAP: dict[str, str] = {
+    # 代码
+    "GB": "UK", "UK": "UK", "US": "US", "DE": "DE", "MX": "MX", "PL": "PL",
+    # 中文名(SDS countryName 可能用)
+    "英国": "UK", "美国": "US", "德国": "DE", "墨西哥": "MX", "波兰": "PL",
+}
+
+
+def country_to_ocr_code(country: Any) -> str | None:
+    """将 ISO 代码或中文国名转为 OCR 服务支持的国家代码,不支持则返回 None。"""
+    if not country or not isinstance(country, str):
+        return None
+    code = _COUNTRY_MAP.get(country.strip().upper()) or _COUNTRY_MAP.get(country.strip())
+    return code if code in SUPPORTED_COUNTRIES else None
diff --git a/flows/sds/label_ocr.py b/flows/sds/label_ocr.py
new file mode 100644
index 0000000..55ccc24
--- /dev/null
+++ b/flows/sds/label_ocr.py
@@ -0,0 +1,98 @@
+"""SDS 面单 OCR enrichment:对已发货的 SDS 订单,用面单 PDF URL 调 OCR 服务回填标签字段。
+
+对应 sds-label-ocr。独立于同步 flow,每 4 小时跑一批。
+"""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+from prefect import flow, task
+
+from api.ocr import OcrClient, country_to_ocr_code
+from flows.shared import SyncDB, build_stats_rows, emit_sync_summary
+
+OCR_CONCURRENCY = 5
+SHIPPED_STATUSES = ["已完成", "部分发货", "备货中", "待确认"]
+
+_OCR_FIELD_MAP = [
+    ("layout", "ocr_layout"), ("courier", "ocr_courier"), ("tracking", "ocr_tracking"),
+    ("marker", "ocr_marker"), ("s_name", "ocr_s_name"), ("s_street", "ocr_s_street"),
+    ("s_city", "ocr_s_city"), ("s_state", "ocr_s_state"), ("s_zip", "ocr_s_zip"),
+    ("r_name", "ocr_r_name"), ("r_street", "ocr_r_street"), ("r_city", "ocr_r_city"),
+    ("r_state", "ocr_r_state"), ("r_zip", "ocr_r_zip"),
+]
+
+
+def _process_one(db: SyncDB, ocr: OcrClient, order_no: str, created_at, pdf_url: str, dest_country: str | None) -> str:
+    country_code = country_to_ocr_code(dest_country)
+    if not country_code:
+        db.update_ocr_fields(order_no, created_at, {"ocr_layout": "UNSUPPORTED"})
+        return "unsupported"
+    try:
+        r = ocr.recognize_url(pdf_url, country_code)
+    except Exception as err:  # noqa: BLE001
+        print(f"[OCR] {order_no} 请求失败: {str(err)[:100]}")
+        db.update_ocr_fields(order_no, created_at, {"ocr_layout": "ERROR"})
+        return "error"
+    if not r.ok:
+        print(f"[OCR] {order_no} 识别失败: {(r.error or '')[:80]}")
+        db.update_ocr_fields(order_no, created_at, {"ocr_layout": "ERROR"})
+        return "error"
+    fields = {col: getattr(r, attr) or "" for attr, col in _OCR_FIELD_MAP}
+    db.update_ocr_fields(order_no, created_at, fields)
+    return "success"
+
+
+@task
+def process_batch(pending: list) -> dict[str, int]:
+    db = SyncDB()
+    ocr = OcrClient()
+    stats = {"success": 0, "error": 0, "unsupported": 0}
+    try:
+        done = 0
+        total = len(pending)
+        with ThreadPoolExecutor(max_workers=min(OCR_CONCURRENCY, total)) as pool:
+            futures = {
+                pool.submit(_process_one, db, ocr, no, ca, url, dc): no
+                for no, ca, url, dc in pending
+            }
+            for fut in as_completed(futures):
+                result = fut.result()
+                stats[result] += 1
+                done += 1
+                if done % 50 == 0 or done == total:
+                    print(f"[OCR] 进度: {done}/{total}  成功 {stats['success']} / 失败 {stats['error']} / 不支持 {stats['unsupported']}")
+        stats["processed"] = total
+        return stats
+    finally:
+        ocr.close()
+
+
+@flow(name="sds-label-ocr", log_prints=True)
+def sds_label_ocr(batch_size: int = 200) -> dict:
+    db = SyncDB()
+    before = db.count_ocr_status()
+    print(f"OCR 状态(开始): {before}")
+
+    pending = db.get_ocr_pending(batch_size)
+    print(f"待 OCR: {len(pending)} 单 (batch_size={batch_size})")
+    if not pending:
+        emit_sync_summary("SDS 面单 OCR", build_stats_rows("SDS", {}, [
+            ("待处理", 0), ("累计已完成", before.get("done", 0)), ("累计失败", before.get("error", 0)),
+        ]))
+        return {"processed": 0, **before}
+
+    stats = process_batch(pending)
+    after = db.count_ocr_status()
+    print(f"完成: {stats}  累计: {after}")
+    emit_sync_summary("SDS 面单 OCR", build_stats_rows("SDS", {}, [
+        ("本批处理", stats["processed"]), ("本批成功", stats["success"]),
+        ("本批失败", stats["error"]), ("本批不支持", stats["unsupported"]),
+        ("累计已完成", after.get("done", 0)), ("累计待处理", after.get("pending", 0)),
+    ]))
+    return {**stats, **after}
+
+
+if __name__ == "__main__":
+    sds_label_ocr.serve(name="sds-label-ocr", cron="0 */4 * * *")
diff --git a/flows/shared/db.py b/flows/shared/db.py
index 53c4657..9fb2db1 100644
--- a/flows/shared/db.py
+++ b/flows/shared/db.py
@@ -98,6 +98,59 @@ class SyncDB:
                 rows = cur.fetchall()
         return [(r[0], r[1]) for r in rows]
 
+    # ---------- OCR 面单 enrichment ----------
+
+    def get_ocr_pending(self, limit: int = 200) -> list[tuple[str, datetime, str, str | None]]:
+        """查待 OCR 的 SDS 订单:(orderNo, createdAt, labelPdfUrl, destinationCountry)。
+
+        条件:SDS、有 labelPdfUrl、ocr_layout 未处理(Null)、已发货/已完成。
+        """
+        with self._conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    '''SELECT "orderNo","createdAt","labelPdfUrl","destinationCountry"
+                       FROM "Orders"
+                       WHERE "systemName"=%s AND "labelPdfUrl" IS NOT NULL
+                         AND "ocr_layout" IS NULL
+                         AND "systemStatus"=ANY(%s)
+                       ORDER BY "orderedAt" DESC LIMIT %s''',
+                    ("SDS", ["已完成", "部分发货", "备货中", "待确认"], limit),
+                )
+                return [(r[0], r[1], r[2], r[3]) for r in cur.fetchall()]
+
+    def update_ocr_fields(
+        self, order_no: str, created_at: datetime, fields: dict[str, str | None]
+    ) -> None:
+        """更新单个订单的 ocr_* 字段。fields 为 {ocr_xxx: value} 字典。"""
+        if not fields:
+            return
+        assignments = ", ".join(f'"{k}"=%s' for k in fields)
+        params = list(fields.values()) + [order_no, created_at]
+        with self._conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f'''UPDATE "Orders" SET {assignments}
+                        WHERE "orderNo"=%s AND "createdAt"=%s''',
+                    params,
+                )
+            conn.commit()
+
+    def count_ocr_status(self) -> dict[str, int]:
+        """统计 OCR 处理状态(done/error/unsupported/pending)。"""
+        with self._conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    '''SELECT
+                        COUNT(*) FILTER (WHERE "ocr_layout" IS NOT NULL AND "ocr_layout" NOT IN ('ERROR','UNSUPPORTED')) AS done,
+                        COUNT(*) FILTER (WHERE "ocr_layout"='ERROR') AS error,
+                        COUNT(*) FILTER (WHERE "ocr_layout"='UNSUPPORTED') AS unsupported,
+                        COUNT(*) FILTER (WHERE "ocr_layout" IS NULL AND "labelPdfUrl" IS NOT NULL) AS pending
+                       FROM "Orders" WHERE "systemName"=%s''',
+                    ("SDS",),
+                )
+                row = cur.fetchone()
+        return {"done": row[0], "error": row[1], "unsupported": row[2], "pending": row[3]} if row else {}
+
     # ---------- 主 upsert ----------
 
     def sync_orders_and_goods(
diff --git a/prefect.yaml b/prefect.yaml
index 75eb44f..4f6a7a6 100644
--- a/prefect.yaml
+++ b/prefect.yaml
@@ -80,3 +80,10 @@ deployments:
     timezone: Asia/Shanghai
   work_pool:
     name: inkreach-pool
+- name: sds-label-ocr
+  entrypoint: flows/sds/label_ocr.py:sds_label_ocr
+  schedules:
+  - cron: 0 */4 * * *
+    timezone: Asia/Shanghai
+  work_pool:
+    name: inkreach-pool