59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
import base64
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.config import Settings
|
|
|
|
|
|
class BaiduOcrAdapter:
|
|
OCR_URL = "https://aip.baidubce.com/rest/2.0/ocr/v1/accurate_basic"
|
|
|
|
def __init__(self, settings: Settings | Mapping[str, Any]) -> None:
|
|
self.settings = settings
|
|
|
|
def _value(self, key: str, default: str | bool = "") -> str | bool:
|
|
if isinstance(self.settings, Mapping):
|
|
return self.settings.get(key, default)
|
|
return getattr(self.settings, key, default)
|
|
|
|
@property
|
|
def configured(self) -> bool:
|
|
return bool(
|
|
self._value("baidu_ocr_enabled")
|
|
and self._value("baidu_ocr_api_key")
|
|
and self._value("baidu_ocr_secret_key")
|
|
)
|
|
|
|
async def _access_token(self) -> str:
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
response = await client.post(
|
|
str(self._value("baidu_ocr_token_url", "https://aip.baidubce.com/oauth/2.0/token")),
|
|
params={
|
|
"grant_type": "client_credentials",
|
|
"client_id": self._value("baidu_ocr_api_key"),
|
|
"client_secret": self._value("baidu_ocr_secret_key"),
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()["access_token"]
|
|
|
|
async def recognize(self, image_bytes: bytes) -> list[dict]:
|
|
if not self.configured:
|
|
raise RuntimeError("Baidu OCR is not configured or enabled.")
|
|
token = await self._access_token()
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
response = await client.post(
|
|
self.OCR_URL,
|
|
params={"access_token": token},
|
|
data={
|
|
"image": base64.b64encode(image_bytes).decode("ascii"),
|
|
"detect_direction": "true",
|
|
"paragraph": "false",
|
|
},
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
)
|
|
response.raise_for_status()
|
|
return response.json().get("words_result", [])
|