915 lines
35 KiB
Python
915 lines
35 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 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.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:
|
|
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")
|
|
description: Optional[str] = None
|
|
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")
|
|
description: Optional[str] = None
|
|
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)
|
|
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)
|
|
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,
|
|
description=body.description,
|
|
tenant_no=tenant_no,
|
|
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"),
|
|
description=payload.get("description"),
|
|
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))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
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
|
|
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)
|
|
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)
|
|
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,
|
|
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"),
|
|
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))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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)
|
|
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)
|
|
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), 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"),
|
|
)
|
|
return success_response(sop_document_svc._row_to_dict(obj), 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)
|
|
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)
|
|
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)
|
|
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)
|
|
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")
|
|
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[str] = "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[str] = 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)
|
|
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)
|
|
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=int(body.status) if body.status is not None else None,
|
|
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=int(payload["status"]) if "status" in payload else None,
|
|
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")
|
|
|
|
|
|
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")
|
|
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")
|
|
|
|
|
|
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] = []
|
|
|
|
|
|
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
|
|
|
|
|
|
@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)
|
|
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)
|
|
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
|
|
# 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,
|
|
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
|
|
# 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,
|
|
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))
|