dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
src/brick_timesheets/domain/sheet.py
"""The timesheet: one person's ISO week, approved as a whole (module.yaml workflows.timesheet).
Plain Python - no SQLAlchemy, FastAPI, MCP or identity. The rules a finance manager checks first live here:
- the state machine only moves FORWARD (a returned week continues as revision n+1; an approved week is superseded, never edited)
- nobody approves, returns or reopens their own week: matched on email OR sub (either one is the same person)
- the approver is derived from hr on the week's Monday and FIXED on the sheet; approvals of one revision never count for the next
- hours are exact (NUMERIC(19,4), steps of the time increment, at most 24 a day); cost is hours x a standard rate, rounded once per line
- a correction is a new revision: every downstream effect (cost, delivery) moves by the DIFFERENCE between revisions
"""
import re
from dataclasses import dataclass, field, replace
from datetime import date, timedelta
from decimal import Decimal
from typing import Callable, Optional
from brick_timesheets.domain.errors import RuleBroken
from brick_timesheets.domain.money import ZERO, hours4, money
STATES = ("open", "submitted", "approved", "returned", "superseded")
TERMINAL = ("returned", "superseded")
LINES_MAX = 200
DAY_MAX = Decimal("24")
RATE_MAX = Decimal("1e7") # an hour of one person costs less than 10 million: 200 lines of a week still fit NUMERIC(18,2)
# transition key -> (states it may start from, the state it ends in). Every arrow goes forward; tests try the reverse of each.
MOVES = {
"submit_timesheet": (("open",), "submitted"),
"approve_timesheet": (("submitted",), "approved"),
"return_timesheet": (("submitted",), "returned"),
"reopen_timesheet": (("approved",), "superseded"),
}
WEEK = re.compile(r"^(\d{4})-W(\d{2})$")
@dataclass(frozen=True)
class Identity:
"""Who a person is to this brick: both of their logins' names. The same person under either is the same person."""
email: str = ""
sub: str = ""
def same_as(self, other: "Identity") -> bool:
a, b = _norm(self.email), _norm(other.email)
s, t = (self.sub or "").strip(), (other.sub or "").strip()
return bool((a and a == b) or (s and s == t))
def _norm(v: Optional[str]) -> str:
return (v or "").strip().lower()
@dataclass
class Entry:
work_date: date
hours: Decimal
note: str = ""
unit_cost: Optional[Decimal] = None # the cost rate in force on work_date for the person's rate group, copied at submission
@dataclass
class Line:
line_no: int
kind: str # project | internal | absence
activity_code: str = ""
project_id: str = ""
item_id: str = ""
cost_centre: str = "" # project: the project's (from its terms); internal: the one booked
billable: bool = False
bill_via: str = "none" # sales_order | ar | none
sales_order_id: str = ""
so_line_no: Optional[int] = None
product_id: str = ""
rate_group: str = ""
unit_cost: Optional[Decimal] = None # the rate when every entry shares it, else None (entries carry their own)
absence_ref: str = ""
description: str = ""
lead_person_id: Optional[int] = None
billing: str = "" # the project's terms: time_and_materials | fixed_price | non_billable
stale: bool = False
stale_reason: str = ""
entries: list = field(default_factory=list)
def hours(self) -> Decimal:
return hours4(sum((e.hours for e in self.entries), Decimal(0)))
def costs(self) -> bool:
return self.kind in ("project", "internal")
@dataclass
class LineApproval:
line_no: int
approver_person_id: int
decided_by: str
decided_sub: str
decided_at: object
@dataclass
class Sheet:
sheet_id: str
person_id: int
company_code: str
week_start: date
person: Identity
revision: int = 1
status: str = "open"
expected: dict = field(default_factory=dict) # ISO date -> contracted minutes (hr's work pattern; 0 on a bank holiday or an unemployed day)
home_cost_centre: str = ""
approval_route: str = ""
approver_person_id: Optional[int] = None
approver: Optional[Identity] = None
rerouted_from: Optional[str] = None
currency: str = ""
submitted_at: object = None
approved_by: str = ""
approved_sub: str = ""
approved_at: object = None
approved_on: Optional[date] = None
journal_id: Optional[int] = None
ended_by: str = ""
ended_sub: str = ""
ended_at: object = None
reason: str = ""
last_line_no: int = 0
lines: list = field(default_factory=list)
approvals: list = field(default_factory=list) # [LineApproval] of THIS revision
# ── derived, never incremented ────────────────────────────────────────────────────────────
def days(self) -> list:
return [self.week_start + timedelta(days=n) for n in range(7)]
def expected_hours(self) -> Decimal:
return hours4(Decimal(sum(self.expected.values())) / 60)
def booked(self, kinds: tuple = ("project", "internal", "absence")) -> dict:
out: dict = {}
for ln in self.lines:
if ln.kind in kinds:
for e in ln.entries:
out[e.work_date] = out.get(e.work_date, Decimal(0)) + e.hours
return out
def total(self, kinds: tuple = ("project", "internal", "absence")) -> Decimal:
return hours4(sum(self.booked(kinds).values(), Decimal(0)))
def line(self, line_no: int) -> Optional[Line]:
return next((ln for ln in self.lines if ln.line_no == line_no), None)
def next_line_no(self) -> int:
"""Line numbers are never reused within a week's revisions: a delivery, a billing claim and a cost reference name a line forever."""
return max([self.last_line_no, *(ln.line_no for ln in self.lines)]) + 1
def lines_needing_lead(self) -> list:
"""The project lines a lead must approve first (route project_lead_then_manager): never the lead's own."""
return [ln for ln in self.lines if ln.kind == "project" and ln.lead_person_id is not None and ln.lead_person_id != self.person_id]
def lines_waiting(self) -> list:
done = {a.line_no for a in self.approvals}
return [ln for ln in self.lines_needing_lead() if ln.line_no not in done]
# ── the state machine ──────────────────────────────────────────────────────────────────
def move(self, key: str) -> None:
frm, to = MOVES[key]
if self.status not in frm:
raise RuleBroken(f"Timesheet {self.sheet_id} revision {self.revision} is {self.status}: {key.replace('_', ' ')} needs it to be {' or '.join(frm)}")
self.status = to
def ensure_open(self) -> None:
if self.status != "open":
raise RuleBroken(f"Timesheet {self.sheet_id} revision {self.revision} is {self.status}: its lines are frozen once it is submitted "
"(a returned or reopened week continues as a new revision)")
# ── weeks and hours ────────────────────────────────────────────────────────────────────────────
def parse_week(text: str) -> date:
"""'2026-W47' -> its Monday. Anything else is not a week."""
m = WEEK.match((text or "").strip())
if not m:
raise ValueError(f"{text!r} is not an ISO week like 2026-W47")
try:
return date.fromisocalendar(int(m.group(1)), int(m.group(2)), 1)
except ValueError:
raise ValueError(f"{text!r} is not an ISO week") from None
def week_id(monday: date) -> str:
y, w, _ = monday.isocalendar()
return f"{y}-W{w:02d}"
def parse_hours(v, increment_minutes: int, *, zero_ok: bool = True) -> Decimal:
"""Hours: an exact decimal, zero or more, at most 24, at most 4 places, a whole number of increments (a quarter hour by default)."""
if isinstance(v, (bool, float)):
raise ValueError("hours are exact: give them as a string such as \"7.50\"")
try:
d = v if isinstance(v, Decimal) else Decimal(str(v).strip())
except Exception:
raise ValueError(f"{v!r} is not a number of hours") from None
if not d.is_finite():
raise ValueError("hours are a finite number")
if d.as_tuple().exponent < -4:
raise ValueError("hours have at most four decimal places")
if d < 0 or (d == 0 and not zero_ok):
raise ValueError("hours are zero or more")
if d > DAY_MAX:
raise ValueError("a day has at most 24.00 hours")
minutes = d * 60
if minutes % max(int(increment_minutes), 1) != 0:
raise ValueError(f"hours are booked in steps of {increment_minutes} minutes")
return hours4(d)
def check_day_totals(sheet: Sheet, day_hours: dict) -> None:
for d, h in sorted(day_hours.items()):
if h > DAY_MAX:
raise ValueError(f"{d} would hold {h:.2f} hours: a day has at most 24.00")
# ── the gates of submit_timesheet ─────────────────────────────────────────────────────────────
def coverage_gaps(sheet: Sheet) -> list:
"""The working days whose contracted hours are not covered by time or absence: [(date, booked, expected)]."""
booked = sheet.booked()
out = []
for d in sheet.days():
exp = hours4(Decimal(sheet.expected.get(d.isoformat(), 0)) / 60)
got = hours4(booked.get(d, Decimal(0)))
if exp > 0 and got < exp:
out.append((d, got, exp))
return out
def long_days(sheet: Sheet, max_day: Decimal) -> list:
"""Days above max_day_hours with no note on any of their entries: they warn, and need a note."""
noted = {e.work_date for ln in sheet.lines for e in ln.entries if e.note.strip()}
return [(d, hours4(h)) for d, h in sorted(sheet.booked(("project", "internal")).items()) if h > max_day and d not in noted]
def stale_lines(sheet: Sheet, absence_days: dict, terms_changed: dict) -> dict:
"""line_no -> why, for lines built on facts that changed: time-off now holds an absence on a day the line also works (the day would hold more
than the contract), or the project's terms are no longer what was copied. Derived from the inputs each time; nothing else is stored.
`absence_days`: ISO date -> absence hours time-off holds now. `terms_changed`: project_id -> what changed."""
out: dict = {}
for ln in sheet.lines:
if ln.kind == "absence":
continue
for e in ln.entries:
a = absence_days.get(e.work_date.isoformat(), Decimal(0))
exp = Decimal(sheet.expected.get(e.work_date.isoformat(), 0)) / 60
others = sum((x.hours for other in sheet.lines if other.kind != "absence" for x in other.entries if x.work_date == e.work_date), Decimal(0))
if a > 0 and a + others > max(exp, a):
out[ln.line_no] = f"time-off holds {a:.2f} hours of absence on {e.work_date}"
break
if ln.line_no not in out and ln.project_id in terms_changed:
out[ln.line_no] = terms_changed[ln.project_id]
return out
def readiness(*, gaps: list, mode: str, notes: list, bookable: list, closed: list, rates: list, stale: dict) -> list:
"""Each gate of submit_timesheet / approve_timesheet as {gate, met, reason, fix}: what the screen shows and what a refusal names."""
covered = not gaps or mode == "warn"
return [
{"gate": "hours_covered", "met": covered,
"reason": "; ".join(f"{d}: {b:.2f} of {x:.2f} hours" for d, b, x in gaps) if not covered else "",
"fix": "book the missing hours, or ask for the absence in time off"},
{"gate": "day_note", "met": not notes, "reason": "; ".join(f"{d} holds {h:.2f} hours and has no note" for d, h in notes),
"fix": "add a note to the long day"},
{"gate": "bookable", "met": not bookable, "reason": "; ".join(bookable), "fix": "book to another project or item"},
{"gate": "cost_centre_open", "met": not closed, "reason": "; ".join(f"cost centre {c} is closed (gl)" for c in closed), "fix": "book to an open cost centre"},
{"gate": "rate_group", "met": not rates, "reason": "; ".join(rates), "fix": "finance sets the person's rate group and its cost rate"},
{"gate": "not_stale", "met": not stale, "reason": "; ".join(f"line {n}: {why}" for n, why in sorted(stale.items())),
"fix": "return the week for a correction"},
]
# ── routing and deciding ────────────────────────────────────────────────────────────────────────
def derive_approver(person: Identity, manager_at: Callable[[int], Optional[dict]]) -> tuple:
"""The approver: the person's manager from hr on the week's Monday. A level whose manager IS the person goes one up (a manager's own week
goes to their manager). Returns (hr person_id, Identity, rerouted_from). No manager above the person is a rule, never a guess."""
skipped: Optional[str] = None
for n in (1, 2):
m = manager_at(n)
if not m:
continue
who = Identity(m.get("email") or "", m.get("pool_user_sub") or "")
if who.same_as(person):
skipped = who.email or who.sub
continue
return m.get("person_id"), who, skipped
raise RuleBroken("hr names no manager above you, so nobody can approve this week: ask HR to set your manager")
def check_approver(sheet: Sheet, who: Identity, what: str = "approve") -> None:
"""Separation of duties: nobody decides their own week, under either login. Then the rules of the state."""
if who.same_as(sheet.person):
raise RuleBroken("This is your own week: it goes to your manager")
if sheet.status != "submitted":
raise RuleBroken(f"Timesheet {sheet.sheet_id} revision {sheet.revision} is {sheet.status}: only a submitted week is {what}d")
def check_reopener(sheet: Sheet, who: Identity) -> None:
"""Finance reopens an approved week - never their own, never one they approved (email AND sub)."""
if who.same_as(sheet.person):
raise RuleBroken("This is your own week: someone else reopens it")
if sheet.approved_by or sheet.approved_sub:
if Identity(sheet.approved_by, sheet.approved_sub).same_as(who):
raise RuleBroken("You approved this week: someone else reopens it")
if sheet.status != "approved":
raise RuleBroken(f"Timesheet {sheet.sheet_id} revision {sheet.revision} is {sheet.status}: only an approved week is reopened")
# ── cost: hours x a standard rate, posted to the project's cost centre, relieved from the person's home ─────
def line_cost(line: Line) -> Decimal:
"""Rounded ONCE per line to cents, from the rate on each entry's own work date."""
total = Decimal(0)
for e in line.entries:
if e.unit_cost is None:
raise RuleBroken(f"Line {line.line_no} has no cost rate on {e.work_date}")
total += e.hours * e.unit_cost
return money(total)
def journal_lines(sheet: Sheet) -> list:
"""The labour journal of an approved revision: a debit per cost centre the person's time was worked for (project lines at the project's cost
centre, internal lines on another centre than home), ONE credit to the person's home cost centre. Time on the home centre and absence post
nothing. Both sides equal to the cent, or it is not posted. [] when nothing posts."""
debits: dict = {}
for ln in sheet.lines:
if not ln.costs() or not ln.entries:
continue
if ln.kind == "internal" and ln.cost_centre == sheet.home_cost_centre:
continue
cost = line_cost(ln)
if cost > 0:
debits[ln.cost_centre] = debits.get(ln.cost_centre, ZERO) + cost
if not debits:
return []
out = [{"purpose": "project_labour", "cost_centre": c, "debit": f"{v:.2f}"} for c, v in sorted(debits.items())]
out.append({"purpose": "labour_recovery", "cost_centre": sheet.home_cost_centre, "credit": f"{sum(debits.values()):.2f}"})
dr = sum(Decimal(x["debit"]) for x in out if "debit" in x)
cr = sum(Decimal(x["credit"]) for x in out if "credit" in x)
if dr != cr:
raise RuleBroken(f"The journal does not balance ({dr:.2f} against {cr:.2f})")
return out
def reversal_of(lines: list) -> list:
"""The mirror of an approved journal: every debit becomes a credit of the same amount and the reverse."""
return [{**{k: v for k, v in ln.items() if k not in ("debit", "credit")},
**({"credit": ln["debit"]} if "debit" in ln else {"debit": ln["credit"]})} for ln in lines]
# ── the hours a sales order is told about: by the DIFFERENCE ────────────────────────────────────────
def delivery_wanted(sheet: Sheet) -> dict:
"""(line_no, sales_order_id, so_line_no) -> approved hours, for the billable project lines a sales order delivers."""
out: dict = {}
for ln in sheet.lines:
if ln.kind == "project" and ln.billable and ln.bill_via == "sales_order" and ln.sales_order_id and ln.so_line_no is not None:
h = ln.hours()
if h > 0:
out[(ln.line_no, ln.sales_order_id, ln.so_line_no)] = h
return out
def delivery_deltas(wanted: dict, delivered: dict) -> list:
"""What the order must be told now: [(line_no, sales_order_id, so_line_no, delta)] where delta = this revision's hours - what the order already
holds from this week's earlier revisions. A line that is gone (or lost its hours) is a negative delta; unchanged lines send nothing."""
out = []
for key in sorted(set(wanted) | set(delivered)):
delta = hours4(wanted.get(key, Decimal(0)) - delivered.get(key, Decimal(0)))
if delta != 0:
out.append((*key, delta))
return out
# ── what payroll and time-off read ───────────────────────────────────────────────────────────────
def split_hours(worked: dict, absence: dict, expected: dict, rule: str) -> dict:
"""date -> (normal, overtime) of one approved week, by the overtime rule in force. weekly_above_contracted: the hours up to the week's contracted
hours (less absence) are normal, the later days' excess is overtime; a week with no contract (irregular hours) has none. daily_above_contracted:
per day. none: everything is normal."""
out: dict = {}
if rule == "none":
return {d: (hours4(h), Decimal(0)) for d, h in worked.items()}
contracted = Decimal(sum(expected.values())) / 60
if rule == "daily_above_contracted":
for d, h in sorted(worked.items()):
if contracted == 0:
out[d] = (hours4(h), Decimal(0))
continue
room = max(Decimal(expected.get(d.isoformat(), 0)) / 60 - absence.get(d, Decimal(0)), Decimal(0))
normal = min(h, room)
out[d] = (hours4(normal), hours4(h - normal))
return out
room = max(contracted - sum(absence.values(), Decimal(0)), Decimal(0)) if contracted > 0 else None
for d, h in sorted(worked.items()):
if room is None:
out[d] = (hours4(h), Decimal(0))
continue
normal = min(h, room)
room -= normal
out[d] = (hours4(normal), hours4(h - normal))
return out
def copy_lines(lines: list) -> list:
"""The lines of the next revision: the same lines, entries and line numbers, the approvals left behind."""
return [replace(ln, entries=[replace(e) for e in ln.entries]) for ln in lines]