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

"""基础信息管理 API（客户、样本类型、采集地区/医院、协议号池、样本编码池）"""
from typing import Any, 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.crud import cell_sample_type as cell_type_svc
from app.crud import collection_region as region_svc
from app.crud import collection_protocol as collection_protocol_svc
from app.crud import collection_protocol_no as protocol_no_svc
from app.crud import customer as customer_svc
from app.crud import sample_code as sample_code_svc
from app.models.models import (
    BizCellSampleType,
    BizCollectionProtocolNo,
    BizCollectionRegion,
    BizCustomer,
    BizSampleCode,
    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


# ---------------------------------------------------------------------------
# 客户
# ---------------------------------------------------------------------------


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

    customer_name: str = Field(..., alias="customerName")
    sex: Optional[str] = Field("2", alias="sex")
    id_card_no: Optional[str] = Field(None, alias="idCardNo")
    age: Optional[int] = None
    phone_no: Optional[str] = Field(None, alias="phoneNo")
    customer_tags: Optional[str] = Field(None, alias="customerTags")
    order_num: Optional[int] = Field(0, alias="orderNum")
    remark: Optional[str] = None


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

    customer_name: Optional[str] = Field(None, alias="customerName")
    sex: Optional[str] = None
    id_card_no: Optional[str] = Field(None, alias="idCardNo")
    age: Optional[int] = None
    phone_no: Optional[str] = Field(None, alias="phoneNo")
    customer_tags: Optional[str] = Field(None, alias="customerTags")
    order_num: Optional[int] = Field(None, alias="orderNum")
    remark: Optional[str] = None


@router.get("/customers")
async def list_customers(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
    keyword: Optional[str] = None,
    user: dict = Depends(require_menu_code("biz:customer:list")),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    query = db.query(BizCustomer).filter(
        BizCustomer.del_flag == 0,
        BizCustomer.tenant_no == tenant_no,
    )
    if keyword:
        query = query.filter(BizCustomer.customer_name.like(f"%{keyword}%"))
    query = apply_data_scope(query, BizCustomer, user, db, tenant_no)
    total = query.count()
    items = query.order_by(BizCustomer.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
    result_list = [customer_svc._row_to_dict(r) for r in items]
    tp = customer_svc.total_pages(total, pageSize)
    return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})


@router.post("/customers")
async def create_customer(
    body: CustomerCreateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        obj = customer_svc.create_customer(
            db,
            customer_name=body.customer_name,
            sex=body.sex or "2",
            id_card_no=body.id_card_no,
            age=body.age,
            phone_no=body.phone_no,
            customer_tags=body.customer_tags,
            tenant_no=tenant_no,
            order_num=body.order_num,
            remark=body.remark,
            user_id=user.get("user_id"),
        )
        return success_response(customer_svc._row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/customers/{customer_id}")
async def update_customer(
    customer_id: str,
    body: CustomerUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
        obj = customer_svc.update_customer(
            db,
            customer_id,
            customer_name=payload.get("customer_name"),
            sex=payload.get("sex"),
            id_card_no=payload.get("id_card_no"),
            age=payload.get("age"),
            phone_no=payload.get("phone_no"),
            customer_tags=payload.get("customer_tags"),
            order_num=payload.get("order_num"),
            remark=payload.get("remark"),
            user_id=user.get("user_id"),
            tenant_no=tenant_no,
        )
        return success_response(customer_svc._row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.delete("/customers/{customer_id}")
async def delete_customer(
    customer_id: str,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        customer_svc.soft_delete_customer(db, customer_id, tenant_no=tenant_no)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))


# ---------------------------------------------------------------------------
# 细胞样本类型
# ---------------------------------------------------------------------------


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

    type_code: str = Field(..., alias="typeCode")
    type_name: str = Field(..., alias="typeName")
    storage_years: Optional[int] = Field(None, alias="storageYears")
    storage_price: Optional[float] = Field(None, alias="storagePrice")
    order_num: Optional[int] = Field(0, alias="orderNum")
    remark: Optional[str] = None


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

    type_code: Optional[str] = Field(None, alias="typeCode")
    type_name: Optional[str] = Field(None, alias="typeName")
    storage_years: Optional[int] = Field(None, alias="storageYears")
    storage_price: Optional[float] = Field(None, alias="storagePrice")
    order_num: Optional[int] = Field(None, alias="orderNum")
    remark: Optional[str] = None


@router.get("/cell-sample-types")
async def list_cell_sample_types(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
    keyword: Optional[str] = None,
    user: dict = Depends(require_menu_code("biz:cellSampleType:list")),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    query = db.query(BizCellSampleType).filter(
        BizCellSampleType.del_flag == 0,
        BizCellSampleType.tenant_no == tenant_no,
    )
    if keyword:
        query = query.filter(BizCellSampleType.type_name.like(f"%{keyword}%"))
    query = apply_data_scope(query, BizCellSampleType, user, db, tenant_no)
    total = query.count()
    items = query.order_by(BizCellSampleType.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
    result_list = [cell_type_svc._row_to_dict(r) for r in items]
    tp = cell_type_svc.total_pages(total, pageSize)
    return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})


@router.post("/cell-sample-types")
async def create_cell_sample_type(
    body: CellSampleTypeCreateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        obj = cell_type_svc.create_cell_sample_type(
            db,
            type_code=body.type_code,
            type_name=body.type_name,
            storage_years=body.storage_years,
            storage_price=body.storage_price,
            tenant_no=tenant_no,
            order_num=body.order_num,
            remark=body.remark,
            user_id=user.get("user_id"),
        )
        return success_response(cell_type_svc._row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/cell-sample-types/{row_id}")
async def update_cell_sample_type(
    row_id: str,
    body: CellSampleTypeUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
        obj = cell_type_svc.update_cell_sample_type(
            db,
            row_id,
            type_code=payload.get("type_code"),
            type_name=payload.get("type_name"),
            storage_years=payload.get("storage_years"),
            storage_price=payload.get("storage_price"),
            order_num=payload.get("order_num"),
            remark=payload.get("remark"),
            user_id=user.get("user_id"),
            tenant_no=tenant_no,
        )
        return success_response(cell_type_svc._row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.delete("/cell-sample-types/{row_id}")
async def delete_cell_sample_type(
    row_id: str,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        cell_type_svc.soft_delete_cell_sample_type(db, row_id, tenant_no=tenant_no)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))


# ---------------------------------------------------------------------------
# 采集地区
# ---------------------------------------------------------------------------


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

    region_name: str = Field(..., alias="regionName")
    order_num: Optional[int] = Field(0, alias="orderNum")
    remark: Optional[str] = None


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

    region_name: Optional[str] = Field(None, alias="regionName")
    order_num: Optional[int] = Field(None, alias="orderNum")
    remark: Optional[str] = None


@router.get("/collection-regions")
async def list_collection_regions(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
    keyword: Optional[str] = None,
    user: dict = Depends(require_menu_code("biz:collectionRegion:list")),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    query = db.query(BizCollectionRegion).filter(
        BizCollectionRegion.del_flag == 0,
        BizCollectionRegion.tenant_no == tenant_no,
    )
    if keyword:
        query = query.filter(BizCollectionRegion.region_name.like(f"%{keyword}%"))
    query = apply_data_scope(query, BizCollectionRegion, user, db, tenant_no)
    total = query.count()
    items = query.order_by(BizCollectionRegion.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
    result_list = [region_svc._region_to_dict(r) for r in items]
    tp = region_svc.total_pages(total, pageSize)
    return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})


@router.post("/collection-regions")
async def create_collection_region(
    body: CollectionRegionCreateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        obj = region_svc.create_collection_region(
            db,
            region_name=body.region_name,
            tenant_no=tenant_no,
            order_num=body.order_num,
            remark=body.remark,
            user_id=user.get("user_id"),
        )
        return success_response(region_svc._region_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/collection-regions/{region_id}")
async def update_collection_region(
    region_id: str,
    body: CollectionRegionUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
        obj = region_svc.update_collection_region(
            db,
            region_id,
            region_name=payload.get("region_name"),
            order_num=payload.get("order_num"),
            remark=payload.get("remark"),
            user_id=user.get("user_id"),
            tenant_no=tenant_no,
        )
        return success_response(region_svc._region_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.delete("/collection-regions/{region_id}")
async def delete_collection_region(
    region_id: str,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        region_svc.soft_delete_collection_region(db, region_id, tenant_no=tenant_no)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))


# ---------------------------------------------------------------------------
# 采集地区 — 下属医院
# ---------------------------------------------------------------------------


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

    hospital_name: str = Field(..., alias="hospitalName")
    order_num: Optional[int] = Field(0, alias="orderNum")
    remark: Optional[str] = None


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

    hospital_name: Optional[str] = Field(None, alias="hospitalName")
    order_num: Optional[int] = Field(None, alias="orderNum")
    remark: Optional[str] = None


@router.get("/collection-regions/{region_id}/hospitals")
async def list_hospitals(
    region_id: str,
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
    keyword: Optional[str] = None,
    user: dict = Depends(require_menu_code("biz:collectionRegion:list")),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items, total = region_svc.list_hospitals(
            db, region_id=region_id, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no
        )
        tp = region_svc.total_pages(total, pageSize)
        return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
    except ValueError as e:
        return error_response(400, str(e))


@router.post("/collection-regions/{region_id}/hospitals")
async def create_hospital(
    region_id: str,
    body: HospitalCreateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        obj = region_svc.create_hospital(
            db,
            region_id=region_id,
            hospital_name=body.hospital_name,
            tenant_no=tenant_no,
            order_num=body.order_num,
            remark=body.remark,
            user_id=user.get("user_id"),
        )
        return success_response(region_svc._hospital_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/collection-regions/{region_id}/hospitals/{hospital_id}")
async def update_hospital(
    region_id: str,
    hospital_id: str,
    body: HospitalUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        h = region_svc.get_hospital(db, hospital_id, tenant_no=tenant_no)
        if not h or h.region_id != region_id:
            return error_response(400, "医院不存在")
        payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
        obj = region_svc.update_hospital(
            db,
            hospital_id,
            hospital_name=payload.get("hospital_name"),
            order_num=payload.get("order_num"),
            remark=payload.get("remark"),
            user_id=user.get("user_id"),
            tenant_no=tenant_no,
        )
        return success_response(region_svc._hospital_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.delete("/collection-regions/{region_id}/hospitals/{hospital_id}")
async def delete_hospital(
    region_id: str,
    hospital_id: str,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        h = region_svc.get_hospital(db, hospital_id, tenant_no=tenant_no)
        if not h or h.region_id != region_id:
            return error_response(400, "医院不存在")
        region_svc.soft_delete_hospital(db, hospital_id, tenant_no=tenant_no)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))


# ---------------------------------------------------------------------------
# 采集协议号池
# ---------------------------------------------------------------------------


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

    protocol_no: str = Field(..., alias="protocolNo")
    issued: int = 0
    order_num: Optional[int] = Field(0, alias="orderNum")
    remark: Optional[str] = None


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

    protocol_no: Optional[str] = Field(None, alias="protocolNo")
    issued: Optional[int] = None
    order_num: Optional[int] = Field(None, alias="orderNum")
    remark: Optional[str] = None
    issued_user_id: Optional[str] = Field(None, alias="issuedUserId")
    issued_user_name: Optional[str] = Field(None, alias="issuedUserName")
    issued_time: Optional[str] = Field(None, alias="issuedTime")


@router.get("/collection-protocol-nos")
async def list_protocol_nos(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
    keyword: Optional[str] = None,
    user: dict = Depends(require_menu_code("biz:collectionProtocolNo:list")),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    query = db.query(BizCollectionProtocolNo).filter(
        BizCollectionProtocolNo.del_flag == 0,
        BizCollectionProtocolNo.tenant_no == tenant_no,
    )
    if keyword:
        query = query.filter(BizCollectionProtocolNo.protocol_no.like(f"%{keyword}%"))
    query = apply_data_scope(query, BizCollectionProtocolNo, user, db, tenant_no)
    total = query.count()
    items = query.order_by(BizCollectionProtocolNo.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
    result_list = [protocol_no_svc._row_to_dict(r) for r in items]
    used_nos = collection_protocol_svc.get_protocol_nos_used_in_protocols(
        db,
        tenant_no=tenant_no,
        protocol_nos=[row["protocolNo"] for row in result_list],
    )
    for row in result_list:
        row["usedInProtocol"] = row.get("protocolNo") in used_nos
    tp = protocol_no_svc.total_pages(total, pageSize)
    return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})


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

    prefix: str = Field("ZKZX", alias="prefix")
    middle: str = Field("34", alias="middle")
    seq_start: int = Field(..., alias="seqStart")
    seq_end: int = Field(..., alias="seqEnd")
    digit_count: int = Field(6, alias="digitCount")


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

    ids: List[str] = Field(..., alias="ids")
    issued_user_id: str = Field(..., alias="issuedUserId")
    issued_user_name: Optional[str] = Field(None, alias="issuedUserName")


@router.post("/collection-protocol-nos/batch-generate")
async def batch_generate_protocol_nos(
    body: ProtocolNoBatchGenerateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        result = protocol_no_svc.batch_generate_protocol_nos(
            db,
            prefix=body.prefix,
            middle=body.middle,
            seq_start=body.seq_start,
            seq_end=body.seq_end,
            digit_count=body.digit_count,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
        )
        msg = f"成功生成 {result['created']} 条"
        if result['skipped'] > 0:
            msg += f"，跳过已存在 {result['skipped']} 条"
        return success_response(result, message=msg)
    except ValueError as e:
        return error_response(400, str(e))


@router.post("/collection-protocol-nos/batch-issue")
async def batch_issue_protocol_nos(
    body: ProtocolNoBatchIssueBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        result = protocol_no_svc.batch_issue_protocol_nos(
            db,
            ids=body.ids,
            issued_user_id=body.issued_user_id,
            issued_user_name=body.issued_user_name,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
        )
        msg = f"成功下发 {result['issuedCount']} 条"
        if result['alreadyIssued']:
            msg += f"，{len(result['alreadyIssued'])} 条已下发被跳过"
        return success_response(result, message=msg)
    except ValueError as e:
        return error_response(400, str(e))


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

    ids: List[str] = Field(..., alias="ids")


@router.post("/collection-protocol-nos/batch-revoke")
async def batch_revoke_protocol_nos(
    body: ProtocolNoBatchRevokeBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        result = protocol_no_svc.batch_revoke_protocol_nos(
            db,
            ids=body.ids,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
        )
        msg = f"成功收回 {result['revokedCount']} 条"
        if result["notIssued"]:
            msg += f"，{len(result['notIssued'])} 条未下发已跳过"
        return success_response(result, message=msg)
    except ValueError as e:
        return error_response(400, str(e))


@router.post("/collection-protocol-nos")
async def create_protocol_no(
    body: ProtocolNoCreateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        obj = protocol_no_svc.create_protocol_no(
            db,
            protocol_no=body.protocol_no,
            issued=body.issued,
            tenant_no=tenant_no,
            order_num=body.order_num,
            remark=body.remark,
            user_id=user.get("user_id"),
        )
        return success_response(protocol_no_svc._row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/collection-protocol-nos/{row_id}")
async def update_protocol_no(
    row_id: str,
    body: ProtocolNoUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
        obj = protocol_no_svc.update_protocol_no(
            db,
            row_id,
            protocol_no=payload.get("protocol_no"),
            issued=payload.get("issued"),
            order_num=payload.get("order_num"),
            remark=payload.get("remark"),
            user_id=user.get("user_id"),
            tenant_no=tenant_no,
            issued_user_id=payload.get("issued_user_id"),
            issued_user_name=payload.get("issued_user_name"),
            issued_time=payload.get("issued_time"),
        )
        return success_response(protocol_no_svc._row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.delete("/collection-protocol-nos/{row_id}")
async def delete_protocol_no(
    row_id: str,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        protocol_no_svc.soft_delete_protocol_no(db, row_id, tenant_no=tenant_no)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))


# ---------------------------------------------------------------------------
# 样本编码池
# ---------------------------------------------------------------------------


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

    sample_code: str = Field(..., alias="sampleCode")
    issued: int = 0
    order_num: Optional[int] = Field(0, alias="orderNum")
    remark: Optional[str] = None


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

    sample_code: Optional[str] = Field(None, alias="sampleCode")
    issued: Optional[int] = None
    order_num: Optional[int] = Field(None, alias="orderNum")
    remark: Optional[str] = None
    issued_user_id: Optional[str] = Field(None, alias="issuedUserId")
    issued_user_name: Optional[str] = Field(None, alias="issuedUserName")
    issued_time: Optional[str] = Field(None, alias="issuedTime")


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

    sample_code_type: str = Field(..., alias="sampleCodeType")
    middle: str = Field("34", alias="middle")
    seq_start: int = Field(..., alias="seqStart")
    seq_end: int = Field(..., alias="seqEnd")
    digit_count: int = Field(6, alias="digitCount")


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

    ids: List[str] = Field(..., alias="ids")
    issued_user_id: str = Field(..., alias="issuedUserId")
    issued_user_name: Optional[str] = Field(None, alias="issuedUserName")


@router.post("/sample-codes/batch-generate")
async def batch_generate_sample_codes(
    body: SampleCodeBatchGenerateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        result = sample_code_svc.batch_generate_sample_codes(
            db,
            sample_code_type=body.sample_code_type,
            middle=body.middle,
            seq_start=body.seq_start,
            seq_end=body.seq_end,
            digit_count=body.digit_count,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
        )
        msg = f"成功生成 {result['created']} 条"
        if result["skipped"] > 0:
            msg += f"，跳过已存在 {result['skipped']} 条"
        return success_response(result, message=msg)
    except ValueError as e:
        return error_response(400, str(e))


@router.post("/sample-codes/batch-issue")
async def batch_issue_sample_codes(
    body: SampleCodeBatchIssueBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        result = sample_code_svc.batch_issue_sample_codes(
            db,
            ids=body.ids,
            issued_user_id=body.issued_user_id,
            issued_user_name=body.issued_user_name,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
        )
        msg = f"成功下发 {result['issuedCount']} 条"
        if result["alreadyIssued"]:
            msg += f"，{len(result['alreadyIssued'])} 条已下发被跳过"
        return success_response(result, message=msg)
    except ValueError as e:
        return error_response(400, str(e))


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

    ids: List[str] = Field(..., alias="ids")


@router.post("/sample-codes/batch-revoke")
async def batch_revoke_sample_codes(
    body: SampleCodeBatchRevokeBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        result = sample_code_svc.batch_revoke_sample_codes(
            db,
            ids=body.ids,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
        )
        msg = f"成功收回 {result['revokedCount']} 条"
        if result["notIssued"]:
            msg += f"，{len(result['notIssued'])} 条未下发已跳过"
        return success_response(result, message=msg)
    except ValueError as e:
        return error_response(400, str(e))


@router.get("/sample-codes")
async def list_sample_codes(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
    keyword: Optional[str] = None,
    user: dict = Depends(require_menu_code("biz:sampleCode:list")),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    query = db.query(BizSampleCode).filter(
        BizSampleCode.del_flag == 0,
        BizSampleCode.tenant_no == tenant_no,
    )
    if keyword:
        query = query.filter(BizSampleCode.sample_code.like(f"%{keyword}%"))
    query = apply_data_scope(query, BizSampleCode, user, db, tenant_no)
    total = query.count()
    items = query.order_by(BizSampleCode.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
    result_list = [sample_code_svc._row_to_dict(r) for r in items]
    used_codes = collection_protocol_svc.get_sample_codes_used_in_protocol_samples(
        db,
        tenant_no=tenant_no,
        sample_codes=[row["sampleCode"] for row in result_list],
    )
    for row in result_list:
        row["usedInProtocol"] = row.get("sampleCode") in used_codes
    tp = sample_code_svc.total_pages(total, pageSize)
    return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})


@router.post("/sample-codes")
async def create_sample_code(
    body: SampleCodeCreateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        obj = sample_code_svc.create_sample_code(
            db,
            sample_code=body.sample_code,
            issued=body.issued,
            tenant_no=tenant_no,
            order_num=body.order_num,
            remark=body.remark,
            user_id=user.get("user_id"),
        )
        return success_response(sample_code_svc._row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/sample-codes/{row_id}")
async def update_sample_code(
    row_id: str,
    body: SampleCodeUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
        obj = sample_code_svc.update_sample_code(
            db,
            row_id,
            sample_code=payload.get("sample_code"),
            issued=payload.get("issued"),
            order_num=payload.get("order_num"),
            remark=payload.get("remark"),
            user_id=user.get("user_id"),
            tenant_no=tenant_no,
            issued_user_id=payload.get("issued_user_id"),
            issued_user_name=payload.get("issued_user_name"),
            issued_time=payload.get("issued_time"),
        )
        return success_response(sample_code_svc._row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.delete("/sample-codes/{row_id}")
async def delete_sample_code(
    row_id: str,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        sample_code_svc.soft_delete_sample_code(db, row_id, tenant_no=tenant_no)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))