retail.promotion-apply

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),
    )