146 lines
4.8 KiB
Python
146 lines
4.8 KiB
Python
"""样本质检短信通知 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))
|