2058 lines
74 KiB
Python
2058 lines
74 KiB
Python
"""生产管理 API 路由（7 个子模块：耗材、试剂、SOP文档、工艺工序分类、产出物类型、耗材包、工艺库）"""
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from sqlalchemy.orm import Session
|
|
from typing import Any, Optional, List
|
|
|
|
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 success_response, error_response
|
|
from app.crud import consumable as consumable_svc
|
|
from app.crud import reagent as reagent_svc
|
|
from app.crud import consumable_stock_log as consumable_stock_log_svc
|
|
from app.crud import reagent_stock_log as reagent_stock_log_svc
|
|
from app.crud import sop_document as sop_document_svc
|
|
from app.crud import process_category as process_category_svc
|
|
from app.crud import output_type as output_type_svc
|
|
from app.crud import consumable_pack as consumable_pack_svc
|
|
from app.crud import process as process_svc
|
|
from app.crud import process_route as route_svc
|
|
from app.schemas.process_route import (
|
|
ProcessRouteCreateBody,
|
|
ProcessRouteUpdateBody,
|
|
SetCurrentBody,
|
|
StepAddBody,
|
|
StepCopyBody,
|
|
StepUpdateBody,
|
|
StepReorderBody,
|
|
StepReagentUpdateBody,
|
|
StepConsumableUpdateBody,
|
|
StepOutputUpdateBody,
|
|
StepDataCollectionUpdateBody,
|
|
)
|
|
from app.models.models import SysUser
|
|
from app.crud import production_work_order as production_work_order_svc
|
|
from app.crud import production_task as production_task_svc
|
|
from app.crud import cell_product as cell_product_svc
|
|
from app.schemas.production_work_order import ProductionWorkOrderListQuery, ProductionWorkOrderComplete
|
|
from app.schemas.production_task import (
|
|
ProductionTaskCreate,
|
|
ProductionTaskUpdate,
|
|
ProductionTaskStatusUpdate,
|
|
TaskStepAdd,
|
|
TaskStepUpdate,
|
|
TaskStepMaterialItem,
|
|
TaskStepOutputItem,
|
|
TaskStepDataCollectionItem,
|
|
TaskInputItem,
|
|
TaskOutputItem,
|
|
TaskStepReorderBody,
|
|
)
|
|
from app.schemas.cell_product import CellProductCreate, CellProductUpdate
|
|
|
|
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:
|
|
return "zkxw"
|
|
sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
|
|
return (sys_user.tenant_no if sys_user else None) or "zkxw"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. 耗材
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ConsumableCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
consumable_code: str = Field(..., alias="consumableCode")
|
|
consumable_name: str = Field(..., alias="consumableName")
|
|
spec_model: Optional[str] = Field(None, alias="specModel")
|
|
standard_usage: Optional[str] = Field(None, alias="standardUsage")
|
|
unit: Optional[str] = None
|
|
description: Optional[str] = None
|
|
warning_qty: Optional[float] = Field(0, alias="warningQty")
|
|
status: Optional[str] = Field("1", alias="status")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ConsumableUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
consumable_code: Optional[str] = Field(None, alias="consumableCode")
|
|
consumable_name: Optional[str] = Field(None, alias="consumableName")
|
|
spec_model: Optional[str] = Field(None, alias="specModel")
|
|
standard_usage: Optional[str] = Field(None, alias="standardUsage")
|
|
unit: Optional[str] = None
|
|
description: Optional[str] = None
|
|
warning_qty: Optional[float] = Field(None, alias="warningQty")
|
|
status: Optional[str] = Field(None, alias="status")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.get("/consumables")
|
|
async def list_consumables(
|
|
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("production:consumable:list")),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = consumable_svc.list_consumables(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
|
tp = consumable_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/consumables/all")
|
|
async def list_all_consumables(
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, _ = consumable_svc.list_consumables(db, page=1, page_size=9999, tenant_no=tenant_no, user_info=user)
|
|
return success_response(items)
|
|
|
|
|
|
@router.post("/consumables")
|
|
async def create_consumable(
|
|
body: ConsumableCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = consumable_svc.create_consumable(
|
|
db,
|
|
consumable_code=body.consumable_code,
|
|
consumable_name=body.consumable_name,
|
|
spec_model=body.spec_model,
|
|
standard_usage=body.standard_usage,
|
|
unit=body.unit,
|
|
description=body.description,
|
|
tenant_no=tenant_no,
|
|
warning_qty=body.warning_qty,
|
|
status=body.status,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(consumable_svc._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/consumables/{consumable_id}")
|
|
async def update_consumable(
|
|
consumable_id: str,
|
|
body: ConsumableUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = consumable_svc.update_consumable(
|
|
db,
|
|
consumable_id,
|
|
consumable_code=payload.get("consumable_code"),
|
|
consumable_name=payload.get("consumable_name"),
|
|
spec_model=payload.get("spec_model"),
|
|
standard_usage=payload.get("standard_usage"),
|
|
unit=payload.get("unit"),
|
|
description=payload.get("description"),
|
|
warning_qty=payload.get("warning_qty"),
|
|
status=payload.get("status"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(consumable_svc._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/consumables/{consumable_id}")
|
|
async def delete_consumable(
|
|
consumable_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
consumable_svc.soft_delete_consumable(db, consumable_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
class StockInBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
qty: float = Field(..., gt=0)
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class StockOutBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
qty: float = Field(..., gt=0)
|
|
receiver: str = Field(..., min_length=1)
|
|
purpose: Optional[str] = None
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class BatchStockInItem(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
materialId: str
|
|
qty: float = Field(..., gt=0)
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class BatchStockInBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
items: List[BatchStockInItem]
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class BatchStockOutItem(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
materialId: str
|
|
qty: float = Field(..., gt=0)
|
|
|
|
|
|
class BatchStockOutBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
items: List[BatchStockOutItem]
|
|
receiver: str = Field(..., min_length=1)
|
|
purpose: Optional[str] = None
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.post("/consumables/{consumable_id}/stock-in")
|
|
async def consumable_stock_in(
|
|
consumable_id: str,
|
|
body: StockInBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
from decimal import Decimal
|
|
obj = consumable_svc.stock_in(
|
|
db, consumable_id,
|
|
qty=Decimal(str(body.qty)),
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
tenant_no=tenant_no,
|
|
)
|
|
return success_response(consumable_svc._row_to_dict(obj), message="入库成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/consumables/{consumable_id}/stock-out")
|
|
async def consumable_stock_out(
|
|
consumable_id: str,
|
|
body: StockOutBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
from decimal import Decimal
|
|
obj = consumable_svc.stock_out(
|
|
db, consumable_id,
|
|
qty=Decimal(str(body.qty)),
|
|
receiver=body.receiver,
|
|
purpose=body.purpose,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
tenant_no=tenant_no,
|
|
)
|
|
return success_response(consumable_svc._row_to_dict(obj), message="出库成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/consumables/batch-stock-in")
|
|
async def consumable_batch_stock_in(
|
|
body: BatchStockInBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump() for it in body.items]
|
|
count = consumable_svc.batch_stock_in(
|
|
db, items=items_data, remark=body.remark,
|
|
tenant_no=tenant_no, user_id=user.get("user_id"),
|
|
)
|
|
return success_response({"count": count}, message=f"成功入库 {count} 条")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/consumables/batch-stock-out")
|
|
async def consumable_batch_stock_out(
|
|
body: BatchStockOutBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump() for it in body.items]
|
|
count = consumable_svc.batch_stock_out(
|
|
db, items=items_data,
|
|
receiver=body.receiver, purpose=body.purpose, remark=body.remark,
|
|
tenant_no=tenant_no, user_id=user.get("user_id"),
|
|
)
|
|
return success_response({"count": count}, message=f"成功出库 {count} 条")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.get("/consumables/{consumable_id}/stock-logs")
|
|
async def list_consumable_stock_logs(
|
|
consumable_id: str,
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
changeType: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = consumable_stock_log_svc.list_stock_logs(
|
|
db, tenant_no, page=page, page_size=pageSize,
|
|
material_id=consumable_id, change_type=changeType,
|
|
user_info=user,
|
|
)
|
|
tp = (total + pageSize - 1) // pageSize if total > 0 else 0
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/consumables/stock-logs")
|
|
async def list_all_consumable_stock_logs(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
keyword: Optional[str] = None,
|
|
changeType: Optional[str] = None,
|
|
startTime: Optional[str] = None,
|
|
endTime: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = consumable_stock_log_svc.list_stock_logs(
|
|
db, tenant_no, page=page, page_size=pageSize,
|
|
keyword=keyword, change_type=changeType,
|
|
start_time=startTime, end_time=endTime,
|
|
user_info=user,
|
|
)
|
|
tp = (total + pageSize - 1) // pageSize if total > 0 else 0
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. 试剂
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ReagentCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
reagent_code: str = Field(..., alias="reagentCode")
|
|
reagent_name: str = Field(..., alias="reagentName")
|
|
spec: Optional[str] = None
|
|
standard_usage: Optional[str] = Field(None, alias="standardUsage")
|
|
unit: Optional[str] = None
|
|
description: Optional[str] = None
|
|
warning_qty: Optional[float] = Field(0, alias="warningQty")
|
|
status: Optional[str] = Field("1", alias="status")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ReagentUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
reagent_code: Optional[str] = Field(None, alias="reagentCode")
|
|
reagent_name: Optional[str] = Field(None, alias="reagentName")
|
|
spec: Optional[str] = None
|
|
standard_usage: Optional[str] = Field(None, alias="standardUsage")
|
|
unit: Optional[str] = None
|
|
description: Optional[str] = None
|
|
warning_qty: Optional[float] = Field(None, alias="warningQty")
|
|
status: Optional[str] = Field(None, alias="status")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.get("/reagents")
|
|
async def list_reagents(
|
|
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("production:reagent:list")),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = reagent_svc.list_reagents(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
|
tp = reagent_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/reagents/all")
|
|
async def list_all_reagents(
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, _ = reagent_svc.list_reagents(db, page=1, page_size=9999, tenant_no=tenant_no, user_info=user)
|
|
return success_response(items)
|
|
|
|
|
|
@router.post("/reagents")
|
|
async def create_reagent(
|
|
body: ReagentCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = reagent_svc.create_reagent(
|
|
db,
|
|
reagent_code=body.reagent_code,
|
|
reagent_name=body.reagent_name,
|
|
spec=body.spec,
|
|
standard_usage=body.standard_usage,
|
|
unit=body.unit,
|
|
description=body.description,
|
|
tenant_no=tenant_no,
|
|
warning_qty=body.warning_qty,
|
|
status=body.status,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(reagent_svc._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/reagents/{reagent_id}")
|
|
async def update_reagent(
|
|
reagent_id: str,
|
|
body: ReagentUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = reagent_svc.update_reagent(
|
|
db,
|
|
reagent_id,
|
|
reagent_code=payload.get("reagent_code"),
|
|
reagent_name=payload.get("reagent_name"),
|
|
spec=payload.get("spec"),
|
|
standard_usage=payload.get("standard_usage"),
|
|
unit=payload.get("unit"),
|
|
description=payload.get("description"),
|
|
warning_qty=payload.get("warning_qty"),
|
|
status=payload.get("status"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(reagent_svc._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/reagents/{reagent_id}")
|
|
async def delete_reagent(
|
|
reagent_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
reagent_svc.soft_delete_reagent(db, reagent_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/reagents/{reagent_id}/stock-in")
|
|
async def reagent_stock_in(
|
|
reagent_id: str,
|
|
body: StockInBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
from decimal import Decimal
|
|
obj = reagent_svc.stock_in(
|
|
db, reagent_id,
|
|
qty=Decimal(str(body.qty)),
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
tenant_no=tenant_no,
|
|
)
|
|
return success_response(reagent_svc._row_to_dict(obj), message="入库成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/reagents/{reagent_id}/stock-out")
|
|
async def reagent_stock_out(
|
|
reagent_id: str,
|
|
body: StockOutBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
from decimal import Decimal
|
|
obj = reagent_svc.stock_out(
|
|
db, reagent_id,
|
|
qty=Decimal(str(body.qty)),
|
|
receiver=body.receiver,
|
|
purpose=body.purpose,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
tenant_no=tenant_no,
|
|
)
|
|
return success_response(reagent_svc._row_to_dict(obj), message="出库成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/reagents/batch-stock-in")
|
|
async def reagent_batch_stock_in(
|
|
body: BatchStockInBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump() for it in body.items]
|
|
count = reagent_svc.batch_stock_in(
|
|
db, items=items_data, remark=body.remark,
|
|
tenant_no=tenant_no, user_id=user.get("user_id"),
|
|
)
|
|
return success_response({"count": count}, message=f"成功入库 {count} 条")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/reagents/batch-stock-out")
|
|
async def reagent_batch_stock_out(
|
|
body: BatchStockOutBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump() for it in body.items]
|
|
count = reagent_svc.batch_stock_out(
|
|
db, items=items_data,
|
|
receiver=body.receiver, purpose=body.purpose, remark=body.remark,
|
|
tenant_no=tenant_no, user_id=user.get("user_id"),
|
|
)
|
|
return success_response({"count": count}, message=f"成功出库 {count} 条")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.get("/reagents/{reagent_id}/stock-logs")
|
|
async def list_reagent_stock_logs(
|
|
reagent_id: str,
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
changeType: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = reagent_stock_log_svc.list_stock_logs(
|
|
db, tenant_no, page=page, page_size=pageSize,
|
|
material_id=reagent_id, change_type=changeType,
|
|
user_info=user,
|
|
)
|
|
tp = (total + pageSize - 1) // pageSize if total > 0 else 0
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/reagents/stock-logs")
|
|
async def list_all_reagent_stock_logs(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
keyword: Optional[str] = None,
|
|
changeType: Optional[str] = None,
|
|
startTime: Optional[str] = None,
|
|
endTime: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = reagent_stock_log_svc.list_stock_logs(
|
|
db, tenant_no, page=page, page_size=pageSize,
|
|
keyword=keyword, change_type=changeType,
|
|
start_time=startTime, end_time=endTime,
|
|
user_info=user,
|
|
)
|
|
tp = (total + pageSize - 1) // pageSize if total > 0 else 0
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. SOP文档
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class SopDocumentCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
sop_code: str = Field(..., alias="sopCode")
|
|
sop_name: str = Field(..., alias="sopName")
|
|
version_no: Optional[str] = Field(None, alias="versionNo")
|
|
sop_attachments: Optional[str] = Field(None, alias="sopAttachments")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class SopDocumentUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
sop_code: Optional[str] = Field(None, alias="sopCode")
|
|
sop_name: Optional[str] = Field(None, alias="sopName")
|
|
version_no: Optional[str] = Field(None, alias="versionNo")
|
|
sop_attachments: Optional[str] = Field(None, alias="sopAttachments")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.get("/sop-documents")
|
|
async def list_sop_documents(
|
|
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("production:sop-document:list")),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = sop_document_svc.list_sop_documents(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
|
tp = sop_document_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/sop-documents/all")
|
|
async def list_all_sop_documents(
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, _ = sop_document_svc.list_sop_documents(db, page=1, page_size=9999, tenant_no=tenant_no, user_info=user)
|
|
return success_response(items)
|
|
|
|
|
|
@router.post("/sop-documents")
|
|
async def create_sop_document(
|
|
body: SopDocumentCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = sop_document_svc.create_sop_document(
|
|
db,
|
|
sop_code=body.sop_code,
|
|
sop_name=body.sop_name,
|
|
version_no=body.version_no,
|
|
sop_attachments=body.sop_attachments,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(
|
|
sop_document_svc._row_to_dict(obj, db, tenant_no),
|
|
message="创建成功",
|
|
)
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/sop-documents/{sop_id}")
|
|
async def update_sop_document(
|
|
sop_id: str,
|
|
body: SopDocumentUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = sop_document_svc.update_sop_document(
|
|
db,
|
|
sop_id,
|
|
sop_code=payload.get("sop_code"),
|
|
sop_name=payload.get("sop_name"),
|
|
version_no=payload.get("version_no"),
|
|
sop_attachments=payload.get("sop_attachments"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
)
|
|
tenant_no = _get_tenant_no(db, user)
|
|
return success_response(
|
|
sop_document_svc._row_to_dict(obj, db, tenant_no),
|
|
message="更新成功",
|
|
)
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/sop-documents/{sop_id}")
|
|
async def delete_sop_document(
|
|
sop_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
sop_document_svc.soft_delete_sop_document(db, sop_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. 工艺工序分类
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ProcessCategoryCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
category_code: str = Field(..., alias="categoryCode")
|
|
category_name: str = Field(..., alias="categoryName")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ProcessCategoryUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
category_code: Optional[str] = Field(None, alias="categoryCode")
|
|
category_name: Optional[str] = Field(None, alias="categoryName")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.get("/process-categories")
|
|
async def list_process_categories(
|
|
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("production:process-category:list")),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = process_category_svc.list_process_categories(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
|
tp = process_category_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/process-categories/all")
|
|
async def list_all_process_categories(
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, _ = process_category_svc.list_process_categories(db, page=1, page_size=9999, tenant_no=tenant_no, user_info=user)
|
|
return success_response(items)
|
|
|
|
|
|
@router.post("/process-categories")
|
|
async def create_process_category(
|
|
body: ProcessCategoryCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = process_category_svc.create_process_category(
|
|
db,
|
|
category_code=body.category_code,
|
|
category_name=body.category_name,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(process_category_svc._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/process-categories/{category_id}")
|
|
async def update_process_category(
|
|
category_id: str,
|
|
body: ProcessCategoryUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = process_category_svc.update_process_category(
|
|
db,
|
|
category_id,
|
|
category_code=payload.get("category_code"),
|
|
category_name=payload.get("category_name"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(process_category_svc._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/process-categories/{category_id}")
|
|
async def delete_process_category(
|
|
category_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
process_category_svc.soft_delete_process_category(db, category_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. 工艺产出物类型
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class OutputTypeCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
type_code: str = Field(..., alias="typeCode")
|
|
type_name: str = Field(..., alias="typeName")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class OutputTypeUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
type_code: Optional[str] = Field(None, alias="typeCode")
|
|
type_name: Optional[str] = Field(None, alias="typeName")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.get("/output-types")
|
|
async def list_output_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("production:output-type:list")),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = output_type_svc.list_output_types(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
|
tp = output_type_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/output-types/all")
|
|
async def list_all_output_types(
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, _ = output_type_svc.list_output_types(db, page=1, page_size=9999, tenant_no=tenant_no, user_info=user)
|
|
return success_response(items)
|
|
|
|
|
|
@router.post("/output-types")
|
|
async def create_output_type(
|
|
body: OutputTypeCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = output_type_svc.create_output_type(
|
|
db,
|
|
type_code=body.type_code,
|
|
type_name=body.type_name,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(output_type_svc._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/output-types/{type_id}")
|
|
async def update_output_type(
|
|
type_id: str,
|
|
body: OutputTypeUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = output_type_svc.update_output_type(
|
|
db,
|
|
type_id,
|
|
type_code=payload.get("type_code"),
|
|
type_name=payload.get("type_name"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(output_type_svc._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/output-types/{type_id}")
|
|
async def delete_output_type(
|
|
type_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
output_type_svc.soft_delete_output_type(db, type_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. 耗材包（主 + 子项）
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class PackItemInput(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
consumable_code: Optional[str] = Field(None, alias="consumableCode")
|
|
consumable_name: Optional[str] = Field(None, alias="consumableName")
|
|
spec_model: Optional[str] = Field(None, alias="specModel")
|
|
standard_usage: Optional[str] = Field(None, alias="standardUsage")
|
|
unit: Optional[str] = None
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ConsumablePackCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
pack_code: str = Field(..., alias="packCode")
|
|
pack_name: str = Field(..., alias="packName")
|
|
usage_scene: Optional[str] = Field(None, alias="usageScene")
|
|
pack_type: Optional[str] = Field(None, alias="packType")
|
|
status: Optional[int] = 1
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
items: List[PackItemInput] = []
|
|
|
|
|
|
class ConsumablePackUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
pack_code: Optional[str] = Field(None, alias="packCode")
|
|
pack_name: Optional[str] = Field(None, alias="packName")
|
|
usage_scene: Optional[str] = Field(None, alias="usageScene")
|
|
pack_type: Optional[str] = Field(None, alias="packType")
|
|
status: Optional[int] = None
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
items: Optional[List[PackItemInput]] = None
|
|
|
|
|
|
@router.get("/consumable-packs")
|
|
async def list_consumable_packs(
|
|
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("production:consumable-pack:list")),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = consumable_pack_svc.list_consumable_packs(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
|
tp = consumable_pack_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/consumable-packs/all")
|
|
async def list_all_consumable_packs(
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items = consumable_pack_svc.list_all_packs(db, tenant_no, user_info=user)
|
|
return success_response(items)
|
|
|
|
|
|
@router.get("/consumable-packs/{pack_id}")
|
|
async def get_consumable_pack_detail(
|
|
pack_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
detail = consumable_pack_svc.get_pack_detail(db, pack_id)
|
|
if not detail:
|
|
return error_response(404, "耗材包不存在")
|
|
return success_response(detail)
|
|
|
|
|
|
@router.post("/consumable-packs")
|
|
async def create_consumable_pack(
|
|
body: ConsumablePackCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump(by_alias=False) for it in body.items] if body.items else None
|
|
obj = consumable_pack_svc.create_consumable_pack(
|
|
db,
|
|
pack_code=body.pack_code,
|
|
pack_name=body.pack_name,
|
|
usage_scene=body.usage_scene,
|
|
pack_type=body.pack_type,
|
|
status=body.status,
|
|
items=items_data,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(consumable_pack_svc._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/consumable-packs/{pack_id}")
|
|
async def update_consumable_pack(
|
|
pack_id: str,
|
|
body: ConsumablePackUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
items_data = [it.model_dump(by_alias=False) for it in body.items] if body.items is not None else None
|
|
obj = consumable_pack_svc.update_consumable_pack(
|
|
db,
|
|
pack_id,
|
|
pack_code=payload.get("pack_code"),
|
|
pack_name=payload.get("pack_name"),
|
|
usage_scene=payload.get("usage_scene"),
|
|
pack_type=payload.get("pack_type"),
|
|
status=payload.get("status"),
|
|
items=items_data,
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(consumable_pack_svc._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/consumable-packs/{pack_id}")
|
|
async def delete_consumable_pack(
|
|
pack_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
consumable_pack_svc.soft_delete_consumable_pack(db, pack_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 7. 工艺库（主 + 3 个子表：试剂、耗材、产出物）
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ProcessReagentInput(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
reagent_code: Optional[str] = Field(None, alias="reagentCode")
|
|
reagent_name: Optional[str] = Field(None, alias="reagentName")
|
|
spec: Optional[str] = None
|
|
standard_usage: Optional[str] = Field(None, alias="standardUsage")
|
|
unit: Optional[str] = None
|
|
is_key_material: Optional[str] = Field("0", alias="isKeyMaterial")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ProcessConsumableInput(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
consumable_code: Optional[str] = Field(None, alias="consumableCode")
|
|
consumable_name: Optional[str] = Field(None, alias="consumableName")
|
|
spec_model: Optional[str] = Field(None, alias="specModel")
|
|
standard_usage: Optional[str] = Field(None, alias="standardUsage")
|
|
unit: Optional[str] = None
|
|
is_required: Optional[str] = Field("1", alias="isRequired")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ProcessOutputInput(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
output_type_id: Optional[str] = Field(None, alias="outputTypeId")
|
|
output_name: Optional[str] = Field(None, alias="outputName")
|
|
output_barcode: Optional[str] = Field(None, alias="outputBarcode")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ProcessDataCollectionInput(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
collection_name: str = Field(..., alias="collectionName")
|
|
unit: Optional[str] = None
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ProcessCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
process_code: str = Field(..., alias="processCode")
|
|
category_id: Optional[str] = Field(None, alias="categoryId")
|
|
process_name: str = Field(..., alias="processName")
|
|
description: Optional[str] = None
|
|
equipment_name: Optional[str] = Field(None, alias="equipmentName")
|
|
estimated_duration: Optional[str] = Field(None, alias="estimatedDuration")
|
|
sop_ids: Optional[str] = Field(None, alias="sopIds")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
reagents: List[ProcessReagentInput] = []
|
|
consumables: List[ProcessConsumableInput] = []
|
|
outputs: List[ProcessOutputInput] = []
|
|
data_collections: List[ProcessDataCollectionInput] = Field([], alias="dataCollections")
|
|
|
|
|
|
class ProcessUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
process_code: Optional[str] = Field(None, alias="processCode")
|
|
category_id: Optional[str] = Field(None, alias="categoryId")
|
|
process_name: Optional[str] = Field(None, alias="processName")
|
|
description: Optional[str] = None
|
|
equipment_name: Optional[str] = Field(None, alias="equipmentName")
|
|
estimated_duration: Optional[str] = Field(None, alias="estimatedDuration")
|
|
sop_ids: Optional[str] = Field(None, alias="sopIds")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
reagents: Optional[List[ProcessReagentInput]] = None
|
|
consumables: Optional[List[ProcessConsumableInput]] = None
|
|
outputs: Optional[List[ProcessOutputInput]] = None
|
|
data_collections: Optional[List[ProcessDataCollectionInput]] = Field(None, alias="dataCollections")
|
|
|
|
|
|
@router.get("/processes")
|
|
async def list_processes(
|
|
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("production:process:list")),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = process_svc.list_processes(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
|
tp = process_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/processes/all")
|
|
async def list_all_processes(
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items = process_svc.list_all_processes(db, tenant_no, user_info=user)
|
|
return success_response(items)
|
|
|
|
|
|
@router.get("/processes/{process_id}")
|
|
async def get_process_detail(
|
|
process_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
detail = process_svc.get_process_detail(db, process_id)
|
|
if not detail:
|
|
return error_response(404, "工艺不存在")
|
|
return success_response(detail)
|
|
|
|
|
|
@router.post("/processes")
|
|
async def create_process(
|
|
body: ProcessCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
reagents_data = [r.model_dump(by_alias=False) for r in body.reagents] if body.reagents else None
|
|
consumables_data = [c.model_dump(by_alias=False) for c in body.consumables] if body.consumables else None
|
|
outputs_data = [o.model_dump(by_alias=False) for o in body.outputs] if body.outputs else None
|
|
data_collections_data = [d.model_dump(by_alias=False) for d in body.data_collections] if body.data_collections else None
|
|
# Convert string fields to appropriate types for sub-records
|
|
for rd in (reagents_data or []):
|
|
if "is_key_material" in rd and isinstance(rd["is_key_material"], str):
|
|
rd["is_key_material"] = int(rd["is_key_material"])
|
|
for cd in (consumables_data or []):
|
|
if "is_required" in cd and isinstance(cd["is_required"], str):
|
|
cd["is_required"] = int(cd["is_required"])
|
|
obj = process_svc.create_process(
|
|
db,
|
|
process_code=body.process_code,
|
|
category_id=body.category_id,
|
|
process_name=body.process_name,
|
|
description=body.description,
|
|
equipment_name=body.equipment_name,
|
|
estimated_duration=body.estimated_duration,
|
|
sop_ids=body.sop_ids,
|
|
reagents=reagents_data,
|
|
consumables=consumables_data,
|
|
outputs=outputs_data,
|
|
data_collections=data_collections_data,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(process_svc._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/processes/{process_id}")
|
|
async def update_process(
|
|
process_id: str,
|
|
body: ProcessUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
reagents_data = [r.model_dump(by_alias=False) for r in body.reagents] if body.reagents is not None else None
|
|
consumables_data = [c.model_dump(by_alias=False) for c in body.consumables] if body.consumables is not None else None
|
|
outputs_data = [o.model_dump(by_alias=False) for o in body.outputs] if body.outputs is not None else None
|
|
data_collections_data = [d.model_dump(by_alias=False) for d in body.data_collections] if body.data_collections is not None else None
|
|
# Convert string fields to appropriate types for sub-records
|
|
for rd in (reagents_data or []):
|
|
if "is_key_material" in rd and isinstance(rd["is_key_material"], str):
|
|
rd["is_key_material"] = int(rd["is_key_material"])
|
|
for cd in (consumables_data or []):
|
|
if "is_required" in cd and isinstance(cd["is_required"], str):
|
|
cd["is_required"] = int(cd["is_required"])
|
|
obj = process_svc.update_process(
|
|
db,
|
|
process_id,
|
|
process_code=payload.get("process_code"),
|
|
category_id=payload.get("category_id"),
|
|
process_name=payload.get("process_name"),
|
|
description=payload.get("description"),
|
|
equipment_name=payload.get("equipment_name"),
|
|
estimated_duration=payload.get("estimated_duration"),
|
|
sop_ids=payload.get("sop_ids"),
|
|
reagents=reagents_data,
|
|
consumables=consumables_data,
|
|
outputs=outputs_data,
|
|
data_collections=data_collections_data,
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(process_svc._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/processes/{process_id}")
|
|
async def delete_process(
|
|
process_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
process_svc.soft_delete_process(db, process_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 8. 工艺路线管理
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/process-routes")
|
|
async def list_process_routes(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
routeName: Optional[str] = None,
|
|
routeCode: Optional[str] = None,
|
|
productTypeId: Optional[str] = None,
|
|
user: dict = Depends(require_menu_code("production:process-route:list")),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = route_svc.list_routes(
|
|
db, page=page, page_size=pageSize,
|
|
route_name=routeName, route_code=routeCode,
|
|
product_type_id=productTypeId, tenant_no=tenant_no,
|
|
user_info=user,
|
|
)
|
|
tp = route_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/process-routes/{route_id}")
|
|
async def get_process_route_detail(
|
|
route_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
detail = route_svc.get_route_detail(db, route_id)
|
|
if not detail:
|
|
return error_response(404, "路线不存在")
|
|
return success_response(detail)
|
|
|
|
|
|
@router.post("/process-routes")
|
|
async def create_process_route(
|
|
body: ProcessRouteCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = route_svc.create_route(
|
|
db,
|
|
route_code=body.route_code,
|
|
route_name=body.route_name,
|
|
product_type_id=body.product_type_id,
|
|
version=body.version,
|
|
remark=body.remark,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(route_svc._route_row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/process-routes/{route_id}")
|
|
async def update_process_route(
|
|
route_id: str,
|
|
body: ProcessRouteUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = route_svc.update_route(
|
|
db, route_id,
|
|
user_id=user.get("user_id"),
|
|
route_code=payload.get("route_code"),
|
|
route_name=payload.get("route_name"),
|
|
product_type_id=payload.get("product_type_id"),
|
|
version=payload.get("version"),
|
|
remark=payload.get("remark"),
|
|
)
|
|
return success_response(route_svc._route_row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/process-routes/set-current")
|
|
async def set_current_process_route(
|
|
body: SetCurrentBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = route_svc.set_current_route(db, body.route_id, tenant_no)
|
|
return success_response(route_svc._route_row_to_dict(obj), message="设置成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/process-routes/{route_id}")
|
|
async def delete_process_route(
|
|
route_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
route_svc.soft_delete_route(db, route_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ----- 步骤操作 -----
|
|
|
|
@router.post("/process-routes/step/add")
|
|
async def add_route_step(
|
|
body: StepAddBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
step_data = route_svc.add_step(
|
|
db,
|
|
route_id=body.route_id,
|
|
process_id=body.process_id,
|
|
step_order=body.step_order,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(step_data, message="添加成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/process-routes/step/copy")
|
|
async def copy_route_step(
|
|
body: StepCopyBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
step_data = route_svc.copy_step(
|
|
db,
|
|
step_id=body.step_id,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(step_data, message="复制成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/process-routes/step/update")
|
|
async def update_route_step(
|
|
body: StepUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload = body.model_dump(exclude_unset=True, by_alias=False)
|
|
payload.pop("step_id", None)
|
|
obj = route_svc.update_step(db, body.step_id, user_id=user.get("user_id"), **payload)
|
|
return success_response(route_svc._step_row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/process-routes/step/reorder")
|
|
async def reorder_route_steps(
|
|
body: StepReorderBody,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
route_svc.reorder_steps(db, body.route_id, body.step_ids)
|
|
return success_response(None, message="排序成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/process-routes/step/{step_id}")
|
|
async def delete_route_step(
|
|
step_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
route_svc.soft_delete_step(db, step_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ----- 步骤子表 -----
|
|
|
|
@router.put("/process-routes/step/reagent/update")
|
|
async def update_step_reagents(
|
|
body: StepReagentUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump(by_alias=False) for it in body.items]
|
|
route_svc.update_step_reagents(db, body.step_id, items_data, tenant_no, user.get("user_id"))
|
|
return success_response(None, message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/process-routes/step/consumable/update")
|
|
async def update_step_consumables(
|
|
body: StepConsumableUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump(by_alias=False) for it in body.items]
|
|
route_svc.update_step_consumables(db, body.step_id, items_data, tenant_no, user.get("user_id"))
|
|
return success_response(None, message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/process-routes/step/output/update")
|
|
async def update_step_outputs(
|
|
body: StepOutputUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump(by_alias=False) for it in body.items]
|
|
route_svc.update_step_outputs(db, body.step_id, items_data, tenant_no, user.get("user_id"))
|
|
return success_response(None, message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/process-routes/step/data-collection/update")
|
|
async def update_step_data_collections(
|
|
body: StepDataCollectionUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump(by_alias=False) for it in body.items]
|
|
route_svc.update_step_data_collections(db, body.step_id, items_data, tenant_no, user.get("user_id"))
|
|
return success_response(None, message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 9. 入库生产工单
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/production-work-orders")
|
|
async def list_production_work_orders(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
workOrderNo: Optional[str] = None,
|
|
workOrderStatus: Optional[str] = None,
|
|
protocolNo: Optional[str] = None,
|
|
sampleCode: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
query = ProductionWorkOrderListQuery(
|
|
page=page, page_size=pageSize,
|
|
work_order_no=workOrderNo, work_order_status=workOrderStatus,
|
|
protocol_no=protocolNo, sample_code=sampleCode,
|
|
)
|
|
result = production_work_order_svc.list_production_work_orders(
|
|
db, tenant_no=tenant_no, query=query, page=page, page_size=pageSize,
|
|
user_info=user,
|
|
)
|
|
items = result.get("items", [])
|
|
total = result.get("total", 0)
|
|
tp = production_task_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/production-work-orders/{wo_id}")
|
|
async def get_production_work_order(
|
|
wo_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
wo = production_work_order_svc.get_production_work_order(db, wo_id, tenant_no)
|
|
if not wo:
|
|
return error_response(404, "工单不存在")
|
|
return success_response(production_work_order_svc._wo_to_response_dict(db, wo))
|
|
|
|
|
|
@router.put("/production-work-orders/{wo_id}/complete")
|
|
async def complete_production_work_order(
|
|
wo_id: str,
|
|
body: ProductionWorkOrderComplete,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
wo = production_work_order_svc.complete_production_work_order(
|
|
db, wo_id, tenant_no, user.get("user_id"), body,
|
|
)
|
|
return success_response(production_work_order_svc._wo_to_response_dict(db, wo), message="操作成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 10. 生产任务
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/production-tasks")
|
|
async def create_production_task(
|
|
body: ProductionTaskCreate,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
data = body.model_dump(by_alias=False)
|
|
task = production_task_svc.create_production_task(
|
|
db, tenant_no=tenant_no, user_id=user.get("user_id"),
|
|
create_dept_id="", data=data,
|
|
)
|
|
db.commit()
|
|
return success_response(production_task_svc._task_to_dict(task), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.get("/production-tasks")
|
|
async def list_production_tasks(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
workOrderId: Optional[str] = None,
|
|
taskNo: Optional[str] = None,
|
|
taskStatus: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = production_task_svc.list_production_tasks(
|
|
db, tenant_no=tenant_no, page=page, page_size=pageSize,
|
|
work_order_id=workOrderId, task_no=taskNo, task_status=taskStatus,
|
|
user_info=user,
|
|
)
|
|
tp = production_task_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/production-tasks/{task_id}")
|
|
async def get_production_task_detail(
|
|
task_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
detail = production_task_svc.get_production_task_detail(db, task_id, tenant_no)
|
|
if not detail:
|
|
return error_response(404, "任务不存在")
|
|
return success_response(detail)
|
|
|
|
|
|
@router.put("/production-tasks/{task_id}")
|
|
async def update_production_task(
|
|
task_id: str,
|
|
body: ProductionTaskUpdate,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
data = body.model_dump(exclude_unset=True, by_alias=False)
|
|
task = production_task_svc.update_production_task(
|
|
db, task_id, tenant_no, user.get("user_id"), data,
|
|
)
|
|
return success_response(production_task_svc._task_to_dict(task), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/production-tasks/{task_id}")
|
|
async def delete_production_task(
|
|
task_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
production_task_svc.delete_production_task(db, task_id, tenant_no, user.get("user_id"))
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/production-tasks/{task_id}/status")
|
|
async def update_production_task_status(
|
|
task_id: str,
|
|
body: ProductionTaskStatusUpdate,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
task = production_task_svc.update_task_status(
|
|
db, task_id, tenant_no, user.get("user_id"), body.task_status,
|
|
)
|
|
return success_response(production_task_svc._task_to_dict(task), message="操作成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/production-tasks/{task_id}/steps")
|
|
async def add_production_task_step(
|
|
task_id: str,
|
|
body: TaskStepAdd,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
data = body.model_dump(by_alias=False)
|
|
step_data = production_task_svc.add_task_step(
|
|
db, task_id, tenant_no, user.get("user_id"), data,
|
|
)
|
|
return success_response(step_data, message="添加成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/production-tasks/{task_id}/steps/{step_id}")
|
|
async def update_production_task_step(
|
|
task_id: str,
|
|
step_id: str,
|
|
body: TaskStepUpdate,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
data = body.model_dump(exclude_unset=True, by_alias=False)
|
|
step = production_task_svc.update_task_step(
|
|
db, step_id, tenant_no, user.get("user_id"), data,
|
|
)
|
|
return success_response(production_task_svc._step_to_dict(step), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/production-tasks/{task_id}/steps/{step_id}")
|
|
async def delete_production_task_step(
|
|
task_id: str,
|
|
step_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
production_task_svc.delete_task_step(db, step_id, tenant_no, user.get("user_id"))
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/production-tasks/step/reorder")
|
|
async def reorder_production_task_steps(
|
|
body: TaskStepReorderBody,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
production_task_svc.reorder_task_steps(db, body.task_id, body.step_ids)
|
|
return success_response(None, message="排序成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/production-tasks/{task_id}/steps/{step_id}/copy")
|
|
async def copy_production_task_step(
|
|
task_id: str,
|
|
step_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
step_data = production_task_svc.copy_task_step(
|
|
db,
|
|
step_id=step_id,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(step_data, message="复制成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/production-tasks/{task_id}/steps/{step_id}/reagents")
|
|
async def replace_task_step_reagents(
|
|
task_id: str,
|
|
step_id: str,
|
|
body: List[TaskStepMaterialItem],
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = []
|
|
for it in body:
|
|
d = it.model_dump(by_alias=False)
|
|
items_data.append({
|
|
"reagent_code": d.get("material_code"),
|
|
"reagent_name": d.get("material_name"),
|
|
"spec": d.get("spec"),
|
|
"standard_usage": d.get("standard_usage"),
|
|
"actual_usage": d.get("actual_usage"),
|
|
"unit": d.get("unit"),
|
|
"is_key_material": d.get("is_key_material", "0"),
|
|
"remark": d.get("remark"),
|
|
})
|
|
production_task_svc.replace_step_reagents(db, step_id, tenant_no, user.get("user_id"), items_data)
|
|
return success_response(None, message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/production-tasks/{task_id}/steps/{step_id}/consumables")
|
|
async def replace_task_step_consumables(
|
|
task_id: str,
|
|
step_id: str,
|
|
body: List[TaskStepMaterialItem],
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = []
|
|
for it in body:
|
|
d = it.model_dump(by_alias=False)
|
|
items_data.append({
|
|
"consumable_code": d.get("material_code"),
|
|
"consumable_name": d.get("material_name"),
|
|
"spec_model": d.get("spec_model"),
|
|
"standard_usage": d.get("standard_usage"),
|
|
"actual_usage": d.get("actual_usage"),
|
|
"unit": d.get("unit"),
|
|
"is_required": d.get("is_required", "1"),
|
|
"remark": d.get("remark"),
|
|
})
|
|
production_task_svc.replace_step_consumables(db, step_id, tenant_no, user.get("user_id"), items_data)
|
|
return success_response(None, message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/production-tasks/{task_id}/steps/{step_id}/data-collection")
|
|
async def replace_task_step_data_collections(
|
|
task_id: str,
|
|
step_id: str,
|
|
body: List[TaskStepDataCollectionItem],
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump(by_alias=False) for it in body]
|
|
production_task_svc.replace_step_data_collections(db, step_id, tenant_no, user.get("user_id"), items_data)
|
|
return success_response(None, message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/production-tasks/{task_id}/steps/{step_id}/outputs")
|
|
async def replace_task_step_outputs(
|
|
task_id: str,
|
|
step_id: str,
|
|
body: List[TaskStepOutputItem],
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump(by_alias=False) for it in body]
|
|
production_task_svc.replace_step_outputs(db, step_id, tenant_no, user.get("user_id"), items_data)
|
|
return success_response(None, message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/production-tasks/{task_id}/inputs")
|
|
async def set_production_task_inputs(
|
|
task_id: str,
|
|
body: List[TaskInputItem],
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump(by_alias=False) for it in body]
|
|
production_task_svc.set_task_inputs(db, task_id, tenant_no, user.get("user_id"), items_data)
|
|
return success_response(None, message="设置成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/production-tasks/{task_id}/outputs")
|
|
async def set_production_task_outputs(
|
|
task_id: str,
|
|
body: List[TaskOutputItem],
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items_data = [it.model_dump(by_alias=False) for it in body]
|
|
production_task_svc.set_task_outputs(db, task_id, tenant_no, user.get("user_id"), items_data)
|
|
return success_response(None, message="设置成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 11. 细胞产品库
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/cell-products/next-code")
|
|
async def get_next_cell_product_code(
|
|
count: int = Query(1, ge=1, le=50),
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
if count <= 1:
|
|
code = cell_product_svc.get_next_code(db, tenant_no)
|
|
return success_response({"code": code})
|
|
codes = cell_product_svc.reserve_next_codes(db, tenant_no, count)
|
|
return success_response({"codes": codes, "code": codes[0] if codes else ""})
|
|
|
|
|
|
@router.get("/cell-products")
|
|
async def list_cell_products(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
keyword: Optional[str] = None,
|
|
productCode: Optional[str] = None,
|
|
productName: Optional[str] = None,
|
|
productTypeId: Optional[str] = None,
|
|
productStatus: Optional[str] = None,
|
|
passage: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
result = cell_product_svc.list_cell_products(
|
|
db, tenant_no=tenant_no, page=page, page_size=pageSize,
|
|
keyword=keyword,
|
|
product_code=productCode, product_name=productName,
|
|
product_type_id=productTypeId, product_status=productStatus,
|
|
passage=passage,
|
|
user_info=user,
|
|
)
|
|
items = result.get("items", [])
|
|
total = result.get("total", 0)
|
|
tp = cell_product_svc.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/cell-products/{product_id}")
|
|
async def get_cell_product_detail(
|
|
product_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = cell_product_svc.get_cell_product(db, product_id, tenant_no)
|
|
if not obj:
|
|
return error_response(404, "细胞产品不存在")
|
|
return success_response(cell_product_svc.product_to_response(db, tenant_no, obj))
|
|
|
|
|
|
@router.post("/cell-products")
|
|
async def create_cell_product(
|
|
body: CellProductCreate,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items = body.items or []
|
|
qty = int(body.quantity) if body.quantity is not None else 1
|
|
qty = max(1, min(qty, 50))
|
|
if items:
|
|
created = cell_product_svc.create_cell_products_batch(
|
|
db,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
create_dept_id="",
|
|
data=body,
|
|
items=items,
|
|
)
|
|
elif qty > 1:
|
|
created = cell_product_svc.create_cell_products_by_quantity(
|
|
db,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
create_dept_id="",
|
|
data=body,
|
|
count=qty,
|
|
)
|
|
else:
|
|
obj = cell_product_svc.create_cell_product(
|
|
db, tenant_no=tenant_no, user_id=user.get("user_id"),
|
|
create_dept_id="", data=body,
|
|
)
|
|
return success_response(
|
|
cell_product_svc.product_to_response(db, tenant_no, obj), message="创建成功"
|
|
)
|
|
resp_list = [
|
|
cell_product_svc.product_to_response(db, tenant_no, obj) for obj in created
|
|
]
|
|
n = len(created)
|
|
return success_response(
|
|
{"list": resp_list, "count": n},
|
|
message=f"成功创建 {n} 份细胞产品",
|
|
)
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/cell-products/{product_id}")
|
|
async def update_cell_product(
|
|
product_id: str,
|
|
body: CellProductUpdate,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = cell_product_svc.update_cell_product(
|
|
db, product_id, tenant_no, user.get("user_id"), data=body,
|
|
)
|
|
return success_response(
|
|
cell_product_svc.product_to_response(db, tenant_no, obj), message="更新成功"
|
|
)
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/cell-products/{product_id}")
|
|
async def delete_cell_product(
|
|
product_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
cell_product_svc.delete_cell_product(db, product_id, tenant_no, user.get("user_id"))
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|