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

"""运输计划 - 必备证书校验"""
from __future__ import annotations

from typing import Iterable, Optional, Sequence

from fastapi import HTTPException
from sqlalchemy.orm import Session

from app.models.pl_models import PlPersonnel
from app.schemas.pl.base_data import CertRow, VehiclePersonnelRow
from app.services.pl_cert_service import list_owner_certs

VEHICLE_TYPE_SELF = "self"
VEHICLE_TYPE_CARRIER = "carrier"

SELF_HEAD_REQUIRED = ("行驶证", "营运证", "保险单")
SELF_TRAILER_REQUIRED = ("行驶证", "营运证", "保险单", "罐检报告")

CARRIER_HEAD_REQUIRED = ("行驶证", "营运证")
CARRIER_TRAILER_REQUIRED = ("行驶证", "营运证")

PERSONNEL_DRIVER_REQUIRED = ("身份证", "驾驶证", "驾驶员从业资格证")
PERSONNEL_ESCORT_REQUIRED = ("身份证", "押运员从业资格证")
PERSONNEL_BOTH_REQUIRED = (
    "身份证",
    "驾驶证",
    "驾驶员从业资格证",
    "押运员从业资格证",
)

VEHICLE_PART_HEAD = "head"
VEHICLE_PART_TRAILER = "trailer"

HUOLALA_TRAILER_NO = "货拉拉"


def is_huolala_trailer_no(trailer_no: str | None) -> bool:
    return (trailer_no or "").strip() == HUOLALA_TRAILER_NO


def should_show_trailer_cert_slots(trailer_no: str | None) -> bool:
    text = (trailer_no or "").strip()
    return bool(text) and not is_huolala_trailer_no(text)


def should_require_trailer_certs(vehicle_type: str | None, trailer_no: str | None) -> bool:
    return (vehicle_type or "").strip() == VEHICLE_TYPE_SELF and should_show_trailer_cert_slots(
        trailer_no
    )


def match_required_cert_name(actual_name: str | None, required_name: str) -> bool:
    name = (actual_name or "").strip()
    if not name:
        return False
    if required_name == "押运员从业资格证":
        return name == "押运员从业资格证" or name == "押运员证"
    return name == required_name


def _cert_row_complete(cert: CertRow | dict, required_name: str) -> bool:
    if isinstance(cert, dict):
        cert_name = cert.get("cert_name") or cert.get("certName")
        expire_date = cert.get("expire_date") or cert.get("expireDate")
    else:
        cert_name = cert.cert_name
        expire_date = cert.expire_date

    if not match_required_cert_name(cert_name, required_name):
        return False
    if not expire_date:
        return False
    # 证书附件非必填（四页车辆/驾押人员统一）
    return True


def _find_cert_for_required(
    cert_rows: Sequence[CertRow | dict],
    required_name: str,
    *,
    vehicle_part: str | None = None,
) -> CertRow | dict | None:
    for cert in cert_rows:
        if isinstance(cert, dict):
            name = cert.get("cert_name") or cert.get("certName")
            part = cert.get("vehicle_part") or cert.get("vehiclePart")
        else:
            name = cert.cert_name
            part = cert.vehicle_part
        if vehicle_part is not None and (part or VEHICLE_PART_HEAD) != vehicle_part:
            continue
        if match_required_cert_name(name, required_name):
            return cert
    return None


def _missing_field_labels(cert: CertRow | dict | None, required_name: str) -> list[str]:
    if cert is None:
        return ["有效期"]
    missing: list[str] = []
    if isinstance(cert, dict):
        expire_date = cert.get("expire_date") or cert.get("expireDate")
        cert_name = cert.get("cert_name") or cert.get("certName")
    else:
        expire_date = cert.expire_date
        cert_name = cert.cert_name

    if not match_required_cert_name(cert_name, required_name):
        return ["有效期"]
    if not expire_date:
        missing.append("有效期")
    # 证书附件非必填
    return missing


def collect_required_cert_errors(
    cert_rows: Sequence[CertRow | dict],
    required_names: Iterable[str],
    *,
    part_label: str = "",
) -> list[str]:
    errors: list[str] = []
    prefix = f"{part_label}" if part_label else ""
    vehicle_part = None
    if part_label in ("主车", "head"):
        vehicle_part = VEHICLE_PART_HEAD
    elif part_label in ("挂车", "trailer"):
        vehicle_part = VEHICLE_PART_TRAILER

    for required_name in required_names:
        cert = _find_cert_for_required(
            cert_rows, required_name, vehicle_part=vehicle_part
        )
        missing = _missing_field_labels(cert, required_name)
        if missing:
            label = f"{prefix}「{required_name}」" if prefix else f"「{required_name}」"
            errors.append(f"{label}缺少：{'、'.join(missing)}")
    return errors


def get_required_personnel_cert_names(personnel_type: str | None) -> tuple[str, ...]:
    ptype = (personnel_type or "").strip()
    if ptype == "escort":
        return PERSONNEL_ESCORT_REQUIRED
    if ptype == "driver":
        return PERSONNEL_DRIVER_REQUIRED
    if ptype == "both":
        return PERSONNEL_BOTH_REQUIRED
    return PERSONNEL_BOTH_REQUIRED


