From a742f4169ea8514d6878aad561a807672d5814db Mon Sep 17 00:00:00 2001 From: yeuimu <2197651308@qq.com> Date: Fri, 31 Jul 2026 01:57:52 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20label-ocr=20=E9=BB=98=E8=AE=A4=20GB+US?= =?UTF-8?q?=20=E8=BF=917=E5=A4=A9;=20refresh=5Fpending=20=E5=90=88?= =?UTF-8?q?=E5=B9=B6=E6=97=A0=E9=9D=A2=E5=8D=95=E8=AE=A2=E5=8D=95(?= =?UTF-8?q?=E8=BF=917=E5=A4=A9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- flows/label_ocr.py | 24 ++++++++++++++++-------- flows/riin/refresh_pending.py | 15 ++++++++++----- flows/sds/refresh_pending.py | 15 ++++++++++----- flows/shared/db.py | 34 +++++++++++++++++++++++++++------- 4 files changed, 63 insertions(+), 25 deletions(-) diff --git a/flows/label_ocr.py b/flows/label_ocr.py index 20aa99f..1e2bbe8 100644 --- a/flows/label_ocr.py +++ b/flows/label_ocr.py @@ -86,20 +86,28 @@ def process_all(pending: list) -> dict[str, int]: @flow(name="label-ocr", log_prints=True) -def label_ocr(destination_country: str | None = 'GB', courier_source: str | None = None) -> dict: - """面单 OCR enrichment。默认只处理英国(GB),可选 courierSource 进一步筛选。 +def label_ocr( + destination_countries: list[str] | None = None, + courier_source: str | None = None, + days: int = 7, +) -> dict: + """面单 OCR enrichment。默认处理英国(GB)+美国(US),近 7 天下单的订单。 - 手动跑其他国家: label_ocr(destination_country='US') + 手动跑其他国家: label_ocr(destination_countries=['DE']) """ + if destination_countries is None: + destination_countries = ['GB', 'US'] db = SyncDB() before = db.count_ocr_status() print(f"OCR 状态(开始): {before}") - pending = db.get_ocr_pending(destination_country=destination_country, courier_source=courier_source) - scope = "" - if destination_country or courier_source: - scope = f" [{destination_country or ''} {courier_source or ''}]".strip() - print(f"待 OCR: {len(pending)} 单{scope} (RIIN+SDS)") + pending = db.get_ocr_pending( + destination_countries=destination_countries, + courier_source=courier_source, + days=days, + ) + scope = f" [{'/'.join(destination_countries)}{f' {courier_source}' if courier_source else ''}]" + print(f"待 OCR: {len(pending)} 单{scope} (RIIN+SDS, 近{days}天)") if not pending: emit_sync_summary(f"面单 OCR{scope}", build_stats_rows("ALL", {}, [ ("待处理", 0), ("累计已完成", before.get("done", 0)), diff --git a/flows/riin/refresh_pending.py b/flows/riin/refresh_pending.py index 42433c1..851d886 100644 --- a/flows/riin/refresh_pending.py +++ b/flows/riin/refresh_pending.py @@ -44,15 +44,20 @@ def upsert_records(raw_orders: list) -> dict: def riin_refresh_pending() -> dict: db = SyncDB() order_nos = db.get_pending_orders("RIIN", RIIN_FINISHED_STATUSES) - print(f"RIIN 未完结订单 {len(order_nos)} 个") - if not order_nos: + no_label_nos = db.get_orders_without_label("RIIN", days=7) + all_nos = list(set(order_nos) | set(no_label_nos)) + print(f"RIIN 未完结 {len(order_nos)} + 无面单(近7天) {len(no_label_nos)} → 合并 {len(all_nos)}") + if not all_nos: return {"pendingCount": 0, "matchedCount": 0} - raw = fetch_pending(order_nos) + raw = fetch_pending(all_nos) stats = upsert_records(raw) print(f"完成: 匹配 {len(raw)} | {stats}") - emit_sync_summary("RIIN 刷新未完结", build_stats_rows("RIIN", stats, [("未完结订单", len(order_nos)), ("匹配", len(raw))])) - return {"pendingCount": len(order_nos), "matchedCount": len(raw), **stats} + emit_sync_summary("RIIN 刷新未完结", build_stats_rows("RIIN", stats, [ + ("未完结订单", len(order_nos)), ("无面单(近7天)", len(no_label_nos)), + ("合并去重", len(all_nos)), ("匹配", len(raw)), + ])) + return {"pendingCount": len(all_nos), "matchedCount": len(raw), **stats} if __name__ == "__main__": diff --git a/flows/sds/refresh_pending.py b/flows/sds/refresh_pending.py index 73ae03c..1a3345e 100644 --- a/flows/sds/refresh_pending.py +++ b/flows/sds/refresh_pending.py @@ -48,15 +48,20 @@ def upsert_records(raw_orders: list) -> dict: def sds_refresh_pending() -> dict: db = SyncDB() order_nos = db.get_pending_orders("SDS", SDS_FINISHED_STATUSES) - print(f"SDS 未完结订单 {len(order_nos)} 个") - if not order_nos: + no_label_nos = db.get_orders_without_label("SDS", days=7) + all_nos = list(set(order_nos) | set(no_label_nos)) + print(f"SDS 未完结 {len(order_nos)} + 无面单(近7天) {len(no_label_nos)} → 合并 {len(all_nos)}") + if not all_nos: return {"pendingCount": 0, "matchedCount": 0} - raw = fetch_pending(order_nos) + raw = fetch_pending(all_nos) stats = upsert_records(raw) print(f"完成: 匹配 {len(raw)} | {stats}") - emit_sync_summary("SDS 刷新未完结", build_stats_rows("SDS", stats, [("未完结订单", len(order_nos)), ("匹配", len(raw))])) - return {"pendingCount": len(order_nos), "matchedCount": len(raw), **stats} + emit_sync_summary("SDS 刷新未完结", build_stats_rows("SDS", stats, [ + ("未完结订单", len(order_nos)), ("无面单(近7天)", len(no_label_nos)), + ("合并去重", len(all_nos)), ("匹配", len(raw)), + ])) + return {"pendingCount": len(all_nos), "matchedCount": len(raw), **stats} if __name__ == "__main__": diff --git a/flows/shared/db.py b/flows/shared/db.py index 6a01f7b..167c5ce 100644 --- a/flows/shared/db.py +++ b/flows/shared/db.py @@ -8,7 +8,7 @@ from __future__ import annotations -from datetime import datetime +from datetime import datetime, timedelta from typing import Any from psycopg2.extras import execute_values @@ -101,12 +101,16 @@ class SyncDB: # ---------- OCR 面单 enrichment ---------- def get_ocr_pending( - self, limit: int = 0, destination_country: str | None = None, courier_source: str | None = None, + self, limit: int = 0, + destination_countries: list[str] | None = None, + courier_source: str | None = None, + days: int = 0, ) -> list[tuple[str, datetime, str, str | None]]: """查待 OCR 订单(所有系统):(orderNo, createdAt, labelPdfUrl, destinationCountry)。 - 条件:有 labelPdfUrl、ocr_layout 未处理(Null)。 - limit=0 表示不限。可选按 destinationCountry / courierSource 筛选。 + 条件:有 labelPdfUrl、ocrLayout 未处理(Null)。 + limit=0 表示不限。可选按 destinationCountries / courierSource 筛选。 + days>0 时只取近 N 天下单的订单。 """ sql = ( 'SELECT "orderNo","createdAt","labelPdfUrl","destinationCountry" ' @@ -114,12 +118,15 @@ class SyncDB: 'WHERE "labelPdfUrl" IS NOT NULL AND "ocrLayout" IS NULL' ) params: list = [] - if destination_country: - sql += ' AND "destinationCountry"=%s' - params.append(destination_country) + if destination_countries: + sql += ' AND "destinationCountry"=ANY(%s)' + params.append(destination_countries) if courier_source: sql += ' AND "courierSource"=%s' params.append(courier_source) + if days > 0: + sql += ' AND "orderedAt" >= %s' + params.append(datetime.now() - timedelta(days=days)) sql += ' ORDER BY "orderedAt" DESC' if limit > 0: sql += " LIMIT %s" @@ -129,6 +136,19 @@ class SyncDB: cur.execute(sql, params) return [(r[0], r[1], r[2], r[3]) for r in cur.fetchall()] + def get_orders_without_label(self, system_name: str, days: int = 7) -> list[str]: + """查近 N 天下单但无 labelPdfUrl 的订单号。""" + cutoff = datetime.now() - timedelta(days=days) + with self._conn() as conn: + with conn.cursor() as cur: + cur.execute( + 'SELECT DISTINCT "orderNo" FROM "Orders" ' + 'WHERE "systemName"=%s AND "labelPdfUrl" IS NULL AND "orderedAt" >= %s', + (system_name, cutoff), + ) + rows = cur.fetchall() + return [r[0] for r in rows] + def update_ocr_fields( self, order_no: str, created_at: datetime, fields: dict[str, str | None] ) -> None: