|
|
@@ -1,13 +1,22 @@
|
|
|
"""车辆维保 API"""
|
|
|
from datetime import date
|
|
|
-from typing import Optional
|
|
|
+from datetime import 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_maintenance import (
|
|
|
VehicleMaintenanceApplyCreate,
|
|
|
VehicleMaintenanceApplyListResponse,
|
|
|
@@ -74,6 +83,76 @@ def list_reminders(
|
|
|
)
|
|
|
|
|
|
|
|
|
+
|
|
|
+MAINT_APPLY_EXPORT_COLUMNS = [
|
|
|
+ ExportColumn(title="申请单号", field="applyNo"),
|
|
|
+ ExportColumn(title="车牌号", field="licensePlate"),
|
|
|
+ ExportColumn(title="维保类型", field="maintenanceType"),
|
|
|
+ ExportColumn(title="申请日期", field="applyDate"),
|
|
|
+ ExportColumn(title="开始日期", field="startDate"),
|
|
|
+ ExportColumn(title="预计取车", field="expectedPickupDate"),
|
|
|
+ ExportColumn(title="申请人", field="applicantName"),
|
|
|
+ ExportColumn(title="状态", field="status"),
|
|
|
+ ExportColumn(title="创建时间", field="createTime"),
|
|
|
+]
|
|
|
+
|
|
|
+
|
|
|
+class MaintenanceApplyExportBody(BaseModel):
|
|
|
+ q: Optional[str] = None
|
|
|
+ maintenance_type: Optional[str] = Field(default=None, alias="maintenanceType")
|
|
|
+ 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_applies(
|
|
|
+ body: MaintenanceApplyExportBody,
|
|
|
+ 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())
|
|
|
+ items, _ = svc.list_applies(
|
|
|
+ db,
|
|
|
+ tenant_no,
|
|
|
+ user,
|
|
|
+ q=body.q,
|
|
|
+ maintenance_type=body.maintenance_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 in ("applyDate", "startDate", "expectedPickupDate", "createTime"):
|
|
|
+ if data.get(k) is not None:
|
|
|
+ data[k] = str(data[k])[:19]
|
|
|
+ rows.append(data)
|
|
|
+ cols = filter_export_columns(MAINT_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=VehicleMaintenanceApplyRead)
|
|
|
def get_apply(
|
|
|
id: str,
|