def validate_personnel_type_role_compat(personnel_type: str | None, role: str | None) -> None:
    ptype = (personnel_type or "").strip()
    r = (role or "").strip()
    if not ptype or not r:
        return
    if ptype == "driver" and r == "escort":
        raise HTTPException(status_code=400, detail="驾驶员类型人员不可关联为押运员")
    if ptype == "escort" and r == "driver":
        raise HTTPException(status_code=400, detail="押运员类型人员不可关联为驾驶员")


def validate_personnel_required_certs(
    cert_rows: Sequence[CertRow | dict],
    personnel_type: str | None,
    *,
    personnel_affiliation: str | None = None,
    subject_label: str = "",
) -> None:
    affiliation = (personnel_affiliation or VEHICLE_TYPE_SELF).strip()
    if affiliation == VEHICLE_TYPE_CARRIER:
        return

    ptype = (personnel_type or "").strip()
    if not ptype:
        raise HTTPException(status_code=400, detail="人员类型不能为空")
    required = get_required_personnel_cert_names(ptype)
    errors = collect_required_cert_errors(cert_rows, required)
    if errors:
        prefix = f"人员{subject_label}" if subject_label else "人员"
        raise HTTPException(
            status_code=400,
            detail=f"必备证书未齐全：{prefix}{'；'.join(errors)}",
        )


def validate_self_vehicle_required_certs(
    cert_rows: Sequence[CertRow | dict],
    *,
    trailer_no: str | None = None,
) -> None:
    errors: list[str] = []
    errors.extend(
        collect_required_cert_errors(
            cert_rows, SELF_HEAD_REQUIRED, part_label="主车"
        )
    )
    if should_require_trailer_certs(VEHICLE_TYPE_SELF, trailer_no):
        errors.extend(
            collect_required_cert_errors(
                cert_rows, SELF_TRAILER_REQUIRED, part_label="挂车"
            )
        )
    if errors:
        raise HTTPException(
            status_code=400,
            detail=f"必备证书未齐全：{'；'.join(errors)}",
        )


def validate_carrier_vehicle_required_certs(
    cert_rows: Sequence[CertRow | dict],
    *,
    trailer_no: str | None = None,
) -> None:
    """三方车辆证件不强制；保留函数供兼容调用，恒为通过。"""
    _ = (cert_rows, trailer_no)
    return


def validate_vehicle_personnel_affiliation(
    db: Session,
    tenant_no: str,
    vehicle_type: str | None,
    personnel_rows: Sequence[VehiclePersonnelRow],
) -> None:
    vtype = (vehicle_type or "").strip()
    if vtype not in (VEHICLE_TYPE_SELF, VEHICLE_TYPE_CARRIER):
        return
    label = "山峡" if vtype == VEHICLE_TYPE_SELF else "非山峡"
    for row in personnel_rows:
        personnel_id = (row.personnel_id or "").strip()
        if not personnel_id:
            continue
        personnel = (
            db.query(PlPersonnel)
            .filter(
                PlPersonnel.id == personnel_id,
                PlPersonnel.tenant_no == tenant_no,
                PlPersonnel.del_flag == 0,
            )
            .first()
        )
        if not personnel:
            continue
        affiliation = (personnel.personnel_affiliation or VEHICLE_TYPE_SELF).strip()
        if affiliation != vtype:
            name = (row.personnel_name or personnel.personnel_name or personnel_id).strip()
            raise HTTPException(
                status_code=400,
                detail=f"关联人员「{name}」归属与车辆类型不匹配，{label}车辆仅可关联{label}人员",
            )


def validate_vehicle_personnel_cert_links(
    db: Session,
    tenant_no: str,
    personnel_rows: Sequence[VehiclePersonnelRow],
) -> None:
    for row in personnel_rows:
        personnel_id = (row.personnel_id or "").strip()
        if not personnel_id:
            continue
        personnel = (
            db.query(PlPersonnel)
            .filter(
                PlPersonnel.id == personnel_id,
                PlPersonnel.tenant_no == tenant_no,
                PlPersonnel.del_flag == 0,
            )
            .first()
        )
        if not personnel:
            continue
        validate_personnel_type_role_compat(personnel.personnel_type, row.role)
        cert_rows = list_owner_certs(db, tenant_no, "personnel", personnel_id)
        role = (row.role or "").strip()
        required = (
            PERSONNEL_DRIVER_REQUIRED
            if role == "driver"
            else PERSONNEL_ESCORT_REQUIRED
        )
        errors = collect_required_cert_errors(cert_rows, required)
        if errors:
            name = (row.personnel_name or personnel.personnel_name or personnel_id).strip()
            raise HTTPException(
                status_code=400,
                detail=(
                    f"关联人员「{name}」必备证书未齐全：{'；'.join(errors)}"
                ),
            )


def validate_vehicle_save_certs(
    db: Session,
    tenant_no: str,
    *,
    vehicle_type: str | None,
    trailer_no: str | None,
    cert_rows: Sequence[CertRow | dict],
    personnel_rows: Sequence[VehiclePersonnelRow] | None = None,
) -> None:
    vtype = (vehicle_type or "").strip()
    if vtype == VEHICLE_TYPE_SELF:
        validate_self_vehicle_required_certs(cert_rows, trailer_no=trailer_no)
    if vtype in (VEHICLE_TYPE_SELF, VEHICLE_TYPE_CARRIER) and personnel_rows:
        validate_vehicle_personnel_affiliation(
            db, tenant_no, vtype, personnel_rows
        )
        if vtype == VEHICLE_TYPE_SELF:
            validate_vehicle_personnel_cert_links(db, tenant_no, personnel_rows)