feat(pl): post brand-level inventory on plan archive

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/pl/inventory.py

@@ -85,10 +85,12 @@ def set_warning_threshold(
 ) -> dict:
     tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
 
+    brand_id = payload.brand_id or ""
     inventory = (
         db.query(PlInventory)
         .filter(
             PlInventory.product_id == payload.product_id,
+            PlInventory.brand_id == brand_id,
             PlInventory.tenant_no == tenant_no,
             PlInventory.del_flag == 0,
         )
@@ -99,12 +101,16 @@ def set_warning_threshold(
         assert_pl_row_in_scope(db, inventory, PlInventory, user, tenant_no)
         inventory.low_stock_warning = payload.low_stock_warning
         inventory.high_stock_warning = payload.high_stock_warning
+        if payload.brand_name is not None:
+            inventory.brand_name = payload.brand_name
     else:
         # 创建库存记录
         inventory = PlInventory(
             id=generate_uuid(),
             tenant_no=tenant_no,
             product_id=payload.product_id,
+            brand_id=brand_id,
+            brand_name=payload.brand_name,
             product_no="",
             product_name="",
             unit="吨",

backend/app/api/pl/plan.py

@@ -58,6 +58,7 @@ from app.services.pl_inventory_link_read import build_transport_inventory_link_r
 from app.services.pl_io_order_service import get_active_io_order_for_plan, void_io_orders_for_plan
 from app.services.pl_site_link_read import build_transport_site_link_read_from_row
 from app.services.pl_transport_site_service import revoke_site_links_for_transport
+from app.services.pl_inventory_service import apply_archive_inventory_for_plan
 from app.services.pl_plan_service import (
     apply_archive_freight,
     apply_archive_product_quantities,
@@ -1177,6 +1178,13 @@ def archive_plan(
     apply_archive_product_quantities(
         db, tenant_no=tenant_no, plan=plan, items=payload.products, user_id=user.id
     )
+    apply_archive_inventory_for_plan(
+        db,
+        tenant_no=tenant_no,
+        plan=plan,
+        user_id=user.id,
+        dept_id=user.dept_id or "",
+    )
     if payload.freight is not None:
         apply_archive_freight(
             db, tenant_no=tenant_no, plan=plan, freight=payload.freight, user_id=user.id

backend/app/schemas/pl/inventory.py

@@ -14,6 +14,8 @@ from pydantic import BaseModel, Field
 class InventoryRead(BaseModel):
     id: str
     product_id: str = Field(alias="productId")
+    brand_id: str = Field(default="", alias="brandId")
+    brand_name: Optional[str] = Field(default=None, alias="brandName")
     product_no: str = Field(alias="productNo")
     product_name: str = Field(alias="productName")
     model: Optional[str] = None
@@ -36,6 +38,8 @@ class InventoryRead(BaseModel):
 class InventoryTransactionRead(BaseModel):
     id: str
     product_id: str = Field(alias="productId")
+    brand_id: str = Field(default="", alias="brandId")
+    brand_name: Optional[str] = Field(default=None, alias="brandName")
     product_name: str = Field(alias="productName")
     transaction_type: str = Field(alias="transactionType")
     plan_id: Optional[str] = Field(default=None, alias="planId")
@@ -57,6 +61,8 @@ class InventoryTransactionRead(BaseModel):
 class InventoryWarningRead(BaseModel):
     id: str
     product_id: str = Field(alias="productId")
+    brand_id: str = Field(default="", alias="brandId")
+    brand_name: Optional[str] = Field(default=None, alias="brandName")
     product_name: str = Field(alias="productName")
     warning_type: str = Field(alias="warningType")
     current_qty: Decimal = Field(alias="currentQty")
@@ -77,6 +83,8 @@ class InventoryWarningUpdate(BaseModel):
 
 class InventoryWarningThreshold(BaseModel):
     product_id: str = Field(..., alias="productId", description="产品ID")
+    brand_id: str = Field(default="", alias="brandId", description="品牌ID，无品牌为空串")
+    brand_name: Optional[str] = Field(default=None, alias="brandName", description="品牌名称")
     low_stock_warning: Optional[Decimal] = Field(default=None, alias="lowStockWarning", description="最低库存预警值")
     high_stock_warning: Optional[Decimal] = Field(default=None, alias="highStockWarning", description="最高库存预警值")
     model_config = {"populate_by_name": True}

backend/app/services/pl_inventory_service.py

@@ -0,0 +1,256 @@
+"""计划归档按产品+品牌写入库存台账与流水。
+
+直销单 (direct) 首期不记账；入库增加、出库扣减；出库不足硬拦截。
+调整归档时先按 plan_id 冲销旧流水再按新量重记。
+"""
+from decimal import Decimal
+from typing import Optional
+
+from fastapi import HTTPException
+from sqlalchemy.orm import Session
+
+from app.core.utils import generate_uuid
+from app.models.pl_models import (
+    PlInventory,
+    PlInventoryTransaction,
+    PlPlan,
+    PlPlanProduct,
+    PlPlanProductBrandQty,
+)
+
+EMPTY_BRAND_ID = ""
+
+
+def qty_delta_for_plan(plan_type: str, qty: Decimal) -> Decimal:
+    if plan_type == "inbound":
+        return Decimal(str(qty))
+    if plan_type == "outbound":
+        return -Decimal(str(qty))
+    return Decimal("0")  # direct: 首期不记账
+
+
+def assert_outbound_enough(*, stock_qty: Decimal, outbound_qty: Decimal, brand_label: str) -> None:
+    if Decimal(str(outbound_qty)) > Decimal(str(stock_qty)):
+        raise ValueError(f"{brand_label}库存不足")
+
+
+def get_or_create_inventory(
+    db: Session,
+    *,
+    tenant_no: str,
+    product_id: str,
+    product_no: str,
+    product_name: str,
+    brand_id: Optional[str],
+    brand_name: Optional[str],
+    unit: Optional[str],
+    model: Optional[str],
+    user_id: str,
+    dept_id: str,
+) -> PlInventory:
+    brand_id = brand_id or EMPTY_BRAND_ID
+    row = (
+        db.query(PlInventory)
+        .filter(
+            PlInventory.tenant_no == tenant_no,
+            PlInventory.product_id == product_id,
+            PlInventory.brand_id == brand_id,
+            PlInventory.del_flag == 0,
+        )
+        .first()
+    )
+    if row:
+        return row
+    row = PlInventory(
+        id=generate_uuid(),
+        tenant_no=tenant_no,
+        product_id=product_id,
+        brand_id=brand_id,
+        brand_name=brand_name,
+        product_no=product_no or "",
+        product_name=product_name or "",
+        model=model,
+        unit=unit or "吨",
+        stock_qty=Decimal("0"),
+        stock_amount=Decimal("0"),
+        create_by=user_id,
+        create_dept_id=dept_id or "",
+        update_by=user_id,
+    )
+    db.add(row)
+    db.flush()
+    return row
+
+
+def apply_brand_stock_change(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan: PlPlan,
+    plan_product: PlPlanProduct,
+    brand_id: Optional[str],
+    brand_name: Optional[str],
+    qty: Decimal,
+    user_id: str,
+    dept_id: str,
+) -> None:
+    delta = qty_delta_for_plan(plan.plan_type, Decimal(str(qty)))
+    if delta == 0:
+        return
+    inv = get_or_create_inventory(
+        db,
+        tenant_no=tenant_no,
+        product_id=plan_product.product_id,
+        product_no=plan_product.product_no,
+        product_name=plan_product.product_name,
+        brand_id=brand_id or EMPTY_BRAND_ID,
+        brand_name=brand_name,
+        unit=plan_product.unit,
+        model=plan_product.model,
+        user_id=user_id,
+        dept_id=dept_id,
+    )
+    if delta < 0:
+        try:
+            assert_outbound_enough(
+                stock_qty=Decimal(str(inv.stock_qty or 0)),
+                outbound_qty=-delta,
+                brand_label=brand_name or plan_product.product_name or "",
+            )
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+    unit_price = (
+        plan_product.inbound_price
+        or plan_product.outbound_price
+        or plan_product.tax_price
+        or Decimal("0")
+    )
+    unit_price = Decimal(str(unit_price))
+    inv.stock_qty = Decimal(str(inv.stock_qty or 0)) + delta
+    inv.stock_amount = Decimal(str(inv.stock_qty or 0)) * unit_price
+    inv.update_by = user_id
+    db.add(
+        PlInventoryTransaction(
+            id=generate_uuid(),
+            tenant_no=tenant_no,
+            product_id=plan_product.product_id,
+            brand_id=brand_id or EMPTY_BRAND_ID,
+            brand_name=brand_name,
+            product_name=plan_product.product_name or "",
+            transaction_type="inbound" if delta > 0 else "outbound",
+            plan_id=plan.id,
+            plan_no=plan.plan_no,
+            qty=delta,
+            unit_price=unit_price,
+            amount=abs(delta) * unit_price,
+            stock_after=inv.stock_qty,
+            create_by=user_id,
+            create_dept_id=dept_id or "",
+            update_by=user_id,
+        )
+    )
+
+
+def reverse_plan_inventory_postings(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan_id: str,
+    user_id: str,
+) -> None:
+    """冲销该计划已产生的库存流水（调整归档前调用）。"""
+    txns = (
+        db.query(PlInventoryTransaction)
+        .filter(
+            PlInventoryTransaction.tenant_no == tenant_no,
+            PlInventoryTransaction.plan_id == plan_id,
+            PlInventoryTransaction.del_flag == 0,
+        )
+        .order_by(PlInventoryTransaction.create_time.asc())
+        .all()
+    )
+    for txn in txns:
+        brand_id = txn.brand_id or EMPTY_BRAND_ID
+        inv = (
+            db.query(PlInventory)
+            .filter(
+                PlInventory.tenant_no == tenant_no,
+                PlInventory.product_id == txn.product_id,
+                PlInventory.brand_id == brand_id,
+                PlInventory.del_flag == 0,
+            )
+            .first()
+        )
+        if inv:
+            inv.stock_qty = Decimal(str(inv.stock_qty or 0)) - Decimal(str(txn.qty or 0))
+            unit_price = Decimal(str(txn.unit_price or 0))
+            inv.stock_amount = Decimal(str(inv.stock_qty or 0)) * unit_price
+            inv.update_by = user_id
+        txn.del_flag = 1
+        txn.update_by = user_id
+
+
+def apply_archive_inventory_for_plan(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan: PlPlan,
+    user_id: str,
+    dept_id: str = "",
+) -> None:
+    """归档后按品牌量（或无品牌 actual_qty）落账；先冲销旧流水再重记。"""
+    reverse_plan_inventory_postings(
+        db, tenant_no=tenant_no, plan_id=plan.id, user_id=user_id
+    )
+    if qty_delta_for_plan(plan.plan_type, Decimal("1")) == 0:
+        return
+
+    products = (
+        db.query(PlPlanProduct)
+        .filter(
+            PlPlanProduct.plan_id == plan.id,
+            PlPlanProduct.tenant_no == tenant_no,
+            PlPlanProduct.del_flag == 0,
+        )
+        .order_by(PlPlanProduct.order_num.asc())
+        .all()
+    )
+    for prod in products:
+        brand_rows = (
+            db.query(PlPlanProductBrandQty)
+            .filter(
+                PlPlanProductBrandQty.plan_product_id == prod.id,
+                PlPlanProductBrandQty.tenant_no == tenant_no,
+                PlPlanProductBrandQty.del_flag == 0,
+            )
+            .order_by(PlPlanProductBrandQty.order_num.asc())
+            .all()
+        )
+        if prod.brand_mode and brand_rows:
+            for bq in brand_rows:
+                apply_brand_stock_change(
+                    db,
+                    tenant_no=tenant_no,
+                    plan=plan,
+                    plan_product=prod,
+                    brand_id=bq.brand_id or EMPTY_BRAND_ID,
+                    brand_name=bq.brand_name,
+                    qty=Decimal(str(bq.qty or 0)),
+                    user_id=user_id,
+                    dept_id=dept_id,
+                )
+            continue
+        actual = Decimal(str(prod.actual_qty or 0))
+        if actual <= 0:
+            continue
+        apply_brand_stock_change(
+            db,
+            tenant_no=tenant_no,
+            plan=plan,
+            plan_product=prod,
+            brand_id=EMPTY_BRAND_ID,
+            brand_name=None,
+            qty=actual,
+            user_id=user_id,
+            dept_id=dept_id,
+        )

backend/tests/test_pl_brand_archive.py

@@ -47,3 +47,18 @@ def test_no_brand_ignores_rows():
         actual_qty=Decimal("5"),
         rows=[],
     )
+
+
+def test_apply_stock_change_outbound_insufficient():
+    from app.services.pl_inventory_service import assert_outbound_enough
+
+    with pytest.raises(ValueError, match="库存不足"):
+        assert_outbound_enough(stock_qty=Decimal("1"), outbound_qty=Decimal("2"), brand_label="君正")
+
+
+def test_delta_for_plan_type():
+    from app.services.pl_inventory_service import qty_delta_for_plan
+
+    assert qty_delta_for_plan("inbound", Decimal("3")) == Decimal("3")
+    assert qty_delta_for_plan("outbound", Decimal("3")) == Decimal("-3")
+    assert qty_delta_for_plan("direct", Decimal("3")) == Decimal("0")