zkxw/backend/app/api/sample_storage_cert_qc.py

"""样本储存证质检信息 API"""
from typing import Any, Dict, List, 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,
    require_strict_menu_code,
    strict_user_has_any_menu_perm,
    strict_user_has_menu_perm,
    user_has_menu_perm,
)
from app.core.response import error_response, success_response
from app.crud import sample_storage_cert_qc as svc
from app.crud import sample_storage_cert_qc_acl as acl_svc
from app.models.models import SysUser
from app.services.sample_storage_cert_qc import ACL_CONFIG_MENU

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

# 市场 / 质检 / 生产三端共用读写
READ_MENUS = (
    "biz:qcWorkOrder:list",
    "biz:myQcTask:list",
    "production:work-order:list",
    "production:my-task:list",
    "cell-storage:sample-info:list",
)
WRITE_MENUS = READ_MENUS


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 _can_configure_acl(db: Session, user: dict) -> bool:
    return user_has_menu_perm(db, user, ACL_CONFIG_MENU) or strict_user_has_menu_perm(
        db, user, ACL_CONFIG_MENU
    )


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")


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

    field_keys: List[str] = Field(default_factory=list, alias="fieldKeys")
    user_ids: List[str] = Field(default_factory=list, alias="userIds")


@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)
    user_id = _get_current_user_id(user)
    detail = svc.get_detail(db, tenant_no, sample_code)
    if detail is None:
        return error_response(404, "样本不存在")
    has_legacy_write = strict_user_has_any_menu_perm(db, user, WRITE_MENUS)
    svc.attach_acl_context(
        db,
        detail,
        tenant_no=tenant_no,
        user_id=user_id,
        has_legacy_write=has_legacy_write,
        can_configure_acl=_can_configure_acl(db, user),
    )
    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)
    has_legacy_write = strict_user_has_any_menu_perm(db, user, WRITE_MENUS)
    # 先用空 sampleTypeKey 解析严格模式名单；类型在 upsert 内再收窄
    preview = svc.get_detail(db, tenant_no, sample_code)
    if preview is None:
        return error_response(404, "样本不存在")
    editable = acl_svc.resolve_editable_field_keys(
        db,
        tenant_no=tenant_no,
        user_id=user_id,
        sample_type_key=preview.get("sampleTypeKey"),
        has_legacy_write=has_legacy_write,
    )
    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,
            editable_field_keys=editable,
        )
        svc.attach_acl_context(
            db,
            detail,
            tenant_no=tenant_no,
            user_id=user_id,
            has_legacy_write=has_legacy_write,
            can_configure_acl=_can_configure_acl(db, user),
        )
        return success_response(detail, message="保存成功")
    except PermissionError as e:
        return error_response(403, str(e))
    except ValueError as e:
        return error_response(400, str(e))


@router.get("/sample-storage-cert-qc-acl/{scope}")
async def get_sample_storage_cert_qc_acl(
    scope: str,
    user: dict = Depends(require_strict_menu_code(ACL_CONFIG_MENU, allow_super_admin=True)),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    try:
        detail = acl_svc.get_acl_detail(db, tenant_no, scope)
        return success_response(detail)
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/sample-storage-cert-qc-acl/{scope}")
async def put_sample_storage_cert_qc_acl(
    scope: str,
    body: AclUpsertBody,
    user: dict = Depends(require_strict_menu_code(ACL_CONFIG_MENU, allow_super_admin=True)),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    user_id = _get_current_user_id(user)
    try:
        detail = acl_svc.upsert_acl(
            db,
            tenant_no=tenant_no,
            user_id=user_id,
            scope=scope,
            field_keys=body.field_keys,
            user_ids=body.user_ids,
        )
        return success_response(detail, message="保存成功")
    except ValueError as e:
        return error_response(400, str(e))