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

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

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

from app.core.database import get_db
from app.core.db_auth import DbUserContext, get_db_current_user
from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope
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
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=100, 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()),
            )
        )

    query = apply_pl_scope(query, PlCarrier, user, db, tenant_no)

    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],
    )


@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=payload.business_license_urls.strip(),
        road_transport_license_urls=payload.road_transport_license_urls.strip(),
        transport_agreement_urls=payload.transport_agreement_urls.strip(),
        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, 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="承运商不存在")
    assert_pl_row_in_scope(db, carrier, PlCarrier, user, tenant_no)
    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=100, 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))

    query = apply_pl_scope(query, PlVehicle, user, db, tenant_no)

    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
        ],
    )


@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,
        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="车辆不存在")
    assert_pl_row_in_scope(db, vehicle, PlVehicle, user, tenant_no)
    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"
    ),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100, 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 q:
        query = query.filter(
            or_(
                func.lower(PlPersonnel.personnel_name).contains(q.lower()),
                PlPersonnel.phone_no.contains(q),
            )
        )

    query = apply_pl_scope(query, PlPersonnel, user, db, tenant_no)

    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
        ],
    )


@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,
        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="人员不存在")
    assert_pl_row_in_scope(db, personnel, PlPersonnel, user, tenant_no)
    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=100, 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),
            )
        )

    query = apply_pl_scope(query, PlPersonnel, user, db, tenant_no)

    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],
    )


@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:
        if isinstance(vd, PlVehiclePersonnel):
            assert_pl_row_in_scope(db, vd, PlVehiclePersonnel, user, tenant_no)
        else:
            assert_pl_row_in_scope(db, vd, PlVehicleDriver, user, tenant_no)
        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