feat(api): expose GET/PUT sample-storage-cert-qc by sample code.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/__init__.py

@@ -40,6 +40,7 @@ from app.api.pre_transit_sample import router as pre_transit_sample_router
 from app.api.oa_approval import router as oa_approval_router
 from app.api.mini_auth import router as mini_auth_router
 from app.api.mini_menu import router as mini_menu_router
+from app.api.sample_storage_cert_qc import router as sample_storage_cert_qc_router
 
 __all__ = [
     "public_router",
@@ -83,4 +84,5 @@ __all__ = [
     "oa_approval_router",
     "mini_auth_router",
     "mini_menu_router",
+    "sample_storage_cert_qc_router",
 ]

backend/app/api/sample_storage_cert_qc.py

@@ -0,0 +1,84 @@
+"""样本储存证质检信息 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
+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_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))

backend/app/main.py

@@ -13,7 +13,7 @@ from app.core.database import SessionLocal
 from app.core.api_request_log_middleware import ApiRequestLogMiddleware
 from app.core.api_crypto_middleware import ApiCryptoMiddleware
 from app.models import models  # noqa: F401
-from app.api import public_router, menu_router, admin_router, workflow_router, basic_info_router, collection_protocol_router, protocol_sample_payment_router, protocol_refund_record_router, presign_order_router, sample_freight_router, leave_router, operations_router, ethnic_router, sample_reception_router, sample_production_reception_router, qc_router, research_donation_router, production_router, pilot_service_router, insurance_company_router, material_router, storage_release_review_router, sample_sms_notification_router, storage_entry_sms_notification_router, qc_sms_notification_router, protocol_sample_collection_router, research_donation_sample_collection_router, pilot_application_router, pilot_checkup_router, pilot_delivery_router, pilot_infusion_confirm_router, pilot_infusion_schedule_mgmt_router, outbound_release_review_router, clinical_payment_router, env_detect_router, cleanroom_garment_router, market_router, pre_transit_sample_router, oa_approval_router, mini_auth_router, mini_menu_router
+from app.api import public_router, menu_router, admin_router, workflow_router, basic_info_router, collection_protocol_router, protocol_sample_payment_router, protocol_refund_record_router, presign_order_router, sample_freight_router, leave_router, operations_router, ethnic_router, sample_reception_router, sample_production_reception_router, qc_router, research_donation_router, production_router, pilot_service_router, insurance_company_router, material_router, storage_release_review_router, sample_sms_notification_router, storage_entry_sms_notification_router, qc_sms_notification_router, protocol_sample_collection_router, research_donation_sample_collection_router, pilot_application_router, pilot_checkup_router, pilot_delivery_router, pilot_infusion_confirm_router, pilot_infusion_schedule_mgmt_router, outbound_release_review_router, clinical_payment_router, env_detect_router, cleanroom_garment_router, market_router, pre_transit_sample_router, oa_approval_router, mini_auth_router, mini_menu_router, sample_storage_cert_qc_router
 
 
 @asynccontextmanager
@@ -152,6 +152,7 @@ app.include_router(pre_transit_sample_router, prefix=settings.API_V1_PREFIX)
 app.include_router(oa_approval_router, prefix=settings.API_V1_PREFIX)
 app.include_router(mini_auth_router, prefix=settings.API_V1_PREFIX)
 app.include_router(mini_menu_router, prefix=settings.API_V1_PREFIX)
+app.include_router(sample_storage_cert_qc_router, prefix=settings.API_V1_PREFIX)
 
 
 def _mask_secret(value: str) -> str: