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}")