Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.
from typing import Dict, List, Sequence, Tuple
from .money_allocate import allocate ← from money.allocate ^1.0.0 · built alongside by fune
from .money_amount import money ← from money.amount ^1.0.0 · built alongside by fune
from .money_apply_rate import apply_rate ← from money.apply-rate ^1.0.0 · built alongside by fune
from .money_sum import sum_money ← from money.sum ^1.0.0 · built alongside by fune
from .retail_promotion_apply_types import AppliedPromotion, PricedLine, PromoLine, Promotion, PromotionResult
_MAX_COMPETING = 6
# A unit is (line index, unit price); a group is (unit indices, discount).
_Group = Tuple[List[int], int]
def _check_promotion(p: Promotion, currency: str) -> None:
if isinstance(p.group_size, bool) or not isinstance(p.group_size, int) or p.group_size < 1:
raise ValueError('promotion "%s": groupSize must be 1 or more' % p.id)
if p.kind == "free-items":
if p.free_items < 1 or p.free_items >= p.group_size:
raise ValueError('promotion "%s": freeItems must be from 1 to groupSize - 1' % p.id)
elif p.kind == "percent-off":
if p.basis_points < 1 or p.basis_points > 10000:
raise ValueError('promotion "%s": basisPoints must be from 1 to 10000' % p.id)
elif p.kind in ("amount-off", "group-price"):
if p.amount is None:
raise ValueError('promotion "%s" needs an amount' % p.id)
if p.amount.currency != currency:
raise ValueError("currency mismatch: %s and %s" % (currency, p.amount.currency))
if p.amount.minor < 0:
raise ValueError('promotion "%s": amount must not be negative' % p.id)
else:
raise ValueError('promotion "%s": unknown kind "%s"' % (p.id, p.kind))
def _evaluate(p: Promotion, units: List[Tuple[int, int]], skus: List[str], claimed: List[bool], currency: str) -> List[_Group]:
# The qualifying unclaimed units from dearest to cheapest (ties in basket
# order), cut into consecutive groups of group_size. Groups that would save
# nothing are skipped.
pool = [i for i, (line, _) in enumerate(units) if not claimed[i] and skus[line] in p.skus]
pool.sort(key=lambda i: (-units[i][1], i))
groups: List[_Group] = []
start = 0
while start + p.group_size <= len(pool):
members = pool[start : start + p.group_size]
prices = [units[i][1] for i in members]
total = sum(prices)
if p.kind == "free-items":
discount = sum(prices[p.group_size - p.free_items :])
elif p.kind == "percent-off":
discount = apply_rate(money(total, currency), p.basis_points, "half-up").minor
elif p.kind == "amount-off":
discount = min(p.amount.minor, total)
else:
discount = max(total - p.amount.minor, 0)
if discount > 0:
groups.append((members, discount))
start += p.group_size
return groups
def _run(order, promotions, units, skus, claimed, currency):
groups: Dict[int, List[_Group]] = {}
total = 0
for index in order:
made = _evaluate(promotions[index], units, skus, claimed, currency)
for members, discount in made:
for u in members:
claimed[u] = True
total += discount
groups[index] = made
return total, groups
def _next_permutation(a: List[int]) -> bool:
i = len(a) - 2
while i >= 0 and a[i] >= a[i + 1]:
i -= 1
if i < 0:
return False
j = len(a) - 1
while a[j] <= a[i]:
j -= 1
a[i], a[j] = a[j], a[i]
a[i + 1 :] = reversed(a[i + 1 :])
return True
def apply_promotions(lines: Sequence[PromoLine], promotions: Sequence[Promotion]) -> PromotionResult:
"""Price a basket under competing promotions: the customer gets the order
of application that saves the most, and each deal's saving is allocated
back to the lines in its groups exactly.
"""
if len(lines) == 0:
raise ValueError("a basket needs at least one line")
currency = lines[0].unit_price.currency
units: List[Tuple[int, int]] = []
skus = [line.sku for line in lines]
for index, line in enumerate(lines):
if line.unit_price.currency != currency:
raise ValueError("currency mismatch: %s and %s" % (currency, line.unit_price.currency))
if isinstance(line.quantity, bool) or not isinstance(line.quantity, int) or line.quantity < 1:
raise ValueError("quantity must be 1 or more, received %r" % (line.quantity,))
if line.unit_price.minor < 0:
raise ValueError("unitPrice must not be negative, received %d" % line.unit_price.minor)
units.extend((index, line.unit_price.minor) for _ in range(line.quantity))
seen = set()
for p in promotions:
if p.id in seen:
raise ValueError('duplicate promotion id "%s"' % p.id)
seen.add(p.id)
_check_promotion(p, currency)
# Promotions with something to act on, joined into sets that share a SKU
# present in the basket. Only promotions in the same set compete.
basket_skus = set(skus)
relevant = [i for i, p in enumerate(promotions) if any(s in basket_skus for s in p.skus)]
parent = {i: i for i in relevant}
def find(i: int) -> int:
while parent[i] != i:
i = parent[i]
return i
for x in range(len(relevant)):
for y in range(x + 1, len(relevant)):
a = promotions[relevant[x]]
b = promotions[relevant[y]]
if any(s in basket_skus and s in b.skus for s in a.skus):
ra, rb = find(relevant[x]), find(relevant[y])
if ra != rb:
parent[max(ra, rb)] = min(ra, rb)
components: List[List[int]] = []
by_root: Dict[int, List[int]] = {}
for i in relevant:
root = find(i)
if root not in by_root:
by_root[root] = []
components.append(by_root[root])
by_root[root].append(i)
claimed = [False] * len(units)
chosen: Dict[int, List[_Group]] = {}
for component in components:
if len(component) > _MAX_COMPETING:
raise ValueError(
"more than %d promotions compete for the same items: %s"
% (_MAX_COMPETING, ", ".join(promotions[i].id for i in component))
)
order = list(component)
best_order = list(order)
best_total = -1
while True:
total, _ = _run(order, promotions, units, skus, list(claimed), currency)
if total > best_total:
best_total = total
best_order = list(order)
if not _next_permutation(order):
break
_, groups = _run(best_order, promotions, units, skus, claimed, currency)
chosen.update(groups)
line_discount = [0] * len(lines)
applied: List[AppliedPromotion] = []
for index, p in enumerate(promotions):
made = chosen.get(index, [])
if not made:
continue
saved = 0
for members, discount in made:
shares = allocate(money(discount, currency), [units[u][1] for u in members])
for u, share in zip(members, shares):
line_discount[units[u][0]] += share.minor
saved += discount
applied.append(AppliedPromotion(id=p.id, groups=len(made), discount=money(saved, currency)))
priced = []
for index, line in enumerate(lines):
gross = line.unit_price.minor * line.quantity
priced.append(
PricedLine(
sku=line.sku,
quantity=line.quantity,
gross=money(gross, currency),
discount=money(line_discount[index], currency),
net=money(gross - line_discount[index], currency),
)
)
subtotal = sum_money([l.gross for l in priced], currency)
discount = sum_money([l.discount for l in priced], currency)
return PromotionResult(
lines=priced,
applied=applied,
subtotal=subtotal,
discount=discount,
total=money(subtotal.minor - discount.minor, currency),
)