"""物料管理业务服务层 将 material_routes 中的价格历史、价格趋势、物料供应商关联等业务编排下沉到此, 路由层只做参数校验与响应组装。 """ from typing import List from fastapi import HTTPException from sqlalchemy import desc, select, update from sqlalchemy.ext.asyncio import AsyncSession from shared.models.identity import User from inventory.models import MaterialPriceHistory, MaterialSupplier, Product, Supplier from ..schemas import ( MaterialPriceHistoryCreate, MaterialPriceHistoryResponse, MaterialPriceTrendResponse, MaterialSupplierCreate, MaterialSupplierResponse, PriceHistoryItem, ) async def _get_material_product(db_session: AsyncSession, product_id: int) -> Product: result = await db_session.execute( select(Product).where(Product.id == product_id, Product.is_active == True) ) product = result.scalar_one_or_none() if not product: raise HTTPException(status_code=404, detail="物料不存在") if product.item_type != "material": raise HTTPException(status_code=400, detail="仅物料类型支持价格历史管理") return product async def _get_active_supplier(db_session: AsyncSession, supplier_id: int, detail: str = "供应商不存在") -> Supplier: result = await db_session.execute( select(Supplier).where(Supplier.id == supplier_id, Supplier.is_active == True) ) supplier = result.scalar_one_or_none() if not supplier: raise HTTPException(status_code=400 if detail == "供应商不存在" else 404, detail=detail) return supplier def _build_price_history_response( product: Product, price_history: MaterialPriceHistory, supplier_name: str | None, ) -> MaterialPriceHistoryResponse: return MaterialPriceHistoryResponse( id=price_history.id, product_id=price_history.product_id, product_sku=product.sku, product_name=product.name, price=price_history.price, effective_date=price_history.effective_date, supplier_id=price_history.supplier_id, supplier_name=supplier_name, remark=price_history.remark, created_at=price_history.created_at, ) def _build_material_supplier_response( product: Product, material_supplier: MaterialSupplier, supplier_name: str | None, ) -> MaterialSupplierResponse: return MaterialSupplierResponse( id=material_supplier.id, product_id=material_supplier.product_id, product_sku=product.sku, product_name=product.name, supplier_id=material_supplier.supplier_id, supplier_name=supplier_name, is_primary=material_supplier.is_primary, contact_person=material_supplier.contact_person, contact_phone=material_supplier.contact_phone, lead_time=material_supplier.lead_time, min_order_quantity=material_supplier.min_order_quantity, created_at=material_supplier.created_at, updated_at=material_supplier.updated_at, ) class MaterialService: """物料价格与供应商关联服务""" @staticmethod async def add_price_history( db_session: AsyncSession, product_id: int, price_data: MaterialPriceHistoryCreate, current_user: User, ) -> MaterialPriceHistoryResponse: product = await _get_material_product(db_session, product_id) supplier_name = None if price_data.supplier_id: supplier = await _get_active_supplier(db_session, price_data.supplier_id) supplier_name = supplier.name price_history = MaterialPriceHistory( product_id=product_id, price=price_data.price, supplier_id=price_data.supplier_id, remark=price_data.remark, ) db_session.add(price_history) product.cost_price = price_data.price await db_session.commit() await db_session.refresh(price_history) return _build_price_history_response(product, price_history, supplier_name) @staticmethod async def get_price_history( db_session: AsyncSession, product_id: int, limit: int, ) -> List[MaterialPriceHistoryResponse]: product = await _get_material_product(db_session, product_id) result = await db_session.execute( select(MaterialPriceHistory, Supplier) .outerjoin(Supplier, MaterialPriceHistory.supplier_id == Supplier.id) .where(MaterialPriceHistory.product_id == product_id) .order_by(desc(MaterialPriceHistory.effective_date)) .limit(limit) ) return [ _build_price_history_response(product, ph, supplier.name if supplier else None) for ph, supplier in result.all() ] @staticmethod async def get_price_trend( db_session: AsyncSession, product_id: int, months: int, ) -> MaterialPriceTrendResponse: product = await _get_material_product(db_session, product_id) result = await db_session.execute( select(MaterialPriceHistory, Supplier) .outerjoin(Supplier, MaterialPriceHistory.supplier_id == Supplier.id) .where(MaterialPriceHistory.product_id == product_id) .order_by(desc(MaterialPriceHistory.effective_date)) .limit(months) ) rows = result.all() price_history_list = [ph for ph, _supplier in rows] if not price_history_list: raise HTTPException(status_code=404, detail="无价格历史记录") current_price = price_history_list[0].price first_price = price_history_list[-1].price price_change = current_price - first_price price_change_percent = (price_change / first_price * 100) if first_price > 0 else 0 return MaterialPriceTrendResponse( product_id=product_id, product_sku=product.sku, product_name=product.name, current_price=current_price, price_change=round(price_change, 2), price_change_percent=round(price_change_percent, 2), price_history=[ PriceHistoryItem( date=ph.effective_date, price=ph.price, supplier_name=supplier.name if supplier else None, ) for ph, supplier in rows ], ) @staticmethod async def add_material_supplier( db_session: AsyncSession, product_id: int, supplier_data: MaterialSupplierCreate, current_user: User, ) -> MaterialSupplierResponse: product = await _get_material_product(db_session, product_id) supplier = await _get_active_supplier(db_session, supplier_data.supplier_id) existing = await db_session.execute( select(MaterialSupplier).where( MaterialSupplier.product_id == product_id, MaterialSupplier.supplier_id == supplier_data.supplier_id, ) ) if existing.scalar_one_or_none(): raise HTTPException(status_code=400, detail="该供应商已关联到该物料") if supplier_data.is_primary: await db_session.execute( update(MaterialSupplier) .where(MaterialSupplier.product_id == product_id) .values(is_primary=False) ) material_supplier = MaterialSupplier( product_id=product_id, supplier_id=supplier_data.supplier_id, is_primary=supplier_data.is_primary, contact_person=supplier_data.contact_person, contact_phone=supplier_data.contact_phone, lead_time=supplier_data.lead_time, min_order_quantity=supplier_data.min_order_quantity, ) db_session.add(material_supplier) await db_session.commit() await db_session.refresh(material_supplier) return _build_material_supplier_response(product, material_supplier, supplier.name) @staticmethod async def get_material_suppliers( db_session: AsyncSession, product_id: int, ) -> List[MaterialSupplierResponse]: product = await _get_material_product(db_session, product_id) result = await db_session.execute( select(MaterialSupplier, Supplier) .outerjoin(Supplier, MaterialSupplier.supplier_id == Supplier.id) .where(MaterialSupplier.product_id == product_id) .order_by(MaterialSupplier.is_primary.desc(), MaterialSupplier.id.asc()) ) return [ _build_material_supplier_response(product, ms, supplier.name if supplier else None) for ms, supplier in result.all() ] @staticmethod async def remove_material_supplier( db_session: AsyncSession, supplier_id: int, current_user: User, ) -> dict: result = await db_session.execute( select(MaterialSupplier).where(MaterialSupplier.id == supplier_id) ) material_supplier = result.scalar_one_or_none() if not material_supplier: raise HTTPException(status_code=404, detail="物料供应商关联不存在") await db_session.delete(material_supplier) await db_session.commit() return {"message": "物料供应商关联已删除"} @staticmethod async def get_supplier_materials( db_session: AsyncSession, supplier_id: int, ) -> List[MaterialSupplierResponse]: supplier = await _get_active_supplier(db_session, supplier_id, detail="供应商不存在") result = await db_session.execute( select(MaterialSupplier, Product) .outerjoin(Product, MaterialSupplier.product_id == Product.id) .where(MaterialSupplier.supplier_id == supplier_id) .order_by(MaterialSupplier.is_primary.desc(), MaterialSupplier.id.asc()) ) return [ MaterialSupplierResponse( id=ms.id, product_id=ms.product_id, product_sku=product.sku if product else None, product_name=product.name if product else None, supplier_id=ms.supplier_id, supplier_name=supplier.name, is_primary=ms.is_primary, contact_person=ms.contact_person, contact_phone=ms.contact_phone, lead_time=ms.lead_time, min_order_quantity=ms.min_order_quantity, created_at=ms.created_at, updated_at=ms.updated_at, ) for ms, product in result.all() ] material_service = MaterialService()