dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
src/brick_timesheets/application/billing.py
"""The billing run (TS-13, TS-14): a project's month of approved billable hours, billed through ar (workflow billing_run).
create the system gathers the approved billable hours of one project (its terms bill via ar) that no run has claimed, prices each line from the product's
price list on its first work day, and writes the run and one claim row per line ONE transaction. The claim (sheet, revision, line, 'ar_run') is the
race: a re-run, a retry or a second worker finds the hours claimed and takes none. A correction (a later approved revision) is billed as its
difference in the next run; hours taken away, or a line made non-billable (a write-off), are a negative line.
release finance releases it (the total not negative; never a person whose hours are in it). The invoice request is written as a PENDING outbox row in the
same transaction and sent after the commit; ar replays on the run's source id, so a retry is one draft.
hold finance parks a draft (a reason). cancel lets the hours go: the claims are removed and the lines fall into the next run.
Nothing here writes ar_ or product tables."""
import json
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from brick_timesheets.application import use_cases as hub
from brick_timesheets.application.inputs import Code, Reason, Ref
from brick_timesheets.domain import billing as bd
from brick_timesheets.domain.errors import RuleBroken
from brick_timesheets.infrastructure import seams
TS = hub.TS
CREATE, RELEASE = "ts:billing_run:create", "ts:billing_run:release"
STATUSES = ("draft", "released", "held", "cancelled")
def _view(run: dict, repo=None) -> dict:
lines = json.loads(run["lines_json"] or "[]")
out = {"run_id": run["run_id"], "project_id": run["project_id"], "period_id": run["period_id"], "company_code": run["company_code"],
"bp_id": None if run["bp_id"] is None else str(run["bp_id"]), "currency": run["currency"], "status": run["status"], "source_id": run["source_id"],
"lines": lines, "total_amount": hub._m(run["total_amount"]), "claims": run["claim_count"], "invoice_id": run["invoice_id"],
"reason": run["reason"], "released_by": run["released_by"],
"released_on": run["released_on"].isoformat() if run["released_on"] else None}
if repo is not None and run["status"] == "released":
row = repo.outbox.get("ar", run["source_id"])
out["invoice_request"] = None if row is None else {"status": row["status"], "last_error": row["last_error"]}
return out
def _need_run(repo, run_id: str) -> dict:
run = repo.runs.get(run_id)
if run is None:
raise Refused(404, f"no billing run {run_id}")
return run
def _period_end(period_id: str) -> date:
try:
y, m = bd.parse_period(period_id)
except RuleBroken as e:
raise hub._bounds(str(e), "a billing month is written 2026-11") from None
return (date(y + (m == 12), m % 12 + 1, 1)) - timedelta(days=1)
def _price(line_want, qty: Decimal, bp_id: int, currency: str, actor: str, cache: dict) -> dict:
"""The product's price of this line's hours on its first work day (TS-14). A missing price, or one not per hour, stops the run - nothing is invented."""
key = (line_want.product_id, line_want.first_day, f"{qty:.4f}")
if key not in cache:
cache[key] = seams.price(line_want.product_id, f"{qty:.4f}", bp_id, currency, line_want.first_day, actor)
answer = cache[key]
if answer is None:
raise Refused(409, f"product has no price for {line_want.product_id} for customer {bp_id} on {line_want.first_day}", gate="price",
fix="give the time product a price in the customer's price list, then run again")
per_hour = bd.price_per_hour(answer)
if per_hour is None:
raise Refused(409, f"product prices {line_want.product_id} per {answer.get('pricing_quantity')} {answer.get('pricing_unit_code')}, not per hour", gate="price_unit",
fix="price the time product per 1 HUR")
return {"unit_price": per_hour, "priced_from": json.dumps(answer.get("priced_from") or {}, sort_keys=True),
"tax_category": str((answer.get("line") or {}).get("tax_category") or "")}
@TS.use_case(kind="command", route="POST /billing-runs", permission=CREATE)
def create_time_billing_run(caller: dict, project_id: Ref, period_id: Code) -> dict:
"""Gather one project's approved billable hours not yet claimed by a run, for billing month `period_id` (2026-11), into a DRAFT run (the project's terms
must bill via ar). Every line is priced from the product price list on its first work day; every line is claimed, so it is billed once however many
times this runs. Nothing to bill answers {created: false}. Corrections of earlier months arrive as negative lines; the run is released by finance."""
end = _period_end(period_id)
actor = hub._who(caller)
with hub.repository() as repo:
today = hub._today(repo)
if bd.period_of(end) > bd.period_of(today):
raise Refused(409, f"{period_id} has not started: nothing of it can be billed yet", gate="period_started", fix="run it in or after the month")
with hub._decide():
terms = seams.time_terms(project_id, min(end, today), actor)
if terms is None:
raise Refused(404, f"projects does not know {project_id}")
if terms.get("bill_via") != "ar":
raise Refused(409, f"{project_id} is not billed through ar (its terms say {terms.get('bill_via') or 'none'})", gate="bills_via_ar",
fix="hours on a sales-order project are billed by the order")
if not terms.get("bp_id") or not terms.get("currency"):
raise Refused(409, f"{project_id} names no customer or currency to invoice", gate="customer", fix="set them on the project")
bp_id, currency = int(terms["bp_id"]), hub._cur(terms["currency"])
with hub.repository() as repo:
sheets = repo.sheets.approved_ar_lines(project_id)
ids = [s.sheet_id for s in sheets]
billed = repo.billing.ar_billed(ids, project_id)
first_price = repo.billing.ar_first_price(ids, project_id)
wants = []
for s in sheets:
wants += bd.wanted(s, project_id, {n for (sid, n) in billed if sid == s.sheet_id})
todo = bd.deltas(wants, billed, end)
if not todo:
return {"created": False, "reason": "no approved billable hours are waiting for a run", "run": None}
lost = bd.unpriced(todo)
if lost:
raise Refused(409, "Lines without a time product cannot be billed: " + "; ".join(f"{w.sheet_id} line {w.line_no}" for w in lost), gate="time_product",
fix="give the project a time product for the person's rate group, then book the week again")
cache: dict = {}
claims, priced = [], []
for w, delta in todo:
if delta > 0:
p = _price(w, delta, bp_id, currency, actor, cache)
product = w.product_id
else:
was = first_price.get((w.sheet_id, w.line_no))
if was is None:
raise Refused(409, f"{w.sheet_id} line {w.line_no}: nothing was billed to take back", gate="price")
product = _product_of(sheets, w)
p = {"unit_price": was[0], "priced_from": was[1], "tax_category": ""}
claims.append({"sheet_id": w.sheet_id, "revision": w.revision, "line_no": w.line_no, "purpose": bd.PURPOSE, "quantity": delta,
"unit_price": p["unit_price"], "priced_from": p["priced_from"], "product_id": product, "tax_category": p["tax_category"] or None})
priced.append({"product_id": product, "quantity": delta, "unit_price": p["unit_price"], "tax_category": p["tax_category"]})
try:
lines, total = bd.invoice_lines(priced, period_id)
except RuleBroken as e:
raise hub._bounds(str(e), "split the hours over two runs") from None
now = hub._now()
with hub.repository() as repo:
runs = repo.runs.of_period(project_id, period_id)
n = len(runs) + 1
rid, source = bd.run_id(project_id, period_id, n), bd.source_id(project_id, period_id, n)
repo.claims.once("billing_create", rid, now, actor)
for c in claims:
c.update(run_id=rid)
repo.billing.add_many(claims, now)
row = {"run_id": rid, "project_id": project_id, "period_id": period_id, "run_no": n, "company_code": sheets[0].company_code, "bp_id": bp_id,
"currency": currency, "status": "draft", "source_id": source, "lines_json": json.dumps(lines, sort_keys=True), "total_amount": total,
"claim_count": len(claims), "invoice_id": None, "reason": None, "created_by": actor, "released_by": None, "released_sub": None,
"released_on": None, "ended_by": None}
repo.runs.add(row, now)
return {"created": True, "run": _view({**row, "released_on": None})}
def _product_of(sheets: list, w) -> str:
"""The time product a taken-back line was billed at: the one on the line of the revision it was billed in (found in the earliest claim)."""
for s in sheets:
if s.sheet_id == w.sheet_id:
for ln in s.lines:
if ln.line_no == w.line_no and ln.product_id:
return ln.product_id
return ""
@TS.use_case(kind="command", route="POST /billing-runs/{run_id}/release", permission=RELEASE, idempotent=True)
def release_time_billing_run(caller: dict, run_id: Ref) -> dict:
"""Finance releases a DRAFT or HELD run to invoicing (-> released, terminal). The total must not be negative (ar refuses it: credit by hand and cancel the
run), and the releaser is never a person whose hours are in the run (email AND sub). The invoice request is written as a pending row with the release and
sent after the commit; ar replays the run's source id, so releasing again sends nothing new. Releasing a released run only sends what is still pending."""
actor = hub._who(caller)
with hub.repository() as repo, hub._decide():
run = _need_run(repo, run_id)
if run["status"] == "released":
pass
elif run["status"] not in ("draft", "held"):
raise Refused(409, f"run {run_id} is {run['status']}: only a draft or held run is released")
else:
if Decimal(run["total_amount"] or 0) < 0:
raise Refused(409, f"run {run_id} totals {hub._m(run['total_amount'])}: a negative total is credited in ar by hand", gate="not_negative",
fix="cancel the run and credit the customer in ar")
me = hub._me(caller)
if any(me.same_as(hub.sh.Identity(email, sub)) for email, sub in repo.billing.people_of_run(run_id)):
raise Refused(409, "your own hours are in this run", gate="second_person", fix="another finance user releases it")
now, today = hub._now(), hub._today(repo)
repo.claims.once("billing_release", run_id, now, actor)
body = bd.invoice_body(run["source_id"], run["bp_id"], run["currency"], json.loads(run["lines_json"]))
repo.outbox.add_many([("ar", run["source_id"], run_id, body)], now)
repo.runs.save({**run, "status": "released", "released_by": me.email or me.sub, "released_sub": me.sub, "released_at": now, "released_on": today}, now)
hub._drain(None, ("ar",), actor)
with hub.repository() as repo:
return _view(_need_run(repo, run_id), repo)
@TS.use_case(kind="command", route="POST /billing-runs/{run_id}/hold", permission=RELEASE)
def hold_time_billing_run(caller: dict, run_id: Ref, reason: Reason) -> dict:
"""Finance holds a DRAFT run (draft -> held; a reason is required): its hours stay claimed, nothing is invoiced, and it can still be released or cancelled."""
if not reason.strip():
raise hub._bounds("a reason is required to hold a run")
with hub.repository() as repo, hub._decide():
run = _need_run(repo, run_id)
if run["status"] != "draft":
raise Refused(409, f"run {run_id} is {run['status']}: only a draft run is held")
now = hub._now()
repo.claims.once("billing_hold", run_id, now, hub._who(caller))
row = {**run, "status": "held", "reason": reason.strip()}
repo.runs.save(row, now)
return _view(row)
@TS.use_case(kind="command", route="POST /billing-runs/{run_id}/cancel", permission=RELEASE, destructive=True)
def cancel_time_billing_run(caller: dict, run_id: Ref, reason: Reason) -> dict:
"""Finance cancels a DRAFT or HELD run (a reason is required; terminal): its claims are released, so the hours fall into the next run. A released run
is not cancelled - ar owns the invoice; credit it there."""
if not reason.strip():
raise hub._bounds("a reason is required to cancel a run")
with hub.repository() as repo, hub._decide():
run = _need_run(repo, run_id)
if run["status"] not in ("draft", "held"):
raise Refused(409, f"run {run_id} is {run['status']}: only a draft or held run is cancelled")
now = hub._now()
repo.claims.once("billing_cancel", run_id, now, hub._who(caller))
repo.billing.delete_run(run_id)
row = {**run, "status": "cancelled", "reason": reason.strip(), "ended_by": hub._who(caller), "ended_at": now, "claim_count": 0}
repo.runs.save(row, now)
return _view(row)
@TS.use_case(kind="query", route="GET /billing-runs/{run_id}", permission=RELEASE)
def get_time_billing_run(caller: dict, run_id: Ref) -> dict:
"""One billing run: its priced invoice lines, its total, its status, and - once released - whether ar has the request (and ar's reason when it refused)."""
with hub.repository() as repo:
return _view(_need_run(repo, run_id), repo)
@TS.use_case(kind="query", route="GET /billing-runs", permission=RELEASE)
def list_time_billing_runs(caller: dict, status: Optional[Code] = None, project_id: Optional[Ref] = None) -> dict:
"""Billing runs, by billing month then project, optionally of one status (draft | held | released | cancelled) or one project."""
if status is not None and status not in STATUSES:
raise hub._bounds(f"status is one of {', '.join(STATUSES)}")
with hub.repository() as repo:
return {"runs": [{k: v for k, v in _view(r).items() if k != "lines"} for r in repo.runs.listing(status, project_id)]}
@TS.use_case(kind="command", route="POST /billing-runs/tick", permission=CREATE, idempotent=True, tool=False)
def tick_time_billing_runs(caller: dict, period_id: Code) -> dict:
"""The worker's month-end pass (no server-side scheduler): a run for every project with approved billable ar hours waiting. A project that cannot be
billed (no price, no time product) is reported with the reason and the others go on. Safe from two workers and run twice: the claims are the race."""
_period_end(period_id)
with hub.repository() as repo:
projects = repo.billing.ar_projects()
done = []
for pid in projects:
try:
got = create_time_billing_run(caller, pid, period_id)
done.append({"project_id": pid, "created": got["created"], "run_id": (got["run"] or {}).get("run_id")})
except Refused as r:
done.append({"project_id": pid, "created": False, "refused": r.reason, "gate": r.gate})
return {"period_id": period_id, "projects": done}
@TS.use_case(kind="command", route="POST /billing-runs/invoices/send", permission=CREATE, idempotent=True, tool=False)
def send_billing_invoices(caller: dict) -> dict:
"""The worker's tick for released runs: send every pending invoice request to ar. ar replays the run's source id; one it refused stays pending with its
reason on the run. Safe at any time and from two workers."""
return {"sent": hub._drain(None, ("ar",), hub._who(caller))}