shanxia2.0/backend/app/api/sample_reception.py

"""质量检测 — 样本接收 API"""
from typing import List, Optional

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

from app.core.data_scope import apply_data_scope
from app.core.database import get_db
from app.core.menu_guard import require_admin_or_super_admin, require_menu_code
from app.core.response import error_response, success_response
from app.core.utils import parse_datetime
from app.crud import sample_reception as sr_svc
from app.crud import sample_reception_check_item as check_item_svc
from app.crud.collection_protocol import total_pages
from app.models.models import BizSampleReception, BizSampleReceptionCheckItem, SysUser

router = APIRouter(prefix="/web/admin", tags=["质量检测"])


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 CheckItemBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    item_code: str = Field(..., alias="itemCode")
    item_name: str = Field(..., alias="itemName")
    status: str = Field("1", alias="status")
    order_num: Optional[int] = Field(None, alias="orderNum")
    remark: Optional[str] = None


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

    item_code: Optional[str] = Field(None, alias="itemCode")
    item_name: Optional[str] = Field(None, alias="itemName")
    status: Optional[str] = None
    order_num: Optional[int] = Field(None, alias="orderNum")
    remark: Optional[str] = None


@router.get("/sample-reception-check-items")
async def list_check_items(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
    keyword: Optional[str] = None,
    status: Optional[str] = None,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:sampleReception:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        query = db.query(BizSampleReceptionCheckItem).filter(
            BizSampleReceptionCheckItem.del_flag == 0,
            BizSampleReceptionCheckItem.tenant_no == tenant_no,
        )
        if keyword:
            query = query.filter(BizSampleReceptionCheckItem.item_name.like(f"%{keyword}%"))
        if status:
            query = query.filter(BizSampleReceptionCheckItem.status == status)
        query = apply_data_scope(query, BizSampleReceptionCheckItem, user, db, tenant_no)
        total = query.count()
        items = query.order_by(BizSampleReceptionCheckItem.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
        result_list = [check_item_svc._row_to_dict(r) for r in items]
        return success_response(_page_payload(result_list, total, page, pageSize))
    except PermissionError as e:
        return error_response(str(e), code=403)
    except Exception as e:
        return error_response(str(e))


@router.get("/sample-reception-check-items/active")
async def list_active_check_items(
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:sampleReception:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items = check_item_svc.list_active_check_items(db, tenant_no)
        return success_response(items)
    except PermissionError as e:
        return error_response(str(e), code=403)
    except Exception as e:
        return error_response(str(e))


@router.post("/sample-reception-check-items")
async def create_check_item(
    body: CheckItemBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_admin_or_super_admin),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
        dept_id = (sys_user.dept_id or "") if sys_user else ""
        obj = check_item_svc.create_check_item(
            db,
            item_code=body.item_code,
            item_name=body.item_name,
            status=body.status,
            tenant_no=tenant_no,
            order_num=body.order_num,
            remark=body.remark,
            user_id=user_id,
            create_dept_id=dept_id,
        )
        return success_response(check_item_svc._row_to_dict(obj))
    except PermissionError as e:
        return error_response(str(e), code=403)
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))


@router.put("/sample-reception-check-items/{row_id}")
async def update_check_item(
    row_id: str,
    body: CheckItemUpdateBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_admin_or_super_admin),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        obj = check_item_svc.update_check_item(
            db,
            row_id,
            item_code=body.item_code,
            item_name=body.item_name,
            status=body.status,
            order_num=body.order_num,
            remark=body.remark,
            user_id=user_id,
            tenant_no=tenant_no,
        )
        return success_response(check_item_svc._row_to_dict(obj))
    except PermissionError as e:
        return error_response(str(e), code=403)
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))


@router.delete("/sample-reception-check-items/{row_id}")
async def delete_check_item(
    row_id: str,
    db: Session = Depends(get_db),
    user: dict = Depends(require_admin_or_super_admin),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        check_item_svc.soft_delete_check_item(db, row_id, tenant_no=tenant_no)
        return success_response(None)
    except PermissionError as e:
        return error_response(str(e), code=403)
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))


# ---------- 样本接收 ----------


