203 lines
7.4 KiB
Python
203 lines
7.4 KiB
Python
"""中试输注计划管理 API"""
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import AliasChoices, BaseModel, ConfigDict, Field
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.database import get_db
|
|
from app.core.menu_guard import require_admin_or_super_admin
|
|
from app.core.response import error_response, success_response
|
|
from app.crud import pilot_infusion_schedule_mgmt as crud
|
|
from app.models.models import SysUser
|
|
|
|
router = APIRouter(prefix="/web/admin", tags=["中试输注计划管理"])
|
|
|
|
|
|
def _get_tenant_no(db: Session, user: dict) -> str:
|
|
user_id = user.get("user_id")
|
|
if not user_id:
|
|
raise PermissionError("用户身份信息缺失")
|
|
sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
|
|
if not sys_user or not sys_user.tenant_no:
|
|
raise PermissionError("用户租户信息缺失")
|
|
return sys_user.tenant_no
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pydantic Schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class UpdateStatusBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
schedule_id: str = Field(..., validation_alias=AliasChoices("scheduleId", "schedule_id"))
|
|
infusion_status: str = Field(..., validation_alias=AliasChoices("infusionStatus", "infusion_status"))
|
|
cancel_reason: Optional[str] = Field(
|
|
None, validation_alias=AliasChoices("cancelReason", "cancel_reason")
|
|
)
|
|
|
|
|
|
class UpdateQcStatusBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
schedule_id: str = Field(..., validation_alias=AliasChoices("scheduleId", "schedule_id"))
|
|
qc_status: str = Field(..., validation_alias=AliasChoices("qcStatus", "qc_status"))
|
|
|
|
|
|
class ProductItemBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
price_id: Optional[str] = Field(None, validation_alias=AliasChoices("priceId", "price_id"))
|
|
project_code: Optional[str] = Field(None, validation_alias=AliasChoices("projectCode", "project_code"))
|
|
project_name: Optional[str] = Field(None, validation_alias=AliasChoices("projectName", "project_name"))
|
|
cell_type_name: Optional[str] = Field(
|
|
None, validation_alias=AliasChoices("cellTypeName", "cell_type_name")
|
|
)
|
|
spec_text: Optional[str] = Field(None, validation_alias=AliasChoices("specText", "spec_text"))
|
|
agreement_price: Optional[float] = Field(
|
|
None, validation_alias=AliasChoices("agreementPrice", "agreement_price")
|
|
)
|
|
actual_price: Optional[float] = Field(
|
|
None, validation_alias=AliasChoices("actualPrice", "actual_price")
|
|
)
|
|
quantity: Optional[int] = Field(None, validation_alias=AliasChoices("quantity", "quantity"))
|
|
remark: Optional[str] = Field(None, validation_alias=AliasChoices("remark", "remark"))
|
|
|
|
|
|
class UpdateScheduleBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
id: str = Field(...)
|
|
cell_type: Optional[str] = Field(None, validation_alias=AliasChoices("cellType", "cell_type"))
|
|
spec_text: Optional[str] = Field(None, validation_alias=AliasChoices("specText", "spec_text"))
|
|
infusion_time: Optional[str] = Field(
|
|
None, validation_alias=AliasChoices("infusionTime", "infusion_time")
|
|
)
|
|
infusion_location: Optional[str] = Field(
|
|
None, validation_alias=AliasChoices("infusionLocation", "infusion_location")
|
|
)
|
|
infusion_method: Optional[str] = Field(
|
|
None, validation_alias=AliasChoices("infusionMethod", "infusion_method")
|
|
)
|
|
products: Optional[List[ProductItemBody]] = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/pilot-infusion-schedule-mgmt/list")
|
|
async def api_list(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
keyword: Optional[str] = Query(None, alias="keyword"),
|
|
infusionStatus: Optional[str] = Query(None, alias="infusionStatus"),
|
|
qcStatus: Optional[str] = Query(None, alias="qcStatus"),
|
|
infusionTimeStart: Optional[str] = Query(None, alias="infusionTimeStart"),
|
|
infusionTimeEnd: Optional[str] = Query(None, alias="infusionTimeEnd"),
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
result = crud.list_infusion_schedules_mgmt(
|
|
db,
|
|
tenant_no=tenant_no,
|
|
page=page,
|
|
page_size=pageSize,
|
|
keyword=keyword,
|
|
infusion_status=infusionStatus,
|
|
qc_status=qcStatus,
|
|
infusion_time_start=infusionTimeStart,
|
|
infusion_time_end=infusionTimeEnd,
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@router.get("/pilot-infusion-schedule-mgmt/detail")
|
|
async def api_detail(
|
|
scheduleId: str = Query(..., alias="scheduleId"),
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
result = crud.get_infusion_schedule_detail(
|
|
db, schedule_id=scheduleId, tenant_no=tenant_no
|
|
)
|
|
if result is None:
|
|
return success_response(None)
|
|
return success_response(result)
|
|
|
|
|
|
@router.put("/pilot-infusion-schedule-mgmt/update-status")
|
|
async def api_update_status(
|
|
body: UpdateStatusBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
result = crud.update_infusion_status(
|
|
db,
|
|
schedule_id=body.schedule_id,
|
|
infusion_status=body.infusion_status,
|
|
cancel_reason=body.cancel_reason,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(result, message="状态更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/pilot-infusion-schedule-mgmt/update-qc-status")
|
|
async def api_update_qc_status(
|
|
body: UpdateQcStatusBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
result = crud.update_qc_status(
|
|
db,
|
|
schedule_id=body.schedule_id,
|
|
qc_status=body.qc_status,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(result, message="质检状态更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/pilot-infusion-schedule-mgmt/update")
|
|
async def api_update(
|
|
body: UpdateScheduleBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
data = {
|
|
"cellType": body.cell_type,
|
|
"specText": body.spec_text,
|
|
"infusionTime": body.infusion_time,
|
|
"infusionLocation": body.infusion_location,
|
|
"infusionMethod": body.infusion_method,
|
|
}
|
|
if body.products is not None:
|
|
data["products"] = [p.model_dump(by_alias=False) for p in body.products]
|
|
|
|
result = crud.update_infusion_schedule_mgmt(
|
|
db,
|
|
schedule_id=body.id,
|
|
data=data,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(result, message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|