shanxia2.0/backend/app/api/duty.py

"""值班排班 · 管理端 API"""
from __future__ import annotations

from datetime import datetime
from typing import Optional

from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session

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.services import duty_schedule as duty_svc
from app.services.duty_schedule import DutyBizError

router = APIRouter(prefix="/web/admin/duty", tags=["值班排班"])


def _enrich_assign_names(db: Session, body: DutyDaySaveBody) -> DutyDaySaveBody:
    """指定用户时补全 personName（用 nick_name）。"""
    for slot in body.slots:
        for item in slot.assignments or []:
            if item.source != "assign":
                continue
            uid = (item.userId or "").strip()
            if not uid:
                continue
            if (item.personName or "").strip():
                continue
            u = user_crud.get_user(db, uid)
            if u:
                item.personName = (u.nick_name or "").strip() or uid
    return body


@router.get("")
def month_summary(
    year: int = Query(..., ge=2000, le=2100),
    month: int = Query(..., ge=1, le=12),
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    tenant_no = _user.get("tenant_no", "") or ""
    data = duty_svc.get_month_summary(db, tenant_no, year, month)
    return success_response(data)


@router.get("/day")
def day_detail(
    dutyDate: str = Query(..., description="YYYY-MM-DD"),
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    tenant_no = _user.get("tenant_no", "") or ""
    try:
        d = datetime.strptime(dutyDate.strip(), "%Y-%m-%d").date()
    except ValueError:
        return error_response(400, "日期格式无效")
    return success_response(duty_svc.get_day_detail(db, tenant_no, d))


@router.post("/day")
def save_day(
    body: DutyDaySaveBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    tenant_no = _user.get("tenant_no", "") or ""
    try:
        body = _enrich_assign_names(db, body)
        data = duty_svc.save_day(db, tenant_no, _user, body)
        return success_response(data, message="保存成功")
    except DutyBizError as exc:
        return error_response(400, str(exc))


@router.post("/day/{day_id}/link")
def set_link(
    day_id: str,
    body: DutyLinkBody,
    _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", "")
    try:
        data = duty_svc.set_link_status(db, tenant_no, day_id, body.linkStatus, operator_id=user_id)
        return success_response(data, message="已更新链接状态")
    except DutyBizError as exc:
        return error_response(400, str(exc))


@router.delete("/assignments/{assignment_id}")
def delete_assignment(
    assignment_id: str,
    _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", "")
    try:
        duty_svc.soft_delete_assignment(db, tenant_no, assignment_id, operator_id=user_id)
        return success_response(None, message="已删除")
    except DutyBizError as exc:
        return error_response(400, str(exc))