diff --git a/backend/app/services/pl_plan_price_mask.py b/backend/app/services/pl_plan_price_mask.py new file mode 100644 index 000000000..b185cb661 --- /dev/null +++ b/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, + } + ) diff --git a/backend/tests/test_pl_plan_price_mask.py b/backend/tests/test_pl_plan_price_mask.py new file mode 100644 index 000000000..f71a10287 --- /dev/null +++ b/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")