"""采购服务：采购单创建/审批/收货入账。收货时写入库流水 + 成本层。""" from decimal import Decimal from fastapi import HTTPException from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.plugins.inventory.services import _default_warehouse_id def next_po_status(status: str, action: str, *, approval_required: bool = False) -> str: if status == "draft" and action == "submit": return "pending_approval" if approval_required else "confirmed" transitions = { ("pending_approval", "approve"): "confirmed", ("pending_approval", "reject"): "draft", ("draft", "cancel"): "cancelled", ("pending_approval", "cancel"): "cancelled", ("confirmed", "cancel"): "cancelled", ("confirmed", "partial_receive"): "partially_received", ("confirmed", "full_receive"): "received", ("partially_received", "partial_receive"): "partially_received", ("partially_received", "full_receive"): "received", ("partially_received", "close"): "closed", ("received", "close"): "closed", } result = transitions.get((status, action)) if result is None: raise HTTPException(status_code=409, detail=f"采购单状态 {status} 不允许执行 {action}") return result def outstanding_qty(ordered: Decimal, received: Decimal) -> Decimal: return max(Decimal("0"), Decimal(str(ordered)) - Decimal(str(received))) def required_approval_levels(rules: list[dict], total_amount: Decimal, warehouse_id: int) -> list[int]: """Return the matching approval ladder; level 1 remains the safe default.""" amount = Decimal(str(total_amount)) levels = { int(rule["approval_level"]) for rule in rules if amount >= Decimal(str(rule["min_amount"])) and int(rule.get("warehouse_id") or 0) in (0, warehouse_id) } return sorted(levels) or [1] def assert_approver_separation(requester_id: int | None, approver_id: int) -> None: if requester_id is not None and int(requester_id) == int(approver_id): raise HTTPException(status_code=409, detail="采购申请人不能审批自己提交的采购单") def assert_approver_not_reused(approver_id: int, prior_approver_ids: set[int]) -> None: if int(approver_id) in prior_approver_ids: raise HTTPException(status_code=409, detail="同一审批人不能审批采购单的多个级别") async def _next_po_no(db: AsyncSession, tenant_id: int) -> str: """按租户生成采购单号。行锁取当年最大序号+1，避免并发撞号（不用 max+1 裸查）。""" from datetime import datetime ym = datetime.utcnow().strftime("%Y%m") prefix = f"PO{ym}" r = await db.execute( text("SELECT po_no FROM inventory_purchase_orders " "WHERE tenant_id=:tid AND po_no LIKE :p ORDER BY po_no DESC LIMIT 1 FOR UPDATE"), {"tid": tenant_id, "p": f"{prefix}%"}, ) row = r.fetchone() seq = (int(row[0][len(prefix):]) + 1) if row and row[0][len(prefix):].isdigit() else 1 return f"{prefix}{seq:04d}" async def _validate_line_products(db: AsyncSession, tenant_id: int, lines: list) -> None: product_ids = sorted({int(line.product_id) for line in lines}) products = await db.execute( text(f"SELECT id FROM products WHERE tenant_id=:tid AND id IN ({','.join(str(pid) for pid in product_ids)})"), {"tid": tenant_id}, ) if {int(row[0]) for row in products.fetchall()} != set(product_ids): raise HTTPException(status_code=400, detail="采购单包含不属于当前租户的商品") expected_variant_pairs = { (int(line.variant_id), int(line.product_id)) for line in lines if line.variant_id } if not expected_variant_pairs: return variant_ids = sorted({variant_id for variant_id, _ in expected_variant_pairs}) variants = await db.execute( text(f"SELECT id, product_id FROM product_variants WHERE tenant_id=:tid " f"AND id IN ({','.join(str(vid) for vid in variant_ids)})"), {"tid": tenant_id}, ) actual_variants = {int(row[0]): int(row[1]) for row in variants.fetchall()} if any(actual_variants.get(variant_id) != product_id for variant_id, product_id in expected_variant_pairs): raise HTTPException(status_code=400, detail="采购单包含不属于当前商品或租户的规格") async def _validate_supplier_catalog(db: AsyncSession, tenant_id: int, supplier_id: int, lines: list) -> None: rows = await db.execute( text("SELECT product_id, variant_id, purchase_uom, uom_factor FROM inventory_supplier_products " "WHERE tenant_id=:tid AND supplier_id=:sid AND is_active=1"), {"tid": tenant_id, "sid": supplier_id}, ) catalog = {(int(row[0]), int(row[1])): row[2:] for row in rows.fetchall()} for line in lines: package = catalog.get((int(line.product_id), int(line.variant_id or 0))) if package is None: raise HTTPException(status_code=400, detail="采购商品或规格未维护在该供应商的供货目录中") if line.purchase_uom != package[0] or Decimal(str(line.uom_factor)) != Decimal(str(package[1])): raise HTTPException(status_code=400, detail="采购包装与供应商供货目录不一致") async def create_po(db: AsyncSession, tenant_id: int, *, supplier_id: int, warehouse_id: int | None, currency: str, fx_rate: Decimal, lines: list, requester_id: int | None = None, supplier_promise_date: str | None = None, planned_arrival_date: str | None = None) -> int: if not lines: raise HTTPException(status_code=400, detail="采购单至少一行") wh = warehouse_id or await _default_warehouse_id(db, tenant_id) supplier = await db.execute( text("SELECT id FROM inventory_suppliers WHERE id=:id AND tenant_id=:tid AND is_active=1"), {"id": supplier_id, "tid": tenant_id}, ) if not supplier.fetchone(): raise HTTPException(status_code=400, detail="供应商不存在、已停用或不属于当前租户") warehouse = await db.execute( text("SELECT id FROM inventory_warehouses WHERE id=:id AND tenant_id=:tid AND is_active=1"), {"id": wh, "tid": tenant_id}, ) if not warehouse.fetchone(): raise HTTPException(status_code=400, detail="收货仓库不存在、已停用或不属于当前租户") await _validate_line_products(db, tenant_id, lines) await _validate_supplier_catalog(db, tenant_id, supplier_id, lines) po_no = await _next_po_no(db, tenant_id) total = sum(Decimal(str(l.qty)) * Decimal(str(l.unit_cost)) for l in lines) await db.execute( text("INSERT INTO inventory_purchase_orders " "(tenant_id, po_no, supplier_id, warehouse_id, status, currency, fx_rate, total_amount, " "requester_id, order_date, supplier_promise_date, planned_arrival_date, created_at, updated_at) " "VALUES (:tid,:no,:sup,:wh,'draft',:cur,:fx,:tot,:requester,CURRENT_DATE,:promise,:arrival,NOW(3),NOW(3))"), {"tid": tenant_id, "no": po_no, "sup": supplier_id, "wh": wh, "cur": currency, "fx": fx_rate, "tot": total, "requester": requester_id, "promise": supplier_promise_date, "arrival": planned_arrival_date}, ) po_id = (await db.execute(text("SELECT LAST_INSERT_ID()"))).scalar() for l in lines: await db.execute( text("INSERT INTO inventory_purchase_order_lines " "(tenant_id, po_id, product_id, variant_id, qty, unit_cost, received_qty, rejected_qty, " "returned_qty, expected_date, purchase_uom, uom_factor, batch_no, expires_on, created_at, updated_at) " "VALUES (:tid,:po,:pid,:vid,:qty,:uc,0,0,0,:expected,:uom,:factor,:bn,:exp,NOW(3),NOW(3))"), {"tid": tenant_id, "po": po_id, "pid": l.product_id, "vid": l.variant_id or 0, "qty": l.qty, "uc": l.unit_cost, "expected": l.expected_date, "uom": l.purchase_uom, "factor": l.uom_factor, "bn": l.batch_no, "exp": l.expires_on}, ) return int(po_id) async def _po_status(db, tenant_id, po_id) -> str | None: r = await db.execute( text("SELECT status FROM inventory_purchase_orders WHERE id=:id AND tenant_id=:tid"), {"id": po_id, "tid": tenant_id}, ) row = r.fetchone() return row[0] if row else None def require_draft(status: str | None) -> None: if status is None: raise HTTPException(status_code=404, detail="采购单不存在") if status != "draft": raise HTTPException(status_code=409, detail=f"采购单状态为 {status}，仅草稿可修改或取消") async def update_po(db: AsyncSession, tenant_id: int, po_id: int, *, supplier_id: int, warehouse_id: int | None, currency: str, fx_rate: Decimal, lines: list, supplier_promise_date: str | None = None, planned_arrival_date: str | None = None) -> None: require_draft(await _po_status(db, tenant_id, po_id)) if not lines: raise HTTPException(status_code=400, detail="采购单至少一行") wh = warehouse_id or await _default_warehouse_id(db, tenant_id) supplier = await db.execute( text("SELECT id FROM inventory_suppliers WHERE id=:id AND tenant_id=:tid AND is_active=1"), {"id": supplier_id, "tid": tenant_id}, ) if not supplier.fetchone(): raise HTTPException(status_code=400, detail="供应商不存在、已停用或不属于当前租户") warehouse = await db.execute( text("SELECT id FROM inventory_warehouses WHERE id=:id AND tenant_id=:tid AND is_active=1"), {"id": wh, "tid": tenant_id}, ) if not warehouse.fetchone(): raise HTTPException(status_code=400, detail="收货仓库不存在、已停用或不属于当前租户") await _validate_line_products(db, tenant_id, lines) await _validate_supplier_catalog(db, tenant_id, supplier_id, lines) total = sum(Decimal(str(line.qty)) * Decimal(str(line.unit_cost)) for line in lines) await db.execute( text("UPDATE inventory_purchase_orders SET supplier_id=:sup, warehouse_id=:wh, currency=:cur, " "fx_rate=:fx, total_amount=:tot, supplier_promise_date=:promise, " "planned_arrival_date=:arrival, updated_at=NOW(3) WHERE id=:id AND tenant_id=:tid"), {"sup": supplier_id, "wh": wh, "cur": currency, "fx": fx_rate, "tot": total, "promise": supplier_promise_date, "arrival": planned_arrival_date, "id": po_id, "tid": tenant_id}, ) await db.execute(text("DELETE FROM inventory_purchase_order_lines WHERE po_id=:po AND tenant_id=:tid"), {"po": po_id, "tid": tenant_id}) for line in lines: await db.execute( text("INSERT INTO inventory_purchase_order_lines " "(tenant_id, po_id, product_id, variant_id, qty, unit_cost, received_qty, rejected_qty, " "returned_qty, expected_date, purchase_uom, uom_factor, batch_no, expires_on, created_at, updated_at) " "VALUES (:tid,:po,:pid,:vid,:qty,:uc,0,0,0,:expected,:uom,:factor,:bn,:exp,NOW(3),NOW(3))"), {"tid": tenant_id, "po": po_id, "pid": line.product_id, "vid": line.variant_id or 0, "qty": line.qty, "uc": line.unit_cost, "expected": line.expected_date, "uom": line.purchase_uom, "factor": line.uom_factor, "bn": line.batch_no, "exp": line.expires_on}, ) async def cancel_po(db: AsyncSession, tenant_id: int, po_id: int) -> None: status = await _po_status(db, tenant_id, po_id) if status is None: raise HTTPException(status_code=404, detail="采购单不存在") next_po_status(status, "cancel") await db.execute( text("UPDATE inventory_purchase_orders SET status='cancelled', updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"id": po_id, "tid": tenant_id}, ) async def _approval_required(db, tenant_id) -> bool: r = await db.execute( text("SELECT purchase_approval FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"), {"tid": tenant_id}, ) row = r.fetchone() return bool(row[0]) if row else False async def submit_po(db: AsyncSession, tenant_id: int, po_id: int) -> str: status = await _po_status(db, tenant_id, po_id) if status is None: raise HTTPException(status_code=404, detail="采购单不存在") target = next_po_status(status, "submit", approval_required=await _approval_required(db, tenant_id)) await db.execute( text("UPDATE inventory_purchase_orders SET status=:status, updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"status": target, "id": po_id, "tid": tenant_id}, ) return target async def approve_po(db: AsyncSession, tenant_id: int, po_id: int, approver_id: int, comment: str | None = None) -> str: po = (await db.execute( text("SELECT status, requester_id, total_amount, warehouse_id FROM inventory_purchase_orders " "WHERE id=:id AND tenant_id=:tid FOR UPDATE"), {"id": po_id, "tid": tenant_id}, )).fetchone() if po is None: raise HTTPException(status_code=404, detail="采购单不存在") status, requester_id, total_amount, warehouse_id = po if status != "pending_approval": raise HTTPException(status_code=409, detail=f"采购单状态为 {status}，不可审批") assert_approver_separation(requester_id, approver_id) rule_rows = (await db.execute( text("SELECT approval_level, min_amount, warehouse_id FROM inventory_purchase_approval_rules " "WHERE tenant_id=:tid AND is_active=1 ORDER BY approval_level"), {"tid": tenant_id}, )).fetchall() rules = [{"approval_level": row[0], "min_amount": row[1], "warehouse_id": row[2]} for row in rule_rows] levels = required_approval_levels(rules, total_amount, int(warehouse_id)) approved_rows = (await db.execute( text("SELECT approval_level, approver_id FROM inventory_purchase_approvals " "WHERE tenant_id=:tid AND po_id=:po AND decision='approved'"), {"tid": tenant_id, "po": po_id}, )).fetchall() approved = {int(row[0]) for row in approved_rows} assert_approver_not_reused(approver_id, {int(row[1]) for row in approved_rows}) level = next((candidate for candidate in levels if candidate not in approved), None) if level is None: raise HTTPException(status_code=409, detail="采购单已完成全部审批") await db.execute( text("INSERT INTO inventory_purchase_approvals " "(tenant_id,po_id,approval_level,approver_id,decision,comment,decided_at,created_at,updated_at) " "VALUES (:tid,:po,:level,:by,'approved',:comment,NOW(3),NOW(3),NOW(3))"), {"tid": tenant_id, "po": po_id, "level": level, "by": approver_id, "comment": comment}, ) target = "confirmed" if all(candidate in approved | {level} for candidate in levels) else "pending_approval" await db.execute( text("UPDATE inventory_purchase_orders SET status=:status, approved_by=:by, " "approved_at=CASE WHEN :status='confirmed' THEN NOW(3) ELSE approved_at END, updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"status": target, "by": approver_id, "id": po_id, "tid": tenant_id}, ) return target async def close_po(db: AsyncSession, tenant_id: int, po_id: int, reason: str | None = None) -> None: status = await _po_status(db, tenant_id, po_id) if status is None: raise HTTPException(status_code=404, detail="采购单不存在") next_po_status(status, "close") await db.execute( text("UPDATE inventory_purchase_orders SET status='closed', closed_reason=:reason, updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"reason": reason, "id": po_id, "tid": tenant_id}, ) async def receive_po(db: AsyncSession, tenant_id: int, po_id: int, operator_id: int) -> None: """Receive every outstanding line through the shared stock-operation writer.""" from app.plugins.inventory.operations import create_full_purchase_receipt, post_operation operation_id = await create_full_purchase_receipt(db, tenant_id, po_id) await post_operation(db, tenant_id, operation_id, operator_id)