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

"""运输计划管理 - 基础数据 API（承运商、车辆、人员）"""
from datetime import datetime
from typing import List, Optional

from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel, Field
from sqlalchemy import case, func, or_
from sqlalchemy.orm import Session

from app.utils.excel_export import (
    EXPORT_MAX_ROWS,
    ExportColumn,
    ExportValidationError,
    export_rows_to_xlsx_response,
    filter_export_columns,
)

from app.core.database import get_db
from app.core.db_auth import DbUserContext, get_db_current_user
from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
from app.core.tenant_context import get_tenant_no
from app.core.utils import generate_uuid
from app.models.pl_models import (
    PlCarrier,
    PlPersonnel,
    PlVehicle,
    PlVehicleDriver,
    PlVehiclePersonnel,
)
from app.schemas.pl import (
    CarrierAuditReject,
    CarrierCreate,
    CarrierListResponse,
    CarrierRead,
    CarrierUpdate,
    carrier_to_read,
    CertRead,
    DriverCreate,
    DriverListResponse,
    DriverRead,
    DriverUpdate,
    PersonnelCreate,
    PersonnelDetailRead,
    PersonnelListResponse,
    PersonnelRead,
    PersonnelUpdate,
    VehicleCreate,
    VehicleDetailRead,
    VehicleDriverCreate,
    VehicleListResponse,
    VehicleRead,
    VehicleUpdate,
)
from app.services.pl_cert_service import (
    certs_to_read,
    list_owner_certs,
    list_owner_certs_batch,
    nearest_expire_date,
    save_owner_certs,
)
from app.services.pl_cert_required_service import (
    validate_personnel_required_certs,
    validate_vehicle_save_certs,
)
from app.services.pl_carrier_service import carrier_eligible_for_transport, normalize_carrier_urls
from app.services.pl_personnel_service import (
    load_vehicle_personnel_rows,
    load_vehicle_personnel_summaries,
    save_vehicle_personnel,
)
from app.schemas.pl.paste_recognize import PasteCommitBody, PasteCommitResult
from app.services.pl_paste_recognize_service import commit_paste_recognize

router = APIRouter(prefix="", tags=["运输计划-基础数据"])


# =============================================================================
# 承运商 API
# =============================================================================


@router.get("/carrier", response_model=CarrierListResponse)
def list_carriers(
    q: Optional[str] = Query(default=None, description="模糊搜索"),
    status: Optional[str] = Query(default=None, description="状态"),
    audit_status: Optional[str] = Query(default=None, alias="auditStatus", description="审核状态"),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> CarrierListResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlCarrier).filter(
        PlCarrier.tenant_no == tenant_no,
        PlCarrier.del_flag == 0,
    )

    if status:
        query = query.filter(PlCarrier.status == status)

    if audit_status:
        query = query.filter(PlCarrier.audit_status == audit_status)

    if q:
        query = query.filter(
            or_(
                func.lower(PlCarrier.carrier_name).contains(q.lower()),
                func.lower(PlCarrier.contact_name).contains(q.lower()),
            )
        )

    # 承运商为物流主数据，列表不做 create_by 数据范围过滤
    query = query.order_by(
        case((PlCarrier.order_num.is_(None), 1), else_=0).asc(),
        PlCarrier.create_time.desc(),
    )
    total = query.count()
    offset = (page - 1) * page_size
    rows = query.offset(offset).limit(page_size).all()
    return CarrierListResponse(
        total=total,
        list=[carrier_to_read(row) for row in rows],
    )




CARRIER_EXPORT_COLUMNS = [
    ExportColumn(title="承运商名称", field="carrierName"),
    ExportColumn(title="联系人", field="contactName"),
    ExportColumn(title="联系电话", field="contactPhone"),
    ExportColumn(title="合作状态", field="statusLabel"),
    ExportColumn(title="资质提示", field="qualificationHint"),
    ExportColumn(title="协议有效期", field="transportAgreementExpireDate"),
    ExportColumn(title="备注", field="remark"),
]


class CarrierExportBody(BaseModel):
    q: Optional[str] = None
    status: Optional[str] = None
    audit_status: Optional[str] = Field(default=None, alias="auditStatus")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None

    model_config = {"populate_by_name": True}


