feat(pl): add feedback brand qty persist helpers and tests

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

backend/app/services/pl_feedback_brand_qty.py

@@ -0,0 +1,217 @@
+"""现场入出库明细与运输回传的品牌数量校验与持久化"""
+from decimal import Decimal
+from typing import Any, Optional, Type
+
+from sqlalchemy.orm import Session
+
+from app.core.utils import generate_uuid
+from app.services.pl_brand_archive import validate_brand_qty_rows
+
+
+def format_brand_qty_summary(rows: list[dict]) -> Optional[str]:
+    parts = []
+    for r in rows:
+        qty = Decimal(str(r.get("qty") or 0))
+        if qty <= 0:
+            continue
+        name = (r.get("brandName") or r.get("brand_name") or "").strip() or "未命名品牌"
+        qty_text = format(qty.normalize(), "f")
+        parts.append(f"{name} {qty_text}")
+    return " / ".join(parts) if parts else None
+
+
+def require_brand_qtys_if_needed(
+    *,
+    brand_mode: Optional[str],
+    required_brand_id: Optional[str],
+    actual_qty: Decimal,
+    rows: list[dict],
+    enforce: bool,
+) -> None:
+    if not brand_mode:
+        return
+    if not enforce:
+        return
+    positive = [r for r in rows if Decimal(str(r.get("qty") or 0)) > 0]
+    if not positive:
+        raise ValueError("有品牌产品须填写至少一个品牌数量")
+    validate_brand_qty_rows(
+        brand_mode=brand_mode,
+        required_brand_id=required_brand_id,
+        actual_qty=Decimal(str(actual_qty)),
+        rows=rows,
+    )
+
+
+def persist_io_item_brand_qtys(
+    db: Session,
+    *,
+    model: Type[Any],
+    tenant_no: str,
+    order_id: str,
+    order_item_id: str,
+    plan_product_id: str,
+    brand_mode: Optional[str],
+    rows: list[dict],
+    user_id: Optional[str] = None,
+) -> None:
+    existing = (
+        db.query(model)
+        .filter(
+            model.order_item_id == order_item_id,
+            model.tenant_no == tenant_no,
+            model.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
+        db.add(
+            model(
+                id=generate_uuid(),
+                tenant_no=tenant_no,
+                order_id=order_id,
+                order_item_id=order_item_id,
+                plan_product_id=plan_product_id,
+                brand_id=item.get("brandId") or item.get("brand_id") or "",
+                brand_name=item.get("brandName") or item.get("brand_name"),
+                qty=qty,
+                order_num=idx,
+                create_by=user_id,
+                create_dept_id="",
+                update_by=user_id,
+            )
+        )
+
+
+def _dict_field(item: dict, *keys: str):
+    for key in keys:
+        if key in item and item[key] is not None:
+            return item[key]
+    return None
+
+
+def persist_transport_product_feedback(
+    db: Session,
+    *,
+    tenant_no: str,
+    transport,
+    plan_id: str,
+    products: list[dict],
+    user_id: Optional[str] = None,
+    enforce_brand: bool = True,
+) -> None:
+    from app.models.pl_models import (
+        PlPlanProduct,
+        PlTransportProductBrandQty,
+        PlTransportProductFeedback,
+    )
+
+    transport_id = transport.id
+
+    for row in (
+        db.query(PlTransportProductFeedback)
+        .filter(
+            PlTransportProductFeedback.transport_id == transport_id,
+            PlTransportProductFeedback.tenant_no == tenant_no,
+            PlTransportProductFeedback.del_flag == 0,
+        )
+        .all()
+    ):
+        row.del_flag = 1
+        row.update_by = user_id
+
+    for row in (
+        db.query(PlTransportProductBrandQty)
+        .filter(
+            PlTransportProductBrandQty.transport_id == transport_id,
+            PlTransportProductBrandQty.tenant_no == tenant_no,
+            PlTransportProductBrandQty.del_flag == 0,
+        )
+        .all()
+    ):
+        row.del_flag = 1
+        row.update_by = user_id
+
+    total_qty = Decimal("0")
+
+    for idx, item in enumerate(products):
+        plan_product_id = _dict_field(item, "planProductId", "plan_product_id")
+        actual_qty = Decimal(str(_dict_field(item, "actualQty", "actual_qty") or 0))
+        brand_qtys = _dict_field(item, "brandQtys", "brand_qtys") or []
+        total_qty += actual_qty
+
+        plan_product = (
+            db.query(PlPlanProduct)
+            .filter(
+                PlPlanProduct.id == plan_product_id,
+                PlPlanProduct.tenant_no == tenant_no,
+                PlPlanProduct.del_flag == 0,
+            )
+            .first()
+        )
+
+        brand_mode = _dict_field(item, "brandMode", "brand_mode")
+        required_brand_id = _dict_field(item, "brandId", "brand_id")
+        if plan_product:
+            brand_mode = brand_mode or plan_product.brand_mode
+            required_brand_id = required_brand_id if required_brand_id is not None else plan_product.brand_id
+
+        require_brand_qtys_if_needed(
+            brand_mode=brand_mode,
+            required_brand_id=required_brand_id,
+            actual_qty=actual_qty,
+            rows=brand_qtys,
+            enforce=enforce_brand,
+        )
+
+        feedback_id = generate_uuid()
+        db.add(
+            PlTransportProductFeedback(
+                id=feedback_id,
+                tenant_no=tenant_no,
+                transport_id=transport_id,
+                plan_id=plan_id,
+                plan_product_id=plan_product_id,
+                actual_qty=actual_qty,
+                order_num=idx,
+                create_by=user_id,
+                create_dept_id="",
+                update_by=user_id,
+            )
+        )
+
+        if not brand_mode:
+            continue
+
+        for bidx, bitem in enumerate(brand_qtys):
+            qty = Decimal(str(bitem.get("qty") or 0))
+            if qty <= 0:
+                continue
+            db.add(
+                PlTransportProductBrandQty(
+                    id=generate_uuid(),
+                    tenant_no=tenant_no,
+                    transport_id=transport_id,
+                    plan_product_id=plan_product_id,
+                    feedback_id=feedback_id,
+                    brand_id=bitem.get("brandId") or bitem.get("brand_id") or "",
+                    brand_name=bitem.get("brandName") or bitem.get("brand_name"),
+                    qty=qty,
+                    order_num=bidx,
+                    create_by=user_id,
+                    create_dept_id="",
+                    update_by=user_id,
+                )
+            )
+
+    transport.feedback_qty = total_qty

backend/tests/test_pl_feedback_brand_qty.py

@@ -0,0 +1,75 @@
+from decimal import Decimal
+
+import pytest
+
+from app.services.pl_feedback_brand_qty import format_brand_qty_summary, require_brand_qtys_if_needed
+
+
+def test_format_brand_qty_summary():
+    s = format_brand_qty_summary([
+        {"brandName": "君正", "qty": Decimal("16")},
+        {"brandName": "红四方", "qty": Decimal("12.000")},
+    ])
+    assert s == "君正 16 / 红四方 12"
+
+
+def test_format_brand_qty_summary_empty():
+    assert format_brand_qty_summary([]) is None
+
+
+def test_format_brand_qty_summary_zero_only():
+    assert format_brand_qty_summary([{"brandName": "君正", "qty": Decimal("0")}]) is None
+
+
+def test_require_skips_when_no_brand_mode():
+    require_brand_qtys_if_needed(
+        brand_mode=None,
+        required_brand_id=None,
+        actual_qty=Decimal("10"),
+        rows=[],
+        enforce=True,
+    )
+
+
+def test_require_skips_when_not_enforced():
+    require_brand_qtys_if_needed(
+        brand_mode="any",
+        required_brand_id=None,
+        actual_qty=Decimal("10"),
+        rows=[],
+        enforce=False,
+    )
+
+
+def test_require_enforces_when_completed():
+    with pytest.raises(ValueError, match="至少一个品牌数量"):
+        require_brand_qtys_if_needed(
+            brand_mode="any",
+            required_brand_id=None,
+            actual_qty=Decimal("10"),
+            rows=[],
+            enforce=True,
+        )
+
+
+def test_require_calls_validate_when_positive_rows(monkeypatch):
+    called = {}
+
+    def mock_validate(**kwargs):
+        called.update(kwargs)
+
+    monkeypatch.setattr(
+        "app.services.pl_feedback_brand_qty.validate_brand_qty_rows",
+        mock_validate,
+    )
+    rows = [{"brandId": "b1", "brandName": "君正", "qty": Decimal("10")}]
+    require_brand_qtys_if_needed(
+        brand_mode="any",
+        required_brand_id=None,
+        actual_qty=Decimal("10"),
+        rows=rows,
+        enforce=True,
+    )
+    assert called["brand_mode"] == "any"
+    assert called["actual_qty"] == Decimal("10")
+    assert called["rows"] == rows