diff --git a/backend/app/schemas/duty.py b/backend/app/schemas/duty.py new file mode 100644 index 000000000..21af9d74b --- /dev/null +++ b/backend/app/schemas/duty.py @@ -0,0 +1,38 @@ +"""值班排班 Pydantic schemas""" +from __future__ import annotations + +from datetime import date +from typing import List, Literal, Optional + +from pydantic import BaseModel, Field + +ShiftType = Literal["morning", "evening", "duty"] +LinkStatus = Literal["open", "closed"] +SourceType = Literal["assign", "grab"] + + +class DutyAssignmentIn(BaseModel): + source: SourceType = "assign" + userId: Optional[str] = None + personName: str = "" + + +class DutySlotSaveIn(BaseModel): + shiftType: ShiftType + capacity: int = Field(ge=1, le=99) + assignments: List[DutyAssignmentIn] = [] + + +class DutyDaySaveBody(BaseModel): + dutyDate: date + slots: List[DutySlotSaveIn] + linkStatus: Optional[LinkStatus] = None + + +class DutyLinkBody(BaseModel): + linkStatus: LinkStatus + + +class DutyGrabBody(BaseModel): + shiftType: ShiftType + personName: str diff --git a/backend/app/services/duty_schedule.py b/backend/app/services/duty_schedule.py new file mode 100644 index 000000000..bc605e6a9 --- /dev/null +++ b/backend/app/services/duty_schedule.py @@ -0,0 +1,473 @@ +"""值班排班领域服务 — docs/superpowers/specs/2026-08-14-duty-schedule-design.md""" +from __future__ import annotations + +import calendar +import secrets +from datetime import date, datetime +from typing import Any, Dict, List, Optional, Sequence + +from sqlalchemy.orm import Session + +from app.core.config import settings +from app.core.utils import generate_id +from app.models.duty_models import ( + LINK_CLOSED, + LINK_OPEN, + SHIFT_LABELS, + SOURCE_ASSIGN, + SOURCE_GRAB, + DutyAssignment, + DutyDay, + DutySlot, +) +from app.schemas.duty import DutyDaySaveBody, DutySlotSaveIn + +PERSON_NAME_MAX_LEN = 20 + + +class DutyBizError(ValueError): + """业务错误（可映射为 HTTP 400）""" + + +def normalize_person_name(name: str) -> str: + raw = (name or "").strip() + if not raw: + raise DutyBizError("姓名不能为空") + if len(raw) > PERSON_NAME_MAX_LEN: + raise DutyBizError(f"姓名不能超过{PERSON_NAME_MAX_LEN}个字符") + return raw + + +def new_access_token() -> str: + return secrets.token_urlsafe(24)[:64] + + +def build_public_url(token: str) -> str: + origin = (settings.PUBLIC_WEB_ORIGIN or "").rstrip("/") + return f"{origin}/#/public/duty/{token}" + + +def _assignment_dict(row: DutyAssignment) -> Dict[str, Any]: + return { + "id": row.id, + "source": row.source, + "userId": row.user_id, + "personName": row.person_name, + } + + +def _slot_dict(slot: DutySlot, assignments: Sequence[DutyAssignment]) -> Dict[str, Any]: + active = [a for a in assignments if a.del_flag == 0] + return { + "id": slot.id, + "shiftType": slot.shift_type, + "shiftLabel": SHIFT_LABELS.get(slot.shift_type, slot.shift_type), + "capacity": slot.capacity, + "taken": len(active), + "assignments": [_assignment_dict(a) for a in active], + } + + +def _day_dict( + day: DutyDay, + slots: Sequence[DutySlot], + assignments_by_slot: Dict[str, List[DutyAssignment]], +) -> Dict[str, Any]: + slot_payload = [ + _slot_dict(s, assignments_by_slot.get(s.id, [])) + for s in sorted(slots, key=lambda x: (x.order_num or 0, x.shift_type or "")) + if s.del_flag == 0 + ] + return { + "id": day.id, + "dutyDate": day.duty_date.isoformat() if day.duty_date else None, + "linkStatus": day.link_status, + "accessToken": day.access_token, + "publicUrl": build_public_url(day.access_token), + "slots": slot_payload, + } + + +def _load_day_by_date(db: Session, tenant_no: str, duty_date: date) -> Optional[DutyDay]: + return ( + db.query(DutyDay) + .filter( + DutyDay.tenant_no == tenant_no, + DutyDay.duty_date == duty_date, + DutyDay.del_flag == 0, + ) + .first() + ) + + +def _load_day_by_id(db: Session, tenant_no: str, day_id: str) -> Optional[DutyDay]: + return ( + db.query(DutyDay) + .filter(DutyDay.id == day_id, DutyDay.tenant_no == tenant_no, DutyDay.del_flag == 0) + .first() + ) + + +def _load_slots(db: Session, day_id: str) -> List[DutySlot]: + return ( + db.query(DutySlot) + .filter(DutySlot.duty_day_id == day_id, DutySlot.del_flag == 0) + .all() + ) + + +def _load_assignments_for_slots( + db: Session, slot_ids: Sequence[str] +) -> Dict[str, List[DutyAssignment]]: + if not slot_ids: + return {} + rows = ( + db.query(DutyAssignment) + .filter(DutyAssignment.duty_slot_id.in_(list(slot_ids)), DutyAssignment.del_flag == 0) + .all() + ) + out: Dict[str, List[DutyAssignment]] = {sid: [] for sid in slot_ids} + for row in rows: + out.setdefault(row.duty_slot_id, []).append(row) + return out + + +def _serialize_day(db: Session, day: DutyDay) -> Dict[str, Any]: + slots = _load_slots(db, day.id) + assigns = _load_assignments_for_slots(db, [s.id for s in slots]) + return _day_dict(day, slots, assigns) + + +def empty_day_skeleton(duty_date: date) -> Dict[str, Any]: + return { + "id": None, + "dutyDate": duty_date.isoformat(), + "linkStatus": LINK_OPEN, + "accessToken": None, + "publicUrl": None, + "slots": [], + } + + +def get_month_summary(db: Session, tenant_no: str, year: int, month: int) -> List[Dict[str, Any]]: + last_day = calendar.monthrange(year, month)[1] + start = date(year, month, 1) + end = date(year, month, last_day) + days = ( + db.query(DutyDay) + .filter( + DutyDay.tenant_no == tenant_no, + DutyDay.del_flag == 0, + DutyDay.duty_date >= start, + DutyDay.duty_date <= end, + ) + .order_by(DutyDay.duty_date.asc()) + .all() + ) + result = [] + for day in days: + payload = _serialize_day(db, day) + # 月历摘要可精简 assignments 人名 + result.append(payload) + return result + + +def get_day_detail(db: Session, tenant_no: str, duty_date: date) -> Dict[str, Any]: + day = _load_day_by_date(db, tenant_no, duty_date) + if not day: + return empty_day_skeleton(duty_date) + return _serialize_day(db, day) + + +def soft_delete_assignment( + db: Session, tenant_no: str, assignment_id: str, operator_id: str = "" +) -> None: + row = ( + db.query(DutyAssignment) + .filter( + DutyAssignment.id == assignment_id, + DutyAssignment.tenant_no == tenant_no, + DutyAssignment.del_flag == 0, + ) + .first() + ) + if not row: + raise DutyBizError("报名记录不存在") + row.del_flag = 1 + row.update_by = operator_id or None + row.update_time = datetime.now() + db.commit() + + +def set_link_status( + db: Session, tenant_no: str, day_id: str, status: str, operator_id: str = "" +) -> Dict[str, Any]: + if status not in (LINK_OPEN, LINK_CLOSED): + raise DutyBizError("链接状态无效") + day = _load_day_by_id(db, tenant_no, day_id) + if not day: + raise DutyBizError("值班日不存在") + day.link_status = status + day.update_by = operator_id or None + day.update_time = datetime.now() + db.commit() + db.refresh(day) + return _serialize_day(db, day) + + +def _count_active_assignments(db: Session, slot_id: str) -> int: + return ( + db.query(DutyAssignment) + .filter(DutyAssignment.duty_slot_id == slot_id, DutyAssignment.del_flag == 0) + .count() + ) + + +def save_day( + db: Session, + tenant_no: str, + user: Dict[str, Any], + body: DutyDaySaveBody, +) -> Dict[str, Any]: + user_id = (user.get("user_id") or user.get("id") or "") or "" + dept_id = (user.get("region_id") or user.get("create_dept_id") or "") or "" + now = datetime.now() + + # 校验：每个 slot 提交的指定人数不超过 capacity + seen_types = set() + for slot_in in body.slots: + if slot_in.shiftType in seen_types: + raise DutyBizError("同一天不能重复配置相同班次") + seen_types.add(slot_in.shiftType) + assign_count = len(slot_in.assignments or []) + if assign_count > slot_in.capacity: + label = SHIFT_LABELS.get(slot_in.shiftType, slot_in.shiftType) + raise DutyBizError(f"{label}报名人数不能超过名额") + + day = _load_day_by_date(db, tenant_no, body.dutyDate) + if not day: + day = DutyDay( + id=generate_id(), + duty_date=body.dutyDate, + access_token=new_access_token(), + link_status=body.linkStatus or LINK_OPEN, + tenant_no=tenant_no, + create_by=user_id, + create_dept_id=dept_id, + create_time=now, + update_by=user_id, + update_time=now, + del_flag=0, + ) + db.add(day) + db.flush() + else: + if body.linkStatus in (LINK_OPEN, LINK_CLOSED): + day.link_status = body.linkStatus + day.update_by = user_id + day.update_time = now + + existing_slots = _load_slots(db, day.id) + existing_by_type = {s.shift_type: s for s in existing_slots} + incoming_types = {s.shiftType for s in body.slots} + + # 停用未再提交且已有报名的班次：禁止 + for shift_type, slot in existing_by_type.items(): + if shift_type not in incoming_types: + taken = _count_active_assignments(db, slot.id) + if taken > 0: + label = SHIFT_LABELS.get(shift_type, shift_type) + raise DutyBizError(f"{label}仍有报名，请先清空后再移除该班次") + slot.del_flag = 1 + slot.update_by = user_id + slot.update_time = now + + order = 0 + for slot_in in body.slots: + order += 1 + slot = existing_by_type.get(slot_in.shiftType) + if slot is None: + slot = DutySlot( + id=generate_id(), + duty_day_id=day.id, + shift_type=slot_in.shiftType, + capacity=slot_in.capacity, + tenant_no=tenant_no, + order_num=order, + create_by=user_id, + create_dept_id=dept_id, + create_time=now, + update_by=user_id, + update_time=now, + del_flag=0, + ) + db.add(slot) + db.flush() + else: + taken = _count_active_assignments(db, slot.id) + # 缩容：不能小于「保留后的报名」——保存时以 body.assignments 为准重建指定项， + # 但抢班记录不在 body 里时需计入 taken 中不在本次提交删除范围的 grab。 + # 简化策略：先统计将保留的 grab + 新 assign 总数，再与 capacity 比。 + existing_assigns = ( + db.query(DutyAssignment) + .filter(DutyAssignment.duty_slot_id == slot.id, DutyAssignment.del_flag == 0) + .all() + ) + # body.assignments 视为该班次最终「指定」名单；grab 全部保留，除非超容时拒绝缩容 + grab_rows = [a for a in existing_assigns if a.source == SOURCE_GRAB] + new_assign_count = len(slot_in.assignments or []) + final_count = len(grab_rows) + new_assign_count + if final_count > slot_in.capacity: + label = SHIFT_LABELS.get(slot_in.shiftType, slot_in.shiftType) + raise DutyBizError( + f"{label}当前已有 {len(grab_rows)} 人抢班 + {new_assign_count} 人指定，" + f"超过名额 {slot_in.capacity}，请先删人再缩容" + ) + slot.capacity = slot_in.capacity + slot.order_num = order + slot.update_by = user_id + slot.update_time = now + + # 软删旧的 assign 指定，再写入新的；grab 保留 + for a in existing_assigns: + if a.source == SOURCE_ASSIGN: + a.del_flag = 1 + a.update_by = user_id + a.update_time = now + + # 新 slot：写入 assignments；已存在 slot：写入新的 assign + for item in slot_in.assignments or []: + person_name = (item.personName or "").strip() + uid = (item.userId or "").strip() or None + if item.source == SOURCE_ASSIGN and not person_name and not uid: + continue + if not person_name: + raise DutyBizError("指定员工须有姓名") + person_name = normalize_person_name(person_name) if person_name else person_name + # 若只有 userId，允许 personName 已由上层填好 + if not person_name: + raise DutyBizError("指定员工姓名不能为空") + db.add( + DutyAssignment( + id=generate_id(), + duty_slot_id=slot.id, + source=SOURCE_ASSIGN if item.source != SOURCE_GRAB else SOURCE_GRAB, + user_id=uid, + person_name=person_name, + tenant_no=tenant_no, + create_by=user_id, + create_dept_id=dept_id, + create_time=now, + update_by=user_id, + update_time=now, + del_flag=0, + ) + ) + + # 新建 slot 时也要校验人数 + if slot.id and _count_active_assignments(db, slot.id) > slot.capacity: + label = SHIFT_LABELS.get(slot.shift_type, slot.shift_type) + raise DutyBizError(f"{label}报名人数不能超过名额") + + db.commit() + db.refresh(day) + return _serialize_day(db, day) + + +def get_public_page(db: Session, token: str) -> Dict[str, Any]: + token = (token or "").strip() + if not token: + raise DutyBizError("链接无效或已失效") + day = ( + db.query(DutyDay) + .filter(DutyDay.access_token == token, DutyDay.del_flag == 0) + .first() + ) + if not day: + raise DutyBizError("链接无效或已失效") + payload = _serialize_day(db, day) + # 公开页不强制返回 accessToken 也可，保留便于调试；可去掉敏感则仅留必要字段 + return { + "dutyDate": payload["dutyDate"], + "linkStatus": payload["linkStatus"], + "slots": payload["slots"], + "closed": day.link_status == LINK_CLOSED, + } + + +def grab(db: Session, token: str, shift_type: str, person_name: str) -> Dict[str, Any]: + name = normalize_person_name(person_name) + shift_type = (shift_type or "").strip() + if shift_type not in SHIFT_LABELS: + raise DutyBizError("班次类型无效") + + day = ( + db.query(DutyDay) + .filter(DutyDay.access_token == token, DutyDay.del_flag == 0) + .with_for_update() + .first() + ) + if not day: + raise DutyBizError("链接无效或已失效") + if day.link_status == LINK_CLOSED: + raise DutyBizError("报名已关闭") + + slot = ( + db.query(DutySlot) + .filter( + DutySlot.duty_day_id == day.id, + DutySlot.shift_type == shift_type, + DutySlot.del_flag == 0, + ) + .with_for_update() + .first() + ) + if not slot: + raise DutyBizError("该班次不存在") + + taken = _count_active_assignments(db, slot.id) + if taken >= slot.capacity: + raise DutyBizError("名额已满") + + now = datetime.now() + row = DutyAssignment( + id=generate_id(), + duty_slot_id=slot.id, + source=SOURCE_GRAB, + user_id=None, + person_name=name, + tenant_no=day.tenant_no, + create_by="", + create_dept_id="", + create_time=now, + update_by="", + update_time=now, + del_flag=0, + ) + db.add(row) + db.commit() + return { + "dutyDate": day.duty_date.isoformat(), + "shiftType": shift_type, + "shiftLabel": SHIFT_LABELS.get(shift_type, shift_type), + "personName": name, + "status": "ok", + } + + +# re-export for tests +__all__ = [ + "SHIFT_LABELS", + "DutyBizError", + "normalize_person_name", + "new_access_token", + "build_public_url", + "get_month_summary", + "get_day_detail", + "empty_day_skeleton", + "save_day", + "set_link_status", + "soft_delete_assignment", + "get_public_page", + "grab", +] diff --git a/backend/tests/test_duty_schedule.py b/backend/tests/test_duty_schedule.py new file mode 100644 index 000000000..80983a145 --- /dev/null +++ b/backend/tests/test_duty_schedule.py @@ -0,0 +1,172 @@ +"""值班排班服务单测""" +from datetime import date +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from app.models.duty_models import ( + LINK_CLOSED, + LINK_OPEN, + SHIFT_EVENING, + SHIFT_LABELS, + SHIFT_MORNING, + SOURCE_GRAB, +) +from app.schemas.duty import DutyAssignmentIn, DutyDaySaveBody, DutySlotSaveIn +from app.services.duty_schedule import ( + DutyBizError, + grab, + normalize_person_name, + save_day, +) + + +def test_normalize_person_name_ok(): + assert normalize_person_name(" 张三 ") == "张三" + + +def test_normalize_person_name_empty(): + with pytest.raises(DutyBizError, match="姓名"): + normalize_person_name(" ") + + +def test_normalize_person_name_too_long(): + with pytest.raises(DutyBizError, match="姓名"): + normalize_person_name("啊" * 21) + + +def test_shift_labels(): + assert SHIFT_LABELS[SHIFT_MORNING] == "早班" + + +def test_grab_rejects_when_full(): + 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, + ) + slot = SimpleNamespace( + id="slot1", + duty_day_id="day1", + shift_type=SHIFT_EVENING, + capacity=1, + del_flag=0, + ) + + day_q = MagicMock() + day_q.filter.return_value.with_for_update.return_value.first.return_value = day + slot_q = MagicMock() + slot_q.filter.return_value.with_for_update.return_value.first.return_value = slot + + def query_side_effect(model): + name = getattr(model, "__name__", str(model)) + if name == "DutyDay": + return day_q + if name == "DutySlot": + return slot_q + # DutyAssignment count + aq = MagicMock() + aq.filter.return_value.count.return_value = 1 + return aq + + db.query.side_effect = query_side_effect + + with pytest.raises(DutyBizError, match="名额已满"): + grab(db, "tok", SHIFT_EVENING, "李四") + + +def test_grab_rejects_when_closed(): + db = MagicMock() + day = SimpleNamespace( + id="day1", + tenant_no="t1", + link_status=LINK_CLOSED, + duty_date=date(2026, 8, 20), + access_token="tok", + del_flag=0, + ) + day_q = MagicMock() + day_q.filter.return_value.with_for_update.return_value.first.return_value = day + db.query.return_value = day_q + + with pytest.raises(DutyBizError, match="关闭"): + grab(db, "tok", SHIFT_EVENING, "李四") + + +def test_save_day_rejects_shrink_below_grab_count(): + """缩容：已有 1 人抢班时 capacity=1 且再指定 1 人应失败。""" + db = MagicMock() + day = SimpleNamespace( + id="day1", + duty_date=date(2026, 8, 20), + access_token="tok", + link_status=LINK_OPEN, + tenant_no="t1", + del_flag=0, + update_by=None, + update_time=None, + ) + slot = SimpleNamespace( + id="slot1", + duty_day_id="day1", + shift_type=SHIFT_EVENING, + capacity=2, + del_flag=0, + order_num=1, + update_by=None, + update_time=None, + ) + grab_row = SimpleNamespace( + id="a1", + duty_slot_id="slot1", + source=SOURCE_GRAB, + del_flag=0, + update_by=None, + update_time=None, + ) + + day_q = MagicMock() + day_q.filter.return_value.first.return_value = day + slot_list_q = MagicMock() + slot_list_q.filter.return_value.all.return_value = [slot] + assign_q = MagicMock() + assign_q.filter.return_value.all.return_value = [grab_row] + count_q = MagicMock() + count_q.filter.return_value.count.return_value = 1 + + def query_side_effect(model): + name = getattr(model, "__name__", "") + if name == "DutyDay": + return day_q + if name == "DutySlot": + return slot_list_q + if name == "DutyAssignment": + # first call in loop may be count, then all + m = MagicMock() + m.filter.return_value.count.return_value = 1 + m.filter.return_value.all.return_value = [grab_row] + return m + return MagicMock() + + db.query.side_effect = query_side_effect + + body = DutyDaySaveBody( + dutyDate=date(2026, 8, 20), + slots=[ + DutySlotSaveIn( + shiftType=SHIFT_EVENING, + capacity=1, + assignments=[ + DutyAssignmentIn(source="assign", userId="u1", personName="张三") + ], + ) + ], + ) + + with pytest.raises(DutyBizError, match="超过名额|缩容|报名"): + save_day(db, "t1", {"user_id": "op1"}, body)