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

"""易制毒购买证管理 API"""
from typing import Any, Literal, Optional

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, 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.core.tenant_context import get_tenant_no
from app.models.pl_models import PlPlan, PlPlanProduct
from app.schemas.pl.purchase_cert import (
    PlanProductCertLinksSaveBody,
    PurchaseCertCreate,
    PurchaseCertUpdate,
    TransportCertBody,
)
from app.services import pl_purchase_cert_service as svc
from app.services.pl_plan_service import confirm_toxic_plan_effective
from app.services.pl_purchase_cert_ocr_service import recognize_hazard_cert

router = APIRouter(prefix="/purchase-cert", tags=["运输计划-购买证管理"])


def _dept_id(user: DbUserContext) -> str:
    return getattr(user, "dept_id", None) or getattr(user, "create_dept_id", None) or ""


class RecognizeHazardCertBody(BaseModel):
    kind: Literal["purchase", "transport"] = Field(..., description="purchase-购买证 transport-运输证")
    attachment_url: str = Field(..., alias="attachmentUrl")

    model_config = {"populate_by_name": True}


@router.post("/recognize")
def recognize_cert_image(
    body: RecognizeHazardCertBody,
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    """DeepSeek 视觉识别购买证 / 运输证并返回可回填字段。"""
    _ = user
    return recognize_hazard_cert(kind=body.kind, attachment_url=body.attachment_url)


@router.get("/stats")
def purchase_cert_stats(
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    return svc.get_dashboard_stats(db, tenant_no=tenant_no)


@router.get("/pending-plans")
def pending_toxic_plans(
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    items = svc.list_pending_toxic_plans(db, tenant_no=tenant_no)
    return {"items": items, "list": items, "total": len(items)}


@router.get("/")
def list_purchase_certs(
    q: Optional[str] = Query(default=None),
    status: Optional[str] = Query(default=None),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=200, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    return svc.list_purchase_certs(
        db,
        tenant_no=tenant_no,
        q=q,
        status=status,
        page=page,
        page_size=page_size,
    )


@router.get("/{cert_id}")
def get_purchase_cert(
    cert_id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    return svc.get_purchase_cert_detail(db, tenant_no=tenant_no, cert_id=cert_id)


@router.post("/")
def create_purchase_cert(
    body: PurchaseCertCreate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    data = body.model_dump(exclude={"trans_list"}, exclude_none=True)
    transports = [t.model_dump(exclude_none=True) for t in (body.trans_list or [])]
    return svc.create_purchase_cert_with_transports(
        db,
        tenant_no=tenant_no,
        data=data,
        transports=transports,
        user_id=user.id,
        dept_id=_dept_id(user),
    )


@router.put("/{cert_id}")
def update_purchase_cert(
    cert_id: str,
    body: PurchaseCertUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    data = body.model_dump(exclude_none=True)
    return svc.update_purchase_cert(
        db,
        tenant_no=tenant_no,
        cert_id=cert_id,
        data=data,
        user_id=user.id,
    )


@router.delete("/{cert_id}")
def delete_purchase_cert(
    cert_id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, str]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    svc.soft_delete_purchase_cert(
        db, tenant_no=tenant_no, cert_id=cert_id, user_id=user.id
    )
    return {"message": "ok"}


@router.post("/{cert_id}/transport-certs")
def add_transport_cert(
    cert_id: str,
    body: TransportCertBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    return svc.add_transport_cert(
        db,
        tenant_no=tenant_no,
        purchase_cert_id=cert_id,
        data=body.model_dump(exclude_none=True),
        user_id=user.id,
        dept_id=_dept_id(user),
    )


@router.post("/links")
def save_plan_product_links(
    body: PlanProductCertLinksSaveBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    links = [lk.model_dump() for lk in body.links]
    items = svc.upsert_plan_product_links(
        db,
        tenant_no=tenant_no,
        plan_id=body.plan_id,
        links=links,
        user_id=user.id,
        dept_id=_dept_id(user),
    )
    return {"items": items, "total": len(items)}


@router.post("/confirm/{plan_id}")
def confirm_plan_with_certs(
    plan_id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict[str, Any]:
    """校验关联 → 占用 → 确认生效。"""
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = (
        db.query(PlPlan)
        .filter(
            PlPlan.id == plan_id,
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
        )
        .first()
    )
    if not plan:
        raise HTTPException(status_code=404, detail="计划不存在")
    products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == plan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        )
        .order_by(PlPlanProduct.order_num)
        .all()
    )
    user_name = getattr(user, "nick_name", None) or getattr(user, "user_no", None)
    plan = confirm_toxic_plan_effective(
        db,
        tenant_no=tenant_no,
        plan=plan,
        products=products,
        user_id=user.id,
        user_name=user_name,
        dept_id=_dept_id(user),
    )
    return {
        "id": plan.id,
        "planNo": plan.plan_no,
        "followStatus": plan.follow_status,
        "hazardCertReady": int(plan.hazard_cert_ready or 0),
    }