Functional Weave
Code in TypeScript

charity.reclaim-batch@1.0.1

impl/python.py

3,944 bytes · the Python implementation · view raw

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, Optional, Sequence

from .charity_gift_aid import gift_aid  ← from charity.gift-aid ^1.0.0 · built alongside by fune
from .charity_reclaim_batch_data import AGGREGATION  ← this capability’s own data, compiled from data/aggregation.json into the same file by fune build
from .charity_reclaim_batch_types import ClaimLine, ExcludedDonation, GiftAidClaimBatch, GiftAidDonation
from .dates_uk_tax_year import uk_tax_year  ← from dates.uk-tax-year ^1.0.0 · built alongside by fune
from .money_amount import money  ← from money.amount ^1.0.0 · built alongside by fune


def gift_aid_claim_batch(
    donations: Sequence[GiftAidDonation],
    period_start: str,
    period_end: str,
    aggregate_small: bool,
) -> GiftAidClaimBatch:
    """The totals of a Gift Aid repayment claim for a period. Every donation is
    either on the claim or listed as excluded with its reason, lines come out
    in a fixed order (donor lines by donorId, then aggregated lines by date),
    and the tax is worked out on each tax year's total.
    """
    uk_tax_year(period_start)
    uk_tax_year(period_end)
    if period_end < period_start:
        raise ValueError("periodEnd %s is before periodStart %s" % (period_end, period_start))
    rule = AGGREGATION[0]
    ids = set()
    excluded: List[ExcludedDonation] = []
    eligible: List[GiftAidDonation] = []
    for d in donations:
        if d.id in ids:
            raise ValueError('duplicate donation id "%s"' % (d.id,))
        ids.add(d.id)
        if d.amount.currency != "GBP":
            raise ValueError('donation "%s" must be in GBP, received %s' % (d.id, d.amount.currency))
        uk_tax_year(d.date)
        reason: Optional[str] = None
        if d.date < period_start or d.date > period_end:
            reason = "outside-period"
        elif d.amount.minor <= 0:
            reason = "not-positive"
        elif not d.has_declaration:
            reason = "no-declaration"
        elif d.donor_id is None:
            reason = "no-donor"
        if reason is None:
            eligible.append(d)
        else:
            excluded.append(ExcludedDonation(id=d.id, reason=reason))

    small = [d for d in eligible if d.amount.minor <= rule.max_donation] if aggregate_small else []
    named = [d for d in eligible if d.amount.minor > rule.max_donation] if aggregate_small else eligible

    by_donor: Dict[str, List[GiftAidDonation]] = {}
    for d in named:
        by_donor.setdefault(d.donor_id, []).append(d)
    lines: List[ClaimLine] = []
    for donor_id in sorted(by_donor.keys()):
        group = by_donor[donor_id]
        lines.append(
            ClaimLine(
                kind="donor",
                donor_id=donor_id,
                last_date=max(d.date for d in group),
                count=len(group),
                total=money(sum(d.amount.minor for d in group), "GBP"),
            )
        )

    # Aggregated lines fill in date order (ties by id) up to the line limit.
    current: List[GiftAidDonation] = []
    current_total = 0

    def flush() -> None:
        nonlocal current, current_total
        if not current:
            return
        lines.append(
            ClaimLine(kind="aggregated", donor_id=None, last_date=current[-1].date, count=len(current), total=money(current_total, "GBP"))
        )
        current = []
        current_total = 0

    for d in sorted(small, key=lambda x: (x.date, x.id)):
        if current_total + d.amount.minor > rule.max_line_total:
            flush()
        current.append(d)
        current_total += d.amount.minor
    flush()

    # One rounding per tax year, since each year has its own basic rate.
    by_year: Dict[str, int] = {}
    for d in eligible:
        start = uk_tax_year(d.date).start
        by_year[start] = by_year.get(start, 0) + d.amount.minor
    claimable = 0
    for start in sorted(by_year.keys()):
        claimable += gift_aid(money(by_year[start], "GBP"), start).tax_reclaimable.minor

    return GiftAidClaimBatch(
        lines=lines,
        eligible_count=len(eligible),
        eligible_total=money(sum(d.amount.minor for d in eligible), "GBP"),
        claimable=money(claimable, "GBP"),
        excluded=excluded,
    )