""" 采购订单路由模块 提供采购订单的管理功能，包括： - 采购订单列表查询（支持分页、状态筛选） - 创建采购订单（自动生成订单号、计算总金额） - 采购订单明细管理 路由前缀: /api/purchase-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 database.database import get_db_session from services.auth_service import get_current_active_user from models.database import ( User, Supplier, Product, Warehouse, Inventory, StockMovement, PurchaseOrder, PurchaseOrderItem ) from .schemas import ( PurchaseOrderCreate, PurchaseOrderResponse, PurchaseOrderDetailResponse, PurchaseOrderItemResponse, PurchaseOrderReceiveRequest, ) from .utils import generate_order_no router = APIRouter(prefix="/purchase-orders", tags=["采购订单"]) def _build_purchase_order_response(order: PurchaseOrder, supplier_name: str) -> PurchaseOrderResponse: return PurchaseOrderResponse( id=order.id, order_no=order.order_no, supplier_name=supplier_name, order_date=order.order_date, expected_date=order.expected_date, status=order.status, total_amount=order.total_amount, paid_amount=order.paid_amount, remark=order.remark, created_at=order.created_at, received_date=order.received_date, paid_date=order.paid_date ) async def _get_order_with_supplier( db_session: AsyncSession, order_id: int ) -> tuple[PurchaseOrder, Supplier]: result = await db_session.execute( select(PurchaseOrder, Supplier) .join(Supplier, PurchaseOrder.supplier_id == Supplier.id) .where(PurchaseOrder.id == order_id) ) row = result.first() if not row: raise HTTPException(status_code=404, detail="采购订单不存在") return row[0], row[1] async def _build_purchase_order_detail( db_session: AsyncSession, order: PurchaseOrder, supplier_name: str ) -> PurchaseOrderDetailResponse: item_result = await db_session.execute( select(PurchaseOrderItem, Product) .join(Product, PurchaseOrderItem.product_id == Product.id) .where(PurchaseOrderItem.order_id == order.id) .order_by(PurchaseOrderItem.id.asc()) ) item_rows = item_result.all() return PurchaseOrderDetailResponse( id=order.id, order_no=order.order_no, supplier_id=order.supplier_id, supplier_name=supplier_name, order_date=order.order_date, expected_date=order.expected_date, status=order.status, total_amount=order.total_amount, paid_amount=order.paid_amount, remark=order.remark, created_at=order.created_at, received_date=order.received_date, paid_date=order.paid_date, items=[ PurchaseOrderItemResponse( id=item.id, product_id=item.product_id, product_sku=product.sku, product_name=product.name, quantity=item.quantity, received_quantity=item.received_quantity, unit_price=item.unit_price, amount=item.amount, remark=item.remark ) for item, product in item_rows ] ) async def _resolve_receive_warehouse( db_session: AsyncSession, warehouse_id: Optional[int] ) -> Warehouse: if warehouse_id: result = await db_session.execute( select(Warehouse).where(Warehouse.id == warehouse_id, Warehouse.is_active == True) ) warehouse = result.scalar_one_or_none() if not warehouse: raise HTTPException(status_code=404, detail="仓库不存在") return warehouse result = await db_session.execute( select(Warehouse).where(Warehouse.is_active == True).order_by(Warehouse.is_default.desc(), Warehouse.id.asc()) ) warehouse = result.scalars().first() if not warehouse: raise HTTPException(status_code=400, detail="未配置可用仓库") return warehouse async def _apply_order_items( db_session: AsyncSession, order: PurchaseOrder, order_data: PurchaseOrderCreate ) -> 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 != "material": raise HTTPException(status_code=400, detail=f"采购单仅允许物料: {product.name}") # 使用物料的成本价格作为单价，忽略前端提交的单价 unit_price = product.cost_price or 0 if unit_price <= 0: raise HTTPException(status_code=400, detail=f"物料 {product.name} 未设置成本价格，请在物料管理界面设置") try: item = PurchaseOrderItem( order_id=order.id, product_id=item_data.product_id, quantity=int(item_data.quantity), unit_price=float(unit_price), amount=float(item_data.quantity) * float(unit_price), remark=item_data.remark or None ) db_session.add(item) total_amount += item.amount except Exception as e: raise HTTPException(status_code=400, detail=f"创建订单明细失败: {str(e)}") return total_amount @router.get("", response_model=List[PurchaseOrderResponse]) async def list_purchase_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(PurchaseOrder, Supplier) .join(Supplier, PurchaseOrder.supplier_id == Supplier.id) .order_by(PurchaseOrder.created_at.desc()) ) if status: query = query.where(PurchaseOrder.status == status) query = query.offset(skip).limit(limit) result = await db_session.execute(query) orders = [] for order, supplier in result.all(): orders.append(_build_purchase_order_response(order, supplier.name)) return orders @router.post("", response_model=PurchaseOrderResponse, status_code=201) async def create_purchase_order( order_data: PurchaseOrderCreate, db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): order = PurchaseOrder( order_no=generate_order_no("PO"), supplier_id=order_data.supplier_id, expected_date=order_data.expected_date, remark=order_data.remark, operator_id=current_user.id, status="pending" ) db_session.add(order) await db_session.flush() order.total_amount = await _apply_order_items(db_session, order, order_data) await db_session.commit() await db_session.refresh(order) supplier = await db_session.execute(select(Supplier).where(Supplier.id == order.supplier_id)) supplier = supplier.scalar_one() return _build_purchase_order_response(order, supplier.name) @router.get("/{order_id}", response_model=PurchaseOrderDetailResponse) async def get_purchase_order_detail( order_id: int, db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): order, supplier = await _get_order_with_supplier(db_session, order_id) return await _build_purchase_order_detail(db_session, order, supplier.name) @router.put("/{order_id}", response_model=PurchaseOrderResponse) async def update_purchase_order( order_id: int, order_data: PurchaseOrderCreate, db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): order, _ = await _get_order_with_supplier(db_session, order_id) if order.paid_amount and order.paid_amount > 0: raise HTTPException(status_code=400, detail="已付款采购单不允许修改") item_result = await db_session.execute( select(PurchaseOrderItem).where(PurchaseOrderItem.order_id == order.id) ) existing_items = item_result.scalars().all() if any((item.received_quantity or 0) > 0 for item in existing_items): raise HTTPException(status_code=400, detail="已发生入库的采购单不允许直接修改") for item in existing_items: await db_session.delete(item) order.supplier_id = order_data.supplier_id order.expected_date = order_data.expected_date order.remark = order_data.remark order.total_amount = await _apply_order_items(db_session, order, order_data) order.status = "pending" await db_session.commit() await db_session.refresh(order) supplier_result = await db_session.execute(select(Supplier).where(Supplier.id == order.supplier_id)) supplier = supplier_result.scalar_one_or_none() return _build_purchase_order_response(order, supplier.name if supplier else "未知供应商") @router.delete("/{order_id}") async def delete_purchase_order( order_id: int, db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): order, _ = await _get_order_with_supplier(db_session, order_id) if order.paid_amount and order.paid_amount > 0: raise HTTPException(status_code=400, detail="已付款采购单不允许删除") item_result = await db_session.execute( select(PurchaseOrderItem).where(PurchaseOrderItem.order_id == order.id) ) if any((item.received_quantity or 0) > 0 for item in item_result.scalars().all()): raise HTTPException(status_code=400, detail="已发生入库的采购单不允许删除") await db_session.delete(order) await db_session.commit() return {"message": "采购订单已删除"} @router.patch("/{order_id}/status") async def update_purchase_order_status( order_id: int, status: dict, db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): order, _ = await _get_order_with_supplier(db_session, order_id) new_status = status.get("status") if not new_status: raise HTTPException(status_code=400, detail="状态不能为空") valid_statuses = ["pending", "received", "paid"] if new_status not in valid_statuses: raise HTTPException(status_code=400, detail=f"无效的状态值，有效值为: {valid_statuses}") # 状态转换逻辑 if order.status == "paid": raise HTTPException(status_code=400, detail="已付款的采购订单禁止修改状态") if order.status == "received" and new_status != "paid": raise HTTPException(status_code=400, detail="已收货的采购订单只能标记为已付款") # 更新状态和对应时间 order.status = new_status if new_status == "received": order.received_date = func.now() elif new_status == "paid": order.paid_date = func.now() await db_session.commit() await db_session.refresh(order) supplier_result = await db_session.execute(select(Supplier).where(Supplier.id == order.supplier_id)) supplier = supplier_result.scalar_one_or_none() return _build_purchase_order_response(order, supplier.name if supplier else "未知供应商") @router.post("/{order_id}/receive", response_model=PurchaseOrderDetailResponse) async def receive_purchase_order( order_id: int, payload: PurchaseOrderReceiveRequest, db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): order, supplier = await _get_order_with_supplier(db_session, order_id) warehouse = await _resolve_receive_warehouse(db_session, payload.warehouse_id) item_result = await db_session.execute( select(PurchaseOrderItem).where(PurchaseOrderItem.order_id == order.id) ) item_map = {item.id: item for item in item_result.scalars().all()} if not item_map: raise HTTPException(status_code=400, detail="采购单无明细，无法入库") if not payload.items: raise HTTPException(status_code=400, detail="请提供本次入库明细") for receive_item in payload.items: item = item_map.get(receive_item.item_id) if not item: raise HTTPException(status_code=400, detail=f"采购明细不存在: {receive_item.item_id}") if receive_item.receive_quantity <= 0: raise HTTPException(status_code=400, detail="入库数量必须大于0") remaining_qty = (item.quantity or 0) - (item.received_quantity or 0) if receive_item.receive_quantity > remaining_qty: raise HTTPException(status_code=400, detail=f"明细{item.id}入库超量，剩余可入库{remaining_qty}") for receive_item in payload.items: item = item_map[receive_item.item_id] product_result = await db_session.execute( select(Product).where(Product.id == item.product_id) ) product = product_result.scalar_one_or_none() if not product: raise HTTPException(status_code=400, detail=f"物料不存在: {item.product_id}") inv_result = await db_session.execute( select(Inventory) .where(Inventory.product_id == item.product_id) .where(Inventory.warehouse_id == warehouse.id) ) inventory = inv_result.scalar_one_or_none() if not inventory: inventory = Inventory( product_id=item.product_id, warehouse_id=warehouse.id, quantity=0, locked_quantity=0 ) db_session.add(inventory) await db_session.flush() before_qty = inventory.quantity inventory.quantity += receive_item.receive_quantity after_qty = inventory.quantity item.received_quantity = (item.received_quantity or 0) + receive_item.receive_quantity movement = StockMovement( product_id=item.product_id, warehouse_id=warehouse.id, movement_type="purchase_in", quantity=receive_item.receive_quantity, before_quantity=before_qty, after_quantity=after_qty, reference_type="purchase_order", reference_id=order.id, reference_no=order.order_no, unit_price=item.unit_price, total_amount=round(float(item.unit_price * receive_item.receive_quantity), 4), remark=payload.remark or f"采购单{order.order_no}到货入库", operator_id=current_user.id ) db_session.add(movement) all_received = all((item.received_quantity or 0) >= (item.quantity or 0) for item in item_map.values()) any_received = any((item.received_quantity or 0) > 0 for item in item_map.values()) if all_received: order.status = "received" order.received_date = func.now() elif any_received: order.status = "partial_received" await db_session.commit() await db_session.refresh(order) return await _build_purchase_order_detail(db_session, order, supplier.name)