dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/domain/schedule.py
"""The Schedule aggregate: one sold line billed over time.
Which period is due, how many are billed and what remains are DERIVED from starts_on, periods and the invoices raised;
nothing increments a counter. A schedule ends early only through stop_billing (dated today, never earlier).
"""
from dataclasses import dataclass, field
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from brick_ar.domain import events
from brick_ar.domain.dates import add_months, end_of_month, first_of_next_month
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.values import BILLING_KINDS, ScheduleState
STEP = {"monthly": 1, "periodic": 1, "quarterly": 3, "annual": 12}
@dataclass
class Schedule:
schedule_id: str
bp_id: int = 0
source_kind: str = ""
source_id: str = ""
line_id: str = ""
product_id: str = ""
description: str = ""
billing_kind: str = "recurring"
billing_plan: str = "monthly"
starts_on: Optional[date] = None
periods: int = 1
quantity: Optional[Decimal] = Decimal(1) # None on a usage schedule: the biller fills each period's quantity in
unit_price: Decimal = Decimal(0)
discount_pct: Decimal = Decimal(0)
unit_code: str = ""
tax_category: str = ""
tax_code: str = ""
currency: str = ""
in_advance: bool = True
billing_day: str = "anniversary" # the settings in force when it was planned: a later change never moves a plan
prorate_first: bool = False
status: str = ScheduleState.ACTIVE.value
stopped_at: Optional[date] = None
stop_reason: str = ""
created_by: str = ""
events: list = field(default_factory=list, repr=False, compare=False)
@classmethod
def plan(cls, schedule_id: str, *, by: str, **fields) -> "Schedule":
s = cls(schedule_id=schedule_id, created_by=by, **fields)
if s.billing_kind not in BILLING_KINDS:
raise RuleBroken(f"a line that repeats is {', '.join(BILLING_KINDS)}; {s.billing_kind!r} is billed once through raise_draft")
if s.billing_plan not in STEP:
raise RuleBroken(f"a billing plan is {', '.join(STEP)}, not {s.billing_plan!r}")
if s.periods < 1 or s.periods > 600:
raise RuleBroken("a schedule has between 1 and 600 periods")
if s.starts_on is None:
raise RuleBroken("a schedule starts on the date its source line records")
if s.billing_kind == "usage":
s.quantity = None
return s
# ── derived: nothing here is counted ─────────────────────────────────────────────────────────
@property
def step(self) -> int:
return STEP[self.billing_plan]
def bounds(self, k: int) -> tuple:
"""(first day, last day) of period k (1-based)."""
if not 1 <= k <= self.periods:
raise RuleBroken(f"{self.schedule_id} has periods 1 to {self.periods}, not {k}")
if self.billing_day == "month_start" and self.starts_on.day != 1:
if k == 1:
return self.starts_on, end_of_month(self.starts_on)
start = add_months(first_of_next_month(self.starts_on), (k - 2) * self.step)
return start, add_months(start, self.step) - timedelta(days=1)
start = add_months(self.starts_on, (k - 1) * self.step)
return start, add_months(self.starts_on, k * self.step) - timedelta(days=1)
def billing_date(self, k: int) -> date:
first, last = self.bounds(k)
return first if self.in_advance else last
def proration(self, k: int) -> Decimal:
"""The share of a full period a part first period covers (1 unless prorate_first and billing at month starts)."""
if k != 1 or not self.prorate_first or self.billing_day != "month_start" or self.starts_on.day == 1:
return Decimal(1)
first, last = self.bounds(1)
month_days = end_of_month(first).day
return Decimal((last - first).days + 1) / Decimal(month_days)
def due_periods(self, today: date, raised: set) -> list:
"""The periods whose billing day has come (today or earlier) and that nobody has raised. A period whose day is
already past is raised in the next pass, covering its own period. A stopped schedule raises only the period in
progress when it stopped and the ones before it."""
if self.status == ScheduleState.FINISHED.value:
return []
out = []
for k in range(1, self.periods + 1):
if k in raised or self.billing_date(k) > today:
continue
if self.status == ScheduleState.STOPPED.value and self.bounds(k)[0] > self.stopped_at:
continue
out.append(k)
return out
def next_due(self, raised: set) -> Optional[date]:
if self.status != ScheduleState.ACTIVE.value:
return None
return next((self.billing_date(k) for k in range(1, self.periods + 1) if k not in raised), None)
def complete(self, raised: set) -> bool:
return all(k in raised for k in range(1, self.periods + 1))
# ── transitions ──────────────────────────────────────────────────────────────────────────────
def stop(self, *, today: date, reason: str, by: str) -> None:
"""Billing stops after the period in progress; issued invoices are untouched (money owed back is a credit note)."""
if self.status == ScheduleState.STOPPED.value:
return
if self.status != ScheduleState.ACTIVE.value:
raise RuleBroken(f"schedule {self.schedule_id} is {self.status}")
self.status, self.stopped_at, self.stop_reason = ScheduleState.STOPPED.value, today, (reason or "").strip()
self.events.append(events.ScheduleStopped(by=by, schedule_id=self.schedule_id, reason=self.stop_reason))
def finish(self, *, raised: set, by: str) -> bool:
"""Finished once every period is raised; returns whether it changed."""
if self.status != ScheduleState.ACTIVE.value:
return False
if not self.complete(raised):
return False
self.status = ScheduleState.FINISHED.value
self.events.append(events.ScheduleFinished(by=by, schedule_id=self.schedule_id))
return True