""" 销售订单路由模块 提供销售订单的管理功能，包括： - 销售订单列表查询（支持分页、状态筛选） - 创建销售订单（自动生成订单号、计算总金额） - 销售订单明细管理 路由前缀: /api/sales-orders """ from fastapi import APIRouter, Depends, Query, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, delete 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, SalesOrderDetailResponse, SalesOrderItemResponse, 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 _build_sales_order_detail_response( db_session: AsyncSession, order: SalesOrder, customer_name: str ) -> SalesOrderDetailResponse: items_result = await db_session.execute( select(SalesOrderItem).where(SalesOrderItem.order_id == order.id).order_by(SalesOrderItem.id.asc()) ) items = items_result.scalars().all() return SalesOrderDetailResponse( id=order.id, order_no=order.order_no, customer_id=order.customer_id, 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, items=[ SalesOrderItemResponse( id=item.id, product_id=item.product_id, quantity=item.quantity, delivered_quantity=item.delivered_quantity, unit_price=item.unit_price, amount=item.amount, remark=item.remark ) for item in items ] ) 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 async def _get_default_warehouse(db_session: AsyncSession) -> Warehouse: warehouse_result = await db_session.execute( select(Warehouse).where(Warehouse.is_active == True).order_by(Warehouse.is_default.desc(), Warehouse.id.asc()) ) warehouse = warehouse_result.scalars().first() if not warehouse: raise HTTPException(status_code=400, detail="未配置可用仓库，无法自动扣减物料") return warehouse async def _issue_materials_for_order_creation( db_session: AsyncSession, order: SalesOrder, current_user: User ) -> tuple[int, float, float]: warehouse = await _get_default_warehouse(db_session) 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 = 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=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) order.status = "pending" return movement_count, planned_material_cost, actual_material_cost async def _rollback_issued_materials( db_session: AsyncSession, order: SalesOrder, current_user: User ): movement_result = await db_session.execute( select(StockMovement) .where(StockMovement.reference_type == "sales_order") .where(StockMovement.reference_id == order.id) .where(StockMovement.movement_type == "issue_to_production") .order_by(StockMovement.id.asc()) ) movements = movement_result.scalars().all() if not movements: return for movement in movements: inv_result = await db_session.execute( select(Inventory) .where(Inventory.product_id == movement.product_id) .where(Inventory.warehouse_id == movement.warehouse_id) ) inventory = inv_result.scalar_one_or_none() if not inventory: inventory = Inventory( product_id=movement.product_id, warehouse_id=movement.warehouse_id, quantity=0, locked_quantity=0 ) db_session.add(inventory) await db_session.flush() before_qty = inventory.quantity inventory.quantity += movement.quantity after_qty = inventory.quantity revert_movement = StockMovement( product_id=movement.product_id, warehouse_id=movement.warehouse_id, movement_type="return_from_production", quantity=movement.quantity, before_quantity=before_qty, after_quantity=after_qty, reference_type="sales_order", reference_id=order.id, reference_no=order.production_no or order.order_no, unit_price=movement.unit_price, total_amount=movement.total_amount, remark=f"销售单{order.order_no}变更/删除，自动回补物料", operator_id=current_user.id ) db_session.add(revert_movement) async def _apply_order_items( db_session: AsyncSession, order: SalesOrder, order_data: SalesOrderCreate ) -> float: total_amount = 0.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 return total_amount @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() order.total_amount = await _apply_order_items(db_session, order, order_data) await _issue_materials_for_order_creation(db_session, order, current_user) 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}", response_model=SalesOrderDetailResponse) async def get_sales_order_detail( 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) return await _build_sales_order_detail_response(db_session, order, customer.name) @router.put("/{order_id}", response_model=SalesOrderResponse) async def update_sales_order( order_id: int, order_data: SalesOrderCreate, 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) await _rollback_issued_materials(db_session, order, current_user) await db_session.execute(delete(SalesOrderItem).where(SalesOrderItem.order_id == order.id)) order.customer_id = order_data.customer_id order.delivery_date = order_data.delivery_date order.remark = order_data.remark order.production_status = "not_started" order.production_no = None order.planned_material_cost = 0 order.actual_material_cost = 0 order.total_amount = await _apply_order_items(db_session, order, order_data) await _issue_materials_for_order_creation(db_session, order, current_user) await db_session.commit() await db_session.refresh(order) customer_result = await db_session.execute(select(Customer).where(Customer.id == order.customer_id)) updated_customer = customer_result.scalar_one_or_none() customer_name = updated_customer.name if updated_customer else "未知客户" return _build_sales_order_response(order, customer_name) @router.delete("/{order_id}") async def delete_sales_order( order_id: int, 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) await _rollback_issued_materials(db_session, order, current_user) await db_session.delete(order) await db_session.commit() return {"message": "销售订单已删除"} @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="该销售单已完成生产") if order.production_status == "material_issued": 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, )