feat(duty): restrict assign and grab to duty staff role

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

backend/app/models/duty_models.py

@@ -17,6 +17,9 @@ LINK_OPEN = "open"
 LINK_CLOSED = "closed"
 SOURCE_ASSIGN = "assign"
 SOURCE_GRAB = "grab"
+# 可排班名单角色（sys_role 固定 id；无菜单，仅标签）
+DUTY_STAFF_ROLE_ID = "R65CAAE057914CCCAE2A81BE627A5101"
+DUTY_STAFF_ROLE_NAME = "值班人员"
 
 
 class DutyDay(Base):

backend/app/services/duty_schedule.py

@@ -6,11 +6,14 @@ import secrets
 from datetime import date, datetime
 from typing import Any, Dict, List, Optional, Sequence
 
+from sqlalchemy import and_
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
 from app.core.utils import generate_id
 from app.models.duty_models import (
+    DUTY_STAFF_ROLE_ID,
+    DUTY_STAFF_ROLE_NAME,
     LINK_CLOSED,
     LINK_OPEN,
     SHIFT_LABELS,
@@ -20,7 +23,7 @@ from app.models.duty_models import (
     DutyDay,
     DutySlot,
 )
-from app.models.models import SysUser
+from app.models.models import SysRole, SysUser, SysUserRole
 from app.schemas.duty import DutyDaySaveBody, DutySlotSaveIn
 
 PERSON_NAME_MAX_LEN = 20
@@ -348,6 +351,8 @@ def save_day(
             # 若只有 userId，允许 personName 已由上层填好
             if not person_name:
                 raise DutyBizError("指定员工姓名不能为空")
+            if item.source == SOURCE_ASSIGN and uid:
+                assert_user_is_duty_staff(db, tenant_no, uid)
             db.add(
                 DutyAssignment(
                     id=generate_id(),
@@ -375,9 +380,48 @@ def save_day(
     return _serialize_day(db, day)
 
 
-def _list_public_users(db: Session, tenant_no: str) -> List[Dict[str, str]]:
+def resolve_duty_staff_role_id(db: Session, tenant_no: str) -> Optional[str]:
+    """按固定 id 查找「值班人员」角色；找不到再按角色名兜底。"""
+    role = (
+        db.query(SysRole)
+        .filter(
+            SysRole.id == DUTY_STAFF_ROLE_ID,
+            SysRole.tenant_no == tenant_no,
+            SysRole.del_flag == 0,
+        )
+        .first()
+    )
+    if role:
+        return role.id
+    role = (
+        db.query(SysRole)
+        .filter(
+            SysRole.role_name == DUTY_STAFF_ROLE_NAME,
+            SysRole.tenant_no == tenant_no,
+            SysRole.del_flag == 0,
+            SysRole.status == "1",
+        )
+        .first()
+    )
+    return role.id if role else None
+
+
+def list_duty_staff_users(db: Session, tenant_no: str) -> List[Dict[str, str]]:
+    """本租户启用且绑定「值班人员」角色的用户（公开页 users / 管理端 options）。"""
+    role_id = resolve_duty_staff_role_id(db, tenant_no)
+    if not role_id:
+        return []
     rows = (
         db.query(SysUser)
+        .join(
+            SysUserRole,
+            and_(
+                SysUserRole.user_id == SysUser.id,
+                SysUserRole.role_id == role_id,
+                SysUserRole.del_flag == 0,
+                SysUserRole.tenant_no == tenant_no,
+            ),
+        )
         .filter(
             SysUser.tenant_no == tenant_no,
             SysUser.del_flag == 0,
@@ -387,7 +431,11 @@ def _list_public_users(db: Session, tenant_no: str) -> List[Dict[str, str]]:
         .all()
     )
     out: List[Dict[str, str]] = []
+    seen = set()
     for r in rows:
+        if r.id in seen:
+            continue
+        seen.add(r.id)
         label = (r.nick_name or "").strip() or (r.user_no or "").strip()
         if not label:
             continue
@@ -395,6 +443,28 @@ def _list_public_users(db: Session, tenant_no: str) -> List[Dict[str, str]]:
     return out
 
 
+def assert_user_is_duty_staff(db: Session, tenant_no: str, user_id: str) -> None:
+    role_id = resolve_duty_staff_role_id(db, tenant_no)
+    if not role_id:
+        raise DutyBizError("未配置「值班人员」角色，请联系管理员")
+    link = (
+        db.query(SysUserRole)
+        .filter(
+            SysUserRole.user_id == user_id,
+            SysUserRole.role_id == role_id,
+            SysUserRole.tenant_no == tenant_no,
+            SysUserRole.del_flag == 0,
+        )
+        .first()
+    )
+    if not link:
+        raise DutyBizError("仅可选择「值班人员」角色的用户")
+
+
+def _list_public_users(db: Session, tenant_no: str) -> List[Dict[str, str]]:
+    return list_duty_staff_users(db, tenant_no)
+
+
 def _resolve_grab_user(db: Session, tenant_no: str, user_id: str) -> SysUser:
     uid = (user_id or "").strip()
     if not uid:
@@ -411,6 +481,7 @@ def _resolve_grab_user(db: Session, tenant_no: str, user_id: str) -> SysUser:
     )
     if not user:
         raise DutyBizError("用户无效或不存在")
+    assert_user_is_duty_staff(db, tenant_no, user.id)
     return user
 
 
@@ -525,4 +596,7 @@ __all__ = [
     "soft_delete_assignment",
     "get_public_page",
     "grab",
+    "resolve_duty_staff_role_id",
+    "list_duty_staff_users",
+    "assert_user_is_duty_staff",
 ]

backend/tests/test_duty_schedule.py

@@ -6,6 +6,8 @@ from unittest.mock import MagicMock
 import pytest
 
 from app.models.duty_models import (
+    DUTY_STAFF_ROLE_ID,
+    DUTY_STAFF_ROLE_NAME,
     LINK_CLOSED,
     LINK_OPEN,
     SHIFT_EVENING,
@@ -16,8 +18,11 @@ from app.models.duty_models import (
 from app.schemas.duty import DutyAssignmentIn, DutyDaySaveBody, DutySlotSaveIn
 from app.services.duty_schedule import (
     DutyBizError,
+    assert_user_is_duty_staff,
     grab,
+    list_duty_staff_users,
     normalize_person_name,
+    resolve_duty_staff_role_id,
     save_day,
 )
 
@@ -72,6 +77,11 @@ def test_grab_rejects_when_full():
     slot_q.filter.return_value.with_for_update.return_value.first.return_value = slot
     user_q = MagicMock()
     user_q.filter.return_value.first.return_value = user
+    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
+    role_q = MagicMock()
+    role_q.filter.return_value.first.return_value = role
+    ur_q = MagicMock()
+    ur_q.filter.return_value.first.return_value = SimpleNamespace(user_id="u1", role_id=DUTY_STAFF_ROLE_ID)
 
     def query_side_effect(model):
         name = getattr(model, "__name__", str(model))
@@ -81,6 +91,10 @@ def test_grab_rejects_when_full():
             return slot_q
         if name == "SysUser":
             return user_q
+        if name == "SysRole":
+            return role_q
+        if name == "SysUserRole":
+            return ur_q
         # DutyAssignment count / duplicate
         aq = MagicMock()
         aq.filter.return_value.count.return_value = 1
@@ -173,6 +187,11 @@ def test_grab_rejects_duplicate_same_shift():
     slot_q.filter.return_value.with_for_update.return_value.first.return_value = slot
     user_q = MagicMock()
     user_q.filter.return_value.first.return_value = user
+    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
+    role_q = MagicMock()
+    role_q.filter.return_value.first.return_value = role
+    ur_q = MagicMock()
+    ur_q.filter.return_value.first.return_value = SimpleNamespace(user_id="u1", role_id=DUTY_STAFF_ROLE_ID)
 
     assign_calls = {"n": 0}
 
@@ -184,6 +203,10 @@ def test_grab_rejects_duplicate_same_shift():
             return slot_q
         if name == "SysUser":
             return user_q
+        if name == "SysRole":
+            return role_q
+        if name == "SysUserRole":
+            return ur_q
         aq = MagicMock()
         # first: duplicate check .first(); later: count
         assign_calls["n"] += 1
@@ -269,3 +292,86 @@ def test_save_day_rejects_shrink_below_grab_count():
 
     with pytest.raises(DutyBizError, match="超过名额|缩容|报名"):
         save_day(db, "t1", {"user_id": "op1"}, body)
+
+
+def test_resolve_duty_staff_role_id_by_fixed_id():
+    db = MagicMock()
+    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID, role_name=DUTY_STAFF_ROLE_NAME, del_flag=0)
+    q = MagicMock()
+    q.filter.return_value.first.return_value = role
+    db.query.return_value = q
+    assert resolve_duty_staff_role_id(db, "shanxia2.0") == DUTY_STAFF_ROLE_ID
+
+
+def test_list_duty_staff_users_empty_when_no_role():
+    db = MagicMock()
+    q = MagicMock()
+    q.filter.return_value.first.return_value = None  # resolve role → None twice
+    db.query.return_value = q
+    assert list_duty_staff_users(db, "t1") == []
+
+
+def test_assert_user_is_duty_staff_rejects_without_role():
+    db = MagicMock()
+    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
+    role_q = MagicMock()
+    role_q.filter.return_value.first.return_value = role
+    ur_q = MagicMock()
+    ur_q.filter.return_value.first.return_value = None
+
+    def query_side_effect(model):
+        name = getattr(model, "__name__", "")
+        if name == "SysRole":
+            return role_q
+        if name == "SysUserRole":
+            return ur_q
+        return MagicMock()
+
+    db.query.side_effect = query_side_effect
+    with pytest.raises(DutyBizError, match="值班人员"):
+        assert_user_is_duty_staff(db, "t1", "u1")
+
+
+def test_grab_rejects_non_duty_staff():
+    db = MagicMock()
+    day = SimpleNamespace(
+        id="day1",
+        tenant_no="t1",
+        link_status=LINK_OPEN,
+        duty_date=date(2026, 8, 20),
+        access_token="tok",
+        del_flag=0,
+    )
+    user = SimpleNamespace(
+        id="u1",
+        tenant_no="t1",
+        nick_name="李四",
+        user_no="lisi",
+        status="1",
+        del_flag=0,
+    )
+    day_q = MagicMock()
+    day_q.filter.return_value.with_for_update.return_value.first.return_value = day
+    user_q = MagicMock()
+    user_q.filter.return_value.first.return_value = user
+    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
+    role_q = MagicMock()
+    role_q.filter.return_value.first.return_value = role
+    ur_q = MagicMock()
+    ur_q.filter.return_value.first.return_value = None
+
+    def query_side_effect(model):
+        name = getattr(model, "__name__", "")
+        if name == "DutyDay":
+            return day_q
+        if name == "SysUser":
+            return user_q
+        if name == "SysRole":
+            return role_q
+        if name == "SysUserRole":
+            return ur_q
+        return MagicMock()
+
+    db.query.side_effect = query_side_effect
+    with pytest.raises(DutyBizError, match="值班人员"):
+        grab(db, "tok", SHIFT_EVENING, "u1")