"""Customer-specific pricing engine for the B2B ordering portal. The backend is the single source of truth for prices — the frontend never computes a final price. :func:`resolve_price` returns a fully-attributed :class:`PriceResolution` so every order line, and any future report, can explain exactly which rule produced the number. Resolution priority (highest wins): 1. **Quote-only** — the product requires a manual quote, or the customer has a ``quote`` price rule for it. No automatic price. 2. **Fixed / contract** — a :class:`CustomerProductPrice` for this customer + product (quantity tiers may refine it). 3. **Price list** — a :class:`PriceListItem` from the customer's assigned price list (quantity tiers may refine it). 4. **Base + discount** — the catalogue list price, optionally reduced by the customer's default discount percentage. 5. **Quote fallback** — no resolvable price ⇒ treated as quote-only. All prices are GST-exclusive. """ from __future__ import annotations from dataclasses import dataclass from sqlalchemy import select from sqlalchemy.orm import Session from app.models.ordering import ( CatalogueProduct, CustomerPriceAssignment, CustomerProductPrice, PriceListItem, PriceTier, ) @dataclass(frozen=True) class PriceResolution: unit_price: float | None price_source: str # one of ordering.PRICE_SOURCES price_rule_id: int | None discount_percent: float requires_quote: bool label: str def line_total(self, quantity: float) -> float | None: if self.unit_price is None: return None return round(self.unit_price * quantity, 4) _SOURCE_LABELS = { "fixed": "Fixed price", "contract": "Contract price", "price_list": "Price list", "tiered": "Tiered price", "base": "List price", "quote": "Quote required", } def _best_tier_price( db: Session, *, customer_product_price_id: int | None = None, price_list_item_id: int | None = None, quantity: float, ) -> float | None: """Return the unit price of the highest qualifying quantity tier, if any.""" stmt = select(PriceTier).where(PriceTier.min_quantity <= quantity) if customer_product_price_id is not None: stmt = stmt.where(PriceTier.customer_product_price_id == customer_product_price_id) elif price_list_item_id is not None: stmt = stmt.where(PriceTier.price_list_item_id == price_list_item_id) else: return None tiers = db.scalars(stmt.order_by(PriceTier.min_quantity.desc())).all() return tiers[0].unit_price if tiers else None def _quote(label: str = "Quote required") -> PriceResolution: return PriceResolution( unit_price=None, price_source="quote", price_rule_id=None, discount_percent=0.0, requires_quote=True, label=label, ) def get_customer_assignment( db: Session, *, client_account_id: int ) -> CustomerPriceAssignment | None: return db.scalar( select(CustomerPriceAssignment).where( CustomerPriceAssignment.client_account_id == client_account_id ) ) def resolve_price( db: Session, *, client_account_id: int, product: CatalogueProduct, quantity: float, ) -> PriceResolution: """Resolve the GST-exclusive unit price for a customer + product + quantity.""" quantity = max(quantity, 0.0) # 1. Product-level manual-quote flag. if product.requires_quote: return _quote() # 2. Customer-specific fixed/contract/quote rule. cpp = db.scalar( select(CustomerProductPrice).where( CustomerProductPrice.client_account_id == client_account_id, CustomerProductPrice.product_id == product.id, CustomerProductPrice.active.is_(True), ) ) if cpp is not None: if cpp.rule_type == "quote": return _quote() tier_price = _best_tier_price( db, customer_product_price_id=cpp.id, quantity=quantity ) if tier_price is not None: return PriceResolution( unit_price=round(tier_price, 4), price_source="tiered", price_rule_id=cpp.id, discount_percent=0.0, requires_quote=False, label=_SOURCE_LABELS["tiered"], ) if cpp.unit_price is not None: source = "contract" if cpp.rule_type == "contract" else "fixed" return PriceResolution( unit_price=round(cpp.unit_price, 4), price_source=source, price_rule_id=cpp.id, discount_percent=0.0, requires_quote=False, label=_SOURCE_LABELS[source], ) assignment = get_customer_assignment(db, client_account_id=client_account_id) # 3. Assigned price list. if assignment is not None and assignment.price_list_id is not None: pli = db.scalar( select(PriceListItem).where( PriceListItem.price_list_id == assignment.price_list_id, PriceListItem.product_id == product.id, ) ) if pli is not None: tier_price = _best_tier_price( db, price_list_item_id=pli.id, quantity=quantity ) if tier_price is not None: return PriceResolution( unit_price=round(tier_price, 4), price_source="tiered", price_rule_id=pli.id, discount_percent=0.0, requires_quote=False, label=_SOURCE_LABELS["tiered"], ) return PriceResolution( unit_price=round(pli.unit_price, 4), price_source="price_list", price_rule_id=pli.id, discount_percent=0.0, requires_quote=False, label=_SOURCE_LABELS["price_list"], ) # 4. Base price (+ optional customer discount). if product.base_price is not None: discount = assignment.discount_percent if assignment else 0.0 discount = min(max(discount, 0.0), 100.0) unit_price = round(product.base_price * (1.0 - discount / 100.0), 4) label = _SOURCE_LABELS["base"] if discount: label = f"List price less {discount:g}%" return PriceResolution( unit_price=unit_price, price_source="base", price_rule_id=product.id, discount_percent=discount, requires_quote=False, label=label, ) # 5. No resolvable price ⇒ quote. return _quote("No price set — quote required")