bonamin/xenia-pos-local
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
34ae328b0d3c7132ecea636d8fea1efc81122c1f
xenia-pos-local/local_backend/services/pricing.py
bonamin 34ae328b0d feat: bump client-services (accumulated feature work + deploy fixes) 
Snapshot of in-progress work across local_backend, manager_dashboard,
and waiter_pwa (pricing, chat, fiscal, prep zones, recovery codes, CRM,
inventory, permissions), plus the nginx/docker-compose deploy fixes for
the Unraid + NPM reverse-proxy setup.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-19 10:00:14 +03:00

728 lines
28 KiB
Python
Raw Blame History

"""
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,
))
Reference in New Issue View Git Blame Copy Permalink