diff --git a/flows/label_ocr.py b/flows/label_ocr.py index b53a056..8e6bd3d 100644 --- a/flows/label_ocr.py +++ b/flows/label_ocr.py @@ -86,15 +86,22 @@ def process_all(pending: list) -> dict[str, int]: @flow(name="label-ocr", log_prints=True) -def label_ocr() -> dict: +def label_ocr(destination_country: str | None = None, courier_source: str | None = None) -> dict: + """面单 OCR enrichment。可选按 destinationCountry / courierSource 筛选优先处理。 + + 例: label_ocr(destination_country='GB', courier_source='手动物流') + """ db = SyncDB() before = db.count_ocr_status() print(f"OCR 状态(开始): {before}") - pending = db.get_ocr_pending() - print(f"待 OCR: {len(pending)} 单 (RIIN+SDS)") + 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)") if not pending: - emit_sync_summary("面单 OCR", build_stats_rows("ALL", {}, [ + emit_sync_summary(f"面单 OCR{scope}", build_stats_rows("ALL", {}, [ ("待处理", 0), ("累计已完成", before.get("done", 0)), ])) return {"processed": 0, **before} @@ -102,10 +109,10 @@ def label_ocr() -> dict: stats = process_all(pending) after = db.count_ocr_status() total_processed = stats["success"] + stats["error"] + stats["unsupported"] - print(f"完成: 处理 {total_processed} 单 / {stats['batches']} 批 | " + print(f"完成{scope}: 处理 {total_processed} 单 / {stats['batches']} 批 | " f"成功 {stats['success']} 失败 {stats['error']} 不支持 {stats['unsupported']}") print(f"累计: {after}") - emit_sync_summary("面单 OCR", build_stats_rows("ALL", {}, [ + emit_sync_summary(f"面单 OCR{scope}", build_stats_rows("ALL", {}, [ ("本批处理", total_processed), ("成功", stats["success"]), ("失败", stats["error"]), ("不支持", stats["unsupported"]), ("累计已完成", after.get("done", 0)), ("累计待处理", after.get("pending", 0)), diff --git a/flows/shared/db.py b/flows/shared/db.py index 7bdeab4..bbd9a3c 100644 --- a/flows/shared/db.py +++ b/flows/shared/db.py @@ -100,19 +100,27 @@ class SyncDB: # ---------- OCR 面单 enrichment ---------- - def get_ocr_pending(self, limit: int = 0) -> list[tuple[str, datetime, str, str | None]]: + def get_ocr_pending( + self, limit: int = 0, destination_country: str | None = None, courier_source: str | None = None, + ) -> list[tuple[str, datetime, str, str | None]]: """查待 OCR 订单(所有系统):(orderNo, createdAt, labelPdfUrl, destinationCountry)。 条件:有 labelPdfUrl、ocr_layout 未处理(Null)。 - limit=0 表示不限。 + limit=0 表示不限。可选按 destinationCountry / courierSource 筛选。 """ sql = ( 'SELECT "orderNo","createdAt","labelPdfUrl","destinationCountry" ' 'FROM "Orders" ' - 'WHERE "labelPdfUrl" IS NOT NULL AND "ocr_layout" IS NULL ' - 'ORDER BY "orderedAt" DESC' + 'WHERE "labelPdfUrl" IS NOT NULL AND "ocr_layout" IS NULL' ) params: list = [] + if destination_country: + sql += ' AND "destinationCountry"=%s' + params.append(destination_country) + if courier_source: + sql += ' AND "courierSource"=%s' + params.append(courier_source) + sql += ' ORDER BY "orderedAt" DESC' if limit > 0: sql += " LIMIT %s" params.append(limit)