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
|