Add Cloud Tour Libo knowledge graph platform

app/aliyun_ocr.py

@@ -0,0 +1,114 @@
+"""阿里云 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