Files
inkreach-prefect/api/ocr/client.py
T

54 lines
1.7 KiB
Python

"""OCR 面单识别 Inference API 客户端(httpx,无鉴权)。
端点: POST /recognize_url JSON {url, country}
"""
from __future__ import annotations
import httpx
from api.ocr.models import OcrResult
DEFAULT_BASE_URL = "http://192.168.124.92:9801"
DEFAULT_TIMEOUT = 60.0
class OcrClient:
"""OCR 推理服务客户端。"""
def __init__(self, base_url: str = DEFAULT_BASE_URL, timeout: float = DEFAULT_TIMEOUT) -> None:
self._http = httpx.Client(base_url=base_url, timeout=timeout)
def recognize_url(self, url: str, country: str = "US") -> OcrResult:
"""通过 PDF URL 识别面单,返回 OcrResult。"""
resp = self._http.post("/recognize_url", json={"url": url, "country": country})
resp.raise_for_status()
return OcrResult.model_validate(resp.json())
def recognize_batch(self, urls: list[dict[str, str]], country: str = "US") -> list[OcrResult]:
"""v2 批量识别:urls=[{"url":"...", "filename":"orderNo"}], 最多 200 项/批。
返回 OcrResult 列表,file 字段 = filename(用于回匹配 orderNo)。
"""
resp = self._http.post("/api/v2/batch", json={"urls": urls, "country": country}, timeout=300.0)
resp.raise_for_status()
data = resp.json()
return [OcrResult.model_validate(r) for r in data.get("results", [])]
def health(self) -> bool:
"""健康检查。"""
try:
resp = self._http.get("/health")
return resp.status_code == 200
except Exception: # noqa: BLE001
return False
def close(self) -> None:
self._http.close()
def __enter__(self) -> "OcrClient":
return self
def __exit__(self, *exc: object) -> None:
self.close()