126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
"""齐套性推演接口路由"""
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.orm import Session
|
||
from decimal import Decimal
|
||
from collections import defaultdict
|
||
|
||
from app.database import get_db_inventory
|
||
from app.models import MaterialBase, BomTable, StockBuy
|
||
|
||
router = APIRouter(prefix="/api/pms", tags=["齐套性推演"])
|
||
|
||
|
||
def calculate_bom_requirements(
|
||
base_id: int,
|
||
quantity: int,
|
||
db: Session,
|
||
bom_no: str | None = None,
|
||
version: str | None = None,
|
||
visited: set | None = None,
|
||
) -> dict[int, Decimal]:
|
||
"""
|
||
递归计算BOM所需物料及用量
|
||
|
||
Args:
|
||
base_id: 目标成品物料ID
|
||
quantity: 目标数量
|
||
db: 数据库会话
|
||
bom_no: BOM编号（精确匹配，可为 None）
|
||
version: BOM版本（精确匹配，可为 None）
|
||
visited: 已访问物料ID集合（防循环）
|
||
"""
|
||
if visited is None:
|
||
visited = set()
|
||
|
||
if base_id in visited:
|
||
return {}
|
||
visited.add(base_id)
|
||
|
||
requirements = defaultdict(Decimal)
|
||
|
||
# 关键：按 bom_no / version 精确匹配 BOM 层级
|
||
bom_query = db.query(BomTable).filter(BomTable.parent_id == base_id)
|
||
if bom_no is not None:
|
||
bom_query = bom_query.filter(BomTable.bom_no == bom_no)
|
||
if version is not None:
|
||
bom_query = bom_query.filter(BomTable.version == version)
|
||
bom_items = bom_query.all()
|
||
|
||
if not bom_items:
|
||
visited.discard(base_id)
|
||
return {base_id: Decimal(quantity)}
|
||
|
||
for item in bom_items:
|
||
child_requirements = calculate_bom_requirements(
|
||
item.child_id,
|
||
int(quantity * item.dosage),
|
||
db,
|
||
bom_no=None,
|
||
version=None,
|
||
visited=visited,
|
||
)
|
||
for child_id, qty in child_requirements.items():
|
||
requirements[child_id] += qty
|
||
|
||
return dict(requirements)
|
||
|
||
|
||
@router.get("/deduce_bom")
|
||
async def deduce_bom(
|
||
target_base_id: int = Query(..., description="目标成品ID"),
|
||
target_quantity: int = Query(..., gt=0, description="计划生产数量"),
|
||
bom_no: str | None = Query(None, description="BOM编号（精确匹配，可为空）"),
|
||
version: str | None = Query(None, description="BOM版本（精确匹配，可为空）"),
|
||
db: Session = Depends(get_db_inventory),
|
||
):
|
||
"""
|
||
齐套性推演接口
|
||
|
||
- 若传 bom_no 和 version，则只查对应版本 BOM
|
||
- 若不传 bom_no/version，则查该成品所有可用 BOM（兼容旧行为）
|
||
- 返回 material_id / material_name / spec_model / unit / required_quantity / current_stock / shortage_quantity
|
||
"""
|
||
target_material = db.query(MaterialBase).filter(MaterialBase.id == target_base_id).first()
|
||
if not target_material:
|
||
raise HTTPException(status_code=404, detail=f"目标物料 ID={target_base_id} 不存在")
|
||
|
||
requirements = calculate_bom_requirements(
|
||
target_base_id, target_quantity, db,
|
||
bom_no=bom_no, version=version,
|
||
)
|
||
|
||
stock_records = {
|
||
r.base_id: r.stock_quantity
|
||
for r in db.query(StockBuy).filter(StockBuy.base_id.in_(requirements.keys())).all()
|
||
}
|
||
|
||
material_requirements = []
|
||
total_shortage = 0
|
||
|
||
for base_id, required_qty in sorted(requirements.items()):
|
||
material = db.query(MaterialBase).filter(MaterialBase.id == base_id).first()
|
||
current_stock = stock_records.get(base_id, Decimal("0"))
|
||
shortage = max(Decimal("0"), required_qty - current_stock)
|
||
if shortage > 0:
|
||
total_shortage += 1
|
||
|
||
material_requirements.append({
|
||
"material_id": base_id,
|
||
"material_name": material.name if material else f"未知物料({base_id})",
|
||
"spec_model": material.spec_model if material else None,
|
||
"unit": material.unit if material else None,
|
||
"required_quantity": required_qty,
|
||
"current_stock": current_stock,
|
||
"shortage_quantity": shortage,
|
||
"is_shortage": shortage > 0,
|
||
})
|
||
|
||
return {
|
||
"target_base_id": target_base_id,
|
||
"target_quantity": target_quantity,
|
||
"bom_no": bom_no,
|
||
"version": version,
|
||
"is_shortage": total_shortage > 0,
|
||
"total_shortage_count": total_shortage,
|
||
"material_requirements": material_requirements,
|
||
} |