115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
"""阿里云 OCR 客户端封装（ocr-api 2021-07-07）。
|
||
|
||
只关注当前需要的能力——RecognizePassport（护照识别）。
|
||
其它证件接口（身份证、行驶证…）按需扩展即可，签名/客户端复用 _build_client。
|
||
|
||
配置项 (app.config.settings):
|
||
- aliyun_ocr_access_key_id
|
||
- aliyun_ocr_access_key_secret
|
||
- aliyun_ocr_endpoint (默认 ocr-api.cn-hangzhou.aliyuncs.com)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from functools import lru_cache
|
||
from io import BytesIO
|
||
from typing import Any
|
||
|
||
from app.config import settings
|
||
|
||
|
||
class OcrConfigError(RuntimeError):
|
||
"""AK/SK 未配置等可恢复错误，路由层应返回 503。"""
|
||
|
||
|
||
class OcrCallError(RuntimeError):
|
||
"""阿里云返回非 200 或网络异常等运行期错误。"""
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def _build_client(): # 延迟导入 SDK，缺包时给出明确提示
|
||
try:
|
||
from alibabacloud_ocr_api20210707.client import Client
|
||
from alibabacloud_tea_openapi.models import Config
|
||
except ImportError as exc: # noqa: BLE001
|
||
raise OcrConfigError(
|
||
"未安装阿里云 OCR SDK，请执行: pip install alibabacloud_ocr_api20210707"
|
||
) from exc
|
||
|
||
ak = settings.aliyun_ocr_access_key_id.strip()
|
||
sk = settings.aliyun_ocr_access_key_secret.strip()
|
||
endpoint = settings.aliyun_ocr_endpoint.strip() or "ocr-api.cn-hangzhou.aliyuncs.com"
|
||
if not ak or not sk:
|
||
raise OcrConfigError(
|
||
"阿里云 OCR 未配置：请在 .env 设置 ALIYUN_OCR_ACCESS_KEY_ID / ALIYUN_OCR_ACCESS_KEY_SECRET"
|
||
)
|
||
config = Config(access_key_id=ak, access_key_secret=sk)
|
||
config.endpoint = endpoint
|
||
return Client(config)
|
||
|
||
|
||
def _parse_data(body: Any) -> dict[str, Any]:
|
||
"""RecognizePassport 返回 body.data 是一段 JSON 字符串，统一解析为 dict。"""
|
||
raw = getattr(body, "data", None)
|
||
if raw is None and isinstance(body, dict):
|
||
raw = body.get("Data") or body.get("data")
|
||
if raw is None:
|
||
return {}
|
||
if isinstance(raw, (dict, list)):
|
||
return raw # type: ignore[return-value]
|
||
try:
|
||
return json.loads(raw)
|
||
except (TypeError, ValueError):
|
||
return {"raw": str(raw)}
|
||
|
||
|
||
def recognize_passport(
|
||
*,
|
||
image_bytes: bytes | None = None,
|
||
image_url: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""识别护照。image_bytes / image_url 至少传一个。
|
||
|
||
返回结构（与控制台字段一致，常见字段示例）:
|
||
{
|
||
"name": "...", # 姓名
|
||
"sex": "M",
|
||
"country": "CHN",
|
||
"passport_no": "E12345678",
|
||
"date_of_birth": "1990-01-01",
|
||
"date_of_expiry": "2030-01-01",
|
||
"place_of_birth": "...",
|
||
...其它字段透传...
|
||
"_request_id": "...", # 便于在阿里云后台溯源
|
||
}
|
||
"""
|
||
if not image_bytes and not image_url:
|
||
raise ValueError("recognize_passport 需要 image_bytes 或 image_url")
|
||
|
||
from alibabacloud_ocr_api20210707.models import RecognizePassportRequest
|
||
from alibabacloud_tea_util.models import RuntimeOptions
|
||
|
||
client = _build_client()
|
||
req_kwargs: dict[str, Any] = {}
|
||
if image_bytes is not None:
|
||
req_kwargs["body"] = BytesIO(image_bytes)
|
||
if image_url:
|
||
req_kwargs["url"] = image_url
|
||
request = RecognizePassportRequest(**req_kwargs)
|
||
|
||
try:
|
||
response = client.recognize_passport_with_options(request, RuntimeOptions())
|
||
except OcrConfigError:
|
||
raise
|
||
except Exception as exc: # noqa: BLE001
|
||
raise OcrCallError(f"调用阿里云 OCR 失败: {exc}") from exc
|
||
|
||
body = getattr(response, "body", response)
|
||
data = _parse_data(body)
|
||
request_id = getattr(body, "request_id", None) or (
|
||
body.get("RequestId") if isinstance(body, dict) else None
|
||
)
|
||
if request_id and isinstance(data, dict):
|
||
data.setdefault("_request_id", request_id)
|
||
return data
|