144 lines
5.0 KiB
Python
144 lines
5.0 KiB
Python
#!/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()
|