from typing import Dict, List, Sequence, Tuple from .money_allocate import allocate from .money_amount import money from .money_apply_rate import apply_rate from .money_sum import sum_money 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), )