@router.get("/sample-receptions/pending")
async def list_pending_samples(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
    keyword: Optional[str] = None,
    sourceType: Optional[str] = None,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:sampleReception:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items, total = sr_svc.list_pending_protocol_samples(
            db, page=page, page_size=pageSize, tenant_no=tenant_no, keyword=keyword,
            source_type=sourceType, user_info=user,
        )
        return success_response(_page_payload(items, total, page, pageSize))
    except PermissionError as e:
        return error_response(str(e), code=403)
    except Exception as e:
        return error_response(str(e))


@router.get("/sample-receptions/received")
async def list_received_samples(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
    keyword: Optional[str] = None,
    sourceType: Optional[str] = None,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:sampleReception:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items, total = sr_svc.list_received_protocol_samples(
            db, page=page, page_size=pageSize, tenant_no=tenant_no, keyword=keyword,
            source_type=sourceType, user_info=user,
        )
        return success_response(_page_payload(items, total, page, pageSize))
    except PermissionError as e:
        return error_response(str(e), code=403)
    except Exception as e:
        return error_response(str(e))


@router.get("/sample-receptions/{reception_id}")
async def get_sample_reception_detail(
    reception_id: str,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:sampleReception:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        item = sr_svc.get_reception_by_id(db, reception_id, tenant_no)
        if not item:
            return error_response("接收记录不存在", code=404)
        return success_response(item)
    except PermissionError as e:
        return error_response(str(e), code=403)
    except Exception as e:
        return error_response(str(e))


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

    check_item_id: str = Field(..., alias="checkItemId")
    checked: bool = False


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

    reception_time: str = Field(..., alias="receptionTime")
    handler_user_id: Optional[str] = Field(None, alias="handlerUserId")
    handler_name: Optional[str] = Field(None, alias="handlerName")
    receive_temperature: Optional[str] = Field(None, alias="receiveTemperature")
    volume_ml: Optional[float] = Field(None, alias="volumeMl")
    net_weight_g: Optional[float] = Field(None, alias="netWeightG")
    checks: List[ReceptionCheckSubmit] = Field(default_factory=list)
    accept_conclusion: str = Field(..., alias="acceptConclusion")
    receive_attachments: Optional[str] = Field(None, alias="receiveAttachments")
    remark: Optional[str] = None


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

    protocol_sample_id: Optional[str] = Field(None, alias="protocolSampleId")
    reception_time: str = Field(..., alias="receptionTime")
    handler_user_id: Optional[str] = Field(None, alias="handlerUserId")
    handler_name: Optional[str] = Field(None, alias="handlerName")
    receive_temperature: Optional[str] = Field(None, alias="receiveTemperature")
    volume_ml: Optional[float] = Field(None, alias="volumeMl")
    net_weight_g: Optional[float] = Field(None, alias="netWeightG")
    checks: List[ReceptionCheckSubmit] = Field(default_factory=list)
    accept_conclusion: str = Field(..., alias="acceptConclusion")
    receive_attachments: Optional[str] = Field(None, alias="receiveAttachments")
    remark: Optional[str] = None
    source_type: Optional[str] = Field("cell_storage", alias="sourceType")
    donation_sample_id: Optional[str] = Field(None, alias="donationSampleId")


@router.post("/sample-receptions")
async def post_sample_reception(
    body: SampleReceptionCreateBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_admin_or_super_admin),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
        dept_id = (sys_user.dept_id or "") if sys_user else ""
        rt = parse_datetime(body.reception_time)
        if not rt:
            return error_response("样本接收时间格式无效")
        checks_payload = [
            {"checkItemId": c.check_item_id, "checked": c.checked} for c in body.checks
        ]
        obj = sr_svc.create_sample_reception(
            db,
            reception_source=body.source_type or "cell_storage",
            protocol_sample_id=body.protocol_sample_id if body.source_type != "research_donation" else None,
            donation_sample_id=body.donation_sample_id if body.source_type == "research_donation" else None,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=dept_id,
            reception_time=rt,
            handler_user_id=body.handler_user_id or user_id,
            handler_name=body.handler_name,
            receive_temperature=body.receive_temperature,
            volume_ml=body.volume_ml,
            net_weight_g=body.net_weight_g,
            checks=checks_payload,
            accept_conclusion=body.accept_conclusion,
            receive_attachments=body.receive_attachments,
            remark=body.remark,
        )
        detail = sr_svc.get_reception_by_id(db, obj.id, tenant_no)
        return success_response(detail or {"id": obj.id})
    except PermissionError as e:
        return error_response(str(e), code=403)
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))


@router.put("/sample-receptions/{reception_id}")
async def put_sample_reception(
    reception_id: str,
    body: SampleReceptionUpdateBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_admin_or_super_admin),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        rt = parse_datetime(body.reception_time)
        if not rt:
            return error_response("样本接收时间格式无效")
        checks_payload = [
            {"checkItemId": c.check_item_id, "checked": c.checked} for c in body.checks
        ]
        obj = sr_svc.update_sample_reception(
            db,
            reception_id=reception_id,
            tenant_no=tenant_no,
            user_id=user_id,
            reception_time=rt,
            handler_user_id=body.handler_user_id or user_id,
            handler_name=body.handler_name,
            receive_temperature=body.receive_temperature,
            volume_ml=body.volume_ml,
            net_weight_g=body.net_weight_g,
            checks=checks_payload,
            accept_conclusion=body.accept_conclusion,
            receive_attachments=body.receive_attachments,
            remark=body.remark,
        )
        detail = sr_svc.get_reception_by_id(db, obj.id, tenant_no)
        return success_response(detail or {"id": obj.id})
    except PermissionError as e:
        return error_response(str(e), code=403)
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))


@router.put("/sample-receptions/{reception_id}")
async def put_sample_reception(
    reception_id: str,
    body: SampleReceptionUpdateBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_admin_or_super_admin),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        rt = parse_datetime(body.reception_time)
        if not rt:
            return error_response("样本接收时间格式无效")
        checks_payload = [
            {"checkItemId": c.check_item_id, "checked": c.checked} for c in body.checks
        ]
        obj = sr_svc.update_sample_reception(
            db,
            reception_id=reception_id,
            tenant_no=tenant_no,
            user_id=user_id,
            reception_time=rt,
            handler_user_id=body.handler_user_id or user_id,
            handler_name=body.handler_name,
            receive_temperature=body.receive_temperature,
            volume_ml=body.volume_ml,
            net_weight_g=body.net_weight_g,
            checks=checks_payload,
            accept_conclusion=body.accept_conclusion,
            receive_attachments=body.receive_attachments,
            remark=body.remark,
        )
        detail = sr_svc.get_reception_by_id(db, obj.id, tenant_no)
        return success_response(detail or {"id": obj.id})
    except PermissionError as e:
        return error_response(str(e), code=403)
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))