shanxia2.0/backend/scripts/test_aliyun_driving_license_ocr.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""本地驾驶证图片 → 阿里云 RecognizeDrivingLicense 冒烟测试。

用法（在 backend 目录）:
  .ocr-venv/bin/python scripts/test_aliyun_driving_license_ocr.py
  .ocr-venv/bin/python scripts/test_aliyun_driving_license_ocr.py ../驾驶证1.png
"""
from __future__ import annotations

import io
import json
import os
import re
import sys
from pathlib import Path

from dotenv import load_dotenv
from PIL import Image

try:
    from alibabacloud_ocr_api20210707.client import Client as OcrClient
    from alibabacloud_ocr_api20210707 import models as ocr_models
    from alibabacloud_tea_openapi import models as open_api_models
    from alibabacloud_tea_util import models as util_models
except ImportError as exc:
    print('请先安装 SDK: .ocr-venv/bin/pip install alibabacloud_ocr_api20210707 -i https://pypi.tuna.tsinghua.edu.cn/simple')
    raise SystemExit(1) from exc

BACKEND_DIR = Path(__file__).resolve().parents[1]
DEFAULT_IMAGE = BACKEND_DIR.parent / '驾驶证1.png'
MAX_UPLOAD_BYTES = 1_500_000


def load_ocr_settings() -> tuple[str, str, str]:
    load_dotenv(BACKEND_DIR / '.env')
    ak = (os.getenv('ALIYUN_OCR_ACCESS_KEY_ID') or '').strip()
    sk = (os.getenv('ALIYUN_OCR_ACCESS_KEY_SECRET') or '').strip()
    endpoint = (os.getenv('ALIYUN_OCR_ENDPOINT') or 'ocr-api.cn-hangzhou.aliyuncs.com').strip()
    if not ak or not sk:
        raise SystemExit('未配置 ALIYUN_OCR_ACCESS_KEY_ID / ALIYUN_OCR_ACCESS_KEY_SECRET')
    return ak, sk, endpoint


def compress_image(path: Path) -> bytes:
    raw = path.read_bytes()
    if len(raw) <= MAX_UPLOAD_BYTES:
        print(f'图片大小 {len(raw)} bytes，直接上传')
        return raw

    print(f'图片较大 ({len(raw)} bytes)，压缩到约 {MAX_UPLOAD_BYTES} bytes 以内…')
    img = Image.open(io.BytesIO(raw))
    if img.mode not in ('RGB', 'L'):
        img = img.convert('RGB')

    quality = 85
    width, height = img.size
    buf = io.BytesIO()
    while True:
        buf.seek(0)
        buf.truncate(0)
        img.save(buf, format='JPEG', quality=quality, optimize=True)
        size = buf.tell()
        if size <= MAX_UPLOAD_BYTES or quality <= 40:
            print(f'压缩后 {size} bytes (quality={quality}, size={img.size})')
            return buf.getvalue()
        quality -= 10
        if size > MAX_UPLOAD_BYTES * 1.5:
            width = int(width * 0.85)
            height = int(height * 0.85)
            img = img.resize((width, height), Image.Resampling.LANCZOS)


def parse_expire_date(valid_period: str | None) -> str | None:
    if not valid_period:
        return None
    text = str(valid_period).strip()
    if '长期' in text:
        return None
    # 支持: 2017-01-26至2027-01-26 / 2017.01.26-2027.01.26 / 2017.01.26至2027.01.26
    m = re.search(
        r'(\d{4})[.\-/年](\d{1,2})[.\-/月](\d{1,2}).*?[至\-—~]\s*(\d{4})[.\-/年](\d{1,2})[.\-/月](\d{1,2})',
        text,
    )
    if m:
        y, mo, d = m.group(4), m.group(5), m.group(6)
        return f'{int(y):04d}-{int(mo):02d}-{int(d):02d}'
    m2 = re.search(r'(\d{4})[.\-/](\d{1,2})[.\-/](\d{1,2})\s*$', text)
    if m2:
        return f'{int(m2.group(1)):04d}-{int(m2.group(2)):02d}-{int(m2.group(3)):02d}'
    return None


def main() -> None:
    image_path = Path(sys.argv[1]).expanduser().resolve() if len(sys.argv) > 1 else DEFAULT_IMAGE
    if not image_path.is_file():
        raise SystemExit(f'图片不存在: {image_path}')

    ak, sk, endpoint = load_ocr_settings()
    image_bytes = compress_image(image_path)

    client = OcrClient(
        open_api_models.Config(
            access_key_id=ak,
            access_key_secret=sk,
            endpoint=endpoint,
        )
    )

    request = ocr_models.RecognizeDrivingLicenseRequest(body=io.BytesIO(image_bytes))
    runtime = util_models.RuntimeOptions()
    print(f'调用 RecognizeDrivingLicense … endpoint={endpoint}')
    resp = client.recognize_driving_license_with_options(request, runtime)

    body = resp.body
    if getattr(body, 'code', None):
        print('识别失败:')
        print('  code:', body.code)
        print('  message:', body.message)
        print('  requestId:', getattr(body, 'request_id', None))
        raise SystemExit(2)

    raw_data = body.data
    data = json.loads(raw_data) if isinstance(raw_data, str) else raw_data
    print('\n===== 原始 Data =====')
    print(json.dumps(data, ensure_ascii=False, indent=2))

    face = (((data or {}).get('data') or {}).get('face') or {}).get('data') or {}
    cert_no = face.get('licenseNumber')
    valid_period = face.get('validPeriod')
    expire_date = parse_expire_date(valid_period)

    print('\n===== 业务回填字段（对应证书表）=====')
    print('证号 certNo     :', cert_no)
    print('有效期原文       :', valid_period)
    print('有效期 expireDate:', expire_date)
    print('姓名             :', face.get('name'))
    print('准驾类型         :', face.get('approvedType'))
    print('requestId        :', getattr(body, 'request_id', None))


if __name__ == '__main__':
    main()