""" Pricing resolution engine. resolve_price() is called once per OrderItem at add-to-order time. The result is the final snapshotted unit_price — it never changes after that. Stacking algorithm (per design spec): 1. Collect all passing modifiers for this product (item-scoped first, then global). 2. Separate into stackable (allow_stack=1) and non-stackable piles. 3. From the non-stackable pile, apply only the first passing one (lowest sort_order). 4. From the stackable pile, apply ALL passing ones in sort_order order, each on the running price result. 5. Apply system-wide rounding to nearest 0.10 as a final pass. Deal evaluation: evaluate_deals() is called after items are added. It returns a list of DealOffer objects that the API caller surfaces to the waiter as prompts. """ import json from dataclasses import dataclass, field from datetime import datetime, timezone, time as dtime from typing import Optional from sqlalchemy.orm import Session from tz import to_local # --------------------------------------------------------------------------- # Data structures (no ORM dependency — safe to import anywhere) # --------------------------------------------------------------------------- @dataclass class OrderContext: """Snapshot of order-level facts needed to evaluate conditions.""" channel: str # 'pos' | 'online' | 'qr' | 'takeaway' now: datetime # current UTC time cart_items: list # list of {product_id, category_id, quantity, unit_price} active_price_group_ids: set # set of PriceGroup.id that are currently active user_tier: Optional[str] = None # 'bronze'|'silver'|'gold' — placeholder @dataclass class PriceEvent: """In-memory representation of a price event, before it's written to DB.""" order_id: int order_item_id: Optional[int] # set after OrderItem is committed event_type: str modifier_id: Optional[int] = None deal_id: Optional[int] = None price_before: Optional[float] = None price_after: Optional[float] = None delta_amount: Optional[float] = None applied_by_user_id: Optional[int] = None waiter_note: Optional[str] = None conditions_snapshot: Optional[dict] = None selected_item_ids: Optional[list] = None @dataclass class DealOffer: """Returned to the API when a deal is triggered and needs waiter confirmation.""" deal_id: int deal_name: str action_type: str # mirrors Deal.action_type # For free_item: the single product to add free_item_id: Optional[int] = None free_item_name: Optional[str] = None # For free_choice: pool of selectable products free_choices: list = field(default_factory=list) # [{id, name}] free_quantity: int = 1 # For price-modifying deals applied to a specific item target_order_item_id: Optional[int] = None price_before: Optional[float] = None price_after: Optional[float] = None # --------------------------------------------------------------------------- # Condition evaluators # --------------------------------------------------------------------------- def _eval_condition(ctype: str, params: dict, ctx: OrderContext, product_id: int, quantity: float) -> tuple[bool, dict]: """ Evaluate one condition. Returns (passed, snapshot_dict). snapshot_dict records the actual runtime value — written to price_event_log for audit. Unknown or placeholder conditions pass silently (forward-compatible). """ now_local = to_local(ctx.now) current_time = now_local.time() current_date = now_local.date() current_weekday = now_local.weekday() # Mon=0 Sun=6 if ctype == "time_range": t_from = dtime.fromisoformat(params["from"]) t_to = dtime.fromisoformat(params["to"]) if t_from <= t_to: passed = t_from <= current_time <= t_to else: # Crosses midnight passed = current_time >= t_from or current_time <= t_to return passed, {"time_range": current_time.strftime("%H:%M")} if ctype == "date_range": from datetime import date d_from = date.fromisoformat(params["from"]) d_to = date.fromisoformat(params["to"]) passed = d_from <= current_date <= d_to return passed, {"date_range": current_date.isoformat()} if ctype == "specific_date": from datetime import date passed = current_date.isoformat() in params.get("dates", []) return passed, {"specific_date": current_date.isoformat()} if ctype == "day_of_week": passed = current_weekday in params.get("days", []) return passed, {"day_of_week": current_weekday} if ctype == "min_item_quantity": cart_qty = sum( ci["quantity"] for ci in ctx.cart_items if ci["product_id"] == product_id ) passed = cart_qty >= params.get("min", 1) return passed, {"min_item_quantity": cart_qty} if ctype == "min_category_qty": cat_id = params.get("category_id") cart_qty = sum( ci["quantity"] for ci in ctx.cart_items if ci.get("category_id") == cat_id ) passed = cart_qty >= params.get("min", 1) return passed, {"min_category_qty": cart_qty} if ctype == "min_order_value": total = sum(ci["quantity"] * ci["unit_price"] for ci in ctx.cart_items) passed = total >= params.get("min", 0.0) return passed, {"min_order_value": round(total, 2)} if ctype == "order_channel": passed = ctx.channel in params.get("channels", []) return passed, {"order_channel": ctx.channel} if ctype == "price_group_active": pgid = params.get("price_group_id") passed = pgid in ctx.active_price_group_ids return passed, {"price_group_active": passed} if ctype == "user_tier": passed = ctx.user_tier in params.get("tiers", []) return passed, {"user_tier": ctx.user_tier} if ctype == "low_stock": # Placeholder until inventory is implemented — always passes silently return True, {"low_stock": "placeholder"} # Unknown condition type — pass silently for forward compatibility return True, {ctype: "unknown_type_skipped"} def _all_conditions_pass(conditions, ctx: OrderContext, product_id: int, quantity: float) -> tuple[bool, dict]: """AND-evaluate all conditions. Returns (all_passed, merged_snapshot).""" snapshot = {} for cond in conditions: params = json.loads(cond.params) if isinstance(cond.params, str) else cond.params passed, snap = _eval_condition(cond.condition_type, params, ctx, product_id, quantity) snapshot.update(snap) if not passed: return False, snapshot return True, snapshot # --------------------------------------------------------------------------- # Modifier action math # --------------------------------------------------------------------------- def _apply_action(price: float, action_type: str, action_value: float, round_to: Optional[str]) -> float: if action_type == "set": result = action_value elif action_type == "add_amount": result = price + action_value elif action_type == "add_percent": result = price * (1 + action_value / 100.0) else: result = price # unknown action — no-op result = max(result, 0.0) # price can never go below 0 if round_to: result = _round_to(result, round_to) return result def _round_to(value: float, mode: str) -> float: if mode == "0.05": return round(value / 0.05) * 0.05 if mode == "0.10": return round(value / 0.10) * 0.10 if mode == "0.20": return round(value / 0.20) * 0.20 if mode == "0.50": return round(value / 0.50) * 0.50 if mode == "x.99": return float(int(value)) + 0.99 if value >= 1 else value if mode == "x.00": return float(round(value)) return value def _system_round(value: float) -> float: """System-wide final rounding: nearest 0.10.""" return round(round(value / 0.10) * 0.10, 2) # --------------------------------------------------------------------------- # PriceGroup auto-schedule evaluation # --------------------------------------------------------------------------- def _price_group_is_active(pg, now: datetime) -> bool: """ Returns True if this price group should currently be considered active. Checks is_active flag, then auto-schedule if configured. """ if pg.auto_enable_time and pg.auto_disable_time and pg.auto_days: now_local = now.astimezone() weekday = now_local.weekday() try: auto_days = json.loads(pg.auto_days) except (ValueError, TypeError): auto_days = [] if weekday in auto_days: t_enable = dtime.fromisoformat(pg.auto_enable_time) t_disable = dtime.fromisoformat(pg.auto_disable_time) current = now_local.time() if t_enable <= t_disable: return t_enable <= current <= t_disable else: return current >= t_enable or current <= t_disable return bool(pg.is_active) def build_active_price_group_ids(db: Session, now: datetime) -> set: """Fetch all price groups once per request and return active IDs.""" from models.pricing import PriceGroup groups = db.query(PriceGroup).all() return {pg.id for pg in groups if _price_group_is_active(pg, now)} # --------------------------------------------------------------------------- # Modifier target matching # --------------------------------------------------------------------------- def _parse_ids(raw) -> list: if not raw: return [] if isinstance(raw, list): return raw try: return json.loads(raw) except Exception: return [] def _modifier_targets_product(modifier, product, db: Session) -> bool: """Return True if any target row on a global modifier covers this product.""" from models.product import Product for t in modifier.targets: if t.target_type == "all": return True if t.target_type == "item": ids = _parse_ids(t.target_ids) or ([t.target_id] if t.target_id else []) if product.id in ids: return True if t.target_type == "category": ids = _parse_ids(t.target_ids) or ([t.target_id] if t.target_id else []) if product.category_id in ids: return True if t.target_type == "prep_zone": ids = _parse_ids(t.target_ids) or ([t.target_id] if t.target_id else []) zone_ids = {pz.id for pz in product.prep_zones} if any(i in zone_ids for i in ids): return True if t.target_type == "tag": prod_tags = set(json.loads(product.tags) if product.tags else []) tags = _parse_ids(t.target_tags) or ([t.target_tag] if t.target_tag else []) if any(tag in prod_tags for tag in tags): return True return False # --------------------------------------------------------------------------- # Main resolver # --------------------------------------------------------------------------- def resolve_price( product, quantity: float, options_extra: float, ctx: OrderContext, db: Session, ) -> tuple[float, list[PriceEvent]]: """ Compute the final unit_price for one order item. Returns (final_unit_price, list_of_PriceEvent_to_log). PriceEvents have order_item_id=None — caller sets it after the OrderItem is committed. order_id must be set on the PriceEvents before writing them; the caller does this. """ from models.pricing import PriceModifier base = product.base_price + options_extra events: list[PriceEvent] = [] # --- Load modifiers --- # Item-scoped modifiers for this product item_modifiers = ( db.query(PriceModifier) .filter( PriceModifier.scope == "item", PriceModifier.item_id == product.id, PriceModifier.is_active == 1, ) .order_by(PriceModifier.sort_order) .all() ) # Global modifiers (all active ones, filtered by target in Python) all_global = ( db.query(PriceModifier) .filter(PriceModifier.scope == "global", PriceModifier.is_active == 1) .order_by(PriceModifier.sort_order) .all() ) global_modifiers = [ m for m in all_global if _modifier_targets_product(m, product, db) ] # Item-scoped take priority: evaluated first, then global appended all_modifiers = item_modifiers + global_modifiers if not all_modifiers: return _system_round(base), events # --- Separate stackable vs non-stackable --- stackable = [] non_stackable = [] for m in all_modifiers: passed, snapshot = _all_conditions_pass(m.conditions, ctx, product.id, quantity) if passed: if m.allow_stack: stackable.append((m, snapshot)) else: non_stackable.append((m, snapshot)) price = base # --- Apply first non-stackable --- if non_stackable: m, snapshot = non_stackable[0] price_before = price price = _apply_action(price, m.action_type, m.action_value, m.round_to) events.append(PriceEvent( order_id=0, # caller will fill in order_item_id=None, event_type="modifier_applied", modifier_id=m.id, price_before=price_before, price_after=price, delta_amount=round(price - price_before, 4), conditions_snapshot=snapshot, )) # --- Apply all stackable modifiers --- for m, snapshot in stackable: price_before = price price = _apply_action(price, m.action_type, m.action_value, m.round_to) events.append(PriceEvent( order_id=0, order_item_id=None, event_type="modifier_applied", modifier_id=m.id, price_before=price_before, price_after=price, delta_amount=round(price - price_before, 4), conditions_snapshot=snapshot, )) final = _system_round(price) # Update the last event's price_after to reflect the final rounded value if events: events[-1].price_after = final events[-1].delta_amount = round(final - events[-1].price_before, 4) return final, events # --------------------------------------------------------------------------- # Deal evaluation # --------------------------------------------------------------------------- def _deal_targets_met(deal, ctx: OrderContext, db: Session) -> bool: """Return True if the cart contains items that satisfy the deal's target rules.""" from models.product import Product if not deal.targets: return True # no target = always matches for target in deal.targets: if target.target_type == "any": if ctx.cart_items: return True if target.target_type == "item": ids = _parse_ids(target.target_ids) or ([target.target_id] if target.target_id else []) if any(ci["product_id"] in ids for ci in ctx.cart_items): return True if target.target_type == "category": ids = _parse_ids(target.target_ids) or ([target.target_id] if target.target_id else []) if any(ci.get("category_id") in ids for ci in ctx.cart_items): return True if target.target_type == "tag": tags = _parse_ids(target.target_tags) or ([target.target_tag] if target.target_tag else []) if tags: for ci in ctx.cart_items: prod = db.query(Product).filter(Product.id == ci["product_id"]).first() if prod: prod_tags = set(json.loads(prod.tags) if prod.tags else []) if any(tag in prod_tags for tag in tags): return True return False def evaluate_deals( order_id: int, ctx: OrderContext, already_fired_deal_ids: set, db: Session, ) -> list[DealOffer]: """ Check all active deals against the current cart. Returns DealOffer objects for deals that are newly triggered (not already fired this order). already_fired_deal_ids: set of deal IDs that have already been offered/applied on this order (derived from price_event_log by the caller). """ from models.pricing import Deal from models.product import Product deals = db.query(Deal).filter(Deal.is_active == 1).order_by(Deal.sort_order).all() offers: list[DealOffer] = [] for deal in deals: if deal.id in already_fired_deal_ids: continue # Check conditions conditions_pass, _ = _all_conditions_pass(deal.conditions, ctx, 0, 0) if not conditions_pass: continue # Check cart targets if not _deal_targets_met(deal, ctx, db): continue # Build the offer offer = DealOffer( deal_id=deal.id, deal_name=deal.name, action_type=deal.action_type, ) if deal.action_type == "free_item" and deal.action_free_item_id: prod = db.query(Product).filter(Product.id == deal.action_free_item_id).first() offer.free_item_id = deal.action_free_item_id offer.free_item_name = prod.name if prod else None offer.free_quantity = deal.action_free_quantity elif deal.action_type == "free_choice" and deal.action_free_target_ids: try: target_ids = json.loads(deal.action_free_target_ids) except (ValueError, TypeError): target_ids = [] if deal.action_free_target_type == "item": prods = db.query(Product).filter(Product.id.in_(target_ids)).all() offer.free_choices = [{"id": p.id, "name": p.name} for p in prods] elif deal.action_free_target_type == "category": prods = db.query(Product).filter( Product.category_id.in_(target_ids), Product.lifecycle_status == "active", Product.is_available == True, # noqa: E712 ).all() offer.free_choices = [{"id": p.id, "name": p.name} for p in prods] elif deal.action_free_target_type == "tag": all_prods = db.query(Product).filter( Product.lifecycle_status == "active", Product.is_available == True, # noqa: E712 ).all() offer.free_choices = [ {"id": p.id, "name": p.name} for p in all_prods if any(tag in (json.loads(p.tags) if p.tags else []) for tag in target_ids) ] offer.free_quantity = deal.action_free_quantity offers.append(offer) return offers # --------------------------------------------------------------------------- # Waiter discount limit enforcement # --------------------------------------------------------------------------- def check_waiter_discount_limits( waiter_id: int, order_id: int, item_ids: list[int], discount_percent: float, item_unit_prices: dict[int, float], # {order_item_id: current_unit_price} db: Session, ) -> tuple[bool, str]: """ Validate that applying the given discount does not exceed any configured limit. Returns (ok, error_message). error_message is empty string when ok=True. """ from models.pricing import WaiterDiscountSettings from models.order import OrderDiscount from models.settings import PosSettings from models.business_day import BusinessDay from models.shift import WaiterShift # --- Global kill-switch --- global_row = db.query(PosSettings).filter(PosSettings.key == "discounts.enabled").first() if global_row and global_row.value == "false": return False, "Discounts are globally disabled" # --- Per-waiter settings --- settings = db.query(WaiterDiscountSettings).filter( WaiterDiscountSettings.user_id == waiter_id ).first() if not settings or not settings.can_apply_discounts: return False, "You do not have permission to apply discounts" # --- Validate percent limits --- if settings.max_discount_percent is not None: if discount_percent > settings.max_discount_percent: return False, f"Discount exceeds your maximum of {settings.max_discount_percent:.0f}%" # --- Compute proposed euro amounts --- proposed_amounts = { item_id: round(unit_price * discount_percent / 100.0, 2) for item_id, unit_price in item_unit_prices.items() if item_id in item_ids } proposed_total = sum(proposed_amounts.values()) proposed_count = len(item_ids) # --- Per-item amount limit --- if settings.max_discount_amount is not None: for item_id, amount in proposed_amounts.items(): if amount > settings.max_discount_amount: return False, f"Single-item discount of €{amount:.2f} exceeds your limit of €{settings.max_discount_amount:.2f}" # --- Items per order --- if settings.max_items_per_order is not None: existing_count_order = db.query(OrderDiscount).filter( OrderDiscount.order_id == order_id ).count() if existing_count_order + proposed_count > settings.max_items_per_order: remaining = settings.max_items_per_order - existing_count_order return False, f"You can only discount {remaining} more item(s) on this order" # --- Current workday --- bd = db.query(BusinessDay).filter(BusinessDay.closed_at == None).first() # noqa: E711 if bd: # Items per workday if settings.max_items_per_workday is not None: from models.order import Order wd_order_ids = [ o.id for o in db.query(Order).filter(Order.business_day_id == bd.id).all() ] wd_count = db.query(OrderDiscount).filter( OrderDiscount.applied_by == waiter_id, OrderDiscount.order_id.in_(wd_order_ids), ).count() if wd_count + proposed_count > settings.max_items_per_workday: remaining = settings.max_items_per_workday - wd_count return False, f"You can only discount {remaining} more item(s) today" # Total value per workday if settings.max_total_value_workday is not None: from models.order import Order wd_order_ids = [ o.id for o in db.query(Order).filter(Order.business_day_id == bd.id).all() ] existing_rows = db.query(OrderDiscount).filter( OrderDiscount.applied_by == waiter_id, OrderDiscount.order_id.in_(wd_order_ids), ).all() existing_wd_total = sum(_discount_euro(r) for r in existing_rows) if existing_wd_total + proposed_total > settings.max_total_value_workday: remaining = settings.max_total_value_workday - existing_wd_total return False, f"You have only €{remaining:.2f} of discount budget left today" # --- Current shift --- shift = db.query(WaiterShift).filter( WaiterShift.waiter_id == waiter_id, WaiterShift.ended_at == None, # noqa: E711 ).order_by(WaiterShift.started_at.desc()).first() if shift: # Items per shift if settings.max_items_per_shift is not None: from models.order import Order shift_order_ids = [ o.id for o in db.query(Order).filter( Order.opened_at >= shift.started_at ).all() ] shift_count = db.query(OrderDiscount).filter( OrderDiscount.applied_by == waiter_id, OrderDiscount.order_id.in_(shift_order_ids), ).count() if shift_count + proposed_count > settings.max_items_per_shift: remaining = settings.max_items_per_shift - shift_count return False, f"You can only discount {remaining} more item(s) this shift" # Total value per shift if settings.max_total_value_shift is not None: from models.order import Order shift_order_ids = [ o.id for o in db.query(Order).filter( Order.opened_at >= shift.started_at ).all() ] existing_rows = db.query(OrderDiscount).filter( OrderDiscount.applied_by == waiter_id, OrderDiscount.order_id.in_(shift_order_ids), ).all() existing_shift_total = sum(_discount_euro(r) for r in existing_rows) if existing_shift_total + proposed_total > settings.max_total_value_shift: remaining = settings.max_total_value_shift - existing_shift_total return False, f"You have only €{remaining:.2f} of discount budget left this shift" # --- Global limits from pos_settings --- def _gs(key): row = db.query(PosSettings).filter(PosSettings.key == key).first() return float(row.value) if row and row.value else None g_max_wd_value = _gs("discounts.max_total_value_workday") g_max_shift_val = _gs("discounts.max_total_value_shift") g_max_shift_cnt = _gs("discounts.max_items_per_shift") if bd and g_max_wd_value is not None: from models.order import Order wd_order_ids = [o.id for o in db.query(Order).filter(Order.business_day_id == bd.id).all()] all_wd_rows = db.query(OrderDiscount).filter( OrderDiscount.order_id.in_(wd_order_ids) ).all() all_wd_total = sum(_discount_euro(r) for r in all_wd_rows) if all_wd_total + proposed_total > g_max_wd_value: remaining = g_max_wd_value - all_wd_total return False, f"Store-wide discount budget has only €{remaining:.2f} left today" if shift and (g_max_shift_val is not None or g_max_shift_cnt is not None): from models.order import Order shift_order_ids = [ o.id for o in db.query(Order).filter(Order.opened_at >= shift.started_at).all() ] all_shift_rows = db.query(OrderDiscount).filter( OrderDiscount.order_id.in_(shift_order_ids) ).all() if g_max_shift_val is not None: total = sum(_discount_euro(r) for r in all_shift_rows) if total + proposed_total > g_max_shift_val: remaining = g_max_shift_val - total return False, f"Store-wide shift discount budget has only €{remaining:.2f} left" if g_max_shift_cnt is not None: count = len(all_shift_rows) if count + proposed_count > int(g_max_shift_cnt): remaining = int(g_max_shift_cnt) - count return False, f"Store-wide shift item limit: only {remaining} discount(s) left" return True, "" def _discount_euro(row) -> float: """Convert an OrderDiscount row to a euro amount.""" if row.price_before is not None and row.price_after is not None: return max(0.0, row.price_before - row.price_after) # Fallback for legacy rows without price_before/after if row.discount_type == "fixed": return row.discount_value return 0.0 # percent rows without snapshots can't be computed retroactively # --------------------------------------------------------------------------- # Log helpers # --------------------------------------------------------------------------- def write_price_events(events: list[PriceEvent], order_id: int, order_item_id: int, db: Session): """Persist PriceEvent objects to price_event_log. Call after OrderItem is committed.""" from models.pricing import PriceEventLog for ev in events: db.add(PriceEventLog( order_id=order_id, order_item_id=order_item_id, event_type=ev.event_type, modifier_id=ev.modifier_id, deal_id=ev.deal_id, price_before=ev.price_before, price_after=ev.price_after, delta_amount=ev.delta_amount, applied_by_user_id=ev.applied_by_user_id, waiter_note=ev.waiter_note, conditions_snapshot=json.dumps(ev.conditions_snapshot) if ev.conditions_snapshot else None, selected_item_ids=json.dumps(ev.selected_item_ids) if ev.selected_item_ids else None, ))