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

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

from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from pydantic import BaseModel, ConfigDict, 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.services.pl_personnel_user_bind import assert_sys_user_bind_available
from app.schemas.pl.paste_recognize import PasteCommitBody, PasteCommitResult
from app.services.pl_paste_recognize_service import commit_paste_recognize

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


def _merge_query(request: Request, **known) -> dict:
    out = {k: v for k, v in known.items() if v is not None and v != ""}
    for k, v in request.query_params.multi_items():
        if k in ("page", "pageSize"):
            continue
        if k not in out and v not in (None, ""):
            out[k] = v
    return out


def _param_get(params: dict, *keys: str) -> Any:
    for key in keys:
        if key in params and params[key] not in (None, ""):
            return params[key]
    return None


def _body_to_params(body: BaseModel) -> dict:
    data = body.model_dump(by_alias=True, exclude_none=True)
    for key in ("ids", "columns"):
        data.pop(key, None)
    extra = getattr(body, "model_extra", None) or {}
    for key, value in extra.items():
        if key not in data and value not in (None, ""):
            data[key] = value
    return data


def _carrier_column_filter_kwargs(params: dict) -> dict:
    return dict(
        carrier_name=_param_get(params, "carrierName", "carrier_name"),
        carrier_name_in=_param_get(params, "carrierNameIn", "carrier_name_in"),
        contact_name=_param_get(params, "contactName", "contact_name"),
        contact_name_in=_param_get(params, "contactNameIn", "contact_name_in"),
        contact_phone=_param_get(params, "contactPhone", "contact_phone"),
        contact_phone_in=_param_get(params, "contactPhoneIn", "contact_phone_in"),
        status=_param_get(params, "status"),
        status_in=_param_get(params, "statusIn", "status_in"),
        remark=_param_get(params, "remark"),
        transport_agreement_expire_date_from=_param_get(
            params,
            "transportAgreementExpireDateFrom",
            "transport_agreement_expire_date_from",
        ),
        transport_agreement_expire_date_to=_param_get(
            params,
            "transportAgreementExpireDateTo",
            "transport_agreement_expire_date_to",
        ),
    )


def _vehicle_column_filter_kwargs(params: dict) -> dict:
    return dict(
        license_plate=_param_get(params, "licensePlate", "license_plate"),
        license_plate_in=_param_get(params, "licensePlateIn", "license_plate_in"),
        vehicle_status=_param_get(params, "vehicleStatus", "vehicle_status"),
        vehicle_status_in=_param_get(params, "vehicleStatusIn", "vehicle_status_in"),
        carrier_name=_param_get(params, "carrierName", "carrier_name"),
        remark=_param_get(params, "remark"),
        load_capacity_op=_param_get(params, "loadCapacityOp", "load_capacity_op"),
        load_capacity=_param_get(params, "loadCapacity", "load_capacity"),
        load_capacity_min=_param_get(params, "loadCapacityMin", "load_capacity_min"),
        load_capacity_max=_param_get(params, "loadCapacityMax", "load_capacity_max"),
    )


def _personnel_column_filter_kwargs(params: dict) -> dict:
    return dict(
        personnel_name=_param_get(params, "personnelName", "personnel_name"),
        personnel_name_in=_param_get(params, "personnelNameIn", "personnel_name_in"),
        phone_no=_param_get(params, "phoneNo", "phone_no"),
        phone_no_in=_param_get(params, "phoneNoIn", "phone_no_in"),
        status=_param_get(params, "status"),
        status_in=_param_get(params, "statusIn", "status_in"),
        personnel_type=_param_get(params, "personnelType", "personnel_type"),
        personnel_type_in=_param_get(params, "personnelTypeIn", "personnel_type_in"),
        id_card_no=_param_get(params, "idCardNo", "id_card_no"),
        remark=_param_get(params, "remark"),
        sys_user_name=_param_get(params, "sysUserName", "sys_user_name"),
        age_op=_param_get(params, "ageOp", "age_op"),
        age=_param_get(params, "age"),
        age_min=_param_get(params, "ageMin", "age_min"),
        age_max=_param_get(params, "ageMax", "age_max"),
    )


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


class CarrierColumnFacetBody(BaseModel):
    """承运商列头可选值（去重；可附带列表上下文）"""

    model_config = ConfigDict(extra="allow")

    field: str = Field(..., description="carrierName | contactName | contactPhone")
    keyword: Optional[str] = Field(default=None, description="可选值关键字")
    limit: Optional[int] = Field(default=500, ge=1, le=500, description="返回条数上限")


_CARRIER_FACET_COLUMNS = {
    "carrierName": PlCarrier.carrier_name,
    "contactName": PlCarrier.contact_name,
    "contactPhone": PlCarrier.contact_phone,
}


