diff --git a/backend/app/models/duty_models.py b/backend/app/models/duty_models.py index 9767fc8eb..6fbae1d5a 100644 --- a/backend/app/models/duty_models.py +++ b/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): diff --git a/backend/app/services/duty_schedule.py b/backend/app/services/duty_schedule.py index ba093f1e1..8163f988a 100644 --- a/backend/app/services/duty_schedule.py +++ b/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", ] diff --git a/backend/tests/test_duty_schedule.py b/backend/tests/test_duty_schedule.py index 6cadb5295..abc675c90 100644 --- a/backend/tests/test_duty_schedule.py +++ b/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")