zkxw/backend/app/api/sample_storage_cert_qc.py

"""样本储存证质检信息 API"""
from typing import Any, Dict, Optional

from fastapi import APIRouter, Depends
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, require_strict_any_menu_code
from app.core.response import error_response, success_response
from app.crud import sample_storage_cert_qc as svc
from app.models.models import SysUser

router = APIRouter(prefix="/web/admin", tags=["储存证质检信息"])

READ_MENUS = (
    "biz:qcWorkOrder:list",
    "biz:myQcTask:list",
    "cell-storage:sample-info:list",
)
WRITE_MENUS = (
    "biz:qcWorkOrder:list",
    "biz:myQcTask: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)


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

    payload: Dict[str, Any] = Field(default_factory=dict)
    qc_work_order_id: Optional[str] = Field(None, alias="qcWorkOrderId")


@router.get("/sample-storage-cert-qc/{sample_code}")
async def get_sample_storage_cert_qc(
    sample_code: str,
    user: dict = Depends(require_any_menu_code(*READ_MENUS)),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    detail = svc.get_detail(db, tenant_no, sample_code)
    if detail is None:
        return error_response(404, "样本不存在")
    return success_response(detail)


@router.put("/sample-storage-cert-qc/{sample_code}")
async def put_sample_storage_cert_qc(
    sample_code: str,
    body: UpsertBody,
    user: dict = Depends(require_strict_any_menu_code(*WRITE_MENUS)),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    user_id = _get_current_user_id(user)
    try:
        detail = svc.upsert(
            db,
            tenant_no=tenant_no,
            user_id=user_id,
            sample_code=sample_code,
            payload=body.payload,
            qc_work_order_id=body.qc_work_order_id,
        )
        return success_response(detail, message="保存成功")
    except ValueError as e:
        return error_response(400, str(e))