提交

backend/app/api/pl/base_data.py

@@ -48,6 +48,10 @@ from app.services.pl_cert_service import (
     nearest_expire_date,
     save_owner_certs,
 )
+from app.services.pl_cert_required_service import (
+    validate_personnel_required_certs,
+    validate_vehicle_save_certs,
+)
 from app.services.pl_carrier_service import (
     carrier_eligible_for_transport,
     carrier_qualification_complete,
@@ -409,6 +413,15 @@ def create_vehicle(
     if existing:
         raise HTTPException(status_code=400, detail="车牌号已存在")
 
+    validate_vehicle_save_certs(
+        db,
+        tenant_no,
+        vehicle_type=payload.vehicle_type,
+        trailer_no=payload.trailer_no,
+        cert_rows=payload.cert_list,
+        personnel_rows=payload.personnel_list,
+    )
+
     vehicle = PlVehicle(
         id=generate_uuid(),
         tenant_no=tenant_no,
@@ -434,6 +447,7 @@ def create_vehicle(
             tenant_no,
             user_id=user.id,
             dept_id=user.dept_id,
+            vehicle_type=payload.vehicle_type,
         )
         save_owner_certs(
             db,
@@ -483,6 +497,26 @@ def update_vehicle(
         exclude_unset=True, exclude={"personnel_list", "cert_list"}
     )
 
+    effective_vehicle_type = data.get("vehicle_type", vehicle.vehicle_type)
+    effective_trailer_no = data.get("trailer_no", vehicle.trailer_no)
+    effective_cert_rows = (
+        cert_list
+        if cert_list is not None
+        else list_owner_certs(db, tenant_no, "vehicle", vehicle.id)
+    )
+    effective_personnel_rows = personnel_list
+    if effective_personnel_rows is None:
+        effective_personnel_rows = load_vehicle_personnel_rows(db, tenant_no, vehicle.id)
+
+    validate_vehicle_save_certs(
+        db,
+        tenant_no,
+        vehicle_type=effective_vehicle_type,
+        trailer_no=effective_trailer_no,
+        cert_rows=effective_cert_rows,
+        personnel_rows=effective_personnel_rows,
+    )
+
     for key, value in data.items():
         setattr(vehicle, key, value)
 
@@ -497,6 +531,7 @@ def update_vehicle(
                 tenant_no,
                 user_id=user.id,
                 dept_id=user.dept_id,
+                vehicle_type=effective_vehicle_type,
             )
         if cert_list is not None:
             save_owner_certs(
@@ -581,6 +616,9 @@ def _personnel_to_detail(
 def list_personnel(
     q: Optional[str] = Query(default=None, description="模糊搜索姓名/手机号"),
     status: Optional[str] = Query(default=None, description="状态"),
+    personnel_affiliation: Optional[str] = Query(
+        default=None, alias="personnelAffiliation", description="人员归属 self/carrier"
+    ),
     page: int = Query(default=1, ge=1),
     page_size: int = Query(default=20, ge=1, le=100, alias="pageSize"),
     db: Session = Depends(get_db),
@@ -595,6 +633,9 @@ def list_personnel(
     if status:
         query = query.filter(PlPersonnel.status == status)
 
+    if personnel_affiliation:
+        query = query.filter(PlPersonnel.personnel_affiliation == personnel_affiliation)
+
     if q:
         query = query.filter(
             or_(
@@ -626,6 +667,12 @@ def create_personnel(
 ) -> PersonnelDetailRead:
     tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
 
+    validate_personnel_required_certs(
+        payload.cert_list,
+        payload.personnel_type,
+        subject_label=f"「{payload.personnel_name}」",
+    )
+
     personnel = PlPersonnel(
         id=generate_uuid(),
         tenant_no=tenant_no,
@@ -633,6 +680,8 @@ def create_personnel(
         phone_no=payload.phone_no,
         age=payload.age,
         id_card_no=payload.id_card_no,
+        personnel_type=payload.personnel_type,
+        personnel_affiliation=payload.personnel_affiliation,
         status=payload.status,
         remark=payload.remark,
         create_by=user.id,
@@ -684,6 +733,20 @@ def update_personnel(
     cert_list = payload.cert_list
     data = payload.model_dump(exclude_unset=True, exclude={"cert_list"})
 
+    effective_personnel_type = data.get("personnel_type", personnel.personnel_type)
+    effective_cert_rows = (
+        cert_list
+        if cert_list is not None
+        else list_owner_certs(db, tenant_no, "personnel", personnel.id)
+    )
+    subject_name = data.get("personnel_name", personnel.personnel_name)
+
+    validate_personnel_required_certs(
+        effective_cert_rows,
+        effective_personnel_type,
+        subject_label=f"「{subject_name}」",
+    )
+
     for key, value in data.items():
         setattr(personnel, key, value)