309 lines
10 KiB
Python
309 lines
10 KiB
Python
"""运输计划 - 必备证书校验"""
|
||
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"
|
||
|
||
|
||
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 (trailer_no or "").strip():
|
||
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:
|
||
errors: list[str] = []
|
||
errors.extend(
|
||
collect_required_cert_errors(
|
||
cert_rows, CARRIER_HEAD_REQUIRED, part_label="主车"
|
||
)
|
||
)
|
||
if (trailer_no or "").strip():
|
||
errors.extend(
|
||
collect_required_cert_errors(
|
||
cert_rows, CARRIER_TRAILER_REQUIRED, part_label="挂车"
|
||
)
|
||
)
|
||
if errors:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"必备证书未齐全：{'；'.join(errors)}",
|
||
)
|
||
|
||
|
||
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)
|