shanxia2.0/backend/app/services/pl_dispatch_calendar_service.py

"""手工派车 CRUD：租户、角色数据范围和关联主数据校验。"""
import re
from datetime import date
from typing import Optional

from fastapi import HTTPException
from sqlalchemy.orm import Session

from app.core.db_auth import DbUserContext
from app.core.utils import generate_uuid
from app.models.crm_models import CrmProduct
from app.models.pl_dispatch_calendar import PlDispatchCalendar
from app.models.pl_models import PlVehicle
from app.schemas.pl.dispatch_calendar import CreateBody, UpdateBody
from app.services.pl_data_scope import apply_pl_scope

CALENDAR_KINDS = ("acid", "alkali", "other")


def _calendar_kind(kind: Optional[str]) -> str:
    value = kind or "acid"
    if value not in CALENDAR_KINDS:
        raise HTTPException(status_code=422, detail="日历类型须为 acid、alkali 或 other")
    return value


def _row_to_dict(row: PlDispatchCalendar) -> dict:
    return {
        "id": row.id,
        "dispatchDate": row.dispatch_date,
        "vehicleId": row.vehicle_id,
        "licensePlate": row.license_plate,
        "productId": row.product_id,
        "productName": row.product_name,
        "origin": row.origin,
        "destination": row.destination,
        "quantity": row.quantity,
        "calendarKind": row.calendar_kind or "acid",
        "remark": row.remark,
        "createTime": row.create_time,
        "updateTime": row.update_time,
    }


def _query(db: Session, tenant_no: str, user: DbUserContext, model=PlDispatchCalendar):
    query = db.query(model).filter(model.tenant_no == tenant_no, model.del_flag == 0)
    return apply_pl_scope(query, model, user, db, tenant_no)


def _month_bounds(month: str):
    if not re.fullmatch(r"[0-9]{4}-[0-9]{2}", month or ""):
        raise HTTPException(status_code=422, detail="月份须为 YYYY-MM")
    year, number = map(int, month.split("-"))
    if not 1900 <= year <= 2100 or not 1 <= number <= 12:
        raise HTTPException(status_code=422, detail="月份须在 1900-01 至 2100-12 范围内")
    return date(year, number, 1), date(year + (number == 12), number % 12 + 1, 1)


def list_records(db: Session, tenant_no: str, user: DbUserContext, *, month: str,
                 vehicle_id: Optional[str] = None, calendar_kind: Optional[str] = None,
                 page: int = 1, page_size: int = 500) -> dict:
    start, end = _month_bounds(month)
    query = _query(db, tenant_no, user).filter(
        PlDispatchCalendar.dispatch_date >= start, PlDispatchCalendar.dispatch_date < end,
        PlDispatchCalendar.calendar_kind == _calendar_kind(calendar_kind),
    )
    if vehicle_id:
        query = query.filter(PlDispatchCalendar.vehicle_id == vehicle_id)
    total = query.count()
    rows = query.order_by(PlDispatchCalendar.create_time.desc(), PlDispatchCalendar.id.desc()).offset(
        (page - 1) * page_size
    ).limit(page_size).all()
    return {"list": [_row_to_dict(row) for row in rows], "total": total}


def get_record(db: Session, tenant_no: str, user: DbUserContext, record_id: str) -> PlDispatchCalendar:
    row = _query(db, tenant_no, user).filter(PlDispatchCalendar.id == record_id).first()
    if row is None:
        raise HTTPException(status_code=404, detail="派车记录不存在或无权访问")
    return row


def _resolve_snapshots(db: Session, tenant_no: str, user: DbUserContext, body: CreateBody) -> dict:
    vehicle = _query(db, tenant_no, user, PlVehicle).filter(PlVehicle.id == body.vehicle_id).first()
    if vehicle is None:
        raise HTTPException(status_code=422, detail="车辆不存在、已删除或无权访问，请重新选择")
    product = (
        _query(db, tenant_no, user, CrmProduct)
        .filter(CrmProduct.id == body.product_id, CrmProduct.status == "on")
        .first()
    )
    if product is None:
        raise HTTPException(status_code=422, detail="产品不存在、已下架、已删除或无权访问，请重新选择")
    return {"license_plate": vehicle.license_plate, "product_name": product.product_name}


def create_record(db: Session, tenant_no: str, user: DbUserContext, body: CreateBody) -> PlDispatchCalendar:
    snapshots = _resolve_snapshots(db, tenant_no, user, body)
    row = PlDispatchCalendar(
        id=generate_uuid(), tenant_no=tenant_no, create_by=user.id,
        create_dept_id=user.dept_id or "", update_by=user.id,
        **body.model_dump(), **snapshots,
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def update_record(db: Session, tenant_no: str, user: DbUserContext, record_id: str,
                  body: UpdateBody) -> PlDispatchCalendar:
    row = get_record(db, tenant_no, user, record_id)
    if body.calendar_kind != (row.calendar_kind or "acid"):
        raise HTTPException(status_code=422, detail="不能把派车计划改到另一本日历")
    values = {**body.model_dump(), **_resolve_snapshots(db, tenant_no, user, body)}
    for key, value in values.items():
        setattr(row, key, value)
    row.update_by = user.id
    db.commit()
    db.refresh(row)
    return row


def delete_record(db: Session, tenant_no: str, user: DbUserContext, record_id: str) -> int:
    row = get_record(db, tenant_no, user, record_id)
    row.del_flag = 1
    row.update_by = user.id
    db.commit()
    return 1


def _list_options(db: Session, tenant_no: str, user: DbUserContext, model, name_field,
                  alias: str, q: Optional[str], page: int, page_size: int) -> dict:
    query = _query(db, tenant_no, user, model)
    keyword = (q or "").strip()
    if keyword:
        query = query.filter(name_field.contains(keyword, autoescape=True))
    total = query.count()
    rows = query.order_by(model.create_time.desc(), model.id.desc()).offset(
        (page - 1) * page_size
    ).limit(page_size).all()
    return {"list": [{"id": row.id, alias: getattr(row, name_field.key)} for row in rows], "total": total}


def list_vehicles(db: Session, tenant_no: str, user: DbUserContext, *, q: Optional[str] = None,
                  page: int = 1, page_size: int = 500) -> dict:
    return _list_options(db, tenant_no, user, PlVehicle, PlVehicle.license_plate,
                         "licensePlate", q, page, page_size)


def list_products(db: Session, tenant_no: str, user: DbUserContext, *, q: Optional[str] = None,
                  page: int = 1, page_size: int = 500) -> dict:
    query = _query(db, tenant_no, user, CrmProduct).filter(CrmProduct.status == "on")
    keyword = (q or "").strip()
    if keyword:
        query = query.filter(CrmProduct.product_name.contains(keyword, autoescape=True))
    total = query.count()
    rows = query.order_by(CrmProduct.create_time.desc(), CrmProduct.id.desc()).offset(
        (page - 1) * page_size
    ).limit(page_size).all()
    return {"list": [{"id": row.id, "productName": row.product_name} for row in rows], "total": total}