446 lines
17 KiB
Python
446 lines
17 KiB
Python
"""车辆维保 API"""
|
|
from datetime import date
|
|
from datetime import datetime
|
|
from typing import Any, List, Optional
|
|
|
|
from pydantic import BaseModel, Field
|
|
from fastapi import APIRouter, Body, 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.schemas.pl.cancel_reason import CancelReasonBody
|
|
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,
|
|
VehicleMaintenanceApplyRead,
|
|
VehicleMaintenanceApplyReminder,
|
|
VehicleMaintenanceApplyUpdate,
|
|
VehicleMaintenanceProcessBoardListResponse,
|
|
VehicleMaintenanceProcessCreate,
|
|
VehicleMaintenanceProcessRead,
|
|
VehicleMaintenanceProcessUpdate,
|
|
)
|
|
from app.services import pl_vehicle_maintenance_service as svc
|
|
|
|
apply_router = APIRouter(prefix="/vehicle-maintenance-apply", tags=["运输计划-车辆维保申请"])
|
|
process_router = APIRouter(prefix="/vehicle-maintenance-process", tags=["运输计划-车辆维保处理"])
|
|
|
|
|
|
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 _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"),
|
|
maintenance_type=_param_get(params, "maintenanceType", "maintenance_type"),
|
|
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"),
|
|
start_date_from=_param_get(params, "startDateFrom", "start_date_from"),
|
|
start_date_to=_param_get(params, "startDateTo", "start_date_to"),
|
|
expected_pickup_date_from=_param_get(
|
|
params, "expectedPickupDateFrom", "expected_pickup_date_from"
|
|
),
|
|
expected_pickup_date_to=_param_get(
|
|
params, "expectedPickupDateTo", "expected_pickup_date_to"
|
|
),
|
|
applicant_id=_param_get(params, "applicantId", "applicant_id"),
|
|
applicant_name=_param_get(params, "applicantName", "applicant_name"),
|
|
create_time_from=_param_get(params, "createTimeFrom", "create_time_from"),
|
|
create_time_to=_param_get(params, "createTimeTo", "create_time_to"),
|
|
sort_field=_param_get(params, "sortField", "sort_field"),
|
|
sort_order=_param_get(params, "sortOrder", "sort_order"),
|
|
)
|
|
|
|
|
|
def _process_board_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"),
|
|
maintenance_type=_param_get(params, "maintenanceType", "maintenance_type"),
|
|
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"),
|
|
process_date_from=_param_get(params, "processDateFrom", "process_date_from"),
|
|
process_date_to=_param_get(params, "processDateTo", "process_date_to"),
|
|
start_date_from=_param_get(params, "startDateFrom", "start_date_from"),
|
|
start_date_to=_param_get(params, "startDateTo", "start_date_to"),
|
|
expected_pickup_date_from=_param_get(
|
|
params, "expectedPickupDateFrom", "expected_pickup_date_from"
|
|
),
|
|
expected_pickup_date_to=_param_get(
|
|
params, "expectedPickupDateTo", "expected_pickup_date_to"
|
|
),
|
|
applicant_name=_param_get(params, "applicantName", "applicant_name"),
|
|
process_no=_param_get(params, "processNo", "process_no"),
|
|
amount_op=_param_get(params, "amountOp", "amount_op"),
|
|
amount=_param_get(params, "amount"),
|
|
amount_min=_param_get(params, "amountMin", "amount_min"),
|
|
amount_max=_param_get(params, "amountMax", "amount_max"),
|
|
handler_name=_param_get(params, "handlerName", "handler_name"),
|
|
sort_field=_param_get(params, "sortField", "sort_field"),
|
|
sort_order=_param_get(params, "sortOrder", "sort_order"),
|
|
)
|
|
|
|
|
|
@apply_router.get("/", response_model=VehicleMaintenanceApplyListResponse)
|
|
def list_applies(
|
|
request: Request,
|
|
q: Optional[str] = Query(default=None),
|
|
maintenance_type: Optional[str] = Query(default=None, alias="maintenanceType"),
|
|
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"),
|
|
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),
|
|
) -> VehicleMaintenanceApplyListResponse:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
params = _merge_query(
|
|
request,
|
|
q=q,
|
|
maintenanceType=maintenance_type,
|
|
status=status,
|
|
applyDateFrom=apply_date_from,
|
|
applyDateTo=apply_date_to,
|
|
applicantId=applicant_id,
|
|
)
|
|
items, total = svc.list_applies(
|
|
db,
|
|
tenant_no,
|
|
user,
|
|
**_apply_list_kwargs(params),
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
return VehicleMaintenanceApplyListResponse(
|
|
items=items, total=total, page=page, pageSize=page_size
|
|
)
|
|
|
|
|
|
@apply_router.get("/reminders", response_model=list[VehicleMaintenanceApplyReminder])
|
|
def list_reminders(
|
|
license_plate: str = Query(..., alias="licensePlate"),
|
|
exclude_apply_id: Optional[str] = Query(default=None, alias="excludeApplyId"),
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> list[VehicleMaintenanceApplyReminder]:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
return svc.list_reminders(
|
|
db,
|
|
tenant_no,
|
|
user,
|
|
license_plate,
|
|
exclude_apply_id=exclude_apply_id,
|
|
)
|
|
|
|
|
|
|
|
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, "extra": "allow"}
|
|
|
|
|
|
@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())
|
|
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)
|
|
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,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleMaintenanceApplyRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
return svc.get_apply(db, tenant_no, user, id)
|
|
|
|
|
|
@apply_router.post("/", response_model=VehicleMaintenanceApplyRead, status_code=status.HTTP_201_CREATED)
|
|
def create_apply(
|
|
payload: VehicleMaintenanceApplyCreate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleMaintenanceApplyRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
return svc.create_apply(db, tenant_no, user, payload)
|
|
|
|
|
|
@apply_router.put("/{id}", response_model=VehicleMaintenanceApplyRead)
|
|
def update_apply(
|
|
id: str,
|
|
payload: VehicleMaintenanceApplyUpdate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleMaintenanceApplyRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
return svc.update_apply(db, tenant_no, user, id, payload)
|
|
|
|
|
|
@apply_router.post("/{id}/cancel", response_model=VehicleMaintenanceApplyRead)
|
|
def cancel_apply(
|
|
id: str,
|
|
body: Optional[CancelReasonBody] = Body(default=None),
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleMaintenanceApplyRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
return svc.cancel_apply(
|
|
db, tenant_no, user, id, cancel_reason=body.cancel_reason if body else None
|
|
)
|
|
|
|
|
|
@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)
|
|
|
|
|
|
@process_router.get("/process-board", response_model=VehicleMaintenanceProcessBoardListResponse)
|
|
def list_process_board(
|
|
request: Request,
|
|
q: Optional[str] = Query(default=None),
|
|
maintenance_type: Optional[str] = Query(default=None, alias="maintenanceType"),
|
|
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"),
|
|
process_date_from: Optional[date] = Query(default=None, alias="processDateFrom"),
|
|
process_date_to: Optional[date] = Query(default=None, alias="processDateTo"),
|
|
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),
|
|
) -> VehicleMaintenanceProcessBoardListResponse:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
params = _merge_query(
|
|
request,
|
|
q=q,
|
|
maintenanceType=maintenance_type,
|
|
applyStatus=apply_status,
|
|
applyDateFrom=apply_date_from,
|
|
applyDateTo=apply_date_to,
|
|
processDateFrom=process_date_from,
|
|
processDateTo=process_date_to,
|
|
)
|
|
items, total = svc.list_process_board(
|
|
db,
|
|
tenant_no,
|
|
user,
|
|
**_process_board_kwargs(params),
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
return VehicleMaintenanceProcessBoardListResponse(
|
|
items=items, total=total, page=page, pageSize=page_size
|
|
)
|
|
|
|
|
|
|
|
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, "extra": "allow"}
|
|
|
|
|
|
@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())
|
|
params = _body_to_params(body)
|
|
items, _ = svc.list_process_board(
|
|
db,
|
|
tenant_no,
|
|
user,
|
|
**_process_board_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)
|
|
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,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleMaintenanceProcessRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
return svc.get_process(db, tenant_no, user, id)
|
|
|
|
|
|
@process_router.post("/", response_model=VehicleMaintenanceProcessRead, status_code=status.HTTP_201_CREATED)
|
|
def create_process(
|
|
payload: VehicleMaintenanceProcessCreate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleMaintenanceProcessRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
return svc.create_process(db, tenant_no, user, payload)
|
|
|
|
|
|
@process_router.put("/{id}", response_model=VehicleMaintenanceProcessRead)
|
|
def update_process(
|
|
id: str,
|
|
payload: VehicleMaintenanceProcessUpdate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleMaintenanceProcessRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
return svc.update_process(db, tenant_no, user, id, payload)
|
|
|
|
|
|
@process_router.delete("/{ids}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_processes(
|
|
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())
|
|
process_ids = [item for item in ids.split(",") if item]
|
|
svc.delete_processes(db, tenant_no, user, process_ids)
|