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

"""车油卡台账 CRUD，租户隔离、数据范围与软删除。"""
from __future__ import annotations

from typing import List, Optional

from fastapi import HTTPException
from sqlalchemy import or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core.db_auth import DbUserContext
from app.core.utils import generate_uuid
from app.models.pl_vehicle_fuel_card import PlVehicleFuelCard
from app.schemas.pl.vehicle_fuel_card import CreateBody, UpdateBody
from app.services.pl_data_scope import apply_pl_scope
from app.services.pl_vehicle_mine_vehicles import (
    license_plates_for_user_personnel,
    should_restrict_mine_vehicles,
)


def _row_to_dict(row: PlVehicleFuelCard) -> dict:
    return {
        "id": row.id,
        "licensePlate": row.license_plate,
        "cardNo": row.card_no,
        "remark": row.remark,
        "createTime": row.create_time,
        "updateTime": row.update_time,
    }


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


def list_cards(db: Session, tenant_no: str, user: DbUserContext, *, q: Optional[str] = None,
               page: int = 1, page_size: int = 20) -> dict:
    query = _query(db, tenant_no, user)
    keyword = (q or "").strip()
    if keyword:
        query = query.filter(or_(
            PlVehicleFuelCard.license_plate.contains(keyword, autoescape=True),
            PlVehicleFuelCard.card_no.contains(keyword, autoescape=True),
            PlVehicleFuelCard.remark.contains(keyword, autoescape=True),
        ))
    total = query.count()
    rows = query.order_by(PlVehicleFuelCard.create_time.desc(), PlVehicleFuelCard.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_card(db: Session, tenant_no: str, user: DbUserContext, record_id: str) -> PlVehicleFuelCard:
    row = _query(db, tenant_no, user).filter(PlVehicleFuelCard.id == record_id).first()
    if not row:
        raise HTTPException(status_code=404, detail="车油卡信息不存在或无权访问")
    return row


def _commit_binding(db: Session) -> None:
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        raise HTTPException(status_code=409, detail="该车牌或油卡卡号已存在有效绑定，请检查后重试")


def create_card(db: Session, tenant_no: str, user: DbUserContext, body: CreateBody) -> PlVehicleFuelCard:
    row = PlVehicleFuelCard(
        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(),
    )
    db.add(row)
    _commit_binding(db)
    db.refresh(row)
    return row


def update_card(db: Session, tenant_no: str, user: DbUserContext, record_id: str,
                body: UpdateBody) -> PlVehicleFuelCard:
    row = get_card(db, tenant_no, user, record_id)
    for key, value in body.model_dump().items():
        setattr(row, key, value)
    row.update_by = user.id
    _commit_binding(db)
    db.refresh(row)
    return row


def delete_cards(db: Session, tenant_no: str, user: DbUserContext, ids: List[str]) -> int:
    unique_ids = set(ids)
    rows = _query(db, tenant_no, user).filter(PlVehicleFuelCard.id.in_(unique_ids)).with_for_update().all()
    if len(rows) != len(unique_ids):
        raise HTTPException(status_code=404, detail="部分车油卡信息不存在或无权访问，未删除任何记录")
    for row in rows:
        row.del_flag = 1
        row.update_by = user.id
    db.commit()
    return len(rows)


def lookup_for_recharge(db: Session, tenant_no: str, user: DbUserContext, license_plate: str,
                        *, mine_vehicles: bool = False) -> dict:
    """充值表单读取关联主数据：权限取决于车辆/单据访问权，而非台账创建人。

    不开放台账列表或管理权限；精确车牌查询只返回卡号，不返回备注等资料。
    小程序继续遵循本人挂靠车辆与管理员放宽规则。
    """
    from app.models.pl_models import PlVehicle, PlVehicleRechargeApply, PlVehicleRechargeRecord

    plate = license_plate.strip().upper()
    if should_restrict_mine_vehicles(db, tenant_no, user.id, mine_vehicles=mine_vehicles):
        permitted = plate in license_plates_for_user_personnel(db, tenant_no, user.id)
    elif mine_vehicles or user.role == "super_admin":
        # mineVehicles=true 能到达这里，表示 helper 已核实管理员角色；与充值列表一致。
        permitted = True
    else:
        permitted = False
        # 历史单据的车牌可能已从车辆档案移除，仍可读取其关联卡号。
        for model in (PlVehicle, PlVehicleRechargeApply, PlVehicleRechargeRecord):
            query = db.query(model).filter(
                model.tenant_no == tenant_no, model.del_flag == 0, model.license_plate == plate
            )
            if apply_pl_scope(query, model, user, db, tenant_no).first() is not None:
                permitted = True
                break
    if not permitted:
        raise HTTPException(status_code=403, detail="无权查询该车辆的油卡信息")
    row = db.query(PlVehicleFuelCard.card_no).filter(
        PlVehicleFuelCard.tenant_no == tenant_no,
        PlVehicleFuelCard.del_flag == 0,
        PlVehicleFuelCard.license_plate == plate,
    ).first()
    return {"licensePlate": plate, "cardNo": row[0] if row else None}


def attach_recharge_list_bindings(db: Session, tenant_no: str, items: list) -> list:
    """为已完成业务权限过滤的当前页补充绑定卡号，单次查询且不覆盖历史发放卡号。"""
    plates = {item.license_plate for item in items if item.recharge_type == "fuel"}
    cards = dict(db.query(PlVehicleFuelCard.license_plate, PlVehicleFuelCard.card_no).filter(
        PlVehicleFuelCard.tenant_no == tenant_no,
        PlVehicleFuelCard.del_flag == 0,
        PlVehicleFuelCard.license_plate.in_(plates),
    ).all()) if plates else {}
    for item in items:
        item.bound_fuel_card_no = cards.get(item.license_plate) if item.recharge_type == "fuel" else None
    return items