dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
src/brick_timesheets/domain/billing.py
"""The billing run (TS-13/TS-14): approved billable hours of a project billed through ar, as the DIFFERENCE from what earlier runs already took.
Plain Python - no SQLAlchemy, FastAPI or MCP. A line is billed once: the claim row of (sheet, revision, line, 'ar_run') is the race, so a re-run, a
retry or two workers bill it at most once. A correction (a later approved revision of the week) bills the difference, negative when hours were taken
away or the line was made non-billable (a write-off is such a correction: there is no other way to unbill hours). The price is the product's on the
line's first work day; a negative difference is priced at the price the line was first billed at, so a correction nets to the cent."""
import re
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_timesheets.domain.errors import RuleBroken
from brick_timesheets.domain.money import ZERO, hours4, money
from brick_timesheets.domain.sheet import Identity, Sheet
PERIOD = re.compile(r"^(\d{4})-(0[1-9]|1[0-2])$")
PURPOSE = "ar_run"
UNIT = "HUR"
PRICE_MAX = Decimal("1e9")
AMOUNT_MAX = Decimal("9999999999999999") # NUMERIC(18,2) holds 16 digits before the point: a run total is refused beyond it, never a 500
def parse_period(text: str) -> tuple:
"""'2026-11' -> (2026, 11): gl's fiscal month, the billing month."""
m = PERIOD.match((text or "").strip())
if not m or not 2000 <= int(m.group(1)) <= 2099:
raise RuleBroken(f"{text!r} is not a billing month like 2026-11")
return int(m.group(1)), int(m.group(2))
def period_of(day: date) -> str:
return f"{day.year}-{day.month:02d}"
def run_id(project_id: str, period_id: str, n: int = 1) -> str:
"""tsbill:<project_id>:<period_id>, a second run of the month ...:<n>."""
return f"tsbill:{project_id}:{period_id}" + (f":{n}" if n > 1 else "")
def source_id(project_id: str, period_id: str, n: int = 1) -> str:
"""The key ar replays on: ts:<project_id>:<period_id>[:<n>]."""
return f"ts:{project_id}:{period_id}" + (f":{n}" if n > 1 else "")
@dataclass(frozen=True)
class Want:
"""What the latest approved revision of a week says one billable line is worth."""
sheet_id: str
revision: int
line_no: int
product_id: str
hours: Decimal
first_day: date
person: Identity
def wanted(sheet: Sheet, project_id: str, billed_lines: set) -> list:
"""The billable ar lines of ONE approved revision on one project. A line without a time product cannot be priced: it is reported, not dropped. A line
already billed that the revision no longer bills (made non-billable, or removed: a write-off) is wanted at zero hours, so its hours are taken back."""
out, seen = [], set()
for ln in sheet.lines:
if ln.kind == "project" and ln.project_id == project_id and (ln.bill_via == "ar" or ln.line_no in billed_lines):
seen.add(ln.line_no)
live = ln.billable and ln.bill_via == "ar"
out.append(Want(sheet.sheet_id, sheet.revision, ln.line_no, ln.product_id if live else "", ln.hours() if live else Decimal(0),
min((e.work_date for e in ln.entries), default=sheet.week_start), sheet.person))
for n in sorted(billed_lines - seen):
out.append(Want(sheet.sheet_id, sheet.revision, n, "", Decimal(0), sheet.week_start, sheet.person))
return out
def deltas(wants: list, billed: dict, period_end: date) -> list:
"""[(Want, delta)] for what is still to bill: hours wanted - hours already claimed on that (sheet, line), signed. A line whose work began after the
billing month is for a later run."""
out = []
for w in wants:
if w.first_day > period_end:
continue
d = hours4(w.hours - billed.get((w.sheet_id, w.line_no), Decimal(0)))
if d != 0:
out.append((w, d))
return out
def unpriced(wants_with_delta: list) -> list:
return [w for w, d in wants_with_delta if d > 0 and not w.product_id]
def invoice_lines(claims: list, period_id: str) -> tuple:
"""(the lines ar is asked for, the total): one line per product and price, the hours added (corrections subtract), each line rounded ONCE.
claims: [{product_id, quantity, unit_price, tax_category}]."""
grouped: dict = {}
for c in claims:
key = (c["product_id"], Decimal(c["unit_price"]), c.get("tax_category") or "")
grouped[key] = grouped.get(key, Decimal(0)) + Decimal(c["quantity"])
lines, total = [], ZERO
for (product, price, tax), qty in sorted(grouped.items(), key=lambda kv: (kv[0][0], kv[0][1])):
if qty == 0:
continue
amount = money(qty * price)
if abs(amount) > AMOUNT_MAX:
raise RuleBroken(f"{product}: {qty} hours at {price} is beyond what an invoice line can hold")
total += amount
line = {"description": f"{product} - time worked {period_id}", "quantity": f"{hours4(qty):.4f}", "unit_code": UNIT, "unit_price": f"{price:.4f}",
"product_id": product}
if tax:
line["tax_category"] = tax
lines.append(line)
if abs(total) > AMOUNT_MAX:
raise RuleBroken("the run's total is beyond what an invoice can hold")
return lines, total
def invoice_body(source: str, bp_id: int, currency: str, lines: list) -> dict:
return {"source_kind": "timesheet", "source_id": source, "bp_id": bp_id, "currency": currency, "kind": "regular", "lines": lines}
def price_per_hour(answer: dict) -> Optional[Decimal]:
"""The product's answer as a price of ONE hour, or None when it is not (per another unit, or per a count of them)."""
try:
price = Decimal(str(answer.get("unit_price")))
except Exception:
return None
if str(answer.get("pricing_unit_code") or UNIT) != UNIT or int(answer.get("pricing_quantity") or 1) != 1:
return None
if not price.is_finite() or not Decimal(0) <= price < PRICE_MAX or price.as_tuple().exponent < -4:
return None
return price