shanxia2.0/backend/app/api/pl/ocr.py

"""运输计划 - 人员证书 OCR"""
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.db_auth import DbUserContext, get_db_current_user
from app.services.pl_ocr_service import recognize_personnel_cert

router = APIRouter(prefix="/ocr", tags=["运输计划-OCR"])


class RecognizeCertBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    cert_name: str = Field(..., alias="certName")
    attachment_url: Optional[str] = Field(None, alias="attachmentUrl")
    attachment_urls: Optional[str] = Field(None, alias="attachmentUrls")


class RecognizeCertResult(BaseModel):
    model_config = ConfigDict(populate_by_name=True, ser_json_by_alias=True)

    cert_no: Optional[str] = Field(None, alias="certNo")
    expire_date: Optional[str] = Field(None, alias="expireDate")
    expire_date_raw: Optional[str] = Field(None, alias="expireDateRaw")
    monthly_used: int = Field(..., alias="monthlyUsed")
    monthly_limit: int = Field(..., alias="monthlyLimit")


@router.post("/recognize-cert")
def recognize_cert(
    body: RecognizeCertBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict:
    _ = user
    urls = (body.attachment_urls or body.attachment_url or "").strip()
    if not urls:
        raise HTTPException(status_code=400, detail="请先上传证书附件")
    result = recognize_personnel_cert(
        db,
        cert_name=body.cert_name,
        attachment_urls=urls,
    )
    return {
        "certNo": result.get("certNo"),
        "expireDate": result.get("expireDate"),
        "expireDateRaw": result.get("expireDateRaw"),
        "monthlyUsed": result["monthlyUsed"],
        "monthlyLimit": result["monthlyLimit"],
    }