""" 销售订单路由模块 提供销售订单的管理功能，包括： - 销售订单列表查询（支持分页、状态筛选） - 创建销售订单（自动生成订单号、计算总金额） - 销售订单明细管理 路由前缀: /api/sales-orders """ from fastapi import APIRouter, Depends, Query, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from typing import Optional, List from math import ceil from database.database import get_db_session from services.auth_service import get_current_active_user from models.database import ( User, Customer, Product, ProductMaterial, Warehouse, Inventory, StockMovement, SalesOrder, SalesOrderItem ) from .schemas import ( SalesOrderCreate, SalesOrderResponse, SalesOrderProductionPlanResponse, ProductionMaterialPlanItemResponse, SalesOrderIssueRequest, SalesOrderIssueResponse ) from .utils import generate_order_no router = APIRouter(prefix="/sales-orders", tags=["销售订单"]) def _build_sales_order_response(order: SalesOrder, customer_name: str) -> SalesOrderResponse: return SalesOrderResponse( id=order.id, order_no=order.order_no, customer_name=customer_name, order_date=order.order_date, delivery_date=order.delivery_date, status=order.status, production_status=order.production_status or "not_started", production_no=order.production_no, planned_material_cost=round(float(order.planned_material_cost or 0), 4), actual_material_cost=round(float(order.actual_material_cost or 0), 4), total_amount=order.total_amount, received_amount=order.received_amount, remark=order.remark, created_at=order.created_at ) async def _get_sales_order_with_customer( db_session: AsyncSession, order_id: int, ) -> tuple[SalesOrder, Customer]: result = await db_session.execute( select(SalesOrder, Customer) .join(Customer, SalesOrder.customer_id == Customer.id) .where(SalesOrder.id == order_id) ) row = result.first() if not row: raise HTTPException(status_code=404, detail="销售订单不存在") return row[0], row[1] async def _build_material_plan(db_session: AsyncSession, order: SalesOrder) -> tuple[List[ProductionMaterialPlanItemResponse], float]: item_result = await db_session.execute( select(SalesOrderItem).where(SalesOrderItem.order_id == order.id) ) order_items = item_result.scalars().all() if not order_items: return [], 0 required_qty_map = {} for order_item in order_items: bom_result = await db_session.execute( select(ProductMaterial, Product) .join(Product, ProductMaterial.material_product_id == Product.id) .where(ProductMaterial.finished_product_id == order_item.product_id) .where(Product.is_active == True) .where(Product.item_type == "material") ) bom_items = bom_result.all() if not bom_items: continue for bom, material in bom_items: qty = float(order_item.quantity) * float(bom.quantity or 0) * (1 + float(bom.loss_rate or 0)) entry = required_qty_map.setdefault( material.id, { "material": material, "required_qty": 0.0 } ) entry["required_qty"] += qty if not required_qty_map: return [], 0 material_ids = list(required_qty_map.keys()) stock_result = await db_session.execute( select(Inventory.product_id, func.coalesce(func.sum(Inventory.quantity), 0)) .where(Inventory.product_id.in_(material_ids)) .group_by(Inventory.product_id) ) stock_map = {row[0]: int(row[1] or 0) for row in stock_result.all()} plan_items = [] planned_material_cost = 0.0 for material_id, entry in required_qty_map.items(): material = entry["material"] required_qty = int(ceil(entry["required_qty"])) available_qty = stock_map.get(material_id, 0) shortage_qty = max(required_qty - available_qty, 0) unit_cost = float(material.cost_price or 0) required_cost = required_qty * unit_cost planned_material_cost += required_cost plan_items.append( ProductionMaterialPlanItemResponse( material_id=material.id, material_sku=material.sku, material_name=material.name, required_quantity=required_qty, available_quantity=available_qty, shortage_quantity=shortage_qty, unit_cost=round(unit_cost, 4), required_cost=round(required_cost, 4), ) ) plan_items = sorted(plan_items, key=lambda x: (x.shortage_quantity, x.required_cost), reverse=True) return plan_items, planned_material_cost @router.get("", response_model=List[SalesOrderResponse]) async def list_sales_orders( status: Optional[str] = None, skip: int = Query(0, ge=0), limit: int = Query(20, ge=1, le=100), db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): query = ( select(SalesOrder, Customer) .join(Customer, SalesOrder.customer_id == Customer.id) .order_by(SalesOrder.created_at.desc()) ) if status: query = query.where(SalesOrder.status == status) query = query.offset(skip).limit(limit) result = await db_session.execute(query) orders = [] for order, customer in result.all(): orders.append(_build_sales_order_response(order, customer.name)) return orders @router.post("", response_model=SalesOrderResponse, status_code=201) async def create_sales_order( order_data: SalesOrderCreate, db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): order = SalesOrder( order_no=generate_order_no("SO"), customer_id=order_data.customer_id, delivery_date=order_data.delivery_date, remark=order_data.remark, operator_id=current_user.id, status="draft" ) db_session.add(order) await db_session.flush() total_amount = 0 for item_data in order_data.items: product_result = await db_session.execute( select(Product).where(Product.id == item_data.product_id, Product.is_active == True) ) product = product_result.scalar_one_or_none() if not product: raise HTTPException(status_code=400, detail=f"产品不存在: {item_data.product_id}") if product.item_type != "finished": raise HTTPException(status_code=400, detail=f"销售单仅允许成品: {product.name}") item = SalesOrderItem( order_id=order.id, product_id=item_data.product_id, quantity=item_data.quantity, unit_price=item_data.unit_price, amount=item_data.quantity * item_data.unit_price, remark=item_data.remark ) db_session.add(item) total_amount += item.amount order.total_amount = total_amount await db_session.commit() await db_session.refresh(order) customer = await db_session.execute(select(Customer).where(Customer.id == order.customer_id)) customer = customer.scalar_one() return _build_sales_order_response(order, customer.name) @router.get("/{order_id}/production-plan", response_model=SalesOrderProductionPlanResponse) async def get_sales_order_production_plan( order_id: int, db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): order, customer = await _get_sales_order_with_customer(db_session, order_id) production_no = order.production_no or generate_order_no("WO") plan_items, planned_material_cost = await _build_material_plan(db_session, order) return SalesOrderProductionPlanResponse( sales_order_id=order.id, order_no=order.order_no, customer_name=customer.name, production_no=production_no, planned_material_cost=round(float(planned_material_cost), 4), items=plan_items, ) @router.post("/{order_id}/issue-materials", response_model=SalesOrderIssueResponse) async def issue_sales_order_materials( order_id: int, payload: SalesOrderIssueRequest, db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): order, _ = await _get_sales_order_with_customer(db_session, order_id) if order.production_status == "completed": raise HTTPException(status_code=400, detail="该销售单已完成生产") warehouse_result = await db_session.execute( select(Warehouse).where(Warehouse.id == payload.warehouse_id, Warehouse.is_active == True) ) warehouse = warehouse_result.scalar_one_or_none() if not warehouse: raise HTTPException(status_code=404, detail="仓库不存在") plan_items, planned_material_cost = await _build_material_plan(db_session, order) if not plan_items: raise HTTPException(status_code=400, detail="该销售单未配置BOM，无法领料") shortage_items = [item for item in plan_items if item.shortage_quantity > 0] if shortage_items: shortage_text = "，".join([f"{item.material_name} 缺 {item.shortage_quantity}" for item in shortage_items]) raise HTTPException(status_code=400, detail=f"物料库存不足：{shortage_text}") production_no = payload.production_no or order.production_no or generate_order_no("WO") actual_material_cost = 0.0 movement_count = 0 for item in plan_items: inv_result = await db_session.execute( select(Inventory) .where(Inventory.product_id == item.material_id) .where(Inventory.warehouse_id == warehouse.id) ) inventory = inv_result.scalar_one_or_none() if not inventory or inventory.quantity < item.required_quantity: raise HTTPException(status_code=400, detail=f"{item.material_name} 在所选仓库库存不足") before_qty = inventory.quantity inventory.quantity -= item.required_quantity after_qty = inventory.quantity total_amount = item.required_quantity * item.unit_cost actual_material_cost += total_amount movement = StockMovement( product_id=item.material_id, warehouse_id=warehouse.id, movement_type="issue_to_production", quantity=item.required_quantity, before_quantity=before_qty, after_quantity=after_qty, reference_type="sales_order", reference_id=order.id, reference_no=production_no, unit_price=item.unit_cost, total_amount=round(float(total_amount), 4), remark=payload.remark or f"销售单{order.order_no}按单生产领料", operator_id=current_user.id ) db_session.add(movement) movement_count += 1 order.production_no = production_no order.production_status = "material_issued" order.planned_material_cost = round(float(planned_material_cost), 4) order.actual_material_cost = round(float(actual_material_cost), 4) if order.status == "draft": order.status = "pending" await db_session.commit() cost_deviation = round(float(actual_material_cost - planned_material_cost), 4) cost_deviation_rate = round((cost_deviation / planned_material_cost), 6) if planned_material_cost > 1e-9 else 0.0 return SalesOrderIssueResponse( sales_order_id=order.id, order_no=order.order_no, production_no=production_no, movement_count=movement_count, planned_material_cost=round(float(planned_material_cost), 4), actual_material_cost=round(float(actual_material_cost), 4), cost_deviation=cost_deviation, cost_deviation_rate=cost_deviation_rate, production_status=order.production_status, )