"""The PayrollExtract aggregate: one period's hand-off to payroll for one legal entity (hr_payroll_batch,
batch_id = <legal_entity_ref>|<period>, with its hr_payroll_line rows).
hr does not calculate payroll (tax, NI, pensions, RTI are the provider's). It publishes RATES, SLICES and
PRORATION DATES per legal entity, and:
* freelancer engagements (paid as suppliers) are never in it — only worker types with payroll: true;
* a sent batch is frozen; a later change to a sent period goes into the next open batch as a RETRO line,
so a row goes out exactly once;
* an amount is computed only where a quantity exists (from a future time brick) or a fixed amount is due,
and is rounded ONCE at the payable line (erp_platform.money.round_once) with the configured mode.
"""
import calendar
import hashlib
import json
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from erp_platform.money import Money, Rate, round_once
from brick_hr.domain.errors import Incomplete, RuleBroken
from brick_hr.domain.person import Person, rate_of
DRAFT, SENT = "draft", "sent"
def period_bounds(period: str) -> tuple:
try:
y, m = (int(x) for x in period.split("-"))
return date(y, m, 1), date(y, m, calendar.monthrange(y, m)[1])
except (ValueError, TypeError):
raise Incomplete(f"a period is YYYY-MM, not {period!r}") from None
def batch_id(legal_entity_ref: str, period: str) -> str:
period_bounds(period)
if not legal_entity_ref:
raise Incomplete("a payroll extract is per legal entity")
return f"{legal_entity_ref}|{period}"
def payable(rate: Rate, *, minutes: Optional[int] = None, days=None, occurrences: Optional[int] = None,
rounding_mode: str = "HALF_UP") -> Money:
"""PayableAmount: rate × quantity, exact, rounded ONCE at the line (13.3333/h × 450 min = 100.00, not 99.99)."""
if minutes is not None:
return round_once([rate.for_minutes(minutes)], rounding_mode=rounding_mode)
if days is not None:
return round_once([rate.for_days(days)], rounding_mode=rounding_mode)
if occurrences is not None:
return round_once([rate.times(occurrences)], rounding_mode=rounding_mode)
raise Incomplete("a payable amount needs a quantity (minutes, days or occurrences)")
@dataclass
class Line:
person_id: int
n: int
component: str
for_period: str
slice_from: date
rate_basis: Optional[str] = None
rate_amount: Optional[Decimal] = None
currency: Optional[str] = None
prorate_from: Optional[date] = None
prorate_to: Optional[date] = None
quantity_minutes: Optional[int] = None
quantity_days: Optional[Decimal] = None
payable_amount: Optional[Decimal] = None
change_kind: str = "pay"
org_unit_id: Optional[str] = None
cost_centre: Optional[str] = None
detail: dict = field(default_factory=dict)
@property
def key(self) -> tuple:
return (self.person_id, self.n, self.component, self.for_period, self.slice_from)
def fingerprint(self) -> str:
body = [self.rate_basis, str(self.rate_amount) if self.rate_amount is not None else None, self.currency,
str(self.prorate_from), str(self.prorate_to), self.org_unit_id, self.cost_centre,
json.dumps(self.detail, sort_keys=True, default=str)]
return hashlib.sha256(json.dumps(body).encode()).hexdigest()[:16]
@dataclass
class Batch:
batch_id: str
legal_entity_ref: str
period: str
state: str = DRAFT
produced_at: Optional[datetime] = None
sent_at: Optional[datetime] = None
sent_by: Optional[str] = None
version: int = 0
lines: list = field(default_factory=list)
def mark_sent(self, by: str, at: datetime) -> bool:
if self.state == SENT:
return False
self.state, self.sent_at, self.sent_by = SENT, at, by
return True
def lines_for(p: Person, period: str, legal_entity_ref: str, payroll_types: set,
bank_changes: Optional[list] = None) -> list:
"""Every pay and component slice touching the period, split where the assignment changes (a mid-month
transfer gives two lines with their proration dates); starters and leavers marked; bank changes listed."""
ps, pe = period_bounds(period)
out = []
for e in sorted(p.engagements.values(), key=lambda e: e.n):
if e.worker_type not in payroll_types or e.legal_entity_ref != legal_entity_ref:
continue
lo, hi = max(ps, e.start_date), min(pe, e.end)
if lo > hi:
continue
assign = [s for s in p.assignments[e.n].overlapping(lo, hi)]
for pay in p.pay[e.n].overlapping(lo, hi):
for a in assign:
seg_lo, seg_hi = max(lo, pay.valid_from, a.valid_from), min(hi, pay.valid_to, a.valid_to)
if seg_lo > seg_hi:
continue
r = rate_of(pay)
kind = "pay"
if e.start_date == seg_lo and ps <= e.start_date <= pe:
kind = "starter"
elif e.actual_end is not None and e.actual_end == seg_hi and ps <= e.actual_end <= pe:
kind = "leaver"
amount = None
if r is not None and r.basis == "fixed":
amount = payable(r, occurrences=1).amount
out.append(Line(p.person_id, e.n, "base", period, seg_lo, r.basis if r else None,
r.amount if r else None, r.currency if r else None, seg_lo, seg_hi,
payable_amount=amount, change_kind=kind, org_unit_id=a.data.get("org_unit_id"),
cost_centre=a.data.get("cost_centre")))
for (m, comp), t in sorted(p.components.items()):
if m != e.n:
continue
for s in t.overlapping(lo, hi):
r = rate_of(s)
seg_lo, seg_hi = max(lo, s.valid_from), min(hi, s.valid_to)
out.append(Line(p.person_id, e.n, comp, period, seg_lo, r.basis, r.amount, r.currency, seg_lo,
seg_hi, change_kind="component"))
for s in bank_changes or []:
if ps <= s.valid_from <= pe and e.active_on(s.valid_from):
out.append(Line(p.person_id, e.n, "bank", period, s.valid_from, change_kind="bank_change",
prorate_from=s.valid_from, detail={"bank": (s.data or {}).get("value")}))
return out
def build(entity: str, period: str, current: dict, sent: list, *, now: datetime, existing: Optional[Batch] = None) -> Batch:
"""The batch for (entity, period): this period's lines, plus a RETRO line for every line of an already-SENT
period that is different now (or new, or gone). `current` is {period: [Line]} for this period and every sent
period; `sent` is every line in a sent batch. A sent batch is never rebuilt."""
if existing is not None and existing.state == SENT:
raise RuleBroken(f"the {period} batch for {entity} was sent {existing.sent_at:%Y-%m-%d}; it is frozen — a "
"later change goes into the next open batch as a retro line")
b = Batch(batch_id(entity, period), entity, period, DRAFT, now, version=existing.version if existing else 0)
b.lines = list(current.get(period, []))
last_sent: dict = {}
for ln in sent:
last_sent[ln.key] = ln # later batches override earlier: sent ordered by period
for q, lines in sorted(current.items()):
if q >= period:
continue
now_keys = {ln.key: ln for ln in lines}
for k, ln in now_keys.items():
before = last_sent.get(k)
if before is None or before.fingerprint() != ln.fingerprint():
retro = Line(**{**ln.__dict__, "change_kind": "retro",
"detail": {**ln.detail, "was": _brief(before)}})
b.lines.append(retro)
for k, before in last_sent.items():
if k[3] == q and k not in now_keys and before.change_kind != "removed":
gone = Line(**{**before.__dict__, "change_kind": "removed", "detail": {"was": _brief(before)}})
b.lines.append(gone)
return b
def _brief(ln: Optional[Line]) -> Optional[dict]:
if ln is None:
return None
return {"rate_amount": str(ln.rate_amount) if ln.rate_amount is not None else None, "rate_basis": ln.rate_basis,
"prorate_from": str(ln.prorate_from), "prorate_to": str(ln.prorate_to)}
__all__ = ["Batch", "Line", "build", "lines_for", "payable", "period_bounds", "batch_id", "DRAFT", "SENT"]