dodil/erp-brick-tax

DODIL ERP brick: tax

src/brick_tax/domain/rules.py

"""Tax rules: which code a (category, treatment, supply) combination gets, from a date (TAX-04). Dated rows, never edited.

A rule names a category (what is sold), a treatment (who buys) and a supply (goods | services | any); `any` also matches
in the category and treatment columns ("(reduced, any) -> R"). Ending a rule is a new row from tomorrow with no code.
When several rules match a line, the one with the higher priority wins, then the more specific, then the later start.
A pair no rule covers is refused: it is never defaulted to 0%.
"""
from dataclasses import dataclass
from datetime import date
from typing import Iterable, Optional

from brick_tax.domain.values import ChangeState

ANY = "any"


@dataclass
class Rule:
    category_code: str
    treatment_code: str
    supply: str
    valid_from: date
    tax_code: str = ""                 # "" = from this date no rule covers the pair
    priority: int = 0
    valid_to: Optional[date] = None
    change_state: str = ChangeState.APPROVED.value
    created_by: str = ""
    approved_by: str = ""
    withdrawn_by: str = ""
    created_sub: str = ""              # the author's sub beside their email (created_by): either identity is the same person
    direction: str = "sale"            # sale | purchase (v3): a purchase pair is chosen only by purchase rules
    consumption: str = "any"           # v6 (TAX-51): eat_in | takeaway | any; part of the rule's key

    @property
    def scope(self) -> tuple:
        return (self.category_code, self.treatment_code, self.supply, self.consumption)

    def describe(self) -> str:
        by = "" if self.consumption == ANY else f", {self.consumption}"
        return f"({self.category_code}, {self.treatment_code}, {self.supply}{by}) from {self.valid_from}"


def live(rows: Iterable[Rule], scope: Optional[tuple] = None) -> list:
    return [r for r in rows if r.change_state != ChangeState.WITHDRAWN.value and (scope is None or r.scope == scope)]


def overlaps(a: tuple, b: tuple) -> bool:
    """Could one (category, treatment, supply) line match both scopes? `any` matches everything in its column."""
    return all(x == y or ANY in (x, y) for x, y in zip(a, b))


def covered(rows: Iterable[Rule], scope: tuple, start: date) -> bool:
    """Does some rule already choose a code for a pair this scope would touch, on `start` or on any date after it? Then the
    scope is not a first row: a new rule over it changes tax somebody is already getting."""
    near = [r for r in live(rows) if overlaps(r.scope, scope)]
    dates = {start} | {r.valid_from for r in near if r.valid_from > start}
    return any(in_force(near, d) for d in dates)


def in_force(rows: Iterable[Rule], on: date) -> list:
    """Per scope the latest APPROVED row that started by `on` (and has not ended); an ending row removes the scope."""
    latest: dict = {}
    for r in rows:
        if r.change_state == ChangeState.APPROVED.value and r.valid_from <= on:
            if r.scope not in latest or r.valid_from > latest[r.scope].valid_from:
                latest[r.scope] = r
    return [r for r in latest.values() if r.tax_code and (r.valid_to is None or r.valid_to >= on)]


def _matching(rows: Iterable[Rule], category: str, treatment: str, supply: str, on: date) -> list:
    return [r for r in in_force(rows, on) if r.category_code in (category, ANY) and r.treatment_code in (treatment, ANY) and r.supply in (supply, ANY)]


def distinguishes(rows: Iterable[Rule], category: str, treatment: str, supply: str, on: date) -> bool:
    """v6 (TAX-51): does a rule in force for this pair tell eat_in from takeaway? Then a line must say which."""
    return any(r.consumption != ANY for r in _matching(rows, category, treatment, supply, on))


def pick(rows: Iterable[Rule], category: str, treatment: str, supply: str, on: date, consumption: str = "") -> Optional[Rule]:
    """The rule that chooses a line's code, or None when the pair is not covered. A rule of consumption `any` covers both."""
    hits = []
    for r in _matching(rows, category, treatment, supply, on):
        if r.consumption not in (consumption, ANY):
            continue
        specificity = ((r.category_code == category) + (r.treatment_code == treatment) + (r.supply == supply and supply != ANY)
                       + (r.consumption == consumption and consumption != ""))
        hits.append((-r.priority, -specificity, -r.valid_from.toordinal(), r.scope, r))
    return min(hits, key=lambda h: h[:4])[4] if hits else None


def pointing_at(rows: Iterable[Rule], tax_code: str, today: date) -> int:
    """Rules in effect today or later that choose `tax_code`: the retire gate."""
    latest: dict = {}
    for r in rows:
        if r.change_state == ChangeState.WITHDRAWN.value:
            continue
        if r.scope not in latest:
            latest[r.scope] = []
        latest[r.scope].append(r)
    count = 0
    for scope_rows in latest.values():
        scope_rows.sort(key=lambda r: r.valid_from)
        in_effect = [r for r in scope_rows if r.valid_from <= today and r.change_state == ChangeState.APPROVED.value]
        candidates = ([in_effect[-1]] if in_effect else []) + [r for r in scope_rows if r.valid_from > today]
        if any(r.tax_code == tax_code for r in candidates):
            count += 1
    return count