inkreachhub/inkreach-prefect
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: label-ocr 默认 GB+US 近7天; refresh_pending 合并无面单订单(近7天)

Browse Source
This commit is contained in:
yeuimu
2026-07-31 01:57:52 +00:00
parent 3e075fd366
commit a742f4169e
4 changed files with 63 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flows/label_ocr.py
+16 -8
View File
@@ -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)),
flows/riin/refresh_pending.py
+10 -5
View File
@@ -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__":
flows/sds/refresh_pending.py
+10 -5
View File
@@ -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__":
flows/shared/db.py
+27 -7
View File
@@ -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: