52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
import base64
|
|
|
|
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) -> None:
|
|
self.settings = settings
|
|
|
|
@property
|
|
def configured(self) -> bool:
|
|
return bool(
|
|
self.settings.baidu_ocr_enabled
|
|
and self.settings.baidu_ocr_api_key
|
|
and self.settings.baidu_ocr_secret_key
|
|
)
|
|
|
|
async def _access_token(self) -> str:
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
response = await client.post(
|
|
self.settings.baidu_ocr_token_url,
|
|
params={
|
|
"grant_type": "client_credentials",
|
|
"client_id": self.settings.baidu_ocr_api_key,
|
|
"client_secret": self.settings.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", [])
|