feat(pl): add Excel export for recharge, urea and maintenance process lists

Support filtered and selected export on apply and disburse/process board pages.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/pl/vehicle_maintenance.py

@@ -239,6 +239,68 @@ def list_process_board(
     )
 
 
+
+MAINT_PROCESS_EXPORT_COLUMNS = [
+    ExportColumn(title="申请单号", field="applyNo"),
+    ExportColumn(title="申请日期", field="applyDate"),
+    ExportColumn(title="申请人", field="applicantName"),
+    ExportColumn(title="申请状态", field="applyStatus"),
+    ExportColumn(title="车牌号", field="licensePlate"),
+    ExportColumn(title="维保类型", field="maintenanceType"),
+    ExportColumn(title="处理单号", field="processNo"),
+    ExportColumn(title="处理日期", field="processDate"),
+    ExportColumn(title="处理人", field="handlerName"),
+    ExportColumn(title="金额(元)", field="amount"),
+]
+
+
+class MaintProcessExportBody(BaseModel):
+    q: Optional[str] = None
+    maintenance_type: Optional[str] = Field(default=None, alias="maintenanceType")
+    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")
+    process_date_from: Optional[date] = Field(default=None, alias="processDateFrom")
+    process_date_to: Optional[date] = Field(default=None, alias="processDateTo")
+    ids: Optional[List[str]] = None
+    columns: Optional[List[str]] = None
+    model_config = {"populate_by_name": True}
+
+
+@process_router.post("/process-board/export")
+def export_process_board(
+    body: MaintProcessExportBody,
+    db: Session = Depends(get_db),
+    user: DbUserContext = Depends(get_db_current_user),
+):
+    try:
+        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
+        items, _ = svc.list_process_board(
+            db, tenant_no, user,
+            q=body.q, maintenance_type=body.maintenance_type, apply_status=body.apply_status,
+            apply_date_from=body.apply_date_from, apply_date_to=body.apply_date_to,
+            process_date_from=body.process_date_from, process_date_to=body.process_date_to,
+            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)
+            for k, v in list(data.items()):
+                if hasattr(v, 'isoformat'):
+                    data[k] = str(v)[:19]
+            rows.append(data)
+
+        cols = filter_export_columns(MAINT_PROCESS_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=rows,
+            filename_stem=f"维保过程_{date_str}",
+        )
+    except ExportValidationError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+
+
 @process_router.get("/{id}", response_model=VehicleMaintenanceProcessRead)
 def get_process(
     id: str,

backend/app/api/pl/vehicle_recharge.py

@@ -1,13 +1,21 @@
 """车辆 ETC/加油充值 API"""
-from datetime import date
-from typing import Optional
+from datetime import date, datetime
+from typing import List, Optional
 
-from fastapi import APIRouter, Depends, Query, status
+from pydantic import BaseModel, Field
+from fastapi import APIRouter, Depends, HTTPException, Query, 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,
@@ -68,6 +76,63 @@ def pending_apply_options(
     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="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
+    model_config = {"populate_by_name": True}
+
+
+@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())
+        items, _ = svc.list_applies(
+            db, tenant_no, user,
+            q=body.q, recharge_type=body.recharge_type, status=body.status,
+            apply_date_from=body.apply_date_from, apply_date_to=body.apply_date_to,
+            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)
+            for k, v in list(data.items()):
+                if hasattr(v, 'isoformat'):
+                    data[k] = str(v)[:19]
+            rows.append(data)
+
+        cols = filter_export_columns(RECHARGE_APPLY_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=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,
@@ -186,6 +251,68 @@ def list_disburse_board(
     )
 
 
+
+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="rechargeDate"),
+    ExportColumn(title="充值金额", field="amount"),
+    ExportColumn(title="加油站", field="gasStation"),
+]
+
+
+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
+    model_config = {"populate_by_name": True}
+
+
+@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())
+        items, _ = svc.list_disburse_board(
+            db, tenant_no, user,
+            q=body.q, recharge_type=body.recharge_type, apply_status=body.apply_status,
+            apply_date_from=body.apply_date_from, apply_date_to=body.apply_date_to,
+            recharge_date_from=body.recharge_date_from, recharge_date_to=body.recharge_date_to,
+            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)
+            for k, v in list(data.items()):
+                if hasattr(v, 'isoformat'):
+                    data[k] = str(v)[:19]
+            rows.append(data)
+
+        cols = filter_export_columns(RECHARGE_DISBURSE_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=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,

backend/app/api/pl/vehicle_urea.py

@@ -1,13 +1,21 @@
 """车辆尿素 API"""
-from datetime import date
-from typing import Optional
+from datetime import date, datetime
+from typing import List, Optional
 
-from fastapi import APIRouter, Depends, Query, status
+from pydantic import BaseModel, Field
+from fastapi import APIRouter, Depends, HTTPException, Query, 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_urea import (
     VehicleUreaApplyCreate,
     VehicleUreaApplyListResponse,
@@ -54,6 +62,62 @@ def list_applies(
     )
 
 
+
+UREA_APPLY_EXPORT_COLUMNS = [
+    ExportColumn(title="申请单号", field="applyNo"),
+    ExportColumn(title="车牌号", field="licensePlate"),
+    ExportColumn(title="申请时间", field="applyDate"),
+    ExportColumn(title="入库时间", field="warehouseInDate"),
+    ExportColumn(title="单位", field="unit"),
+    ExportColumn(title="入库数量", field="quantity"),
+    ExportColumn(title="申请人", field="applicantName"),
+    ExportColumn(title="状态", field="status"),
+]
+
+
+class UreaApplyExportBody(BaseModel):
+    q: Optional[str] = None
+    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
+    model_config = {"populate_by_name": True}
+
+
+@apply_router.post("/export")
+def export_urea_applies(
+    body: UreaApplyExportBody,
+    db: Session = Depends(get_db),
+    user: DbUserContext = Depends(get_db_current_user),
+):
+    try:
+        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
+        items, _ = svc.list_applies(
+            db, tenant_no, user,
+            q=body.q, status=body.status,
+            apply_date_from=body.apply_date_from, apply_date_to=body.apply_date_to,
+            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)
+            for k, v in list(data.items()):
+                if hasattr(v, 'isoformat'):
+                    data[k] = str(v)[:19]
+            rows.append(data)
+
+        cols = filter_export_columns(UREA_APPLY_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=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=VehicleUreaApplyRead)
 def get_apply(
     id: str,
@@ -138,6 +202,67 @@ def list_disburse_board(
     )
 
 
+
+UREA_DISBURSE_EXPORT_COLUMNS = [
+    ExportColumn(title="申请单号", field="applyNo"),
+    ExportColumn(title="车牌号", field="licensePlate"),
+    ExportColumn(title="申请时间", field="applyDate"),
+    ExportColumn(title="入库时间", field="warehouseInDate"),
+    ExportColumn(title="申请人", field="applicantName"),
+    ExportColumn(title="单位", field="unit"),
+    ExportColumn(title="入库数量", field="quantity"),
+    ExportColumn(title="申请状态", field="applyStatus"),
+    ExportColumn(title="发放日期", field="disburseDate"),
+    ExportColumn(title="发放金额", field="amount"),
+]
+
+
+class UreaDisburseExportBody(BaseModel):
+    q: Optional[str] = None
+    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")
+    disburse_date_from: Optional[date] = Field(default=None, alias="disburseDateFrom")
+    disburse_date_to: Optional[date] = Field(default=None, alias="disburseDateTo")
+    ids: Optional[List[str]] = None
+    columns: Optional[List[str]] = None
+    model_config = {"populate_by_name": True}
+
+
+@disburse_router.post("/disburse-board/export")
+def export_urea_disburse_board(
+    body: UreaDisburseExportBody,
+    db: Session = Depends(get_db),
+    user: DbUserContext = Depends(get_db_current_user),
+):
+    try:
+        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
+        items, _ = svc.list_disburse_board(
+            db, tenant_no, user,
+            q=body.q, apply_status=body.apply_status,
+            apply_date_from=body.apply_date_from, apply_date_to=body.apply_date_to,
+            disburse_date_from=body.disburse_date_from, disburse_date_to=body.disburse_date_to,
+            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)
+            for k, v in list(data.items()):
+                if hasattr(v, 'isoformat'):
+                    data[k] = str(v)[:19]
+            rows.append(data)
+
+        cols = filter_export_columns(UREA_DISBURSE_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=rows,
+            filename_stem=f"尿素发放_{date_str}",
+        )
+    except ExportValidationError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+
+
 @disburse_router.get("/{id}", response_model=VehicleUreaDisburseRead)
 def get_disburse(
     id: str,