"""运输计划管理 - 基础数据 API（承运商、车辆、人员）""" from datetime import datetime from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import BaseModel, Field from sqlalchemy import case, func, or_ from sqlalchemy.orm import Session from app.utils.excel_export import ( EXPORT_MAX_ROWS, ExportColumn, ExportValidationError, export_rows_to_xlsx_response, filter_export_columns, ) from app.core.database import get_db from app.core.db_auth import DbUserContext, get_db_current_user from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context from app.core.tenant_context import get_tenant_no from app.core.utils import generate_uuid from app.models.pl_models import ( PlCarrier, PlPersonnel, PlVehicle, PlVehicleDriver, PlVehiclePersonnel, ) from app.schemas.pl import ( CarrierAuditReject, CarrierCreate, CarrierListResponse, CarrierRead, CarrierUpdate, carrier_to_read, CertRead, DriverCreate, DriverListResponse, DriverRead, DriverUpdate, PersonnelCreate, PersonnelDetailRead, PersonnelListResponse, PersonnelRead, PersonnelUpdate, VehicleCreate, VehicleDetailRead, VehicleDriverCreate, VehicleListResponse, VehicleRead, VehicleUpdate, ) from app.services.pl_cert_service import ( certs_to_read, list_owner_certs, list_owner_certs_batch, nearest_expire_date, save_owner_certs, ) from app.services.pl_cert_required_service import ( validate_personnel_required_certs, validate_vehicle_save_certs, ) from app.services.pl_carrier_service import carrier_eligible_for_transport, normalize_carrier_urls from app.services.pl_personnel_service import ( load_vehicle_personnel_rows, load_vehicle_personnel_summaries, save_vehicle_personnel, ) from app.schemas.pl.paste_recognize import PasteCommitBody, PasteCommitResult from app.services.pl_paste_recognize_service import commit_paste_recognize router = APIRouter(prefix="", tags=["运输计划-基础数据"]) # ============================================================================= # 承运商 API # ============================================================================= @router.get("/carrier", response_model=CarrierListResponse) def list_carriers( q: Optional[str] = Query(default=None, description="模糊搜索"), status: Optional[str] = Query(default=None, description="状态"), audit_status: Optional[str] = Query(default=None, alias="auditStatus", description="审核状态"), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> CarrierListResponse: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) query = db.query(PlCarrier).filter( PlCarrier.tenant_no == tenant_no, PlCarrier.del_flag == 0, ) if status: query = query.filter(PlCarrier.status == status) if audit_status: query = query.filter(PlCarrier.audit_status == audit_status) if q: query = query.filter( or_( func.lower(PlCarrier.carrier_name).contains(q.lower()), func.lower(PlCarrier.contact_name).contains(q.lower()), ) ) # 承运商为物流主数据，列表不做 create_by 数据范围过滤 query = query.order_by( case((PlCarrier.order_num.is_(None), 1), else_=0).asc(), PlCarrier.create_time.desc(), ) total = query.count() offset = (page - 1) * page_size rows = query.offset(offset).limit(page_size).all() return CarrierListResponse( total=total, list=[carrier_to_read(row) for row in rows], ) CARRIER_EXPORT_COLUMNS = [ ExportColumn(title="承运商名称", field="carrierName"), ExportColumn(title="联系人", field="contactName"), ExportColumn(title="联系电话", field="contactPhone"), ExportColumn(title="合作状态", field="statusLabel"), ExportColumn(title="资质提示", field="qualificationHint"), ExportColumn(title="协议有效期", field="transportAgreementExpireDate"), ExportColumn(title="备注", field="remark"), ] class CarrierExportBody(BaseModel): q: Optional[str] = None status: Optional[str] = None audit_status: Optional[str] = Field(default=None, alias="auditStatus") ids: Optional[List[str]] = None columns: Optional[List[str]] = None model_config = {"populate_by_name": True} @router.post("/carrier/export") def export_carriers( body: CarrierExportBody, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ): """导出承运商列表（当前筛选全量，或 ids 已选）。""" try: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) query = db.query(PlCarrier).filter( PlCarrier.tenant_no == tenant_no, PlCarrier.del_flag == 0, ) if body.status: query = query.filter(PlCarrier.status == body.status) if body.audit_status: query = query.filter(PlCarrier.audit_status == body.audit_status) if body.q: query = query.filter( or_( func.lower(PlCarrier.carrier_name).contains(body.q.lower()), func.lower(PlCarrier.contact_name).contains(body.q.lower()), ) ) if body.ids: query = query.filter(PlCarrier.id.in_(body.ids)) query = query.order_by( case((PlCarrier.order_num.is_(None), 1), else_=0).asc(), PlCarrier.create_time.desc(), ) rows = query.limit(EXPORT_MAX_ROWS + 1).all() items = [] for row in rows: data = carrier_to_read(row).model_dump(by_alias=True) data["statusLabel"] = "合作中" if str(data.get("status")) == "1" else "已终止" if data.get("transportAgreementExpired"): data["qualificationHint"] = "协议已过期" elif data.get("qualificationComplete") is False: data["qualificationHint"] = "资质不全" else: data["qualificationHint"] = "资质齐全" items.append(data) requested = None if body.columns: requested = [] for key in body.columns: if key in ( "qualificationAttachments", "qualificationAttachment", "photoUrls", ): continue if key == "status": requested.append("statusLabel") elif key == "qualificationHint": requested.append("qualificationHint") else: requested.append(key) cols = filter_export_columns(CARRIER_EXPORT_COLUMNS, requested) date_str = datetime.now().strftime("%Y%m%d") return export_rows_to_xlsx_response( db, tenant_no=tenant_no, sheet_title="承运商列表", columns=cols, rows=items, filename_stem=f"承运商列表_{date_str}", ) except ExportValidationError as e: raise HTTPException(status_code=400, detail=str(e)) from e @router.post("/carrier", response_model=CarrierRead, status_code=status.HTTP_201_CREATED) def create_carrier( payload: CarrierCreate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> CarrierRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) carrier = PlCarrier( id=generate_uuid(), tenant_no=tenant_no, carrier_name=payload.carrier_name, contact_name=payload.contact_name, contact_phone=payload.contact_phone, status=payload.status, business_license_urls=normalize_carrier_urls(payload.business_license_urls), road_transport_license_urls=normalize_carrier_urls(payload.road_transport_license_urls), transport_agreement_urls=normalize_carrier_urls(payload.transport_agreement_urls), transport_agreement_expire_date=payload.transport_agreement_expire_date, audit_status="pending", remark=payload.remark, create_by=user.id, create_dept_id=user.dept_id, ) db.add(carrier) db.commit() db.refresh(carrier) return carrier_to_read(carrier) @router.get("/carrier/{id}", response_model=CarrierRead) def get_carrier( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> CarrierRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) carrier = _get_carrier(db, tenant_no, id, user) return carrier_to_read(carrier) @router.put("/carrier/{id}", response_model=CarrierRead) def update_carrier( id: str, payload: CarrierUpdate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> CarrierRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) carrier = _get_carrier(db, tenant_no, id, user) payload_data = payload.model_dump(exclude_unset=True) for key in ( "business_license_urls", "road_transport_license_urls", "transport_agreement_urls", ): if key in payload_data: payload_data[key] = normalize_carrier_urls(payload_data[key]) for key, value in payload_data.items(): setattr(carrier, key, value) carrier.update_by = user.id db.commit() db.refresh(carrier) return carrier_to_read(carrier) @router.delete("/carrier/{id}", status_code=status.HTTP_204_NO_CONTENT) def delete_carrier( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> None: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) carrier = _get_carrier(db, tenant_no, id, user) assert_row_deletable_by_creator(carrier, user_info_from_context(user)) carrier.del_flag = 1 carrier.update_by = user.id db.commit() @router.post("/carrier/{id}/approve", response_model=CarrierRead) def approve_carrier( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> CarrierRead: raise HTTPException(status_code=400, detail="承运商审核流程已停用") @router.post("/carrier/{id}/reject", response_model=CarrierRead) def reject_carrier( id: str, payload: CarrierAuditReject, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> CarrierRead: raise HTTPException(status_code=400, detail="承运商审核流程已停用") def _get_carrier(db: Session, tenant_no: str, carrier_id: str, user: DbUserContext) -> PlCarrier: carrier = ( db.query(PlCarrier) .filter( PlCarrier.id == carrier_id, PlCarrier.tenant_no == tenant_no, PlCarrier.del_flag == 0, ) .first() ) if not carrier: raise HTTPException(status_code=404, detail="承运商不存在") return carrier # ============================================================================= # 车辆 API # ============================================================================= def _vehicle_to_read( vehicle: PlVehicle, summary: Optional[str] = None, default_driver_name: Optional[str] = None, *, cert_list: Optional[list[CertRead]] = None, ) -> VehicleRead: data = VehicleRead.model_validate(vehicle) if summary is not None: data.personnel_summary = summary if default_driver_name is not None: data.default_driver_name = default_driver_name if cert_list is not None: data.cert_list = cert_list return data def _vehicle_to_detail( db: Session, tenant_no: str, vehicle: PlVehicle ) -> VehicleDetailRead: summaries = load_vehicle_personnel_summaries(db, tenant_no, [vehicle.id]) summary, default_name = summaries.get(vehicle.id, ("", None)) personnel_list = load_vehicle_personnel_rows(db, tenant_no, vehicle.id) cert_list = certs_to_read( list_owner_certs(db, tenant_no, "vehicle", vehicle.id) ) base = _vehicle_to_read(vehicle, summary, default_name) return VehicleDetailRead( **base.model_dump(by_alias=False, exclude={"cert_list"}), personnel_list=personnel_list, cert_list=cert_list, ) @router.get("/vehicle", response_model=VehicleListResponse) def list_vehicles( q: Optional[str] = Query(default=None, description="模糊搜索车牌号"), vehicle_type: Optional[str] = Query(default=None, alias="vehicleType", description="车辆类型"), carrier_id: Optional[str] = Query(default=None, alias="carrierId", description="承运商ID"), vehicle_status: Optional[str] = Query(default=None, alias="vehicleStatus", description="车辆状态"), exclude_ineligible_carrier: Optional[bool] = Query( default=False, alias="excludeIneligibleCarrier", description="排除资质不全或协议过期的承运商车辆", ), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleListResponse: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) query = db.query(PlVehicle).filter( PlVehicle.tenant_no == tenant_no, PlVehicle.del_flag == 0, ) if vehicle_type: query = query.filter(PlVehicle.vehicle_type == vehicle_type) if carrier_id: query = query.filter(PlVehicle.carrier_id == carrier_id) if vehicle_status: query = query.filter(PlVehicle.vehicle_status == vehicle_status) if q: query = query.filter(PlVehicle.license_plate.contains(q)) # 车辆为主数据，列表不做 create_by 数据范围过滤 query = query.order_by( case((PlVehicle.order_num.is_(None), 1), else_=0).asc(), PlVehicle.create_time.desc(), ) if exclude_ineligible_carrier: vehicles = query.all() carrier_ids = { v.carrier_id for v in vehicles if v.vehicle_type == "carrier" and v.carrier_id } carrier_map: dict[str, PlCarrier] = {} if carrier_ids: carriers = ( db.query(PlCarrier) .filter( PlCarrier.id.in_(carrier_ids), PlCarrier.tenant_no == tenant_no, PlCarrier.del_flag == 0, ) .all() ) carrier_map = {c.id: c for c in carriers} vehicles = [ v for v in vehicles if v.vehicle_type != "carrier" or not v.carrier_id or ( v.carrier_id in carrier_map and carrier_eligible_for_transport(carrier_map[v.carrier_id]) ) ] summaries = load_vehicle_personnel_summaries( db, tenant_no, [v.id for v in vehicles] ) certs_by_owner = list_owner_certs_batch( db, tenant_no, "vehicle", [v.id for v in vehicles] ) items = [ _vehicle_to_read( v, *summaries.get(v.id, ("", None)), cert_list=certs_to_read(certs_by_owner.get(v.id, [])), ) for v in vehicles ] return VehicleListResponse(total=len(items), list=items) total = query.count() offset = (page - 1) * page_size vehicles = query.offset(offset).limit(page_size).all() summaries = load_vehicle_personnel_summaries( db, tenant_no, [v.id for v in vehicles] ) certs_by_owner = list_owner_certs_batch( db, tenant_no, "vehicle", [v.id for v in vehicles] ) return VehicleListResponse( total=total, list=[ _vehicle_to_read( v, *summaries.get(v.id, ("", None)), cert_list=certs_to_read(certs_by_owner.get(v.id, [])), ) for v in vehicles ], ) VEHICLE_EXPORT_COLUMNS = [ ExportColumn(title="车牌号", field="licensePlate"), ExportColumn(title="关联人员", field="personnelSummary"), ExportColumn(title="车辆类型", field="vehicleTypeLabel"), ExportColumn(title="车辆状态", field="vehicleStatusLabel"), ExportColumn(title="载重(吨)", field="loadCapacity"), ExportColumn(title="承运商", field="carrierName"), ExportColumn(title="备注", field="remark"), ] _VEHICLE_TYPE_LABEL = {"self": "自有车", "carrier": "承运商车"} _VEHICLE_STATUS_LABEL = {"1": "启用", "0": "停用", "idle": "空闲", "busy": "在途", "maintain": "维保"} class VehicleExportBody(BaseModel): q: Optional[str] = None vehicle_type: Optional[str] = Field(default=None, alias="vehicleType") carrier_id: Optional[str] = Field(default=None, alias="carrierId") vehicle_status: Optional[str] = Field(default=None, alias="vehicleStatus") ids: Optional[List[str]] = None columns: Optional[List[str]] = None model_config = {"populate_by_name": True} @router.post("/vehicle/export") def export_vehicles( body: VehicleExportBody, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ): """导出车辆列表（当前筛选全量，或 ids 已选）。""" try: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) query = db.query(PlVehicle).filter( PlVehicle.tenant_no == tenant_no, PlVehicle.del_flag == 0, ) if body.vehicle_type: query = query.filter(PlVehicle.vehicle_type == body.vehicle_type) if body.carrier_id: query = query.filter(PlVehicle.carrier_id == body.carrier_id) if body.vehicle_status: query = query.filter(PlVehicle.vehicle_status == body.vehicle_status) if body.q: query = query.filter(PlVehicle.license_plate.contains(body.q)) if body.ids: query = query.filter(PlVehicle.id.in_(body.ids)) query = query.order_by( case((PlVehicle.order_num.is_(None), 1), else_=0).asc(), PlVehicle.create_time.desc(), ) vehicles = query.limit(EXPORT_MAX_ROWS + 1).all() summaries = load_vehicle_personnel_summaries(db, tenant_no, [v.id for v in vehicles]) items = [] for v in vehicles: data = _vehicle_to_read(v, *summaries.get(v.id, ("", None))).model_dump(by_alias=True) vt = str(data.get("vehicleType") or "") data["vehicleTypeLabel"] = _VEHICLE_TYPE_LABEL.get(vt, vt) vs = str(data.get("vehicleStatus") or "") data["vehicleStatusLabel"] = _VEHICLE_STATUS_LABEL.get(vs, vs) items.append(data) requested = None if body.columns: requested = [] for key in body.columns: if key in ("photoUrls", "certAttachments", "certList"): continue if key == "vehicleType": requested.append("vehicleTypeLabel") elif key == "vehicleStatus": requested.append("vehicleStatusLabel") else: requested.append(key) cols = filter_export_columns(VEHICLE_EXPORT_COLUMNS, requested) date_str = datetime.now().strftime("%Y%m%d") return export_rows_to_xlsx_response( db, tenant_no=tenant_no, sheet_title="车辆列表", columns=cols, rows=items, filename_stem=f"车辆列表_{date_str}", ) except ExportValidationError as e: raise HTTPException(status_code=400, detail=str(e)) from e @router.post("/vehicle", response_model=VehicleDetailRead, status_code=status.HTTP_201_CREATED) def create_vehicle( payload: VehicleCreate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleDetailRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) existing = ( db.query(PlVehicle) .filter( PlVehicle.license_plate == payload.license_plate, PlVehicle.del_flag == 0, ) .first() ) if existing: raise HTTPException(status_code=400, detail="车牌号已存在") validate_vehicle_save_certs( db, tenant_no, vehicle_type=payload.vehicle_type, trailer_no=payload.trailer_no, cert_rows=payload.cert_list, personnel_rows=payload.personnel_list, ) vehicle = PlVehicle( id=generate_uuid(), tenant_no=tenant_no, vehicle_type=payload.vehicle_type, carrier_id=payload.carrier_id, carrier_name=payload.carrier_name, license_plate=payload.license_plate, trailer_no=payload.trailer_no, load_capacity=payload.load_capacity, vehicle_status=payload.vehicle_status, photo_urls=payload.photo_urls, remark=payload.remark, create_by=user.id, create_dept_id=user.dept_id, ) db.add(vehicle) db.flush() try: save_vehicle_personnel( db, vehicle.id, payload.personnel_list, tenant_no, user_id=user.id, dept_id=user.dept_id, vehicle_type=payload.vehicle_type, ) save_owner_certs( db, "vehicle", vehicle.id, payload.cert_list, tenant_no, user_id=user.id, dept_id=user.dept_id, ) db.commit() except HTTPException: db.rollback() raise except Exception: db.rollback() raise db.refresh(vehicle) return _vehicle_to_detail(db, tenant_no, vehicle) @router.get("/vehicle/{id}", response_model=VehicleDetailRead) def get_vehicle( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleDetailRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) vehicle = _get_vehicle(db, tenant_no, id, user) return _vehicle_to_detail(db, tenant_no, vehicle) @router.put("/vehicle/{id}", response_model=VehicleDetailRead) def update_vehicle( id: str, payload: VehicleUpdate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleDetailRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) vehicle = _get_vehicle(db, tenant_no, id, user) personnel_list = payload.personnel_list cert_list = payload.cert_list data = payload.model_dump( exclude_unset=True, exclude={"personnel_list", "cert_list"} ) effective_vehicle_type = data.get("vehicle_type", vehicle.vehicle_type) effective_trailer_no = data.get("trailer_no", vehicle.trailer_no) effective_cert_rows = ( cert_list if cert_list is not None else list_owner_certs(db, tenant_no, "vehicle", vehicle.id) ) effective_personnel_rows = personnel_list if effective_personnel_rows is None: effective_personnel_rows = load_vehicle_personnel_rows(db, tenant_no, vehicle.id) validate_vehicle_save_certs( db, tenant_no, vehicle_type=effective_vehicle_type, trailer_no=effective_trailer_no, cert_rows=effective_cert_rows, personnel_rows=effective_personnel_rows, ) for key, value in data.items(): setattr(vehicle, key, value) vehicle.update_by = user.id try: if personnel_list is not None: save_vehicle_personnel( db, vehicle.id, personnel_list, tenant_no, user_id=user.id, dept_id=user.dept_id, vehicle_type=effective_vehicle_type, ) if cert_list is not None: save_owner_certs( db, "vehicle", vehicle.id, cert_list, tenant_no, user_id=user.id, dept_id=user.dept_id, ) db.commit() except HTTPException: db.rollback() raise except Exception: db.rollback() raise db.refresh(vehicle) return _vehicle_to_detail(db, tenant_no, vehicle) @router.delete("/vehicle/{id}", status_code=status.HTTP_204_NO_CONTENT) def delete_vehicle( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> None: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) vehicle = _get_vehicle(db, tenant_no, id, user) assert_row_deletable_by_creator(vehicle, user_info_from_context(user)) vehicle.del_flag = 1 vehicle.update_by = user.id db.commit() def _get_vehicle(db: Session, tenant_no: str, vehicle_id: str, user: DbUserContext) -> PlVehicle: vehicle = ( db.query(PlVehicle) .filter( PlVehicle.id == vehicle_id, PlVehicle.tenant_no == tenant_no, PlVehicle.del_flag == 0, ) .first() ) if not vehicle: raise HTTPException(status_code=404, detail="车辆不存在") return vehicle # ============================================================================= # 人员 API # ============================================================================= def _personnel_to_read( db: Session, tenant_no: str, personnel: PlPersonnel, *, cert_list: Optional[list[CertRead]] = None, ) -> PersonnelRead: nearest = nearest_expire_date(db, tenant_no, "personnel", personnel.id) data = PersonnelRead.model_validate(personnel) data.nearest_expire_date = nearest if cert_list is not None: data.cert_list = cert_list return data def _personnel_to_detail( db: Session, tenant_no: str, personnel: PlPersonnel ) -> PersonnelDetailRead: base = _personnel_to_read(db, tenant_no, personnel) cert_list = certs_to_read( list_owner_certs(db, tenant_no, "personnel", personnel.id) ) return PersonnelDetailRead( **base.model_dump(by_alias=False, exclude={"cert_list"}), cert_list=cert_list, ) @router.get("/personnel", response_model=PersonnelListResponse) def list_personnel( q: Optional[str] = Query(default=None, description="模糊搜索姓名/手机号"), status: Optional[str] = Query(default=None, description="状态"), personnel_affiliation: Optional[str] = Query( default=None, alias="personnelAffiliation", description="人员归属 self/carrier" ), personnel_type: Optional[str] = Query( default=None, alias="personnelType", description="人员类型 driver/escort" ), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> PersonnelListResponse: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) query = db.query(PlPersonnel).filter( PlPersonnel.tenant_no == tenant_no, PlPersonnel.del_flag == 0, ) if status: query = query.filter(PlPersonnel.status == status) if personnel_affiliation: query = query.filter(PlPersonnel.personnel_affiliation == personnel_affiliation) if personnel_type: query = query.filter(PlPersonnel.personnel_type == personnel_type) if q: query = query.filter( or_( func.lower(PlPersonnel.personnel_name).contains(q.lower()), PlPersonnel.phone_no.contains(q), ) ) # 人员为主数据，列表不做 create_by 数据范围过滤 query = query.order_by( case((PlPersonnel.order_num.is_(None), 1), else_=0).asc(), PlPersonnel.create_time.desc(), ) total = query.count() offset = (page - 1) * page_size rows = query.offset(offset).limit(page_size).all() personnel_ids = [p.id for p in rows] certs_by_owner = list_owner_certs_batch(db, tenant_no, "personnel", personnel_ids) return PersonnelListResponse( total=total, list=[ _personnel_to_read( db, tenant_no, p, cert_list=certs_to_read(certs_by_owner.get(p.id, [])), ) for p in rows ], ) PERSONNEL_EXPORT_COLUMNS = [ ExportColumn(title="姓名", field="personnelName"), ExportColumn(title="人员归属", field="personnelAffiliationLabel"), ExportColumn(title="人员类型", field="personnelTypeLabel"), ExportColumn(title="手机号", field="phoneNo"), ExportColumn(title="年龄", field="age"), ExportColumn(title="身份证号", field="idCardNo"), ExportColumn(title="状态", field="statusLabel"), ExportColumn(title="最近到期", field="nearestExpireDate"), ExportColumn(title="备注", field="remark"), ] _AFFILIATION_LABEL = {"self": "自有", "carrier": "承运商"} _PERSONNEL_TYPE_LABEL = {"driver": "司机", "escort": "押运员"} class PersonnelExportBody(BaseModel): q: Optional[str] = None status: Optional[str] = None personnel_affiliation: Optional[str] = Field(default=None, alias="personnelAffiliation") personnel_type: Optional[str] = Field(default=None, alias="personnelType") ids: Optional[List[str]] = None columns: Optional[List[str]] = None model_config = {"populate_by_name": True} @router.post("/personnel/export") def export_personnel( body: PersonnelExportBody, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ): """导出人员列表（当前筛选全量，或 ids 已选）。""" try: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) query = db.query(PlPersonnel).filter( PlPersonnel.tenant_no == tenant_no, PlPersonnel.del_flag == 0, ) if body.status: query = query.filter(PlPersonnel.status == body.status) if body.personnel_affiliation: query = query.filter(PlPersonnel.personnel_affiliation == body.personnel_affiliation) if body.personnel_type: query = query.filter(PlPersonnel.personnel_type == body.personnel_type) if body.q: query = query.filter( or_( func.lower(PlPersonnel.personnel_name).contains(body.q.lower()), PlPersonnel.phone_no.contains(body.q), ) ) if body.ids: query = query.filter(PlPersonnel.id.in_(body.ids)) query = query.order_by( case((PlPersonnel.order_num.is_(None), 1), else_=0).asc(), PlPersonnel.create_time.desc(), ) rows = query.limit(EXPORT_MAX_ROWS + 1).all() items = [] for p in rows: data = _personnel_to_read(db, tenant_no, p).model_dump(by_alias=True) af = str(data.get("personnelAffiliation") or "") data["personnelAffiliationLabel"] = _AFFILIATION_LABEL.get(af, af) pt = str(data.get("personnelType") or "") data["personnelTypeLabel"] = _PERSONNEL_TYPE_LABEL.get(pt, pt) st = str(data.get("status") or "") data["statusLabel"] = "启用" if st == "1" else ("停用" if st == "0" else st) items.append(data) requested = None if body.columns: requested = [] for key in body.columns: if key in ("photoUrls", "certAttachments", "certList"): continue if key == "personnelAffiliation": requested.append("personnelAffiliationLabel") elif key == "personnelType": requested.append("personnelTypeLabel") elif key == "status": requested.append("statusLabel") else: requested.append(key) cols = filter_export_columns(PERSONNEL_EXPORT_COLUMNS, requested) date_str = datetime.now().strftime("%Y%m%d") return export_rows_to_xlsx_response( db, tenant_no=tenant_no, sheet_title="人员列表", columns=cols, rows=items, filename_stem=f"人员列表_{date_str}", ) except ExportValidationError as e: raise HTTPException(status_code=400, detail=str(e)) from e @router.post("/personnel", response_model=PersonnelDetailRead, status_code=status.HTTP_201_CREATED) def create_personnel( payload: PersonnelCreate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> PersonnelDetailRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) validate_personnel_required_certs( payload.cert_list, payload.personnel_type, personnel_affiliation=payload.personnel_affiliation, subject_label=f"「{payload.personnel_name}」", ) personnel = PlPersonnel( id=generate_uuid(), tenant_no=tenant_no, personnel_name=payload.personnel_name, phone_no=payload.phone_no, age=payload.age, id_card_no=payload.id_card_no, personnel_type=payload.personnel_type, personnel_affiliation=payload.personnel_affiliation, status=payload.status, photo_urls=payload.photo_urls, remark=payload.remark, create_by=user.id, create_dept_id=user.dept_id, ) db.add(personnel) db.flush() try: save_owner_certs( db, "personnel", personnel.id, payload.cert_list, tenant_no, user_id=user.id, dept_id=user.dept_id, ) db.commit() except Exception: db.rollback() raise db.refresh(personnel) return _personnel_to_detail(db, tenant_no, personnel) @router.get("/personnel/{id}", response_model=PersonnelDetailRead) def get_personnel( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> PersonnelDetailRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) personnel = _get_personnel(db, tenant_no, id, user) return _personnel_to_detail(db, tenant_no, personnel) @router.put("/personnel/{id}", response_model=PersonnelDetailRead) def update_personnel( id: str, payload: PersonnelUpdate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> PersonnelDetailRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) personnel = _get_personnel(db, tenant_no, id, user) cert_list = payload.cert_list data = payload.model_dump(exclude_unset=True, exclude={"cert_list"}) effective_personnel_type = data.get("personnel_type", personnel.personnel_type) effective_personnel_affiliation = data.get( "personnel_affiliation", personnel.personnel_affiliation ) effective_cert_rows = ( cert_list if cert_list is not None else list_owner_certs(db, tenant_no, "personnel", personnel.id) ) subject_name = data.get("personnel_name", personnel.personnel_name) validate_personnel_required_certs( effective_cert_rows, effective_personnel_type, personnel_affiliation=effective_personnel_affiliation, subject_label=f"「{subject_name}」", ) for key, value in data.items(): setattr(personnel, key, value) personnel.update_by = user.id try: if cert_list is not None: save_owner_certs( db, "personnel", personnel.id, cert_list, tenant_no, user_id=user.id, dept_id=user.dept_id, ) db.commit() except Exception: db.rollback() raise db.refresh(personnel) return _personnel_to_detail(db, tenant_no, personnel) @router.delete("/personnel/{id}", status_code=status.HTTP_204_NO_CONTENT) def delete_personnel( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> None: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) personnel = _get_personnel(db, tenant_no, id, user) assert_row_deletable_by_creator(personnel, user_info_from_context(user)) personnel.del_flag = 1 personnel.update_by = user.id db.commit() def _get_personnel(db: Session, tenant_no: str, personnel_id: str, user: DbUserContext) -> PlPersonnel: personnel = ( db.query(PlPersonnel) .filter( PlPersonnel.id == personnel_id, PlPersonnel.tenant_no == tenant_no, PlPersonnel.del_flag == 0, ) .first() ) if not personnel: raise HTTPException(status_code=404, detail="人员不存在") return personnel # ============================================================================= # 司机 API（兼容别名，内部使用 pl_personnel） # ============================================================================= def _personnel_to_driver_read(personnel: PlPersonnel) -> DriverRead: return DriverRead( id=personnel.id, driver_name=personnel.personnel_name, phone_no=personnel.phone_no, id_card_no=personnel.id_card_no, tenant_no=personnel.tenant_no, order_num=personnel.order_num, remark=personnel.remark, create_by=personnel.create_by, create_dept_id=personnel.create_dept_id, create_time=personnel.create_time, update_by=personnel.update_by, update_time=personnel.update_time, del_flag=personnel.del_flag, ) @router.get("/driver", response_model=DriverListResponse) def list_drivers( q: Optional[str] = Query(default=None, description="模糊搜索姓名/手机号"), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> DriverListResponse: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) query = db.query(PlPersonnel).filter( PlPersonnel.tenant_no == tenant_no, PlPersonnel.del_flag == 0, ) if q: query = query.filter( or_( func.lower(PlPersonnel.personnel_name).contains(q.lower()), PlPersonnel.phone_no.contains(q), ) ) # 司机列表为主数据选项，不做 create_by 数据范围过滤 query = query.order_by( case((PlPersonnel.order_num.is_(None), 1), else_=0).asc(), PlPersonnel.create_time.desc(), ) total = query.count() offset = (page - 1) * page_size rows = query.offset(offset).limit(page_size).all() return DriverListResponse( total=total, list=[_personnel_to_driver_read(p) for p in rows], ) DRIVER_EXPORT_COLUMNS = [ ExportColumn(title="司机姓名", field="driverName"), ExportColumn(title="手机号", field="phoneNo"), ExportColumn(title="身份证号", field="idCardNo"), ExportColumn(title="备注", field="remark"), ] class DriverExportBody(BaseModel): q: Optional[str] = None ids: Optional[List[str]] = None columns: Optional[List[str]] = None @router.post("/driver/export") def export_drivers( body: DriverExportBody, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ): """导出司机列表（当前筛选全量，或 ids 已选）。""" try: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) query = db.query(PlPersonnel).filter( PlPersonnel.tenant_no == tenant_no, PlPersonnel.del_flag == 0, ) if body.q: query = query.filter( or_( func.lower(PlPersonnel.personnel_name).contains(body.q.lower()), PlPersonnel.phone_no.contains(body.q), ) ) if body.ids: query = query.filter(PlPersonnel.id.in_(body.ids)) query = query.order_by( case((PlPersonnel.order_num.is_(None), 1), else_=0).asc(), PlPersonnel.create_time.desc(), ) rows = query.limit(EXPORT_MAX_ROWS + 1).all() items = [_personnel_to_driver_read(p).model_dump(by_alias=True) for p in rows] cols = filter_export_columns(DRIVER_EXPORT_COLUMNS, body.columns) date_str = datetime.now().strftime("%Y%m%d") return export_rows_to_xlsx_response( db, tenant_no=tenant_no, sheet_title="司机列表", columns=cols, rows=items, filename_stem=f"司机列表_{date_str}", ) except ExportValidationError as e: raise HTTPException(status_code=400, detail=str(e)) from e @router.post("/driver", response_model=DriverRead, status_code=status.HTTP_201_CREATED) def create_driver( payload: DriverCreate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> DriverRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) personnel = PlPersonnel( id=generate_uuid(), tenant_no=tenant_no, personnel_name=payload.driver_name, phone_no=payload.phone_no, id_card_no=payload.id_card_no, status="1", remark=payload.remark, create_by=user.id, create_dept_id=user.dept_id, ) db.add(personnel) db.commit() db.refresh(personnel) return _personnel_to_driver_read(personnel) @router.get("/driver/{id}", response_model=DriverRead) def get_driver( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> DriverRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) personnel = _get_personnel(db, tenant_no, id, user) return _personnel_to_driver_read(personnel) @router.put("/driver/{id}", response_model=DriverRead) def update_driver( id: str, payload: DriverUpdate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> DriverRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) personnel = _get_personnel(db, tenant_no, id, user) if payload.driver_name is not None: personnel.personnel_name = payload.driver_name if payload.phone_no is not None: personnel.phone_no = payload.phone_no if payload.id_card_no is not None: personnel.id_card_no = payload.id_card_no if payload.remark is not None: personnel.remark = payload.remark personnel.update_by = user.id db.commit() db.refresh(personnel) return _personnel_to_driver_read(personnel) @router.delete("/driver/{id}", status_code=status.HTTP_204_NO_CONTENT) def delete_driver( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> None: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) personnel = _get_personnel(db, tenant_no, id, user) assert_row_deletable_by_creator(personnel, user_info_from_context(user)) personnel.del_flag = 1 personnel.update_by = user.id db.commit() # ============================================================================= # 车辆-司机绑定 API（兼容，写入 pl_vehicle_personnel） # ============================================================================= @router.post("/vehicle-driver", status_code=status.HTTP_201_CREATED) def bind_vehicle_driver( payload: VehicleDriverCreate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> dict: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) _get_vehicle(db, tenant_no, payload.vehicle_id, user) _get_personnel(db, tenant_no, payload.driver_id, user) if payload.is_default == 1: db.query(PlVehiclePersonnel).filter( PlVehiclePersonnel.vehicle_id == payload.vehicle_id, PlVehiclePersonnel.role == "driver", PlVehiclePersonnel.del_flag == 0, ).update({"is_default": 0}) vd = PlVehiclePersonnel( id=generate_uuid(), tenant_no=tenant_no, vehicle_id=payload.vehicle_id, personnel_id=payload.driver_id, role="driver", is_default=payload.is_default, create_by=user.id, create_dept_id=user.dept_id, ) db.add(vd) db.commit() return {"id": vd.id} @router.delete("/vehicle-driver/{id}", status_code=status.HTTP_204_NO_CONTENT) def unbind_vehicle_driver( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> None: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) vd = ( db.query(PlVehiclePersonnel) .filter( PlVehiclePersonnel.id == id, PlVehiclePersonnel.tenant_no == tenant_no, PlVehiclePersonnel.del_flag == 0, ) .first() ) if not vd: vd = ( db.query(PlVehicleDriver) .filter( PlVehicleDriver.id == id, PlVehicleDriver.tenant_no == tenant_no, PlVehicleDriver.del_flag == 0, ) .first() ) if vd: assert_row_deletable_by_creator(vd, user_info_from_context(user)) vd.del_flag = 1 db.commit() @router.post("/paste-recognize/commit", response_model=PasteCommitResult) def paste_recognize_commit( payload: PasteCommitBody, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> PasteCommitResult: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) try: result = commit_paste_recognize( db, tenant_no, payload, user_id=user.id, dept_id=user.dept_id, ) db.commit() return result except HTTPException: db.rollback() raise except Exception: db.rollback() raise