@router.post("/carrier/export")
def export_carriers(
    body: CarrierExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """导出承运商列表（当前筛选全量，或 ids 已选）。"""
    try:
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        query = db.query(PlCarrier).filter(
            PlCarrier.tenant_no == tenant_no,
            PlCarrier.del_flag == 0,
        )
        if body.status:
            query = query.filter(PlCarrier.status == body.status)
        if body.audit_status:
            query = query.filter(PlCarrier.audit_status == body.audit_status)
        if body.q:
            query = query.filter(
                or_(
                    func.lower(PlCarrier.carrier_name).contains(body.q.lower()),
                    func.lower(PlCarrier.contact_name).contains(body.q.lower()),
                )
            )
        if body.ids:
            query = query.filter(PlCarrier.id.in_(body.ids))
        query = query.order_by(
            case((PlCarrier.order_num.is_(None), 1), else_=0).asc(),
            PlCarrier.create_time.desc(),
        )
        rows = query.limit(EXPORT_MAX_ROWS + 1).all()
        items = []
        for row in rows:
            data = carrier_to_read(row).model_dump(by_alias=True)
            data["statusLabel"] = "合作中" if str(data.get("status")) == "1" else "已终止"
            if data.get("transportAgreementExpired"):
                data["qualificationHint"] = "协议已过期"
            elif data.get("qualificationComplete") is False:
                data["qualificationHint"] = "资质不全"
            else:
                data["qualificationHint"] = "资质齐全"
            items.append(data)
        requested = None
        if body.columns:
            requested = []
            for key in body.columns:
                if key in (
                    "qualificationAttachments",
                    "qualificationAttachment",
                    "photoUrls",
                ):
                    continue
                if key == "status":
                    requested.append("statusLabel")
                elif key == "qualificationHint":
                    requested.append("qualificationHint")
                else:
                    requested.append(key)
        cols = filter_export_columns(CARRIER_EXPORT_COLUMNS, requested)
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="承运商列表",
            columns=cols,
            rows=items,
            filename_stem=f"承运商列表_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e


@router.post("/carrier", response_model=CarrierRead, status_code=status.HTTP_201_CREATED)
def create_carrier(
    payload: CarrierCreate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> CarrierRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    carrier = PlCarrier(
        id=generate_uuid(),
        tenant_no=tenant_no,
        carrier_name=payload.carrier_name,
        contact_name=payload.contact_name,
        contact_phone=payload.contact_phone,
        status=payload.status,
        business_license_urls=normalize_carrier_urls(payload.business_license_urls),
        road_transport_license_urls=normalize_carrier_urls(payload.road_transport_license_urls),
        transport_agreement_urls=normalize_carrier_urls(payload.transport_agreement_urls),
        transport_agreement_expire_date=payload.transport_agreement_expire_date,
        audit_status="pending",
        remark=payload.remark,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(carrier)
    db.commit()
    db.refresh(carrier)
    return carrier_to_read(carrier)


@router.get("/carrier/{id}", response_model=CarrierRead)
def get_carrier(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> CarrierRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    carrier = _get_carrier(db, tenant_no, id, user)
    return carrier_to_read(carrier)


@router.put("/carrier/{id}", response_model=CarrierRead)
def update_carrier(
    id: str,
    payload: CarrierUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> CarrierRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    carrier = _get_carrier(db, tenant_no, id, user)

    payload_data = payload.model_dump(exclude_unset=True)

    for key in (
        "business_license_urls",
        "road_transport_license_urls",
        "transport_agreement_urls",
    ):
        if key in payload_data:
            payload_data[key] = normalize_carrier_urls(payload_data[key])

    for key, value in payload_data.items():
        setattr(carrier, key, value)

    carrier.update_by = user.id
    db.commit()
    db.refresh(carrier)
    return carrier_to_read(carrier)


@router.delete("/carrier/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_carrier(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> None:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    carrier = _get_carrier(db, tenant_no, id, user)
    assert_row_deletable_by_creator(carrier, user_info_from_context(user))
    carrier.del_flag = 1
    carrier.update_by = user.id
    db.commit()


@router.post("/carrier/{id}/approve", response_model=CarrierRead)
def approve_carrier(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> CarrierRead:
    raise HTTPException(status_code=400, detail="承运商审核流程已停用")


@router.post("/carrier/{id}/reject", response_model=CarrierRead)
def reject_carrier(
    id: str,
    payload: CarrierAuditReject,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> CarrierRead:
    raise HTTPException(status_code=400, detail="承运商审核流程已停用")


def _get_carrier(db: Session, tenant_no: str, carrier_id: str, user: DbUserContext) -> PlCarrier:
    carrier = (
        db.query(PlCarrier)
        .filter(
            PlCarrier.id == carrier_id,
            PlCarrier.tenant_no == tenant_no,
            PlCarrier.del_flag == 0,
        )
        .first()
    )
    if not carrier:
        raise HTTPException(status_code=404, detail="承运商不存在")
    return carrier


# =============================================================================
# 车辆 API
# =============================================================================


def _vehicle_to_read(
    vehicle: PlVehicle,
    summary: Optional[str] = None,
    default_driver_name: Optional[str] = None,
    *,
    cert_list: Optional[list[CertRead]] = None,
) -> VehicleRead:
    data = VehicleRead.model_validate(vehicle)
    if summary is not None:
        data.personnel_summary = summary
    if default_driver_name is not None:
        data.default_driver_name = default_driver_name
    if cert_list is not None:
        data.cert_list = cert_list
    return data


def _vehicle_to_detail(
    db: Session, tenant_no: str, vehicle: PlVehicle
) -> VehicleDetailRead:
    summaries = load_vehicle_personnel_summaries(db, tenant_no, [vehicle.id])
    summary, default_name = summaries.get(vehicle.id, ("", None))
    personnel_list = load_vehicle_personnel_rows(db, tenant_no, vehicle.id)
    cert_list = certs_to_read(
        list_owner_certs(db, tenant_no, "vehicle", vehicle.id)
    )
    base = _vehicle_to_read(vehicle, summary, default_name)
    return VehicleDetailRead(
        **base.model_dump(by_alias=False, exclude={"cert_list"}),
        personnel_list=personnel_list,
        cert_list=cert_list,
    )


@router.get("/vehicle", response_model=VehicleListResponse)
def list_vehicles(
    q: Optional[str] = Query(default=None, description="模糊搜索车牌号"),
    vehicle_type: Optional[str] = Query(default=None, alias="vehicleType", description="车辆类型"),
    carrier_id: Optional[str] = Query(default=None, alias="carrierId", description="承运商ID"),
    vehicle_status: Optional[str] = Query(default=None, alias="vehicleStatus", description="车辆状态"),
    exclude_ineligible_carrier: Optional[bool] = Query(
        default=False,
        alias="excludeIneligibleCarrier",
        description="排除资质不全或协议过期的承运商车辆",
    ),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> VehicleListResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlVehicle).filter(
        PlVehicle.tenant_no == tenant_no,
        PlVehicle.del_flag == 0,
    )

    if vehicle_type:
        query = query.filter(PlVehicle.vehicle_type == vehicle_type)

    if carrier_id:
        query = query.filter(PlVehicle.carrier_id == carrier_id)

    if vehicle_status:
        query = query.filter(PlVehicle.vehicle_status == vehicle_status)

    if q:
        query = query.filter(PlVehicle.license_plate.contains(q))

    # 车辆为主数据，列表不做 create_by 数据范围过滤
    query = query.order_by(
        case((PlVehicle.order_num.is_(None), 1), else_=0).asc(),
        PlVehicle.create_time.desc(),
    )

    if exclude_ineligible_carrier:
        vehicles = query.all()
        carrier_ids = {
            v.carrier_id for v in vehicles if v.vehicle_type == "carrier" and v.carrier_id
        }
        carrier_map: dict[str, PlCarrier] = {}
        if carrier_ids:
            carriers = (
                db.query(PlCarrier)
                .filter(
                    PlCarrier.id.in_(carrier_ids),
                    PlCarrier.tenant_no == tenant_no,
                    PlCarrier.del_flag == 0,
                )
                .all()
            )
            carrier_map = {c.id: c for c in carriers}
        vehicles = [
            v
            for v in vehicles
            if v.vehicle_type != "carrier"
            or not v.carrier_id
            or (
                v.carrier_id in carrier_map
                and carrier_eligible_for_transport(carrier_map[v.carrier_id])
            )
        ]
        summaries = load_vehicle_personnel_summaries(
            db, tenant_no, [v.id for v in vehicles]
        )
        certs_by_owner = list_owner_certs_batch(
            db, tenant_no, "vehicle", [v.id for v in vehicles]
        )
        items = [
            _vehicle_to_read(
                v,
                *summaries.get(v.id, ("", None)),
                cert_list=certs_to_read(certs_by_owner.get(v.id, [])),
            )
            for v in vehicles
        ]
        return VehicleListResponse(total=len(items), list=items)

    total = query.count()
    offset = (page - 1) * page_size
    vehicles = query.offset(offset).limit(page_size).all()
    summaries = load_vehicle_personnel_summaries(
        db, tenant_no, [v.id for v in vehicles]
    )
    certs_by_owner = list_owner_certs_batch(
        db, tenant_no, "vehicle", [v.id for v in vehicles]
    )
    return VehicleListResponse(
        total=total,
        list=[
            _vehicle_to_read(
                v,
                *summaries.get(v.id, ("", None)),
                cert_list=certs_to_read(certs_by_owner.get(v.id, [])),
            )
            for v in vehicles
        ],
    )



VEHICLE_EXPORT_COLUMNS = [
    ExportColumn(title="车牌号", field="licensePlate"),
    ExportColumn(title="关联人员", field="personnelSummary"),
    ExportColumn(title="车辆类型", field="vehicleTypeLabel"),
    ExportColumn(title="车辆状态", field="vehicleStatusLabel"),
    ExportColumn(title="载重(吨)", field="loadCapacity"),
    ExportColumn(title="承运商", field="carrierName"),
    ExportColumn(title="备注", field="remark"),
]

_VEHICLE_TYPE_LABEL = {"self": "自有车", "carrier": "承运商车"}
_VEHICLE_STATUS_LABEL = {"1": "启用", "0": "停用", "idle": "空闲", "busy": "在途", "maintain": "维保"}


class VehicleExportBody(BaseModel):
    q: Optional[str] = None
    vehicle_type: Optional[str] = Field(default=None, alias="vehicleType")
    carrier_id: Optional[str] = Field(default=None, alias="carrierId")
    vehicle_status: Optional[str] = Field(default=None, alias="vehicleStatus")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None

    model_config = {"populate_by_name": True}


@router.post("/vehicle/export")
def export_vehicles(
    body: VehicleExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """导出车辆列表（当前筛选全量，或 ids 已选）。"""
    try:
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        query = db.query(PlVehicle).filter(
            PlVehicle.tenant_no == tenant_no,
            PlVehicle.del_flag == 0,
        )
        if body.vehicle_type:
            query = query.filter(PlVehicle.vehicle_type == body.vehicle_type)
        if body.carrier_id:
            query = query.filter(PlVehicle.carrier_id == body.carrier_id)
        if body.vehicle_status:
            query = query.filter(PlVehicle.vehicle_status == body.vehicle_status)
        if body.q:
            query = query.filter(PlVehicle.license_plate.contains(body.q))
        if body.ids:
            query = query.filter(PlVehicle.id.in_(body.ids))
        query = query.order_by(
            case((PlVehicle.order_num.is_(None), 1), else_=0).asc(),
            PlVehicle.create_time.desc(),
        )
        vehicles = query.limit(EXPORT_MAX_ROWS + 1).all()
        summaries = load_vehicle_personnel_summaries(db, tenant_no, [v.id for v in vehicles])
        items = []
        for v in vehicles:
            data = _vehicle_to_read(v, *summaries.get(v.id, ("", None))).model_dump(by_alias=True)
            vt = str(data.get("vehicleType") or "")
            data["vehicleTypeLabel"] = _VEHICLE_TYPE_LABEL.get(vt, vt)
            vs = str(data.get("vehicleStatus") or "")
            data["vehicleStatusLabel"] = _VEHICLE_STATUS_LABEL.get(vs, vs)
            items.append(data)
        requested = None
        if body.columns:
            requested = []
            for key in body.columns:
                if key in ("photoUrls", "certAttachments", "certList"):
                    continue
                if key == "vehicleType":
                    requested.append("vehicleTypeLabel")
                elif key == "vehicleStatus":
                    requested.append("vehicleStatusLabel")
                else:
                    requested.append(key)
        cols = filter_export_columns(VEHICLE_EXPORT_COLUMNS, requested)
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="车辆列表",
            columns=cols,
            rows=items,
            filename_stem=f"车辆列表_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e



@router.post("/vehicle", response_model=VehicleDetailRead, status_code=status.HTTP_201_CREATED)
def create_vehicle(
    payload: VehicleCreate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> VehicleDetailRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    existing = (
        db.query(PlVehicle)
        .filter(
            PlVehicle.license_plate == payload.license_plate,
            PlVehicle.del_flag == 0,
        )
        .first()
    )
    if existing:
        raise HTTPException(status_code=400, detail="车牌号已存在")

    validate_vehicle_save_certs(
        db,
        tenant_no,
        vehicle_type=payload.vehicle_type,
        trailer_no=payload.trailer_no,
        cert_rows=payload.cert_list,
        personnel_rows=payload.personnel_list,
    )

    vehicle = PlVehicle(
        id=generate_uuid(),
        tenant_no=tenant_no,
        vehicle_type=payload.vehicle_type,
        carrier_id=payload.carrier_id,
        carrier_name=payload.carrier_name,
        license_plate=payload.license_plate,
        trailer_no=payload.trailer_no,
        load_capacity=payload.load_capacity,
        vehicle_status=payload.vehicle_status,
        photo_urls=payload.photo_urls,
        remark=payload.remark,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(vehicle)
    db.flush()

    try:
        save_vehicle_personnel(
            db,
            vehicle.id,
            payload.personnel_list,
            tenant_no,
            user_id=user.id,
            dept_id=user.dept_id,
            vehicle_type=payload.vehicle_type,
        )
        save_owner_certs(
            db,
            "vehicle",
            vehicle.id,
            payload.cert_list,
            tenant_no,
            user_id=user.id,
            dept_id=user.dept_id,
        )
        db.commit()
    except HTTPException:
        db.rollback()
        raise
    except Exception:
        db.rollback()
        raise

    db.refresh(vehicle)
    return _vehicle_to_detail(db, tenant_no, vehicle)


@router.get("/vehicle/{id}", response_model=VehicleDetailRead)
def get_vehicle(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> VehicleDetailRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    vehicle = _get_vehicle(db, tenant_no, id, user)
    return _vehicle_to_detail(db, tenant_no, vehicle)


@router.put("/vehicle/{id}", response_model=VehicleDetailRead)
def update_vehicle(
    id: str,
    payload: VehicleUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> VehicleDetailRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    vehicle = _get_vehicle(db, tenant_no, id, user)

    personnel_list = payload.personnel_list
    cert_list = payload.cert_list
    data = payload.model_dump(
        exclude_unset=True, exclude={"personnel_list", "cert_list"}
    )

    effective_vehicle_type = data.get("vehicle_type", vehicle.vehicle_type)
    effective_trailer_no = data.get("trailer_no", vehicle.trailer_no)
    effective_cert_rows = (
        cert_list
        if cert_list is not None
        else list_owner_certs(db, tenant_no, "vehicle", vehicle.id)
    )
    effective_personnel_rows = personnel_list
    if effective_personnel_rows is None:
        effective_personnel_rows = load_vehicle_personnel_rows(db, tenant_no, vehicle.id)

    validate_vehicle_save_certs(
        db,
        tenant_no,
        vehicle_type=effective_vehicle_type,
        trailer_no=effective_trailer_no,
        cert_rows=effective_cert_rows,
        personnel_rows=effective_personnel_rows,
    )

    for key, value in data.items():
        setattr(vehicle, key, value)

    vehicle.update_by = user.id

    try:
        if personnel_list is not None:
            save_vehicle_personnel(
                db,
                vehicle.id,
                personnel_list,
                tenant_no,
                user_id=user.id,
                dept_id=user.dept_id,
                vehicle_type=effective_vehicle_type,
            )
        if cert_list is not None:
            save_owner_certs(
                db,
                "vehicle",
                vehicle.id,
                cert_list,
                tenant_no,
                user_id=user.id,
                dept_id=user.dept_id,
            )
        db.commit()
    except HTTPException:
        db.rollback()
        raise
    except Exception:
        db.rollback()
        raise

    db.refresh(vehicle)
    return _vehicle_to_detail(db, tenant_no, vehicle)


@router.delete("/vehicle/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_vehicle(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> None:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    vehicle = _get_vehicle(db, tenant_no, id, user)
    assert_row_deletable_by_creator(vehicle, user_info_from_context(user))
    vehicle.del_flag = 1
    vehicle.update_by = user.id
    db.commit()


def _get_vehicle(db: Session, tenant_no: str, vehicle_id: str, user: DbUserContext) -> PlVehicle:
    vehicle = (
        db.query(PlVehicle)
        .filter(
            PlVehicle.id == vehicle_id,
            PlVehicle.tenant_no == tenant_no,
            PlVehicle.del_flag == 0,
        )
        .first()
    )
    if not vehicle:
        raise HTTPException(status_code=404, detail="车辆不存在")
    return vehicle


# =============================================================================
# 人员 API
# =============================================================================


def _personnel_to_read(
    db: Session,
    tenant_no: str,
    personnel: PlPersonnel,
    *,
    cert_list: Optional[list[CertRead]] = None,
) -> PersonnelRead:
    nearest = nearest_expire_date(db, tenant_no, "personnel", personnel.id)
    data = PersonnelRead.model_validate(personnel)
    data.nearest_expire_date = nearest
    if cert_list is not None:
        data.cert_list = cert_list
    return data


def _personnel_to_detail(
    db: Session, tenant_no: str, personnel: PlPersonnel
) -> PersonnelDetailRead:
    base = _personnel_to_read(db, tenant_no, personnel)
    cert_list = certs_to_read(
        list_owner_certs(db, tenant_no, "personnel", personnel.id)
    )
    return PersonnelDetailRead(
        **base.model_dump(by_alias=False, exclude={"cert_list"}),
        cert_list=cert_list,
    )


@router.get("/personnel", response_model=PersonnelListResponse)
def list_personnel(
    q: Optional[str] = Query(default=None, description="模糊搜索姓名/手机号"),
    status: Optional[str] = Query(default=None, description="状态"),
    personnel_affiliation: Optional[str] = Query(
        default=None, alias="personnelAffiliation", description="人员归属 self/carrier"
    ),
    personnel_type: Optional[str] = Query(
        default=None, alias="personnelType", description="人员类型 driver/escort"
    ),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PersonnelListResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlPersonnel).filter(
        PlPersonnel.tenant_no == tenant_no,
        PlPersonnel.del_flag == 0,
    )

    if status:
        query = query.filter(PlPersonnel.status == status)

    if personnel_affiliation:
        query = query.filter(PlPersonnel.personnel_affiliation == personnel_affiliation)

    if personnel_type:
        query = query.filter(PlPersonnel.personnel_type == personnel_type)

    if q:
        query = query.filter(
            or_(
                func.lower(PlPersonnel.personnel_name).contains(q.lower()),
                PlPersonnel.phone_no.contains(q),
            )
        )

    # 人员为主数据，列表不做 create_by 数据范围过滤
    query = query.order_by(
        case((PlPersonnel.order_num.is_(None), 1), else_=0).asc(),
        PlPersonnel.create_time.desc(),
    )
    total = query.count()
    offset = (page - 1) * page_size
    rows = query.offset(offset).limit(page_size).all()
    personnel_ids = [p.id for p in rows]
    certs_by_owner = list_owner_certs_batch(db, tenant_no, "personnel", personnel_ids)
    return PersonnelListResponse(
        total=total,
        list=[
            _personnel_to_read(
                db,
                tenant_no,
                p,
                cert_list=certs_to_read(certs_by_owner.get(p.id, [])),
            )
            for p in rows
        ],
    )



PERSONNEL_EXPORT_COLUMNS = [
    ExportColumn(title="姓名", field="personnelName"),
    ExportColumn(title="人员归属", field="personnelAffiliationLabel"),
    ExportColumn(title="人员类型", field="personnelTypeLabel"),
    ExportColumn(title="手机号", field="phoneNo"),
    ExportColumn(title="年龄", field="age"),
    ExportColumn(title="身份证号", field="idCardNo"),
    ExportColumn(title="状态", field="statusLabel"),
    ExportColumn(title="最近到期", field="nearestExpireDate"),
    ExportColumn(title="备注", field="remark"),
]

_AFFILIATION_LABEL = {"self": "自有", "carrier": "承运商"}
_PERSONNEL_TYPE_LABEL = {"driver": "司机", "escort": "押运员"}


class PersonnelExportBody(BaseModel):
    q: Optional[str] = None
    status: Optional[str] = None
    personnel_affiliation: Optional[str] = Field(default=None, alias="personnelAffiliation")
    personnel_type: Optional[str] = Field(default=None, alias="personnelType")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None

    model_config = {"populate_by_name": True}


@router.post("/personnel/export")
def export_personnel(
    body: PersonnelExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """导出人员列表（当前筛选全量，或 ids 已选）。"""
    try:
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        query = db.query(PlPersonnel).filter(
            PlPersonnel.tenant_no == tenant_no,
            PlPersonnel.del_flag == 0,
        )
        if body.status:
            query = query.filter(PlPersonnel.status == body.status)
        if body.personnel_affiliation:
            query = query.filter(PlPersonnel.personnel_affiliation == body.personnel_affiliation)
        if body.personnel_type:
            query = query.filter(PlPersonnel.personnel_type == body.personnel_type)
        if body.q:
            query = query.filter(
                or_(
                    func.lower(PlPersonnel.personnel_name).contains(body.q.lower()),
                    PlPersonnel.phone_no.contains(body.q),
                )
            )
        if body.ids:
            query = query.filter(PlPersonnel.id.in_(body.ids))
        query = query.order_by(
            case((PlPersonnel.order_num.is_(None), 1), else_=0).asc(),
            PlPersonnel.create_time.desc(),
        )
        rows = query.limit(EXPORT_MAX_ROWS + 1).all()
        items = []
        for p in rows:
            data = _personnel_to_read(db, tenant_no, p).model_dump(by_alias=True)
            af = str(data.get("personnelAffiliation") or "")
            data["personnelAffiliationLabel"] = _AFFILIATION_LABEL.get(af, af)
            pt = str(data.get("personnelType") or "")
            data["personnelTypeLabel"] = _PERSONNEL_TYPE_LABEL.get(pt, pt)
            st = str(data.get("status") or "")
            data["statusLabel"] = "启用" if st == "1" else ("停用" if st == "0" else st)
            items.append(data)
        requested = None
        if body.columns:
            requested = []
            for key in body.columns:
                if key in ("photoUrls", "certAttachments", "certList"):
                    continue
                if key == "personnelAffiliation":
                    requested.append("personnelAffiliationLabel")
                elif key == "personnelType":
                    requested.append("personnelTypeLabel")
                elif key == "status":
                    requested.append("statusLabel")
                else:
                    requested.append(key)
        cols = filter_export_columns(PERSONNEL_EXPORT_COLUMNS, requested)
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="人员列表",
            columns=cols,
            rows=items,
            filename_stem=f"人员列表_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e



@router.post("/personnel", response_model=PersonnelDetailRead, status_code=status.HTTP_201_CREATED)
def create_personnel(
    payload: PersonnelCreate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PersonnelDetailRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    validate_personnel_required_certs(
        payload.cert_list,
        payload.personnel_type,
        personnel_affiliation=payload.personnel_affiliation,
        subject_label=f"「{payload.personnel_name}」",
    )

    personnel = PlPersonnel(
        id=generate_uuid(),
        tenant_no=tenant_no,
        personnel_name=payload.personnel_name,
        phone_no=payload.phone_no,
        age=payload.age,
        id_card_no=payload.id_card_no,
        personnel_type=payload.personnel_type,
        personnel_affiliation=payload.personnel_affiliation,
        status=payload.status,
        photo_urls=payload.photo_urls,
        remark=payload.remark,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(personnel)
    db.flush()

    try:
        save_owner_certs(
            db,
            "personnel",
            personnel.id,
            payload.cert_list,
            tenant_no,
            user_id=user.id,
            dept_id=user.dept_id,
        )
        db.commit()
    except Exception:
        db.rollback()
        raise

    db.refresh(personnel)
    return _personnel_to_detail(db, tenant_no, personnel)


@router.get("/personnel/{id}", response_model=PersonnelDetailRead)
def get_personnel(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PersonnelDetailRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    personnel = _get_personnel(db, tenant_no, id, user)
    return _personnel_to_detail(db, tenant_no, personnel)


@router.put("/personnel/{id}", response_model=PersonnelDetailRead)
def update_personnel(
    id: str,
    payload: PersonnelUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PersonnelDetailRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    personnel = _get_personnel(db, tenant_no, id, user)

    cert_list = payload.cert_list
    data = payload.model_dump(exclude_unset=True, exclude={"cert_list"})

    effective_personnel_type = data.get("personnel_type", personnel.personnel_type)
    effective_personnel_affiliation = data.get(
        "personnel_affiliation", personnel.personnel_affiliation
    )
    effective_cert_rows = (
        cert_list
        if cert_list is not None
        else list_owner_certs(db, tenant_no, "personnel", personnel.id)
    )
    subject_name = data.get("personnel_name", personnel.personnel_name)

    validate_personnel_required_certs(
        effective_cert_rows,
        effective_personnel_type,
        personnel_affiliation=effective_personnel_affiliation,
        subject_label=f"「{subject_name}」",
    )

    for key, value in data.items():
        setattr(personnel, key, value)

    personnel.update_by = user.id

    try:
        if cert_list is not None:
            save_owner_certs(
                db,
                "personnel",
                personnel.id,
                cert_list,
                tenant_no,
                user_id=user.id,
                dept_id=user.dept_id,
            )
        db.commit()
    except Exception:
        db.rollback()
        raise

    db.refresh(personnel)
    return _personnel_to_detail(db, tenant_no, personnel)


@router.delete("/personnel/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_personnel(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> None:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    personnel = _get_personnel(db, tenant_no, id, user)
    assert_row_deletable_by_creator(personnel, user_info_from_context(user))
    personnel.del_flag = 1
    personnel.update_by = user.id
    db.commit()


def _get_personnel(db: Session, tenant_no: str, personnel_id: str, user: DbUserContext) -> PlPersonnel:
    personnel = (
        db.query(PlPersonnel)
        .filter(
            PlPersonnel.id == personnel_id,
            PlPersonnel.tenant_no == tenant_no,
            PlPersonnel.del_flag == 0,
        )
        .first()
    )
    if not personnel:
        raise HTTPException(status_code=404, detail="人员不存在")
    return personnel


# =============================================================================
# 司机 API（兼容别名，内部使用 pl_personnel）
# =============================================================================


def _personnel_to_driver_read(personnel: PlPersonnel) -> DriverRead:
    return DriverRead(
        id=personnel.id,
        driver_name=personnel.personnel_name,
        phone_no=personnel.phone_no,
        id_card_no=personnel.id_card_no,
        tenant_no=personnel.tenant_no,
        order_num=personnel.order_num,
        remark=personnel.remark,
        create_by=personnel.create_by,
        create_dept_id=personnel.create_dept_id,
        create_time=personnel.create_time,
        update_by=personnel.update_by,
        update_time=personnel.update_time,
        del_flag=personnel.del_flag,
    )


@router.get("/driver", response_model=DriverListResponse)
def list_drivers(
    q: Optional[str] = Query(default=None, description="模糊搜索姓名/手机号"),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> DriverListResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlPersonnel).filter(
        PlPersonnel.tenant_no == tenant_no,
        PlPersonnel.del_flag == 0,
    )

    if q:
        query = query.filter(
            or_(
                func.lower(PlPersonnel.personnel_name).contains(q.lower()),
                PlPersonnel.phone_no.contains(q),
            )
        )

    # 司机列表为主数据选项，不做 create_by 数据范围过滤
    query = query.order_by(
        case((PlPersonnel.order_num.is_(None), 1), else_=0).asc(),
        PlPersonnel.create_time.desc(),
    )
    total = query.count()
    offset = (page - 1) * page_size
    rows = query.offset(offset).limit(page_size).all()
    return DriverListResponse(
        total=total,
        list=[_personnel_to_driver_read(p) for p in rows],
    )



DRIVER_EXPORT_COLUMNS = [
    ExportColumn(title="司机姓名", field="driverName"),
    ExportColumn(title="手机号", field="phoneNo"),
    ExportColumn(title="身份证号", field="idCardNo"),
    ExportColumn(title="备注", field="remark"),
]


class DriverExportBody(BaseModel):
    q: Optional[str] = None
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None


@router.post("/driver/export")
def export_drivers(
    body: DriverExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """导出司机列表（当前筛选全量，或 ids 已选）。"""
    try:
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        query = db.query(PlPersonnel).filter(
            PlPersonnel.tenant_no == tenant_no,
            PlPersonnel.del_flag == 0,
        )
        if body.q:
            query = query.filter(
                or_(
                    func.lower(PlPersonnel.personnel_name).contains(body.q.lower()),
                    PlPersonnel.phone_no.contains(body.q),
                )
            )
        if body.ids:
            query = query.filter(PlPersonnel.id.in_(body.ids))
        query = query.order_by(
            case((PlPersonnel.order_num.is_(None), 1), else_=0).asc(),
            PlPersonnel.create_time.desc(),
        )
        rows = query.limit(EXPORT_MAX_ROWS + 1).all()
        items = [_personnel_to_driver_read(p).model_dump(by_alias=True) for p in rows]
        cols = filter_export_columns(DRIVER_EXPORT_COLUMNS, body.columns)
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="司机列表",
            columns=cols,
            rows=items,
            filename_stem=f"司机列表_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e



@router.post("/driver", response_model=DriverRead, status_code=status.HTTP_201_CREATED)
def create_driver(
    payload: DriverCreate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> DriverRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    personnel = PlPersonnel(
        id=generate_uuid(),
        tenant_no=tenant_no,
        personnel_name=payload.driver_name,
        phone_no=payload.phone_no,
        id_card_no=payload.id_card_no,
        status="1",
        remark=payload.remark,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(personnel)
    db.commit()
    db.refresh(personnel)
    return _personnel_to_driver_read(personnel)


@router.get("/driver/{id}", response_model=DriverRead)
def get_driver(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> DriverRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    personnel = _get_personnel(db, tenant_no, id, user)
    return _personnel_to_driver_read(personnel)


@router.put("/driver/{id}", response_model=DriverRead)
def update_driver(
    id: str,
    payload: DriverUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> DriverRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    personnel = _get_personnel(db, tenant_no, id, user)

    if payload.driver_name is not None:
        personnel.personnel_name = payload.driver_name
    if payload.phone_no is not None:
        personnel.phone_no = payload.phone_no
    if payload.id_card_no is not None:
        personnel.id_card_no = payload.id_card_no
    if payload.remark is not None:
        personnel.remark = payload.remark

    personnel.update_by = user.id
    db.commit()
    db.refresh(personnel)
    return _personnel_to_driver_read(personnel)


@router.delete("/driver/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_driver(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> None:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    personnel = _get_personnel(db, tenant_no, id, user)
    assert_row_deletable_by_creator(personnel, user_info_from_context(user))
    personnel.del_flag = 1
    personnel.update_by = user.id
    db.commit()


# =============================================================================
# 车辆-司机绑定 API（兼容，写入 pl_vehicle_personnel）
# =============================================================================


@router.post("/vehicle-driver", status_code=status.HTTP_201_CREATED)
def bind_vehicle_driver(
    payload: VehicleDriverCreate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    _get_vehicle(db, tenant_no, payload.vehicle_id, user)
    _get_personnel(db, tenant_no, payload.driver_id, user)

    if payload.is_default == 1:
        db.query(PlVehiclePersonnel).filter(
            PlVehiclePersonnel.vehicle_id == payload.vehicle_id,
            PlVehiclePersonnel.role == "driver",
            PlVehiclePersonnel.del_flag == 0,
        ).update({"is_default": 0})

    vd = PlVehiclePersonnel(
        id=generate_uuid(),
        tenant_no=tenant_no,
        vehicle_id=payload.vehicle_id,
        personnel_id=payload.driver_id,
        role="driver",
        is_default=payload.is_default,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    db.add(vd)
    db.commit()
    return {"id": vd.id}


@router.delete("/vehicle-driver/{id}", status_code=status.HTTP_204_NO_CONTENT)
def unbind_vehicle_driver(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> None:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    vd = (
        db.query(PlVehiclePersonnel)
        .filter(
            PlVehiclePersonnel.id == id,
            PlVehiclePersonnel.tenant_no == tenant_no,
            PlVehiclePersonnel.del_flag == 0,
        )
        .first()
    )
    if not vd:
        vd = (
            db.query(PlVehicleDriver)
            .filter(
                PlVehicleDriver.id == id,
                PlVehicleDriver.tenant_no == tenant_no,
                PlVehicleDriver.del_flag == 0,
            )
            .first()
        )
    if vd:
        assert_row_deletable_by_creator(vd, user_info_from_context(user))
        vd.del_flag = 1
        db.commit()


@router.post("/paste-recognize/commit", response_model=PasteCommitResult)
def paste_recognize_commit(
    payload: PasteCommitBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PasteCommitResult:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    try:
        result = commit_paste_recognize(
            db,
            tenant_no,
            payload,
            user_id=user.id,
            dept_id=user.dept_id,
        )
        db.commit()
        return result
    except HTTPException:
        db.rollback()
        raise
    except Exception:
        db.rollback()
        raise