From 68f0d9439ca094d8bd493a8d8f1d44fa119d01e0 Mon Sep 17 00:00:00 2001
From: yeuimu <2197651308@qq.com>
Date: Fri, 31 Jul 2026 05:43:02 +0000
Subject: [PATCH] =?UTF-8?q?feat:=20label-ocr=20=E9=98=B2=E9=87=8D=E5=85=A5?=
 =?UTF-8?q?(=E5=90=8C=E4=B8=80=E6=97=B6=E9=97=B4=E5=8F=AA=E8=B7=91?=
 =?UTF-8?q?=E4=B8=80=E4=B8=AA)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 flows/label_ocr.py | 30 ++++++++++++++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/flows/label_ocr.py b/flows/label_ocr.py
index 1e2bbe8..5e1ceff 100644
--- a/flows/label_ocr.py
+++ b/flows/label_ocr.py
@@ -6,7 +6,11 @@
 
 from __future__ import annotations
 
+import os
+
+import httpx
 from prefect import flow, task
+from prefect.runtime import deployment, flow_run
 
 from api.ocr import OcrClient, country_to_ocr_code
 from flows.shared import SyncDB, build_stats_rows, emit_sync_summary
@@ -28,6 +32,29 @@ def _result_to_fields(r) -> dict[str, str | None]:
     return {col: (getattr(r, attr) or "") for attr, col in _OCR_FIELD_MAP}
 
 
+def _has_running_peer() -> bool:
+    """检查是否有同类 flow run 正在运行(排除自身)。"""
+    api_url = os.environ.get("PREFECT_API_URL", "http://localhost:4200/api")
+    cur_id = str(flow_run.id) if flow_run.id else ""
+    dep_id = str(deployment.id) if deployment.id else ""
+    if not dep_id:
+        return False
+    try:
+        resp = httpx.post(
+            f"{api_url}/flow_runs/filter",
+            json={
+                "flow_runs": {"deployment_id": {"any_": [dep_id]}},
+                "state": {"type": {"any_": ["RUNNING"]}},
+            },
+            params={"limit": 10},
+            timeout=10,
+        )
+        resp.raise_for_status()
+        return any(r["id"] != cur_id for r in resp.json())
+    except Exception:
+        return False
+
+
 @task
 def process_all(pending: list) -> dict[str, int]:
     db = SyncDB()
@@ -97,6 +124,9 @@ def label_ocr(
     """
     if destination_countries is None:
         destination_countries = ['GB', 'US']
+    if _has_running_peer():
+        print("已有 label-ocr 在运行,跳过本次")
+        return {"skipped": True}
     db = SyncDB()
     before = db.count_ocr_status()
     print(f"OCR 状态(开始): {before}")