feat(backend): add plan archive price mask helpers

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

backend/app/services/pl_plan_price_mask.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from decimal import Decimal
+
+from sqlalchemy.orm import Session
+
+from app.core.menu_guard import SUPER_ADMIN_ROLE, get_user_menu_codes
+from app.schemas.pl.plan import PlanWithProducts
+
+ARCHIVE_PRICE_PERM = "pl:plan-archive:price"
+ARCHIVE_PERM = "pl:plan-archive:archive"
+
+
+def user_has_menu_perm(db: Session, user, perm: str) -> bool:
+    if getattr(user, "role", None) == SUPER_ADMIN_ROLE:
+        return True
+    codes = get_user_menu_codes(db, user.id)
+    return perm in codes
+
+
+def user_can_view_archive_prices(db: Session, user) -> bool:
+    return user_has_menu_perm(db, user, ARCHIVE_PRICE_PERM)
+
+
+def user_can_archive_plan(db: Session, user) -> bool:
+    return user_has_menu_perm(db, user, ARCHIVE_PERM)
+
+
+def _mask_brand_qtys(brand_qtys):
+    if not brand_qtys:
+        return brand_qtys
+    return [bq.model_copy(update={"unit_price": None}) for bq in brand_qtys]
+
+
+def _mask_product_like(item):
+    return item.model_copy(
+        update={
+            "inbound_price": None,
+            "inbound_amount": None,
+            "outbound_price": None,
+            "outbound_amount": None,
+            "brand_qtys": _mask_brand_qtys(getattr(item, "brand_qtys", None)),
+        }
+    )
+
+
+def mask_plan_with_products_prices(
+    plan: PlanWithProducts, *, enabled: bool = True
+) -> PlanWithProducts:
+    if not enabled:
+        return plan
+    products = [_mask_product_like(p) for p in (plan.products or [])]
+    load = plan.load
+    if load is not None and getattr(load, "products", None):
+        load = load.model_copy(
+            update={"products": [_mask_product_like(p) for p in load.products]}
+        )
+    deliveries = []
+    for d in plan.deliveries or []:
+        if getattr(d, "products", None):
+            deliveries.append(
+                d.model_copy(
+                    update={"products": [_mask_product_like(p) for p in d.products]}
+                )
+            )
+        else:
+            deliveries.append(d)
+    return plan.model_copy(
+        update={
+            "total_amount": Decimal("0"),
+            "freight_amount": Decimal("0"),
+            "products": products,
+            "load": load,
+            "deliveries": deliveries,
+        }
+    )

backend/tests/test_pl_plan_price_mask.py

@@ -0,0 +1,96 @@
+from datetime import date, datetime
+from decimal import Decimal
+
+from app.schemas.pl.plan import (
+    PlanLoadRead,
+    PlanProductRead,
+    PlanStopProductRead,
+    PlanWithProducts,
+)
+from app.services.pl_plan_price_mask import mask_plan_with_products_prices
+
+
+def _minimal_plan(**kwargs):
+    base = dict(
+        id="p1",
+        plan_no="P001",
+        belongs_company="shanxia",
+        create_date=date(2026, 9, 1),
+        total_amount=Decimal("100.00"),
+        freight_amount=Decimal("50.00"),
+        tenant_no="t1",
+        create_time=datetime(2026, 9, 1),
+        update_time=datetime(2026, 9, 1),
+        products=[],
+        deliveries=[],
+    )
+    base.update(kwargs)
+    return PlanWithProducts.model_validate(base)
+
+
+def test_mask_clears_plan_and_product_prices():
+    product = PlanProductRead.model_validate(
+        {
+            "id": "pp1",
+            "planId": "p1",
+            "tenantNo": "t1",
+            "createTime": "2026-09-01T00:00:00",
+            "updateTime": "2026-09-01T00:00:00",
+            "productId": "g1",
+            "productNo": "N1",
+            "productName": "尿素",
+            "inboundPrice": "12.5",
+            "inboundAmount": "125",
+            "outboundPrice": "13",
+            "outboundAmount": "130",
+            "brandQtys": [
+                {"brandId": "b1", "brandName": "君正", "qty": "10", "unitPrice": "12.5"}
+            ],
+        }
+    )
+    stop_product = PlanStopProductRead.model_validate(
+        {
+            "id": "sp1",
+            "planId": "p1",
+            "tenantNo": "t1",
+            "createTime": "2026-09-01T00:00:00",
+            "updateTime": "2026-09-01T00:00:00",
+            "productId": "g1",
+            "productNo": "N1",
+            "productName": "尿素",
+            "inboundPrice": "12.5",
+            "inboundAmount": "125",
+            "brandQtys": [{"brandId": "b1", "qty": "10", "unitPrice": "12.5"}],
+        }
+    )
+    plan = _minimal_plan(
+        products=[product],
+        load=PlanLoadRead.model_validate(
+            {
+                "id": "l1",
+                "planId": "p1",
+                "stopKind": "load",
+                "products": [stop_product],
+                "tenantNo": "t1",
+                "createTime": "2026-09-01T00:00:00",
+                "updateTime": "2026-09-01T00:00:00",
+            }
+        ),
+    )
+    out = mask_plan_with_products_prices(plan)
+    assert out.total_amount == Decimal("0")
+    assert out.freight_amount == Decimal("0")
+    assert out.products[0].inbound_price is None
+    assert out.products[0].inbound_amount is None
+    assert out.products[0].outbound_price is None
+    assert out.products[0].outbound_amount is None
+    assert out.products[0].brand_qtys[0].unit_price is None
+    assert out.load.products[0].inbound_price is None
+    assert out.load.products[0].brand_qtys[0].unit_price is None
+
+
+def test_mask_is_noop_when_disabled():
+    plan = _minimal_plan(total_amount=Decimal("9"), freight_amount=Decimal("8"))
+    out = mask_plan_with_products_prices(plan, enabled=False)
+    assert out.total_amount == Decimal("9")
+    assert out.freight_amount == Decimal("8")