feat(pl): bind sys user on personnel and snapshot selectedEscortId

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/pl/base_data.py

@@ -68,6 +68,7 @@ from app.services.pl_personnel_service import (
     load_vehicle_personnel_summaries,
     save_vehicle_personnel,
 )
+from app.services.pl_personnel_user_bind import assert_sys_user_bind_available
 from app.schemas.pl.paste_recognize import PasteCommitBody, PasteCommitResult
 from app.services.pl_paste_recognize_service import commit_paste_recognize
 
@@ -865,6 +866,35 @@ def _personnel_to_read(
     return data
 
 
+def _assert_personnel_sys_user_bind(
+    db: Session,
+    tenant_no: str,
+    *,
+    personnel_id: str,
+    sys_user_id: Optional[str],
+) -> None:
+    uid = (sys_user_id or "").strip()
+    if not uid:
+        return
+    existing = (
+        db.query(PlPersonnel)
+        .filter(
+            PlPersonnel.tenant_no == tenant_no,
+            PlPersonnel.sys_user_id == uid,
+            PlPersonnel.del_flag == 0,
+        )
+        .all()
+    )
+    try:
+        assert_sys_user_bind_available(
+            existing=existing,
+            personnel_id=personnel_id,
+            sys_user_id=uid,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
 def _personnel_to_detail(
     db: Session, tenant_no: str, personnel: PlPersonnel
 ) -> PersonnelDetailRead:
@@ -1083,8 +1113,16 @@ def create_personnel(
         subject_label=f"「{payload.personnel_name}」",
     )
 
+    personnel_id = generate_uuid()
+    _assert_personnel_sys_user_bind(
+        db,
+        tenant_no,
+        personnel_id=personnel_id,
+        sys_user_id=payload.sys_user_id,
+    )
+
     personnel = PlPersonnel(
-        id=generate_uuid(),
+        id=personnel_id,
         tenant_no=tenant_no,
         personnel_name=payload.personnel_name,
         phone_no=payload.phone_no,
@@ -1094,6 +1132,7 @@ def create_personnel(
         personnel_affiliation=payload.personnel_affiliation,
         status=payload.status,
         photo_urls=payload.photo_urls,
+        sys_user_id=(payload.sys_user_id or "").strip() or None,
         remark=payload.remark,
         create_by=user.id,
         create_dept_id=user.dept_id,
@@ -1162,6 +1201,16 @@ def update_personnel(
         subject_label=f"「{subject_name}」",
     )
 
+    if "sys_user_id" in data:
+        raw_uid = data.get("sys_user_id")
+        data["sys_user_id"] = (raw_uid or "").strip() or None
+        _assert_personnel_sys_user_bind(
+            db,
+            tenant_no,
+            personnel_id=personnel.id,
+            sys_user_id=data["sys_user_id"],
+        )
+
     for key, value in data.items():
         setattr(personnel, key, value)
 

backend/app/schemas/pl/base_data.py

@@ -258,6 +258,9 @@ class PersonnelBase(BaseModel):
     photo_urls: Optional[str] = Field(
         default=None, max_length=2000, alias="photoUrls", description="人员照片URL，逗号分隔（默认1张）"
     )
+    sys_user_id: Optional[str] = Field(
+        default=None, alias="sysUserId", max_length=64, description="关联系统用户ID"
+    )
     remark: Optional[str] = None
 
 
@@ -279,6 +282,7 @@ class PersonnelUpdate(BaseModel):
     photo_urls: Optional[str] = Field(
         default=None, max_length=2000, alias="photoUrls", description="人员照片URL，逗号分隔（默认1张）"
     )
+    sys_user_id: Optional[str] = Field(default=None, alias="sysUserId", max_length=64)
     remark: Optional[str] = None
     cert_list: Optional[list[CertRow]] = Field(default=None, alias="certList")
     model_config = {"populate_by_name": True}

backend/app/schemas/pl/resource_snapshot.py

@@ -58,6 +58,7 @@ class TransportResourceSnapshot(BaseModel):
     personnel_list: list[SnapshotPersonnelRow] = Field(default_factory=list, alias="personnelList")
     personnel_summary: Optional[str] = Field(default=None, alias="personnelSummary")
     selected_driver_id: Optional[str] = Field(default=None, alias="selectedDriverId")
+    selected_escort_id: Optional[str] = Field(default=None, alias="selectedEscortId")
 
     model_config = {"populate_by_name": True}
 

backend/app/services/pl_personnel_user_bind.py

@@ -0,0 +1,17 @@
+from typing import Iterable, Optional
+
+
+def assert_sys_user_bind_available(
+    *, existing: Iterable, personnel_id: str, sys_user_id: Optional[str]
+) -> None:
+    uid = (sys_user_id or "").strip()
+    if not uid:
+        return
+    for row in existing:
+        if getattr(row, "del_flag", 0) == 1:
+            continue
+        if (getattr(row, "sys_user_id", None) or "").strip() != uid:
+            continue
+        if row.id == personnel_id:
+            continue
+        raise ValueError("该系统用户已关联其他驾押人员")

backend/app/services/pl_transport_resource_snapshot_service.py

@@ -43,6 +43,7 @@ def build_transport_resource_snapshot(
     vehicle_id: str,
     *,
     selected_driver_id: Optional[str] = None,
+    selected_escort_id: Optional[str] = None,
 ) -> TransportResourceSnapshot:
     """读取车辆当前主数据，生成安排运输用整包快照。"""
     vehicle = (
@@ -116,6 +117,7 @@ def build_transport_resource_snapshot(
         personnel_list=personnel_list,
         personnel_summary=personnel_summary or None,
         selected_driver_id=selected_driver_id,
+        selected_escort_id=selected_escort_id,
     )
 
 

backend/tests/test_pl_personnel_sys_user_bind.py

@@ -0,0 +1,34 @@
+from app.services.pl_personnel_user_bind import assert_sys_user_bind_available
+
+
+class FakeRow:
+    def __init__(self, id, sys_user_id, del_flag=0):
+        self.id = id
+        self.sys_user_id = sys_user_id
+        self.del_flag = del_flag
+
+
+def test_bind_ok_when_empty():
+    assert_sys_user_bind_available(
+        existing=[], personnel_id="p1", sys_user_id="u1"
+    )
+
+
+def test_bind_rejects_other_personnel():
+    try:
+        assert_sys_user_bind_available(
+            existing=[FakeRow("p2", "u1")],
+            personnel_id="p1",
+            sys_user_id="u1",
+        )
+        assert False, "expected ValueError"
+    except ValueError as e:
+        assert "已关联" in str(e)
+
+
+def test_bind_allows_self():
+    assert_sys_user_bind_available(
+        existing=[FakeRow("p1", "u1")],
+        personnel_id="p1",
+        sys_user_id="u1",
+    )