zkxw/backend/app/api/qc_sms_notification.py

"""样本质检短信通知 API"""
from typing import Optional

from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.menu_guard import require_any_menu_code
from app.core.response import error_response, success_response
from app.crud import qc_sms_notification as svc
from app.crud.collection_protocol import total_pages
from app.models.models import SysUser

router = APIRouter(prefix="/web/admin", tags=["样本质检短信通知"])

_QC_SMS_MENU = require_any_menu_code(
    "cell-storage:sample-info:list",
)


def _get_tenant_no(db: Session, user: dict) -> str:
    user_id = user.get("user_id")
    if not user_id:
        raise PermissionError("用户身份信息缺失，无法确定租户")
    sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
    if not sys_user or not sys_user.tenant_no:
        raise PermissionError("用户租户信息缺失")
    return sys_user.tenant_no


def _get_current_user_id(user: dict) -> str:
    user_id = user.get("user_id")
    if not user_id:
        raise PermissionError("用户身份信息缺失")
    return str(user_id)


def _page_payload(items: list, total: int, page: int, page_size: int) -> dict:
    return {
        "list": items,
        "total": total,
        "page": page,
        "pageSize": page_size,
        "totalPages": total_pages(total, page_size),
    }


class ToggleStatusBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    sms_notified: int = Field(..., alias="smsNotified")
    qc_work_order_id: Optional[str] = Field(None, alias="qcWorkOrderId")
    protocol_sample_id: Optional[str] = Field(None, alias="protocolSampleId")


@router.get("/qc-sms-notifications/list")
async def list_notifications(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
    keyword: Optional[str] = None,
    qcStatus: Optional[str] = None,
    qcConclusion: Optional[str] = None,
    smsNotified: Optional[int] = None,
    qcCompleteTimeStart: Optional[str] = Query(None, alias="qcCompleteTimeStart"),
    qcCompleteTimeEnd: Optional[str] = Query(None, alias="qcCompleteTimeEnd"),
    db: Session = Depends(get_db),
    user: dict = Depends(_QC_SMS_MENU),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items, total = svc.list_notifications(
            db,
            page=page,
            page_size=pageSize,
            tenant_no=tenant_no,
            keyword=keyword,
            qc_status=qcStatus,
            qc_conclusion=qcConclusion,
            sms_notified=smsNotified,
            qc_complete_time_start=qcCompleteTimeStart,
            qc_complete_time_end=qcCompleteTimeEnd,
            user_info=user,
        )
        return success_response(_page_payload(items, total, page, pageSize))
    except PermissionError as e:
        return error_response(403, str(e))
    except Exception as e:
        return error_response(400, str(e))


@router.put("/qc-sms-notifications/{notification_id}/toggle-status")
async def toggle_status(
    notification_id: str,
    body: ToggleStatusBody,
    db: Session = Depends(get_db),
    user: dict = Depends(_QC_SMS_MENU),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        result = svc.toggle_notification_status(
            db,
            notification_id=notification_id,
            tenant_no=tenant_no,
            sms_notified=body.sms_notified,
            user_id=user_id,
            qc_work_order_id=body.qc_work_order_id,
            protocol_sample_id=body.protocol_sample_id,
        )
        if not result:
            return error_response(404, "通知记录不存在")
        return success_response(None)
    except PermissionError as e:
        return error_response(403, str(e))
    except Exception as e:
        return error_response(400, str(e))


@router.post("/qc-sms-notifications/{notification_id}/generate-sms")
async def generate_sms(
    notification_id: str,
    qcWorkOrderId: Optional[str] = Query(None),
    protocolSampleId: Optional[str] = Query(None),
    db: Session = Depends(get_db),
    user: dict = Depends(_QC_SMS_MENU),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        content = svc.generate_sms_content(
            db,
            notification_id=notification_id,
            tenant_no=tenant_no,
            qc_work_order_id=qcWorkOrderId,
            protocol_sample_id=protocolSampleId,
            user_id=user_id,
        )
        if content is None:
            return error_response(404, "通知记录不存在")
        return success_response({"smsContent": content})
    except PermissionError as e:
        return error_response(403, str(e))
    except Exception as e:
        return error_response(400, str(e))