@router.post("/carrier/column-facet-values")
def carrier_column_facet_values(
    body: CarrierColumnFacetBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """承运商列头可选值：在列表上下文（排除本列）下去重。"""
    from app.services.column_facet import list_column_facet_values
    from app.services.pl.carrier_list_filters import build_carrier_column_filters
    from app.services.pl.facet_context import strip_simple_facet_self_params

    column = _CARRIER_FACET_COLUMNS.get(body.field)
    if column is None:
        raise HTTPException(status_code=400, detail="不支持的列")
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    params = strip_simple_facet_self_params(_body_to_params(body), body.field)
    extra_filters = build_carrier_column_filters(
        **_carrier_column_filter_kwargs(params)
    )
    audit = _param_get(params, "auditStatus", "audit_status")
    if audit:
        extra_filters.append(PlCarrier.audit_status == audit)
    qv = _param_get(params, "q")
    if qv:
        extra_filters.append(
            or_(
                func.lower(PlCarrier.carrier_name).contains(str(qv).lower()),
                func.lower(PlCarrier.contact_name).contains(str(qv).lower()),
                PlCarrier.contact_phone.contains(str(qv)),
            )
        )
    rows = list_column_facet_values(
        db,
        PlCarrier,
        column,
        tenant_no=tenant_no,
        keyword=body.keyword,
        limit=body.limit if body.limit is not None else 500,
        extra_filters=extra_filters or None,
    )
    return {"list": rows}



class VehicleColumnFacetBody(BaseModel):
    """车辆列头可选值（去重；可附带列表上下文）"""

    model_config = ConfigDict(extra="allow")

    field: str = Field(..., description="licensePlate")
    keyword: Optional[str] = Field(default=None, description="可选值关键字")
    limit: Optional[int] = Field(default=500, ge=1, le=500, description="返回条数上限")


_VEHICLE_FACET_COLUMNS = {
    "licensePlate": PlVehicle.license_plate,
}


@router.post("/vehicle/column-facet-values")
def vehicle_column_facet_values(
    body: VehicleColumnFacetBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """车辆列头可选值：在列表上下文（排除本列）下去重（含 vehicleType）。"""
    from app.services.column_facet import list_column_facet_values
    from app.services.pl.facet_context import strip_simple_facet_self_params
    from app.services.pl.vehicle_list_filters import build_vehicle_column_filters

    column = _VEHICLE_FACET_COLUMNS.get(body.field)
    if column is None:
        raise HTTPException(status_code=400, detail="不支持的列")
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    params = strip_simple_facet_self_params(_body_to_params(body), body.field)
    extra_filters = build_vehicle_column_filters(
        **_vehicle_column_filter_kwargs(params)
    )
    vt = _param_get(params, "vehicleType", "vehicle_type")
    if vt:
        extra_filters.append(PlVehicle.vehicle_type == vt)
    cid = _param_get(params, "carrierId", "carrier_id")
    if cid:
        extra_filters.append(PlVehicle.carrier_id == cid)
    qv = _param_get(params, "q")
    if qv:
        extra_filters.append(PlVehicle.license_plate.contains(str(qv)))
    mine_flag = _param_get(params, "mineVehicles", "mine_vehicles")
    if mine_flag is True or (
        mine_flag is not None
        and str(mine_flag).strip().lower() in {"1", "true", "yes", "y"}
    ):
        from app.services.pl_vehicle_mine_vehicles import (
            license_plates_for_user_personnel,
            should_restrict_mine_vehicles,
        )

        if should_restrict_mine_vehicles(
            db, tenant_no, user.id, mine_vehicles=True
        ):
            plates = license_plates_for_user_personnel(db, tenant_no, user.id)
            if not plates:
                return {"list": []}
            extra_filters.append(PlVehicle.license_plate.in_(list(plates)))
    rows = list_column_facet_values(
        db,
        PlVehicle,
        column,
        tenant_no=tenant_no,
        keyword=body.keyword,
        limit=body.limit if body.limit is not None else 500,
        extra_filters=extra_filters or None,
    )
    return {"list": rows}



class PersonnelColumnFacetBody(BaseModel):
    """人员列头可选值（去重；可附带列表上下文）"""

    model_config = ConfigDict(extra="allow")

    field: str = Field(..., description="personnelName | phoneNo")
    keyword: Optional[str] = Field(default=None, description="可选值关键字")
    limit: Optional[int] = Field(default=500, ge=1, le=500, description="返回条数上限")


_PERSONNEL_FACET_COLUMNS = {
    "personnelName": PlPersonnel.personnel_name,
    "phoneNo": PlPersonnel.phone_no,
}


@router.post("/personnel/column-facet-values")
def personnel_column_facet_values(
    body: PersonnelColumnFacetBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """人员列头可选值：在列表上下文（排除本列）下去重（含 personnelAffiliation）。"""
    from app.services.column_facet import list_column_facet_values
    from app.services.pl.facet_context import strip_simple_facet_self_params
    from app.services.pl.personnel_list_filters import build_personnel_column_filters

    column = _PERSONNEL_FACET_COLUMNS.get(body.field)
    if column is None:
        raise HTTPException(status_code=400, detail="不支持的列")
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    params = strip_simple_facet_self_params(_body_to_params(body), body.field)
    extra_filters = build_personnel_column_filters(
        **_personnel_column_filter_kwargs(params)
    )
    affiliation = _param_get(params, "personnelAffiliation", "personnel_affiliation")
    if affiliation:
        extra_filters.append(PlPersonnel.personnel_affiliation == affiliation)
    qv = _param_get(params, "q")
    if qv:
        extra_filters.append(
            or_(
                func.lower(PlPersonnel.personnel_name).contains(str(qv).lower()),
                PlPersonnel.phone_no.contains(str(qv)),
            )
        )
    rows = list_column_facet_values(
        db,
        PlPersonnel,
        column,
        tenant_no=tenant_no,
        keyword=body.keyword,
        limit=body.limit if body.limit is not None else 500,
        extra_filters=extra_filters or None,
    )
    return {"list": rows}


@router.get("/carrier", response_model=CarrierListResponse)
def list_carriers(
    request: Request,
    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())
    params = _merge_query(request, q=q, status=status, auditStatus=audit_status)
    query = db.query(PlCarrier).filter(
        PlCarrier.tenant_no == tenant_no,
        PlCarrier.del_flag == 0,
    )

    audit = _param_get(params, "auditStatus", "audit_status")
    if audit:
        query = query.filter(PlCarrier.audit_status == audit)

    qv = _param_get(params, "q")
    if qv:
        query = query.filter(
            or_(
                func.lower(PlCarrier.carrier_name).contains(str(qv).lower()),
                func.lower(PlCarrier.contact_name).contains(str(qv).lower()),
                PlCarrier.contact_phone.contains(str(qv)),
            )
        )

    from app.services.pl.carrier_list_filters import build_carrier_column_filters

    column_filters = build_carrier_column_filters(
        **_carrier_column_filter_kwargs(params)
    )
    if column_filters:
        query = query.filter(*column_filters)

    # 承运商为物流主数据，列表不做 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, "extra": "allow"}


@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())
        params = _body_to_params(body)
        query = db.query(PlCarrier).filter(
            PlCarrier.tenant_no == tenant_no,
            PlCarrier.del_flag == 0,
        )
        audit = _param_get(params, "auditStatus", "audit_status")
        if audit:
            query = query.filter(PlCarrier.audit_status == audit)
        qv = _param_get(params, "q")
        if qv:
            query = query.filter(
                or_(
                    func.lower(PlCarrier.carrier_name).contains(str(qv).lower()),
                    func.lower(PlCarrier.contact_name).contains(str(qv).lower()),
                    PlCarrier.contact_phone.contains(str(qv)),
                )
            )
        from app.services.pl.carrier_list_filters import build_carrier_column_filters

        column_filters = build_carrier_column_filters(
            **_carrier_column_filter_kwargs(params)
        )
        if column_filters:
            query = query.filter(*column_filters)
        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(
    request: Request,
    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="排除资质不全或协议过期的承运商车辆",
    ),
    mine_vehicles: Optional[bool] = Query(
        default=None,
        alias="mineVehicles",
        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())
    params = _merge_query(
        request,
        q=q,
        vehicleType=vehicle_type,
        carrierId=carrier_id,
        vehicleStatus=vehicle_status,
        mineVehicles=mine_vehicles,
    )
    query = db.query(PlVehicle).filter(
        PlVehicle.tenant_no == tenant_no,
        PlVehicle.del_flag == 0,
    )

    vt = _param_get(params, "vehicleType", "vehicle_type")
    if vt:
        query = query.filter(PlVehicle.vehicle_type == vt)

    cid = _param_get(params, "carrierId", "carrier_id")
    if cid:
        query = query.filter(PlVehicle.carrier_id == cid)

    qv = _param_get(params, "q")
    if qv:
        query = query.filter(PlVehicle.license_plate.contains(str(qv)))

    from app.services.pl.vehicle_list_filters import build_vehicle_column_filters

    column_filters = build_vehicle_column_filters(
        **_vehicle_column_filter_kwargs(params)
    )
    if column_filters:
        query = query.filter(*column_filters)

    mine_flag = _param_get(params, "mineVehicles", "mine_vehicles")
    if mine_flag is True or (
        mine_flag is not None
        and str(mine_flag).strip().lower() in {"1", "true", "yes", "y"}
    ):
        from app.services.pl_vehicle_mine_vehicles import (
            license_plates_for_user_personnel,
            should_restrict_mine_vehicles,
        )

        if should_restrict_mine_vehicles(
            db, tenant_no, user.id, mine_vehicles=True
        ):
            plates = license_plates_for_user_personnel(db, tenant_no, user.id)
            if not plates:
                return VehicleListResponse(total=0, list=[])
            query = query.filter(PlVehicle.license_plate.in_(list(plates)))

    # 车辆为主数据，列表不做 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, "extra": "allow"}


@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())
        params = _body_to_params(body)
        query = db.query(PlVehicle).filter(
            PlVehicle.tenant_no == tenant_no,
            PlVehicle.del_flag == 0,
        )
        vt = _param_get(params, "vehicleType", "vehicle_type")
        if vt:
            query = query.filter(PlVehicle.vehicle_type == vt)
        cid = _param_get(params, "carrierId", "carrier_id")
        if cid:
            query = query.filter(PlVehicle.carrier_id == cid)
        qv = _param_get(params, "q")
        if qv:
            query = query.filter(PlVehicle.license_plate.contains(str(qv)))
        from app.services.pl.vehicle_list_filters import build_vehicle_column_filters

        column_filters = build_vehicle_column_filters(
            **_vehicle_column_filter_kwargs(params)
        )
        if column_filters:
            query = query.filter(*column_filters)
        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 _assert_personnel_sys_user_bind(
    db: Session,
    tenant_no: str,
    *,
    personnel_id: str,
    sys_user_id: Optional[str],
) -> None:
    uid = (sys_user_id or "").strip()
    if not uid:
        return
    existing = (
        db.query(PlPersonnel)
        .filter(
            PlPersonnel.tenant_no == tenant_no,
            PlPersonnel.sys_user_id == uid,
            PlPersonnel.del_flag == 0,
        )
        .all()
    )
    try:
        assert_sys_user_bind_available(
            existing=existing,
            personnel_id=personnel_id,
            sys_user_id=uid,
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc


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(
    request: Request,
    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())
    params = _merge_query(
        request,
        q=q,
        status=status,
        personnelAffiliation=personnel_affiliation,
        personnelType=personnel_type,
    )
    query = db.query(PlPersonnel).filter(
        PlPersonnel.tenant_no == tenant_no,
        PlPersonnel.del_flag == 0,
    )

    affiliation = _param_get(params, "personnelAffiliation", "personnel_affiliation")
    if affiliation:
        query = query.filter(PlPersonnel.personnel_affiliation == affiliation)

    qv = _param_get(params, "q")
    if qv:
        query = query.filter(
            or_(
                func.lower(PlPersonnel.personnel_name).contains(str(qv).lower()),
                PlPersonnel.phone_no.contains(str(qv)),
            )
        )

    from app.services.pl.personnel_list_filters import build_personnel_column_filters

    column_filters = build_personnel_column_filters(
        **_personnel_column_filter_kwargs(params)
    )
    if column_filters:
        query = query.filter(*column_filters)

    # 人员为主数据，列表不做 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, "extra": "allow"}


@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())
        params = _body_to_params(body)
        query = db.query(PlPersonnel).filter(
            PlPersonnel.tenant_no == tenant_no,
            PlPersonnel.del_flag == 0,
        )
        affiliation = _param_get(params, "personnelAffiliation", "personnel_affiliation")
        if affiliation:
            query = query.filter(PlPersonnel.personnel_affiliation == affiliation)
        qv = _param_get(params, "q")
        if qv:
            query = query.filter(
                or_(
                    func.lower(PlPersonnel.personnel_name).contains(str(qv).lower()),
                    PlPersonnel.phone_no.contains(str(qv)),
                )
            )
        from app.services.pl.personnel_list_filters import build_personnel_column_filters

        column_filters = build_personnel_column_filters(
            **_personnel_column_filter_kwargs(params)
        )
        if column_filters:
            query = query.filter(*column_filters)
        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_id = generate_uuid()
    _assert_personnel_sys_user_bind(
        db,
        tenant_no,
        personnel_id=personnel_id,
        sys_user_id=payload.sys_user_id,
    )

    personnel = PlPersonnel(
        id=personnel_id,
        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,
        sys_user_id=(payload.sys_user_id or "").strip() or None,
        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}」",
    )

    if "sys_user_id" in data:
        raw_uid = data.get("sys_user_id")
        data["sys_user_id"] = (raw_uid or "").strip() or None
        _assert_personnel_sys_user_bind(
            db,
            tenant_no,
            personnel_id=personnel.id,
            sys_user_id=data["sys_user_id"],
        )

    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