shanxia2.0/backend/app/services/duty_schedule.py

"""值班排班领域服务 — docs/superpowers/specs/2026-08-14-duty-schedule-design.md"""
from __future__ import annotations

import calendar
import re
import secrets
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, generate_uuid
from app.models.duty_models import (
    DUTY_STAFF_ROLE_ID,
    DUTY_STAFF_ROLE_NAME,
    DICT_SHIFT_TYPE,
    DICT_TIME_RANGE,
    DICT_WORK_PLAN,
    LINK_CLOSED,
    LINK_OPEN,
    SHIFT_LABELS,
    SOURCE_ASSIGN,
    SOURCE_GRAB,
    DutyAssignment,
    DutyDay,
    DutySlot,
)
from app.models.models import SysDictData, SysRole, SysUser, SysUserRole
from app.schemas.duty import DutyDaySaveBody, DutySlotSaveIn

PERSON_NAME_MAX_LEN = 20


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")
    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)


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 _pad_hhmm(h: int, m: int) -> str:
    return f"{h:02d}:{m:02d}"


def parse_time_range_input(raw: str) -> Dict[str, Any]:
    """解析「09:00-17:00」「18:00-次日07:00」等，返回 start/end/crossDay/label/value。"""
    text = (raw or "").strip().replace("—", "-").replace("–", "-").replace("~", "-")
    text = re.sub(r"\s+", "", text)
    if not text:
        raise DutyBizError("时段不能为空")
    m = re.match(
        r"^(\d{1,2}):(\d{2})[-至到](次日)?(\d{1,2}):(\d{2})$",
        text,
    )
    if not m:
        raise DutyBizError("时段格式须为 09:00-17:00 或 18:00-次日07:00")
    sh, sm, cross_flag, eh, em = m.groups()
    try:
        start_h, start_m = int(sh), int(sm)
        end_h, end_m = int(eh), int(em)
    except ValueError as exc:
        raise DutyBizError("时段格式须为 09:00-17:00 或 18:00-次日07:00") from exc
    if not (0 <= start_h <= 23 and 0 <= start_m <= 59 and 0 <= end_h <= 23 and 0 <= end_m <= 59):
        raise DutyBizError("时段格式须为 09:00-17:00 或 18:00-次日07:00")
    start = _pad_hhmm(start_h, start_m)
    end = _pad_hhmm(end_h, end_m)
    cross = 1 if cross_flag or end < start else 0
    label = f"{start}–次日{end}" if cross else f"{start}–{end}"
    value = f"{start}|{end}|{cross}"
    return {
        "startTime": start,
        "endTime": end,
        "crossDay": bool(cross),
        "label": label,
        "value": value,
    }


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, DICT_TIME_RANGE):
        raise DutyBizError("不支持的字典类型")

    preferred = (preferred_value or "").strip() or None
    if dict_type == DICT_TIME_RANGE:
        parsed = parse_time_range_input(label)
        label = parsed["label"]
        preferred = preferred or parsed["value"]

    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}

    if preferred:
        by_value = (
            db.query(SysDictData)
            .filter(
                SysDictData.tenant_no == tenant_no,
                SysDictData.dict_type == dict_type,
                SysDictData.dict_value == preferred,
                SysDictData.del_flag == 0,
            )
            .first()
        )
        if by_value:
            return {"label": by_value.dict_label, "value": by_value.dict_value}

    value = preferred 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:
        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],
    *,
    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)
        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],
    }


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, []),
            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
    ]
    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])
    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]:
    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)
        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:
            slot = DutySlot(
                id=generate_id(),
                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,
                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()
            )
            grandfathered_user_ids = {
                a.user_id
                for a in existing_assigns
                if a.source == SOURCE_ASSIGN and a.user_id
            }
            # 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.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
            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("指定员工姓名不能为空")
            if item.source == SOURCE_ASSIGN and uid:
                if uid not in grandfathered_user_ids:
                    assert_user_is_duty_staff(db, tenant_no, uid)
            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 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,
            SysUser.status == "1",
        )
        .order_by(SysUser.nick_name.asc(), SysUser.user_no.asc())
        .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
        out.append({"id": r.id, "label": label})
    return out


def assert_user_is_duty_staff(db: Session, tenant_no: str, user_id: str) -> None:
    uid = (user_id or "").strip()
    if not uid:
        raise DutyBizError("用户无效或不具备值班人员角色")
    user = (
        db.query(SysUser)
        .filter(
            SysUser.id == uid,
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == "1",
        )
        .first()
    )
    if not user:
        raise DutyBizError("用户无效或不具备值班人员角色")
    role_id = resolve_duty_staff_role_id(db, tenant_no)
    if not role_id:
        raise DutyBizError("未配置「值班人员」角色，请联系管理员")
    link = (
        db.query(SysUserRole)
        .filter(
            SysUserRole.user_id == uid,
            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:
        raise DutyBizError("请选择有效用户")
    user = (
        db.query(SysUser)
        .filter(
            SysUser.id == uid,
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == "1",
        )
        .first()
    )
    if not user:
        raise DutyBizError("用户无效或不存在")
    assert_user_is_duty_staff(db, tenant_no, user.id)
    return user


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)
    return {
        "dutyDate": payload["dutyDate"],
        "linkStatus": payload["linkStatus"],
        "slots": payload["slots"],
        "closed": day.link_status == LINK_CLOSED,
        "users": _list_public_users(db, day.tenant_no),
    }


def grab(db: Session, token: str, shift_type: str, user_id: str) -> Dict[str, Any]:
    shift_type = (shift_type or "").strip()
    if not shift_type:
        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("报名已关闭")

    user = _resolve_grab_user(db, day.tenant_no, user_id)
    name = normalize_person_name((user.nick_name or "").strip() or (user.user_no or "").strip())

    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("该班次不存在")

    dup = (
        db.query(DutyAssignment)
        .filter(
            DutyAssignment.duty_slot_id == slot.id,
            DutyAssignment.user_id == user.id,
            DutyAssignment.del_flag == 0,
        )
        .first()
    )
    if dup:
        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=user.id,
        person_name=name,
        tenant_no=day.tenant_no,
        create_by=user.id,
        create_dept_id="",
        create_time=now,
        update_by=user.id,
        update_time=now,
        del_flag=0,
    )
    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)
        or SHIFT_LABELS.get(shift_type, shift_type),
        "personName": name,
        "userId": user.id,
        "status": "ok",
    }


# re-export for tests
__all__ = [
    "SHIFT_LABELS",
    "DutyBizError",
    "parse_hhmm",
    "parse_time_range_input",
    "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",
    "get_month_summary",
    "get_day_detail",
    "empty_day_skeleton",
    "save_day",
    "set_link_status",
    "soft_delete_assignment",
    "get_public_page",
    "grab",
    "resolve_duty_staff_role_id",
    "list_duty_staff_users",
    "assert_user_is_duty_staff",
]