1170 lines
42 KiB
Python
1170 lines
42 KiB
Python
"""基础信息管理 API（客户、样本类型、采集地区/医院、协议号池、样本编码池）"""
|
||
from typing import Any, List, Optional
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
from pydantic import BaseModel, ConfigDict, Field
|
||
from sqlalchemy import or_
|
||
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,
|
||
)
|
||
from app.services.customer_relations import build_customer_relations
|
||
|
||
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
|
||
|
||
|
||
def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
|
||
val = (value or "").strip()
|
||
if len(val) == 10:
|
||
return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00"
|
||
return val
|
||
|
||
|
||
@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,
|
||
createTimeStart: Optional[str] = Query(None, alias="createTimeStart"),
|
||
createTimeEnd: Optional[str] = Query(None, alias="createTimeEnd"),
|
||
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:
|
||
kw = keyword.strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
query = query.filter(
|
||
or_(
|
||
BizCustomer.customer_name.like(like),
|
||
BizCustomer.phone_no.like(like),
|
||
BizCustomer.id_card_no.like(like),
|
||
BizCustomer.customer_tags.like(like),
|
||
)
|
||
)
|
||
if createTimeStart:
|
||
query = query.filter(
|
||
BizCustomer.create_time >= _normalize_date_bound(createTimeStart)
|
||
)
|
||
if createTimeEnd:
|
||
query = query.filter(
|
||
BizCustomer.create_time <= _normalize_date_bound(createTimeEnd, end_of_day=True)
|
||
)
|
||
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.get("/customers/{customer_id}/relations")
|
||
async def get_customer_relations(
|
||
customer_id: str,
|
||
user: dict = Depends(require_menu_code("biz:customer:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
data = build_customer_relations(db, customer_id, tenant_no)
|
||
if not data:
|
||
return error_response(404, "客户不存在")
|
||
return success_response(data)
|
||
|
||
|
||
@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")
|
||
type_logo: Optional[str] = Field(None, alias="typeLogo")
|
||
storage_years: Optional[int] = Field(None, alias="storageYears")
|
||
storage_price: Optional[float] = Field(None, alias="storagePrice")
|
||
sample_code_filter: Optional[str] = Field(None, alias="sampleCodeFilter")
|
||
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")
|
||
type_logo: Optional[str] = Field(None, alias="typeLogo")
|
||
storage_years: Optional[int] = Field(None, alias="storageYears")
|
||
storage_price: Optional[float] = Field(None, alias="storagePrice")
|
||
sample_code_filter: Optional[str] = Field(None, alias="sampleCodeFilter")
|
||
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:
|
||
kw = keyword.strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
query = query.filter(
|
||
or_(
|
||
BizCellSampleType.type_code.like(like),
|
||
BizCellSampleType.type_name.like(like),
|
||
)
|
||
)
|
||
query = apply_data_scope(query, BizCellSampleType, user, db, tenant_no)
|
||
total = query.count()
|
||
items = query.order_by(BizCellSampleType.order_num.asc(), 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,
|
||
type_logo=body.type_logo,
|
||
storage_years=body.storage_years,
|
||
storage_price=body.storage_price,
|
||
sample_code_filter=body.sample_code_filter,
|
||
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"),
|
||
type_logo=payload.get("type_logo"),
|
||
storage_years=payload.get("storage_years"),
|
||
storage_price=payload.get("storage_price"),
|
||
sample_code_filter=payload.get("sample_code_filter"),
|
||
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()
|
||
region_ids = [r.id for r in items]
|
||
hospital_map = region_svc._hospital_names_by_region_ids(db, region_ids)
|
||
staff_map = region_svc._staff_names_by_region_ids(db, region_ids)
|
||
result_list = [
|
||
region_svc._region_to_dict(
|
||
r,
|
||
hospital_names=hospital_map.get(r.id, []),
|
||
staff_names=staff_map.get(r.id, ([], []))[0],
|
||
leader_names=staff_map.get(r.id, ([], []))[1],
|
||
)
|
||
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-hospitals/search")
|
||
async def search_collection_hospitals(
|
||
keyword: str = Query(..., min_length=1),
|
||
limit: int = Query(20, ge=1, le=50),
|
||
user: dict = Depends(require_menu_code("biz:collectionRegion:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
items = region_svc.search_hospitals_by_keyword(db, keyword=keyword, tenant_no=tenant_no, limit=limit)
|
||
return success_response({"list": items})
|
||
|
||
|
||
@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 RegionStaffCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
user_id: str = Field(..., alias="userId")
|
||
is_leader: Optional[bool] = Field(False, alias="isLeader")
|
||
order_num: Optional[int] = Field(0, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class RegionStaffUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
is_leader: Optional[bool] = Field(None, alias="isLeader")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
@router.get("/collection-regions/{region_id}/staff")
|
||
async def list_region_staff(
|
||
region_id: str,
|
||
page: int = Query(1, ge=1),
|
||
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
||
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_staff(
|
||
db, region_id=region_id, page=page, page_size=pageSize, 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}/staff")
|
||
async def create_region_staff(
|
||
region_id: str,
|
||
body: RegionStaffCreateBody,
|
||
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_staff(
|
||
db,
|
||
region_id=region_id,
|
||
user_id=body.user_id,
|
||
is_leader=bool(body.is_leader),
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id_actor=user.get("user_id"),
|
||
)
|
||
# enrich response with user display fields
|
||
items, _ = region_svc.list_staff(db, region_id=region_id, page=1, page_size=100, tenant_no=tenant_no)
|
||
matched = next((x for x in items if x["id"] == obj.id), region_svc._staff_to_dict(obj))
|
||
return success_response(matched, message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/collection-regions/{region_id}/staff/{staff_id}")
|
||
async def update_region_staff(
|
||
region_id: str,
|
||
staff_id: str,
|
||
body: RegionStaffUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
s = region_svc.get_staff(db, staff_id, tenant_no=tenant_no)
|
||
if not s or s.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_staff(
|
||
db,
|
||
staff_id,
|
||
is_leader=payload.get("is_leader"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id=user.get("user_id"),
|
||
tenant_no=tenant_no,
|
||
)
|
||
items, _ = region_svc.list_staff(db, region_id=region_id, page=1, page_size=100, tenant_no=tenant_no)
|
||
matched = next((x for x in items if x["id"] == obj.id), region_svc._staff_to_dict(obj))
|
||
return success_response(matched, message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/collection-regions/{region_id}/staff/{staff_id}")
|
||
async def delete_region_staff(
|
||
region_id: str,
|
||
staff_id: str,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
s = region_svc.get_staff(db, staff_id, tenant_no=tenant_no)
|
||
if not s or s.region_id != region_id:
|
||
return error_response(400, "绑定记录不存在")
|
||
region_svc.soft_delete_staff(db, staff_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
|
||
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 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,
|
||
issued_user_id: Optional[str] = Query(None, alias="issuedUserId"),
|
||
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,
|
||
)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
query = query.filter(
|
||
or_(
|
||
BizCollectionProtocolNo.protocol_no.like(like),
|
||
BizCollectionProtocolNo.remark.like(like),
|
||
)
|
||
)
|
||
uid = (issued_user_id or "").strip()
|
||
if uid:
|
||
query = query.filter(BizCollectionProtocolNo.issued_user_id == uid)
|
||
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]
|
||
protocol_no_svc.enrich_issued_user_names(db, result_list)
|
||
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"),
|
||
issued_user_id=body.issued_user_id,
|
||
issued_user_name=body.issued_user_name,
|
||
issued_time=body.issued_time,
|
||
)
|
||
row = protocol_no_svc._row_to_dict(obj)
|
||
protocol_no_svc.enrich_issued_user_names(db, [row])
|
||
return success_response(row, 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")
|
||
sample_code_type: Optional[str] = Field(None, alias="sampleCodeType")
|
||
issued: int = 0
|
||
order_num: Optional[int] = Field(0, 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 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,
|
||
sample_code_type: Optional[str] = Query(None, alias="sampleCodeType"),
|
||
code_filter: Optional[str] = Query(None, alias="codeFilter"),
|
||
issued_user_id: Optional[str] = Query(None, alias="issuedUserId"),
|
||
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,
|
||
)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
query = query.filter(
|
||
or_(
|
||
BizSampleCode.sample_code.like(like),
|
||
BizSampleCode.remark.like(like),
|
||
)
|
||
)
|
||
code_type = (sample_code_type or "").strip()
|
||
if code_type:
|
||
query = query.filter(BizSampleCode.sample_code_type == code_type)
|
||
cf = (code_filter or "").strip()
|
||
if cf:
|
||
query = query.filter(BizSampleCode.sample_code.like(f"%{cf}%"))
|
||
uid = (issued_user_id or "").strip()
|
||
if uid:
|
||
query = query.filter(BizSampleCode.issued_user_id == uid)
|
||
query = apply_data_scope(query, BizSampleCode, user, db, tenant_no)
|
||
total = query.count()
|
||
items = (
|
||
query.order_by(BizSampleCode.create_time.desc(), BizSampleCode.sample_code.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,
|
||
sample_code_type=body.sample_code_type,
|
||
issued=body.issued,
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
issued_user_id=body.issued_user_id,
|
||
issued_user_name=body.issued_user_name,
|
||
issued_time=body.issued_time,
|
||
)
|
||
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))
|