"""产品管理业务服务层 将 product_routes 中不跨模块的产品 CRUD / BOM 编排下沉到此， 路由层只做参数校验与响应组装。 """ from decimal import Decimal from pathlib import Path from typing import Dict, List, Optional from fastapi import HTTPException from sqlalchemy import delete, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from shared.models.identity import User from moldinsight.models import ProcessingTask, STPFile from inventory.models import Product, ProductMaterial from ..schemas import ( ProductBOMResponse, ProductBOMUpdate, ProductCreate, ProductMaterialItemResponse, ProductResponse, ) async def _calculate_material_cost_map(db_session: AsyncSession, product_ids: List[int]) -> Dict[int, Decimal]: if not product_ids: return {} result = await db_session.execute( select( ProductMaterial.finished_product_id, func.coalesce( func.sum(Product.cost_price * ProductMaterial.quantity), 0, ), ) .join(Product, ProductMaterial.material_product_id == Product.id) .where(ProductMaterial.finished_product_id.in_(product_ids)) .group_by(ProductMaterial.finished_product_id) ) return {row[0]: Decimal(str(row[1] or 0)) for row in result.all()} def _build_product_response(product: Product, material_cost: Decimal = Decimal("0")) -> ProductResponse: return ProductResponse( id=product.id, sku=product.sku, name=product.name, description=product.description, category=product.category, unit=product.unit, item_type=product.item_type, cost_price=Decimal(str(product.cost_price or 0)), sale_price=Decimal(str(product.sale_price or 0)), min_stock=product.min_stock, max_stock=product.max_stock, material_cost=Decimal(str(material_cost)).quantize(Decimal("0.0001")), is_active=product.is_active, created_at=product.created_at, ) async def _get_product_or_404(db_session: AsyncSession, product_id: int, active_only: bool = False) -> Product: query = select(Product).where(Product.id == product_id) if active_only: query = query.where(Product.is_active == True) result = await db_session.execute(query) product = result.scalar_one_or_none() if not product: raise HTTPException(status_code=404, detail="产品不存在") return product def _validate_item_type(item_type: str) -> None: if item_type not in ["material", "finished"]: raise HTTPException(status_code=400, detail="item_type 必须为 material 或 finished") async def _build_bom_response(db_session: AsyncSession, product: Product) -> ProductBOMResponse: bom_result = await db_session.execute( select(ProductMaterial, Product) .join(Product, ProductMaterial.material_product_id == Product.id) .where(ProductMaterial.finished_product_id == product.id) .order_by(ProductMaterial.id.asc()) ) items: List[ProductMaterialItemResponse] = [] total_material_cost = Decimal("0") for bom, material in bom_result.all(): line_cost = Decimal(str(material.cost_price or 0)) * Decimal(str(bom.quantity)) total_material_cost += line_cost items.append( ProductMaterialItemResponse( material_id=material.id, material_sku=material.sku, material_name=material.name, quantity=Decimal(str(bom.quantity)), unit_cost=Decimal(str(material.cost_price or 0)), line_cost=line_cost, ) ) return ProductBOMResponse( product_id=product.id, product_name=product.name, total_material_cost=total_material_cost, items=items, ) class ProductService: """产品 CRUD 与 BOM 服务""" @staticmethod async def list_products( db_session: AsyncSession, skip: int, limit: int, search: Optional[str], category: Optional[str], item_type: Optional[str], ) -> List[ProductResponse]: query = select(Product).where(Product.is_active == True) if search: query = query.where(or_(Product.name.ilike(f"%{search}%"), Product.sku.ilike(f"%{search}%"))) if category: query = query.where(Product.category == category) if item_type: query = query.where(Product.item_type == item_type) query = query.offset(skip).limit(limit).order_by(Product.created_at.desc()) result = await db_session.execute(query) products = result.scalars().all() finished_product_ids = [p.id for p in products if p.item_type == "finished"] material_cost_map = await _calculate_material_cost_map(db_session, finished_product_ids) return [_build_product_response(p, material_cost_map.get(p.id, 0)) for p in products] @staticmethod async def create_product( db_session: AsyncSession, product_data: ProductCreate, current_user: User, ) -> ProductResponse: _validate_item_type(product_data.item_type) existing = await db_session.execute(select(Product).where(Product.sku == product_data.sku)) if existing.scalar_one_or_none(): raise HTTPException(status_code=400, detail="SKU已存在") product_dict = product_data.model_dump() if product_data.item_type == "finished": product_dict["min_stock"] = 0 product_dict["max_stock"] = 0 product = Product(**product_dict) db_session.add(product) await db_session.commit() await db_session.refresh(product) return _build_product_response(product, 0) @staticmethod async def create_product_from_task( db_session: AsyncSession, task_id: str, current_user: User, ) -> ProductResponse: task_result = await db_session.execute(select(ProcessingTask).where(ProcessingTask.task_id == task_id)) task = task_result.scalar_one_or_none() if not task: raise HTTPException(status_code=404, detail="分析任务不存在") stp_result = await db_session.execute(select(STPFile).where(STPFile.id == task.stp_file_id)) stp_file = stp_result.scalar_one_or_none() if not stp_file: raise HTTPException(status_code=404, detail="STP 分析记录不存在") if stp_file.product_id: existed = await db_session.execute(select(Product).where(Product.id == stp_file.product_id)) product = existed.scalar_one_or_none() if product: return _build_product_response(product, 0) base_sku = f"MI{stp_file.id}" sku = base_sku n = 1 while True: conflict = await db_session.execute(select(Product).where(Product.sku == sku)) if not conflict.scalar_one_or_none(): break n += 1 sku = f"{base_sku}-{n}" name = Path(stp_file.original_filename or f"mold_{stp_file.id}").stem or f"模具分析-{stp_file.id}" desc_parts = [] if stp_file.volume: desc_parts.append(f"体积 {stp_file.volume:.1f} mm³") if stp_file.product_weight: desc_parts.append(f"重量 {stp_file.product_weight:.2f} g") if stp_file.surface_area: desc_parts.append(f"表面积 {stp_file.surface_area:.1f} mm²") description = "由模具分析创建" + ("：" + "；".join(desc_parts) if desc_parts else "") product = Product( sku=sku, name=name, description=description, category="模具成品", unit="件", item_type="finished", cost_price=0, sale_price=0, min_stock=0, max_stock=0, ) db_session.add(product) await db_session.flush() stp_file.product_id = product.id await db_session.commit() await db_session.refresh(product) return _build_product_response(product, 0) @staticmethod async def update_product( db_session: AsyncSession, product_id: int, product_data: ProductCreate, current_user: User, ) -> ProductResponse: product = await _get_product_or_404(db_session, product_id) _validate_item_type(product_data.item_type) conflict = await db_session.execute( select(Product).where(Product.sku == product_data.sku, Product.id != product_id) ) if conflict.scalar_one_or_none(): raise HTTPException(status_code=400, detail="SKU已存在") product_dict = product_data.model_dump() if product_data.item_type == "finished": product_dict["min_stock"] = 0 product_dict["max_stock"] = 0 for key, value in product_dict.items(): setattr(product, key, value) await db_session.commit() await db_session.refresh(product) material_cost_map = await _calculate_material_cost_map(db_session, [product.id]) return _build_product_response(product, material_cost_map.get(product.id, 0)) @staticmethod async def delete_product( db_session: AsyncSession, product_id: int, current_user: User, ) -> dict: product = await _get_product_or_404(db_session, product_id) product.is_active = False await db_session.commit() return {"message": "产品已删除"} @staticmethod async def get_product_bom( db_session: AsyncSession, product_id: int, ) -> ProductBOMResponse: product = await _get_product_or_404(db_session, product_id, active_only=True) if product.item_type != "finished": raise HTTPException(status_code=400, detail="仅成品支持配置物料BOM") return await _build_bom_response(db_session, product) @staticmethod async def replace_product_bom( db_session: AsyncSession, product_id: int, payload: ProductBOMUpdate, current_user: User, ) -> ProductBOMResponse: product = await _get_product_or_404(db_session, product_id, active_only=True) if product.item_type != "finished": raise HTTPException(status_code=400, detail="仅成品支持配置物料BOM") material_ids = [item.material_id for item in payload.items] if len(material_ids) != len(set(material_ids)): raise HTTPException(status_code=400, detail="BOM 物料不允许重复") if material_ids: material_result = await db_session.execute( select(Product).where(Product.id.in_(material_ids), Product.is_active == True) ) materials = material_result.scalars().all() material_map = {m.id: m for m in materials} if len(material_map) != len(material_ids): raise HTTPException(status_code=400, detail="存在无效物料") invalid_materials = [m.name for m in materials if m.item_type != "material"] if invalid_materials: raise HTTPException(status_code=400, detail=f"以下条目不是物料：{', '.join(invalid_materials)}") for item in payload.items: if item.quantity <= 0: raise HTTPException(status_code=400, detail="物料数量必须大于 0") await db_session.execute(delete(ProductMaterial).where(ProductMaterial.finished_product_id == product_id)) for item in payload.items: db_session.add( ProductMaterial( finished_product_id=product_id, material_product_id=item.material_id, quantity=item.quantity, loss_rate=item.loss_rate, ) ) await db_session.commit() return await _build_bom_response(db_session, product) product_service = ProductService()