58 lines
2.0 KiB
Python
58 lines
2.0 KiB
Python
"""OCR 相关接口——目前供旅行社"名单"模块识别护照使用。"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
|
||
|
||
from app.aliyun_ocr import OcrCallError, OcrConfigError, recognize_passport
|
||
|
||
router = APIRouter()
|
||
|
||
MAX_IMAGE_BYTES = 10 * 1024 * 1024 # 阿里云接口单图上限按 10MB 兜底
|
||
|
||
|
||
@router.post("/ocr/passport")
|
||
async def ocr_passport(
|
||
file: UploadFile | None = File(default=None),
|
||
url: str | None = Form(default=None),
|
||
) -> dict[str, Any]:
|
||
"""护照识别。任选其一：
|
||
- multipart 上传 `file`（图片二进制）
|
||
- 表单字段 `url`（图片可公网访问的 URL，如阿里云 OSS）
|
||
|
||
成功返回 OCR 解析后的字段 dict，失败按错误类型映射到不同 HTTP 状态：
|
||
- 400：参数缺失/图片过大
|
||
- 503：AK/SK 未配置（运维介入）
|
||
- 502：阿里云返回错误或网络故障
|
||
"""
|
||
if file is None and not (url and url.strip()):
|
||
raise HTTPException(status_code=400, detail="请上传图片文件或提供图片 URL")
|
||
|
||
image_bytes: bytes | None = None
|
||
if file is not None:
|
||
image_bytes = await file.read()
|
||
if not image_bytes:
|
||
raise HTTPException(status_code=400, detail="上传的图片为空")
|
||
if len(image_bytes) > MAX_IMAGE_BYTES:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"图片过大（{len(image_bytes)} 字节），请压缩到 10MB 以内",
|
||
)
|
||
|
||
try:
|
||
data = recognize_passport(
|
||
image_bytes=image_bytes,
|
||
image_url=url.strip() if url else None,
|
||
)
|
||
except OcrConfigError as exc:
|
||
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
||
except OcrCallError as exc:
|
||
raise HTTPException(status_code=502, detail=str(exc)) from exc
|
||
|
||
return {
|
||
"ok": True,
|
||
"filename": file.filename if file else None,
|
||
"data": data,
|
||
}
|