"""车辆 ETC/加油充值 API""" from datetime import date, datetime from typing import Any, List, Optional from pydantic import BaseModel, Field from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from sqlalchemy.orm import Session from app.core.database import get_db from app.core.db_auth import DbUserContext, get_db_current_user from app.core.tenant_context import get_tenant_no from app.utils.excel_export import ( EXPORT_MAX_ROWS, ExportColumn, ExportValidationError, export_rows_to_xlsx_response, filter_export_columns, ) from app.schemas.pl.vehicle_recharge import ( VehicleRechargeApplyCreate, VehicleRechargeApplyListResponse, VehicleRechargeApplyOption, VehicleRechargeApplyRead, VehicleRechargeApplyUpdate, VehicleRechargeDisburseBoardListResponse, VehicleRechargeRecordCreate, VehicleRechargeRecordListResponse, VehicleRechargeRecordRead, VehicleRechargeRecordUpdate, ) from app.services import pl_vehicle_recharge_service as svc apply_router = APIRouter(prefix="/vehicle-recharge-apply", tags=["运输计划-ETC加油申请"]) record_router = APIRouter(prefix="/vehicle-recharge-record", tags=["运输计划-ETC加油发放"]) RECHARGE_TYPE_LABELS = {"etc": "ETC", "fuel": "加油"} RECHARGE_APPLY_STATUS_LABELS = { "draft": "草稿", "pending": "待发放", "in_approval": "审批中", "rejected": "已驳回", "disbursed": "已发放", "cancelled": "已作废", } 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 _as_bool(value: Any) -> bool: if isinstance(value, bool): return value if value is None: return False return str(value).strip().lower() in {"1", "true", "yes", "y"} 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 _apply_list_kwargs(params: dict) -> dict: return dict( q=_param_get(params, "q"), apply_no=_param_get(params, "applyNo", "apply_no"), license_plate=_param_get(params, "licensePlate", "license_plate"), recharge_type=_param_get(params, "rechargeType", "recharge_type"), recharge_type_in=_param_get(params, "rechargeTypeIn", "recharge_type_in"), status=_param_get(params, "status"), status_in=_param_get(params, "statusIn", "status_in"), apply_date_from=_param_get(params, "applyDateFrom", "apply_date_from"), apply_date_to=_param_get(params, "applyDateTo", "apply_date_to"), applicant_id=_param_get(params, "applicantId", "applicant_id"), sort_field=_param_get(params, "sortField", "sort_field"), sort_order=_param_get(params, "sortOrder", "sort_order"), mine_vehicles=_as_bool(_param_get(params, "mineVehicles", "mine_vehicles")), ) def _disburse_list_kwargs(params: dict) -> dict: return dict( q=_param_get(params, "q"), apply_no=_param_get(params, "applyNo", "apply_no"), license_plate=_param_get(params, "licensePlate", "license_plate"), recharge_type=_param_get(params, "rechargeType", "recharge_type"), recharge_type_in=_param_get(params, "rechargeTypeIn", "recharge_type_in"), apply_status=_param_get(params, "applyStatus", "apply_status"), apply_status_in=_param_get(params, "applyStatusIn", "apply_status_in"), apply_date_from=_param_get(params, "applyDateFrom", "apply_date_from"), apply_date_to=_param_get(params, "applyDateTo", "apply_date_to"), recharge_date_from=_param_get(params, "rechargeDateFrom", "recharge_date_from"), recharge_date_to=_param_get(params, "rechargeDateTo", "recharge_date_to"), sort_field=_param_get(params, "sortField", "sort_field"), sort_order=_param_get(params, "sortOrder", "sort_order"), ) @apply_router.get("/", response_model=VehicleRechargeApplyListResponse) def list_applies( request: Request, q: Optional[str] = Query(default=None), recharge_type: Optional[str] = Query(default=None, alias="rechargeType"), status: Optional[str] = Query(default=None), apply_date_from: Optional[date] = Query(default=None, alias="applyDateFrom"), apply_date_to: Optional[date] = Query(default=None, alias="applyDateTo"), applicant_id: Optional[str] = Query(default=None, alias="applicantId"), mine_vehicles: Optional[bool] = Query(default=None, alias="mineVehicles"), 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), ) -> VehicleRechargeApplyListResponse: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) params = _merge_query( request, q=q, rechargeType=recharge_type, status=status, applyDateFrom=apply_date_from, applyDateTo=apply_date_to, applicantId=applicant_id, mineVehicles=mine_vehicles, ) items, total = svc.list_applies( db, tenant_no, user, **_apply_list_kwargs(params), page=page, page_size=page_size, ) return VehicleRechargeApplyListResponse( items=items, total=total, page=page, pageSize=page_size ) @apply_router.get("/pending-options", response_model=list[VehicleRechargeApplyOption]) def pending_apply_options( db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> list[VehicleRechargeApplyOption]: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) rows = svc.list_pending_apply_options(db, tenant_no, user) return [VehicleRechargeApplyOption.model_validate(row) for row in rows] RECHARGE_APPLY_EXPORT_COLUMNS = [ ExportColumn(title="申请单号", field="applyNo"), ExportColumn(title="车牌号", field="licensePlate"), ExportColumn(title="充值类型", field="rechargeType"), ExportColumn(title="申请日期", field="applyDate"), ExportColumn(title="期初公里数", field="odometerKm"), ExportColumn(title="申请人", field="applicantName"), ExportColumn(title="附件", field="attachmentUrls"), ExportColumn(title="备注", field="remark"), ExportColumn(title="状态", field="status"), ExportColumn(title="创建时间", field="createTime"), ] class RechargeApplyExportBody(BaseModel): q: Optional[str] = None recharge_type: Optional[str] = Field(default=None, alias="rechargeType") status: Optional[str] = None apply_date_from: Optional[date] = Field(default=None, alias="applyDateFrom") apply_date_to: Optional[date] = Field(default=None, alias="applyDateTo") ids: Optional[List[str]] = None columns: Optional[List[str]] = None export_format: Optional[str] = Field( default=None, alias="exportFormat", description="空=Excel；archive_zip=数据及附件 ZIP", ) model_config = {"populate_by_name": True, "extra": "allow"} @apply_router.post("/export") def export_recharge_applies( body: RechargeApplyExportBody, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ): try: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) params = _body_to_params(body) items, _ = svc.list_applies( db, tenant_no, user, **_apply_list_kwargs(params), ids=body.ids, page=1, page_size=EXPORT_MAX_ROWS, max_page_size=EXPORT_MAX_ROWS, ) rows = [] for item in items: data = item.model_dump(by_alias=True) data["rechargeType"] = RECHARGE_TYPE_LABELS.get( data.get("rechargeType") or "", data.get("rechargeType") ) data["status"] = RECHARGE_APPLY_STATUS_LABELS.get( data.get("status") or "", data.get("status") ) for k, v in list(data.items()): if hasattr(v, 'isoformat'): data[k] = str(v)[:19] rows.append(data) export_format = (body.export_format or "").strip().lower() cols = filter_export_columns(RECHARGE_APPLY_EXPORT_COLUMNS, body.columns) if export_format == "archive_zip": from app.services.export_attachment_zip import ( build_attachment_zip_streaming_response, urls_to_attachment_items, ) if not rows: raise ExportValidationError("没有可导出的数据") # ZIP 总表用超链接列，去掉原始 URL 列 zip_cols = [c for c in cols if c.field != "attachmentUrls"] attachments_by_id: dict = {} for i, data in enumerate(rows, start=1): data["rowNo"] = i rid = str(data.get("id") or "").strip() if not rid: continue attachments_by_id[rid] = urls_to_attachment_items( data.get("attachmentUrls") ) stamp = datetime.now().strftime("%Y%m%d%H%M") return build_attachment_zip_streaming_response( db=db, tenant_no=tenant_no, columns=zip_cols, rows=rows, attachments_by_id=attachments_by_id, stamp=stamp, package_label="充值申请", id_field="id", sheet_title="充值申请", xlsx_basename="(总表)充值申请.xlsx", ) date_str = datetime.now().strftime("%Y%m%d") return export_rows_to_xlsx_response( db, tenant_no=tenant_no, sheet_title="充值申请", columns=cols, rows=rows, filename_stem=f"充值申请_{date_str}", ) except ExportValidationError as e: raise HTTPException(status_code=400, detail=str(e)) from e @apply_router.get("/{id}", response_model=VehicleRechargeApplyRead) def get_apply( id: str, mine_vehicles: bool = Query(default=False, alias="mineVehicles"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleRechargeApplyRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) return svc.get_apply(db, tenant_no, user, id, mine_vehicles=mine_vehicles) @apply_router.post("/{id}/submit", response_model=VehicleRechargeApplyRead) def submit_apply( id: str, mine_vehicles: bool = Query(default=False, alias="mineVehicles"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleRechargeApplyRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) return svc.submit_apply(db, tenant_no, user, id, mine_vehicles=mine_vehicles) @apply_router.post("/", response_model=VehicleRechargeApplyRead, status_code=status.HTTP_201_CREATED) def create_apply( payload: VehicleRechargeApplyCreate, mine_vehicles: bool = Query(default=False, alias="mineVehicles"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleRechargeApplyRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) return svc.create_apply( db, tenant_no, user, payload, mine_vehicles=mine_vehicles ) @apply_router.put("/{id}", response_model=VehicleRechargeApplyRead) def update_apply( id: str, payload: VehicleRechargeApplyUpdate, mine_vehicles: bool = Query(default=False, alias="mineVehicles"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleRechargeApplyRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) return svc.update_apply( db, tenant_no, user, id, payload, mine_vehicles=mine_vehicles ) @apply_router.post("/{id}/cancel", response_model=VehicleRechargeApplyRead) def cancel_apply( id: str, mine_vehicles: bool = Query(default=False, alias="mineVehicles"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleRechargeApplyRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) return svc.cancel_apply(db, tenant_no, user, id, mine_vehicles=mine_vehicles) @apply_router.post("/{id}/withdraw", response_model=VehicleRechargeApplyRead) def withdraw_apply( id: str, mine_vehicles: bool = Query(default=False, alias="mineVehicles"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleRechargeApplyRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) return svc.withdraw_apply( db, tenant_no, user, id, mine_vehicles=mine_vehicles ) @apply_router.delete("/{ids}", status_code=status.HTTP_204_NO_CONTENT) def delete_applies( ids: 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()) apply_ids = [item for item in ids.split(",") if item] svc.delete_applies(db, tenant_no, user, apply_ids) @record_router.get("/", response_model=VehicleRechargeRecordListResponse) def list_records( q: Optional[str] = Query(default=None), recharge_type: Optional[str] = Query(default=None, alias="rechargeType"), recharge_date_from: Optional[date] = Query(default=None, alias="rechargeDateFrom"), recharge_date_to: Optional[date] = Query(default=None, alias="rechargeDateTo"), gas_station: Optional[str] = Query(default=None, alias="gasStation"), apply_no: Optional[str] = Query(default=None, alias="applyNo"), 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), ) -> VehicleRechargeRecordListResponse: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) items, total = svc.list_records( db, tenant_no, user, q=q, recharge_type=recharge_type, recharge_date_from=recharge_date_from, recharge_date_to=recharge_date_to, gas_station=gas_station, apply_no=apply_no, page=page, page_size=page_size, ) return VehicleRechargeRecordListResponse( items=items, total=total, page=page, pageSize=page_size ) @record_router.get("/disburse-board", response_model=VehicleRechargeDisburseBoardListResponse) def list_disburse_board( request: Request, q: Optional[str] = Query(default=None), recharge_type: Optional[str] = Query(default=None, alias="rechargeType"), apply_status: Optional[str] = Query(default=None, alias="applyStatus"), apply_date_from: Optional[date] = Query(default=None, alias="applyDateFrom"), apply_date_to: Optional[date] = Query(default=None, alias="applyDateTo"), recharge_date_from: Optional[date] = Query(default=None, alias="rechargeDateFrom"), recharge_date_to: Optional[date] = Query(default=None, alias="rechargeDateTo"), 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), ) -> VehicleRechargeDisburseBoardListResponse: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) params = _merge_query( request, q=q, rechargeType=recharge_type, applyStatus=apply_status, applyDateFrom=apply_date_from, applyDateTo=apply_date_to, rechargeDateFrom=recharge_date_from, rechargeDateTo=recharge_date_to, ) items, total = svc.list_disburse_board( db, tenant_no, user, **_disburse_list_kwargs(params), page=page, page_size=page_size, ) return VehicleRechargeDisburseBoardListResponse( items=items, total=total, page=page, pageSize=page_size ) RECHARGE_DISBURSE_EXPORT_COLUMNS = [ ExportColumn(title="申请单号", field="applyNo"), ExportColumn(title="车牌号", field="licensePlate"), ExportColumn(title="充值类型", field="rechargeType"), ExportColumn(title="申请日期", field="applyDate"), ExportColumn(title="申请人", field="applicantName"), ExportColumn(title="申请状态", field="applyStatus"), ExportColumn(title="期初公里数", field="odometerKm"), ExportColumn(title="充值订单号", field="rechargeOrderNo"), ExportColumn(title="充值日期", field="rechargeDate"), ExportColumn(title="充值金额", field="amount"), ExportColumn(title="卡号", field="cardNo"), ExportColumn(title="加油站", field="gasStation"), ExportColumn(title="发放附件", field="recordAttachmentUrls"), ExportColumn(title="登记人", field="recordCreateByName"), ExportColumn(title="登记备注", field="recordRemark"), ] class RechargeDisburseExportBody(BaseModel): q: Optional[str] = None recharge_type: Optional[str] = Field(default=None, alias="rechargeType") apply_status: Optional[str] = Field(default=None, alias="applyStatus") apply_date_from: Optional[date] = Field(default=None, alias="applyDateFrom") apply_date_to: Optional[date] = Field(default=None, alias="applyDateTo") recharge_date_from: Optional[date] = Field(default=None, alias="rechargeDateFrom") recharge_date_to: Optional[date] = Field(default=None, alias="rechargeDateTo") ids: Optional[List[str]] = None columns: Optional[List[str]] = None export_format: Optional[str] = Field( default=None, alias="exportFormat", description="空=Excel；archive_zip=数据及附件 ZIP", ) model_config = {"populate_by_name": True, "extra": "allow"} @record_router.post("/disburse-board/export") def export_recharge_disburse_board( body: RechargeDisburseExportBody, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ): try: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) params = _body_to_params(body) items, _ = svc.list_disburse_board( db, tenant_no, user, **_disburse_list_kwargs(params), ids=body.ids, page=1, page_size=EXPORT_MAX_ROWS, max_page_size=EXPORT_MAX_ROWS, ) rows = [] for item in items: data = item.model_dump(by_alias=True) data["rechargeType"] = RECHARGE_TYPE_LABELS.get( data.get("rechargeType") or "", data.get("rechargeType") ) data["applyStatus"] = RECHARGE_APPLY_STATUS_LABELS.get( data.get("applyStatus") or "", data.get("applyStatus") ) for k, v in list(data.items()): if hasattr(v, 'isoformat'): data[k] = str(v)[:19] rows.append(data) export_format = (body.export_format or "").strip().lower() cols = filter_export_columns(RECHARGE_DISBURSE_EXPORT_COLUMNS, body.columns) if export_format == "archive_zip": from app.services.export_attachment_zip import ( build_attachment_zip_streaming_response, urls_to_attachment_items, ) if not rows: raise ExportValidationError("没有可导出的数据") # ZIP 总表用超链接列，去掉原始 URL 列 zip_cols = [c for c in cols if c.field != "recordAttachmentUrls"] attachments_by_id: dict = {} for i, data in enumerate(rows, start=1): data["rowNo"] = i rid = str(data.get("applyId") or "").strip() if not rid: continue attachments_by_id[rid] = urls_to_attachment_items( data.get("recordAttachmentUrls") ) stamp = datetime.now().strftime("%Y%m%d%H%M") return build_attachment_zip_streaming_response( db=db, tenant_no=tenant_no, columns=zip_cols, rows=rows, attachments_by_id=attachments_by_id, stamp=stamp, package_label="充值发放", id_field="applyId", sheet_title="充值发放", xlsx_basename="(总表)充值发放.xlsx", ) date_str = datetime.now().strftime("%Y%m%d") return export_rows_to_xlsx_response( db, tenant_no=tenant_no, sheet_title="充值发放", columns=cols, rows=rows, filename_stem=f"充值发放_{date_str}", ) except ExportValidationError as e: raise HTTPException(status_code=400, detail=str(e)) from e @record_router.get("/{id}", response_model=VehicleRechargeRecordRead) def get_record( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleRechargeRecordRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) return svc.get_record(db, tenant_no, user, id) @record_router.post("/", response_model=VehicleRechargeRecordRead, status_code=status.HTTP_201_CREATED) def create_record( payload: VehicleRechargeRecordCreate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleRechargeRecordRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) return svc.create_record(db, tenant_no, user, payload) @record_router.put("/{id}", response_model=VehicleRechargeRecordRead) def update_record( id: str, payload: VehicleRechargeRecordUpdate, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> VehicleRechargeRecordRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) return svc.update_record(db, tenant_no, user, id, payload) @record_router.delete("/{ids}", status_code=status.HTTP_204_NO_CONTENT) def delete_records( ids: 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()) record_ids = [item for item in ids.split(",") if item] svc.delete_records(db, tenant_no, user, record_ids)