feat(duty): persist work plans and times; relax grab shift types

Map invalid HH:mm to DutyBizError, serialize workPlans/timeLabel from dict,
ensure dict items on save, and allow custom shift types on public grab.

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

backend/app/api/duty.py

@@ -11,7 +11,7 @@ from app.core.database import get_db
 from app.core.menu_guard import require_admin_or_super_admin
 from app.core.response import error_response, success_response
 from app.crud import user as user_crud
-from app.schemas.duty import DutyDaySaveBody, DutyLinkBody
+from app.schemas.duty import DutyDaySaveBody, DutyDictEnsureBody, DutyLinkBody
 from app.services import duty_schedule as duty_svc
 from app.services.duty_schedule import DutyBizError
 
@@ -73,6 +73,28 @@ def duty_user_options(
     return success_response(out)
 
 
+@router.post("/dict/ensure")
+def ensure_duty_dict(
+    body: DutyDictEnsureBody,
+    _user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    tenant_no = _user.get("tenant_no", "") or ""
+    user_id = _user.get("user_id") or _user.get("id", "") or ""
+    try:
+        data = duty_svc.ensure_dict_data(
+            db,
+            tenant_no=tenant_no,
+            dict_type=body.dictType,
+            label=body.label,
+            user_id=user_id,
+        )
+        db.commit()
+        return success_response(data, message="已保存")
+    except DutyBizError as exc:
+        return error_response(400, str(exc))
+
+
 @router.post("/day")
 def save_day(
     body: DutyDaySaveBody,

backend/app/services/duty_schedule.py

@@ -43,7 +43,10 @@ def parse_hhmm(value: Optional[str]) -> Optional[time]:
     parts = raw.split(":")
     if len(parts) < 2:
         raise DutyBizError("时间格式须为 HH:mm")
-    h, m = int(parts[0]), int(parts[1])
+    try:
+        h, m = int(parts[0]), int(parts[1])
+    except ValueError as exc:
+        raise DutyBizError("时间格式须为 HH:mm") from exc
     if not (0 <= h <= 23 and 0 <= m <= 59):
         raise DutyBizError("时间格式须为 HH:mm")
     return time(hour=h, minute=m)
@@ -198,15 +201,32 @@ def _assignment_dict(row: DutyAssignment) -> Dict[str, Any]:
     }
 
 
-def _slot_dict(slot: DutySlot, assignments: Sequence[DutyAssignment]) -> Dict[str, Any]:
+def _slot_dict(
+    slot: DutySlot,
+    assignments: Sequence[DutyAssignment],
+    *,
+    shift_labels: Dict[str, str],
+    work_plan_labels: Dict[str, str],
+) -> Dict[str, Any]:
     active = [a for a in assignments if a.del_flag == 0]
+    plans = split_work_plans(getattr(slot, "work_plans", None))
+    cross = bool(getattr(slot, "cross_day", 0) or 0)
+    start = getattr(slot, "start_time", None)
+    end = getattr(slot, "end_time", None)
     return {
         "id": slot.id,
         "shiftType": slot.shift_type,
-        "shiftLabel": SHIFT_LABELS.get(slot.shift_type, slot.shift_type),
+        "shiftLabel": shift_labels.get(slot.shift_type)
+        or SHIFT_LABELS.get(slot.shift_type, slot.shift_type),
         "capacity": slot.capacity,
         "taken": len(active),
+        "workPlans": plans,
+        "workPlanLabels": [work_plan_labels.get(v, v) for v in plans],
         "remark": (slot.remark or "").strip() or None,
+        "startTime": format_hhmm(start),
+        "endTime": format_hhmm(end),
+        "crossDay": cross,
+        "timeLabel": format_time_label(start, end, cross),
         "assignments": [_assignment_dict(a) for a in active],
     }
 
@@ -215,9 +235,17 @@ def _day_dict(
     day: DutyDay,
     slots: Sequence[DutySlot],
     assignments_by_slot: Dict[str, List[DutyAssignment]],
+    *,
+    shift_labels: Dict[str, str],
+    work_plan_labels: Dict[str, str],
 ) -> Dict[str, Any]:
     slot_payload = [
-        _slot_dict(s, assignments_by_slot.get(s.id, []))
+        _slot_dict(
+            s,
+            assignments_by_slot.get(s.id, []),
+            shift_labels=shift_labels,
+            work_plan_labels=work_plan_labels,
+        )
         for s in sorted(slots, key=lambda x: (x.order_num or 0, x.shift_type or ""))
         if s.del_flag == 0
     ]
@@ -278,7 +306,15 @@ def _load_assignments_for_slots(
 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)
+    shift_labels = _load_dict_label_map(db, day.tenant_no, DICT_SHIFT_TYPE)
+    work_plan_labels = _load_dict_label_map(db, day.tenant_no, DICT_WORK_PLAN)
+    return _day_dict(
+        day,
+        slots,
+        assigns,
+        shift_labels=shift_labels,
+        work_plan_labels=work_plan_labels,
+    )
 
 
 def empty_day_skeleton(duty_date: date) -> Dict[str, Any]:
@@ -430,6 +466,59 @@ def save_day(
         order += 1
         slot = existing_by_type.get(slot_in.shiftType)
         slot_remark = (slot_in.remark or "").strip() or None
+        plans_raw = list(slot_in.workPlans or [])
+        for v in plans_raw:
+            v = (v or "").strip()
+            if not v:
+                continue
+            row = (
+                db.query(SysDictData)
+                .filter(
+                    SysDictData.tenant_no == tenant_no,
+                    SysDictData.dict_type == DICT_WORK_PLAN,
+                    SysDictData.dict_value == v,
+                    SysDictData.del_flag == 0,
+                )
+                .first()
+            )
+            if not row:
+                ensure_dict_data(
+                    db,
+                    tenant_no=tenant_no,
+                    dict_type=DICT_WORK_PLAN,
+                    label=v,
+                    preferred_value=v,
+                    user_id=user_id,
+                )
+        slot_work_plans = join_work_plans(plans_raw)
+
+        if slot_in.shiftType not in SHIFT_LABELS:
+            row = (
+                db.query(SysDictData)
+                .filter(
+                    SysDictData.tenant_no == tenant_no,
+                    SysDictData.dict_type == DICT_SHIFT_TYPE,
+                    SysDictData.dict_value == slot_in.shiftType,
+                    SysDictData.del_flag == 0,
+                )
+                .first()
+            )
+            if not row:
+                ensure_dict_data(
+                    db,
+                    tenant_no=tenant_no,
+                    dict_type=DICT_SHIFT_TYPE,
+                    label=SHIFT_LABELS.get(slot_in.shiftType, slot_in.shiftType),
+                    preferred_value=slot_in.shiftType,
+                    user_id=user_id,
+                )
+
+        start_t = parse_hhmm(slot_in.startTime)
+        end_t = parse_hhmm(slot_in.endTime)
+        if (start_t is None) ^ (end_t is None):
+            raise DutyBizError("值班时间须同时填写开始与结束，或全部留空")
+        cross = 1 if slot_in.crossDay else 0
+
         # 已有 slot 上历史 assign 用户可再保存（D4）；新建 slot 必须具备值班人员角色（D7）
         grandfathered_user_ids: set = set()
         if slot is None:
@@ -438,6 +527,10 @@ def save_day(
                 duty_day_id=day.id,
                 shift_type=slot_in.shiftType,
                 capacity=slot_in.capacity,
+                work_plans=slot_work_plans,
+                start_time=start_t,
+                end_time=end_t,
+                cross_day=cross,
                 remark=slot_remark,
                 tenant_no=tenant_no,
                 order_num=order,
@@ -476,6 +569,10 @@ def save_day(
                     f"超过名额 {slot_in.capacity}，请先删人再缩容"
                 )
             slot.capacity = slot_in.capacity
+            slot.work_plans = slot_work_plans
+            slot.start_time = start_t
+            slot.end_time = end_t
+            slot.cross_day = cross
             slot.remark = slot_remark
             slot.order_num = order
             slot.update_by = user_id
@@ -673,7 +770,7 @@ def get_public_page(db: Session, token: str) -> Dict[str, Any]:
 
 def grab(db: Session, token: str, shift_type: str, user_id: str) -> Dict[str, Any]:
     shift_type = (shift_type or "").strip()
-    if shift_type not in SHIFT_LABELS:
+    if not shift_type:
         raise DutyBizError("班次类型无效")
 
     day = (
@@ -736,10 +833,12 @@ def grab(db: Session, token: str, shift_type: str, user_id: str) -> Dict[str, An
     )
     db.add(row)
     db.commit()
+    shift_labels = _load_dict_label_map(db, day.tenant_no, DICT_SHIFT_TYPE)
     return {
         "dutyDate": day.duty_date.isoformat(),
         "shiftType": shift_type,
-        "shiftLabel": SHIFT_LABELS.get(shift_type, shift_type),
+        "shiftLabel": shift_labels.get(shift_type)
+        or SHIFT_LABELS.get(shift_type, shift_type),
         "personName": name,
         "userId": user.id,
         "status": "ok",

backend/tests/test_duty_schedule.py

@@ -42,6 +42,11 @@ def test_parse_hhmm():
     assert parse_hhmm("") is None
 
 
+def test_parse_hhmm_invalid_raises():
+    with pytest.raises(DutyBizError, match="时间格式"):
+        parse_hhmm("08:xx")
+
+
 def test_format_time_label_same_day():
     assert format_time_label(parse_hhmm("08:00"), parse_hhmm("18:00"), False) == "08:00–18:00"
 
@@ -180,6 +185,74 @@ def test_grab_rejects_when_full():
         grab(db, "tok", SHIFT_EVENING, "u1")
 
 
+def test_grab_allows_custom_shift_type_when_slot_exists():
+    """自定义 shift_type 只要当日存在 slot 即可，不必在 SHIFT_LABELS。"""
+    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="zhongban",
+        capacity=2,
+        del_flag=0,
+    )
+    user = SimpleNamespace(
+        id="u1",
+        tenant_no="t1",
+        nick_name="王五",
+        user_no="wangwu",
+        status="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
+    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))
+        if name == "DutyDay":
+            return day_q
+        if name == "DutySlot":
+            return slot_q
+        if name == "SysUser":
+            return user_q
+        if name == "SysRole":
+            return role_q
+        if name == "SysUserRole":
+            return ur_q
+        if name == "SysDictData":
+            dq = MagicMock()
+            dq.filter.return_value.all.return_value = []
+            dq.filter.return_value.first.return_value = None
+            return dq
+        aq = MagicMock()
+        aq.filter.return_value.count.return_value = 0
+        aq.filter.return_value.first.return_value = None
+        return aq
+
+    db.query.side_effect = query_side_effect
+    result = grab(db, "tok", "zhongban", "u1")
+    assert result["shiftType"] == "zhongban"
+    assert result["status"] == "ok"
+
+
 def test_grab_rejects_when_closed():
     db = MagicMock()
     day = SimpleNamespace(