zkxw/backend/app/api/pre_transit_sample.py

"""质量检测 — 预产及在途样本 API"""
from typing import Optional

from fastapi import APIRouter, Depends, Query
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 pre_transit_sample as pts_svc
from app.models.models import 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 _page_payload(items: list, total: int, page: int, page_size: int) -> dict:
    return {
        "list": items,
        "total": total,
        "page": page,
        "pageSize": page_size,
        "totalPages": pts_svc.total_pages(total, page_size),
    }


@router.get("/pre-transit-samples/in-transit")
async def list_in_transit_samples(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
    keyword: Optional[str] = None,
    sourceType: Optional[str] = None,
    createTimeStart: Optional[str] = Query(None, alias="createTimeStart"),
    createTimeEnd: Optional[str] = Query(None, alias="createTimeEnd"),
    sampleCollectTimeStart: Optional[str] = Query(None, alias="sampleCollectTimeStart"),
    sampleCollectTimeEnd: Optional[str] = Query(None, alias="sampleCollectTimeEnd"),
    db: Session = Depends(get_db),
    user: dict = Depends(
        require_any_menu_code(
            "biz:preTransitSample:list",
            "production:preTransitSample:list",
        )
    ),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items, total = pts_svc.list_in_transit_protocols(
            db,
            page=page,
            page_size=pageSize,
            tenant_no=tenant_no,
            keyword=keyword,
            source_type=sourceType,
            create_time_start=createTimeStart,
            create_time_end=createTimeEnd,
            sample_collect_time_start=sampleCollectTimeStart,
            sample_collect_time_end=sampleCollectTimeEnd,
            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.get("/pre-transit-samples/pre-delivery")
async def list_pre_delivery_samples(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
    keyword: Optional[str] = None,
    sourceType: Optional[str] = None,
    createTimeStart: Optional[str] = Query(None, alias="createTimeStart"),
    createTimeEnd: Optional[str] = Query(None, alias="createTimeEnd"),
    sampleCollectTimeStart: Optional[str] = Query(None, alias="sampleCollectTimeStart"),
    sampleCollectTimeEnd: Optional[str] = Query(None, alias="sampleCollectTimeEnd"),
    db: Session = Depends(get_db),
    user: dict = Depends(
        require_any_menu_code(
            "biz:preTransitSample:list",
            "production:preTransitSample:list",
        )
    ),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items, total = pts_svc.list_pre_delivery_protocols(
            db,
            page=page,
            page_size=pageSize,
            tenant_no=tenant_no,
            keyword=keyword,
            source_type=sourceType,
            create_time_start=createTimeStart,
            create_time_end=createTimeEnd,
            sample_collect_time_start=sampleCollectTimeStart,
            sample_collect_time_end=sampleCollectTimeEnd,
            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))