geMoldInsight/src/inventory/services/product_service.py

"""产品管理业务服务层

将 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()