feat(duty): add time/work-plan helpers and dict ensure

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

backend/app/models/duty_models.py

@@ -61,7 +61,7 @@ class DutySlot(Base):
     cross_day = Column(SmallInteger, nullable=False, default=0, comment="结束是否次日")
     tenant_no = Column(String(64), nullable=False, default="shanxia2.0", comment="租户编号")
     order_num = Column(Integer, nullable=True, comment="排序号")
-    remark = Column(String(500), nullable=True, comment="班次工作安排")
+    remark = Column(String(500), nullable=True, comment="备注")
     create_by = Column(String(64), nullable=True, comment="创建者")
     create_dept_id = Column(String(64), nullable=False, default="", comment="创建部门")
     create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")

backend/app/services/duty_schedule.py

@@ -3,17 +3,20 @@ from __future__ import annotations
 
 import calendar
 import secrets
-from datetime import date, datetime
+from datetime import date, datetime, time
 from typing import Any, Dict, List, Optional, Sequence
 
+from pypinyin import Style, lazy_pinyin
 from sqlalchemy import and_
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
-from app.core.utils import generate_id
+from app.core.utils import generate_id, generate_uuid
 from app.models.duty_models import (
     DUTY_STAFF_ROLE_ID,
     DUTY_STAFF_ROLE_NAME,
+    DICT_SHIFT_TYPE,
+    DICT_WORK_PLAN,
     LINK_CLOSED,
     LINK_OPEN,
     SHIFT_LABELS,
@@ -23,7 +26,7 @@ from app.models.duty_models import (
     DutyDay,
     DutySlot,
 )
-from app.models.models import SysRole, SysUser, SysUserRole
+from app.models.models import SysDictData, SysRole, SysUser, SysUserRole
 from app.schemas.duty import DutyDaySaveBody, DutySlotSaveIn
 
 PERSON_NAME_MAX_LEN = 20
@@ -33,6 +36,141 @@ class DutyBizError(ValueError):
     """业务错误（可映射为 HTTP 400）"""
 
 
+def parse_hhmm(value: Optional[str]) -> Optional[time]:
+    raw = (value or "").strip()
+    if not raw:
+        return None
+    parts = raw.split(":")
+    if len(parts) < 2:
+        raise DutyBizError("时间格式须为 HH:mm")
+    h, m = int(parts[0]), int(parts[1])
+    if not (0 <= h <= 23 and 0 <= m <= 59):
+        raise DutyBizError("时间格式须为 HH:mm")
+    return time(hour=h, minute=m)
+
+
+def format_hhmm(t: Optional[time]) -> Optional[str]:
+    if t is None:
+        return None
+    return f"{t.hour:02d}:{t.minute:02d}"
+
+
+def format_time_label(
+    start: Optional[time], end: Optional[time], cross_day: bool
+) -> Optional[str]:
+    if start is None or end is None:
+        return None
+    s, e = format_hhmm(start), format_hhmm(end)
+    if cross_day:
+        return f"{s}–次日{e}"
+    return f"{s}–{e}"
+
+
+def split_work_plans(raw: Optional[str]) -> List[str]:
+    if not raw:
+        return []
+    return [p.strip() for p in str(raw).replace("，", ",").split(",") if p.strip()]
+
+
+def join_work_plans(values: Sequence[str]) -> Optional[str]:
+    parts = [v.strip() for v in values if (v or "").strip()]
+    return ",".join(parts) if parts else None
+
+
+def build_dict_value_from_label(label: str) -> str:
+    text = (label or "").strip()
+    parts: List[str] = []
+    for ch in text:
+        if "\u4e00" <= ch <= "\u9fff":
+            pys = lazy_pinyin(ch, style=Style.NORMAL)
+            if pys and pys[0]:
+                parts.append(str(pys[0]).lower())
+        elif ch.isalnum():
+            parts.append(ch.lower())
+    base = "".join(parts)[:32] or "item"
+    return base
+
+
+def ensure_dict_data(
+    db: Session,
+    *,
+    tenant_no: str,
+    dict_type: str,
+    label: str,
+    user_id: str = "",
+    preferred_value: Optional[str] = None,
+) -> Dict[str, str]:
+    label = (label or "").strip()
+    if not label:
+        raise DutyBizError("字典标签不能为空")
+    if dict_type not in (DICT_SHIFT_TYPE, DICT_WORK_PLAN):
+        raise DutyBizError("不支持的字典类型")
+
+    existing_label = (
+        db.query(SysDictData)
+        .filter(
+            SysDictData.tenant_no == tenant_no,
+            SysDictData.dict_type == dict_type,
+            SysDictData.dict_label == label,
+            SysDictData.del_flag == 0,
+        )
+        .first()
+    )
+    if existing_label:
+        return {"label": existing_label.dict_label, "value": existing_label.dict_value}
+
+    value = (preferred_value or "").strip() or build_dict_value_from_label(label)
+    for _ in range(8):
+        clash = (
+            db.query(SysDictData)
+            .filter(
+                SysDictData.tenant_no == tenant_no,
+                SysDictData.dict_type == dict_type,
+                SysDictData.dict_value == value,
+                SysDictData.del_flag == 0,
+            )
+            .first()
+        )
+        if not clash:
+            break
+        value = f"{value[:27]}_{secrets.token_hex(2)}"
+    else:
+        raise DutyBizError("无法生成唯一字典值")
+
+    obj = SysDictData(
+        id=generate_uuid(),
+        dict_label=label,
+        dict_value=value,
+        dict_type=dict_type,
+        tag_type="",
+        tenant_no=tenant_no,
+        order_num=0,
+        remark=None,
+        create_by=user_id or None,
+        create_dept_id="",
+        create_time=datetime.now(),
+        update_by=user_id or None,
+        update_time=datetime.now(),
+        del_flag=0,
+    )
+    db.add(obj)
+    db.flush()
+    return {"label": obj.dict_label, "value": obj.dict_value}
+
+
+def _load_dict_label_map(db: Session, tenant_no: str, dict_type: str) -> Dict[str, str]:
+    rows = (
+        db.query(SysDictData)
+        .filter(
+            SysDictData.tenant_no == tenant_no,
+            SysDictData.dict_type == dict_type,
+            SysDictData.del_flag == 0,
+        )
+        .all()
+    )
+    return {r.dict_value: r.dict_label for r in rows}
+
+
 def normalize_person_name(name: str) -> str:
     raw = (name or "").strip()
     if not raw:
@@ -612,6 +750,13 @@ def grab(db: Session, token: str, shift_type: str, user_id: str) -> Dict[str, An
 __all__ = [
     "SHIFT_LABELS",
     "DutyBizError",
+    "parse_hhmm",
+    "format_hhmm",
+    "format_time_label",
+    "split_work_plans",
+    "join_work_plans",
+    "build_dict_value_from_label",
+    "ensure_dict_data",
     "normalize_person_name",
     "new_access_token",
     "build_public_url",

backend/tests/test_duty_schedule.py

@@ -17,17 +17,89 @@ from app.models.duty_models import (
     SOURCE_GRAB,
 )
 from app.schemas.duty import DutyAssignmentIn, DutyDaySaveBody, DutySlotSaveIn
+from app.models.models import SysDictData
 from app.services.duty_schedule import (
     DutyBizError,
     assert_user_is_duty_staff,
+    build_dict_value_from_label,
+    ensure_dict_data,
+    format_time_label,
     grab,
+    join_work_plans,
     list_duty_staff_users,
     normalize_person_name,
+    parse_hhmm,
     resolve_duty_staff_role_id,
     save_day,
+    split_work_plans,
 )
 
 
+def test_parse_hhmm():
+    assert parse_hhmm("08:00").hour == 8
+    assert parse_hhmm("18:30").minute == 30
+    assert parse_hhmm(None) is None
+    assert parse_hhmm("") is None
+
+
+def test_format_time_label_same_day():
+    assert format_time_label(parse_hhmm("08:00"), parse_hhmm("18:00"), False) == "08:00–18:00"
+
+
+def test_format_time_label_cross_day():
+    assert format_time_label(parse_hhmm("18:00"), parse_hhmm("07:00"), True) == "18:00–次日07:00"
+
+
+def test_format_time_label_empty():
+    assert format_time_label(None, None, False) is None
+
+
+def test_work_plans_join_split():
+    assert join_work_plans(["peisuan", "zhuangxie"]) == "peisuan,zhuangxie"
+    assert split_work_plans("peisuan,zhuangxie") == ["peisuan", "zhuangxie"]
+    assert split_work_plans(None) == []
+
+
+def test_build_dict_value_from_label():
+    assert build_dict_value_from_label("配酸") == "peisuan"
+    assert build_dict_value_from_label("装卸") == "zhuangxie"
+    assert build_dict_value_from_label("  ") == "item"
+
+
+def test_ensure_dict_data_reuses_existing_label():
+    db = MagicMock()
+    existing = SimpleNamespace(dict_label="配酸", dict_value="peisuan")
+    q = MagicMock()
+    q.filter.return_value.first.return_value = existing
+    db.query.return_value = q
+
+    result = ensure_dict_data(
+        db, tenant_no="t1", dict_type="duty_work_plan", label="配酸", user_id="u1"
+    )
+    assert result == {"label": "配酸", "value": "peisuan"}
+    db.add.assert_not_called()
+
+
+def test_ensure_dict_data_inserts_new_without_commit():
+    db = MagicMock()
+    q = MagicMock()
+    q.filter.return_value.first.side_effect = [None, None]
+    db.query.return_value = q
+
+    result = ensure_dict_data(
+        db, tenant_no="t1", dict_type="duty_work_plan", label="巡检", user_id="u1"
+    )
+    assert result["label"] == "巡检"
+    assert result["value"] == "xunjian"
+    db.add.assert_called_once()
+    db.flush.assert_called_once()
+    db.commit.assert_not_called()
+    added = db.add.call_args[0][0]
+    assert isinstance(added, SysDictData)
+    assert added.dict_type == "duty_work_plan"
+    assert added.tenant_no == "t1"
+
+
 def test_normalize_person_name_ok():
     assert normalize_person_name("  张三  ") == "张三"