921 lines
33 KiB
Python
921 lines
33 KiB
Python
"""基础信息管理 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))
|