feat(pl): validate and persist archive quantities by brand

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

backend/app/schemas/pl/plan.py

@@ -301,9 +301,17 @@ class PlanProductCertsUpdate(BaseModel):
     model_config = {"populate_by_name": True}
 
 
+class PlanArchiveBrandQtyItem(BaseModel):
+    brand_id: str = Field(default="", alias="brandId")
+    brand_name: Optional[str] = Field(default=None, alias="brandName")
+    qty: Decimal = Field(..., ge=0)
+    model_config = {"populate_by_name": True}
+
+
 class PlanArchiveProductItem(BaseModel):
     id: str = Field(..., description="计划产品明细ID pl_plan_product.id")
     actual_qty: Decimal = Field(..., alias="actualQty", ge=0, description="实际数量")
+    brand_qtys: Optional[list[PlanArchiveBrandQtyItem]] = Field(default=None, alias="brandQtys")
     model_config = {"populate_by_name": True}
 
 

backend/app/services/pl_brand_archive.py

@@ -0,0 +1,84 @@
+"""计划归档按品牌数量校验与持久化"""
+from decimal import Decimal
+from typing import Any, Optional
+
+from sqlalchemy.orm import Session
+
+from app.core.utils import generate_uuid
+from app.models.pl_models import PlPlanProductBrandQty
+
+
+def validate_brand_qty_rows(
+    *,
+    brand_mode: Optional[str],
+    required_brand_id: Optional[str],
+    actual_qty: Decimal,
+    rows: list[dict],
+) -> None:
+    actual_qty = Decimal(str(actual_qty))
+    if not brand_mode:
+        return
+    positive = [r for r in rows if Decimal(str(r.get("qty") or 0)) > 0]
+    if brand_mode == "required":
+        if len(positive) != 1 or positive[0].get("brandId") != required_brand_id:
+            raise ValueError("指定品牌时只能填写该品牌数量")
+        if Decimal(str(positive[0]["qty"])) != actual_qty:
+            raise ValueError("指定品牌数量必须等于归档实际数量")
+        return
+    if brand_mode == "any":
+        total = sum((Decimal(str(r.get("qty") or 0)) for r in rows), Decimal("0"))
+        if total != actual_qty:
+            raise ValueError("各品牌数量合计必须等于归档实际数量")
+        return
+    raise ValueError("无效的 brand_mode")
+
+
+def persist_plan_product_brand_qtys(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan_id: str,
+    plan_product_id: str,
+    brand_mode: Optional[str],
+    rows: list[dict[str, Any]],
+    user_id: str,
+    dept_id: str = "",
+) -> None:
+    """软删旧品牌量行后按归档结果插入；无品牌约定则只软删不插入。"""
+    existing = (
+        db.query(PlPlanProductBrandQty)
+        .filter(
+            PlPlanProductBrandQty.plan_product_id == plan_product_id,
+            PlPlanProductBrandQty.tenant_no == tenant_no,
+            PlPlanProductBrandQty.del_flag == 0,
+        )
+        .all()
+    )
+    for row in existing:
+        row.del_flag = 1
+        row.update_by = user_id
+
+    if not brand_mode:
+        return
+
+    for idx, item in enumerate(rows):
+        qty = Decimal(str(item.get("qty") or 0))
+        if qty <= 0:
+            continue
+        brand_id = item.get("brandId") or item.get("brand_id") or ""
+        brand_name = item.get("brandName") or item.get("brand_name")
+        db.add(
+            PlPlanProductBrandQty(
+                id=generate_uuid(),
+                tenant_no=tenant_no,
+                plan_id=plan_id,
+                plan_product_id=plan_product_id,
+                brand_id=brand_id,
+                brand_name=brand_name,
+                qty=qty,
+                order_num=idx,
+                create_by=user_id,
+                create_dept_id=dept_id or "",
+                update_by=user_id,
+            )
+        )

backend/app/services/pl_plan_service.py

@@ -31,6 +31,7 @@ from app.schemas.pl.plan import (
     PlanSiteFeedbackUpdateBody,
     PlanTransportCreateBlock,
 )
+from app.services.pl_brand_archive import persist_plan_product_brand_qtys, validate_brand_qty_rows
 from app.services.pl_freight_settlement_service import sync_draft_settlement_freight_for_transport
 from app.services.pl_transport_resource_snapshot_service import (
     apply_resource_snapshot_json,
@@ -1138,6 +1139,36 @@ def apply_archive_product_quantities(
                 prod.outbound_amount = actual * Decimal(str(price_out))
         prod.update_by = user_id
 
+        brand_rows = []
+        if item.brand_qtys:
+            brand_rows = [
+                {
+                    "brandId": bq.brand_id or "",
+                    "brandName": bq.brand_name,
+                    "qty": bq.qty,
+                }
+                for bq in item.brand_qtys
+            ]
+        try:
+            validate_brand_qty_rows(
+                brand_mode=prod.brand_mode,
+                required_brand_id=prod.brand_id,
+                actual_qty=actual,
+                rows=brand_rows,
+            )
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+        persist_plan_product_brand_qtys(
+            db,
+            tenant_no=tenant_no,
+            plan_id=plan.id,
+            plan_product_id=prod.id,
+            brand_mode=prod.brand_mode,
+            rows=brand_rows,
+            user_id=user_id,
+            dept_id=getattr(prod, "create_dept_id", "") or "",
+        )
+
 
 def apply_archive_freight(
     db: Session,

backend/tests/test_pl_brand_archive.py

@@ -0,0 +1,49 @@
+from decimal import Decimal
+
+import pytest
+
+from app.services.pl_brand_archive import validate_brand_qty_rows
+
+
+def test_required_allows_only_that_brand():
+    with pytest.raises(ValueError, match="指定品牌"):
+        validate_brand_qty_rows(
+            brand_mode="required",
+            required_brand_id="b1",
+            actual_qty=Decimal("10"),
+            rows=[{"brandId": "b2", "brandName": "红四方", "qty": Decimal("10")}],
+        )
+
+
+def test_any_sum_must_match():
+    with pytest.raises(ValueError, match="合计"):
+        validate_brand_qty_rows(
+            brand_mode="any",
+            required_brand_id=None,
+            actual_qty=Decimal("10"),
+            rows=[
+                {"brandId": "b1", "brandName": "君正", "qty": Decimal("6")},
+                {"brandId": "b2", "brandName": "红四方", "qty": Decimal("3")},
+            ],
+        )
+
+
+def test_any_ok():
+    validate_brand_qty_rows(
+        brand_mode="any",
+        required_brand_id=None,
+        actual_qty=Decimal("10"),
+        rows=[
+            {"brandId": "b1", "brandName": "君正", "qty": Decimal("6")},
+            {"brandId": "b2", "brandName": "红四方", "qty": Decimal("4")},
+        ],
+    )
+
+
+def test_no_brand_ignores_rows():
+    validate_brand_qty_rows(
+        brand_mode=None,
+        required_brand_id=None,
+        actual_qty=Decimal("5"),
+        rows=[],
+    )