shanxia2.0/backend/app/api/pl/vehicle_maintenance.py

"""车辆维保 API"""
from datetime import date
from datetime import 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_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"),
        applicant_id=_param_get(params, "applicantId", "applicant_id"),
        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"),
        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,
    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)


@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)