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 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.models.models import SysUser
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 _list_public_users(db: Session, tenant_no: str) -> List[Dict[str, str]]:
    rows = (
        db.query(SysUser)
        .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]] = []
    for r in rows:
        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 _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("用户无效或不存在")
    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 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("报名已关闭")

    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()
    return {
        "dutyDate": day.duty_date.isoformat(),
        "shiftType": shift_type,
        "shiftLabel": SHIFT_LABELS.get(shift_type, shift_type),
        "personName": name,
        "userId": user.id,
        "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",
]