dodil/erp-brick-payroll
DODIL ERP brick: payroll
src/brick_payroll/application/ytd.py
"""The year-to-date load (PRL-24, workflow prl_ytd_run): for an owner moving payroll in-house mid-year, or leaving Odoo, each employment's year to date
from the previous provider, loaded, reconciled to the penny against that provider's last FPS, signed by the owner, or discarded.
load a reader (infrastructure/readers.py) answers the provider's last FPS and its report; ONE opening row per employment and tax year is written,
keyed "ytd:<payroll_id>:<tax year>" (a plain INSERT: loading twice writes nothing, and a second run cannot overwrite a held row)
reconcile per employee and in total: taxable pay, tax, employee NI and employer NI equal to the FPS, or the check says what differs
sign the owner, only when every check is equal; the people become payroll employees with the provider's tax code dated from the cut-over, and the
first run here calculates cumulatively from the opening figures (they count only once SIGNED)
discard the owner, before signing: the opening rows are deleted; a signed load is never undone (a correction is a later run)
Payroll IDs are kept as HMRC knows them: "<person_id>-<n>" gives the person (hr offers no migration map for it yet - AGENTS.md)."""
import json
from decimal import Decimal
from erp_platform.bricks import Refused
from brick_payroll.application import use_cases as hub
from brick_payroll.application.inputs import Code, Company, Day, Reason, Ref, parse_money
from brick_payroll.domain import pay_run as pr
from brick_payroll.domain.errors import RuleBroken
from brick_payroll.domain.money import ZERO, money
from brick_payroll.infrastructure import readers
PRL = hub.PRL
MIG_LOAD, MIG_RECONCILE, MIG_READ, MIG_SIGN = "prl:migration:load", "prl:migration:reconcile", "prl:migration:read", "prl:migration:sign"
FIELDS = ("taxable", "tax", "ee_ni", "er_ni")
LABELS = {"taxable": "taxable pay to date", "tax": "tax to date", "ee_ni": "employee NI to date", "er_ni": "employer NI to date"}
def _amount(v, what: str, negative_ok: bool = False) -> Decimal:
try:
return parse_money(v if v is not None else "0.00", what, negative_ok=negative_ok)
except ValueError as e:
raise hub._bounds(str(e)) from None
def _person(pid: str) -> tuple:
try:
person, n = pid.split("-")
if not (person.isdigit() and n.isdigit() and 1 <= int(n) <= 99):
raise ValueError
return int(person), int(n)
except ValueError:
raise hub._bounds(f"payroll id {pid!r} is <person_id>-<n> like 4410-1") from None
def _run_view(repo, run: dict, with_rows: bool = False) -> dict:
checks = json.loads(run["checks_json"] or "[]")
rows = repo.ytd.of_run(run["run_id"])
out = {"run_id": run["run_id"], "tax_year": run["tax_year"], "paye_ref": run["paye_ref"], "company_code": run["company_code"], "schedule_code": run["schedule_code"],
"cutover_date": run["cutover_date"].isoformat(), "source_ref": run["source_ref"], "status": run["status"], "employees": len(rows),
"totals": {k: f"{sum((Decimal(r[c] or 0) for r in rows), ZERO):.2f}" for k, c in
(("taxable", "taxable_pay_amount"), ("tax", "tax_amount"), ("ee_ni", "ee_ni_amount"), ("er_ni", "er_ni_amount"))},
"checks": checks, "loaded_by": run["loaded_by"], "signed_by": run["signed_by"], "signed_on": run["signed_on"].isoformat() if run["signed_on"] else None,
"discard_reason": run["discard_reason"]}
if run["status"] == "reconciled":
out["readiness"] = [{"gate": "every_total_equal", "met": all(c["equal"] for c in checks) and bool(checks),
"reason": "; ".join(f"{c['check']}: {c['previous']} against {c['ours']}" for c in checks if not c["equal"]), "fix": "discard it and load again"}]
if with_rows:
out["openings"] = [{"ytd_ref": r["ytd_ref"], "payroll_id": r["payroll_id"], "taxable": f"{Decimal(r['taxable_pay_amount'] or 0):.2f}",
"tax": f"{Decimal(r['tax_amount'] or 0):.2f}", "employee_ni": f"{Decimal(r['ee_ni_amount'] or 0):.2f}",
"employer_ni": f"{Decimal(r['er_ni_amount'] or 0):.2f}", "pension": f"{Decimal(r['pension_amount'] or 0):.2f}",
"tax_code": r["tax_code"], "basis": r["basis"]} for r in rows]
return out
def _get(repo, run_id: str) -> dict:
run = repo.ytd.run(run_id)
if run is None:
raise Refused(404, f"no year-to-date load {run_id}")
return run
@PRL.use_case(kind="command", route="POST /migration/ytd/load", permission=MIG_LOAD, idempotent=True, tool=False)
def load_ytd(caller: dict, run_id: Code, tax_year: Code, paye_ref: Code, company_code: Company, schedule_code: Code, cutover_date: Day, source_ref: Ref,
hr_migration_signed: bool) -> dict:
"""Load the previous provider's year to date (loaded): the reader reads its last FPS and its report for source_ref, and one opening row per employment is
written keyed ytd:<payroll_id>:<tax year>. Needs hr's migration signed first (attested: no route says so) - the people come from hr. cutover_date is the
first day the first run here pays from (the start of a tax month, never mid-period). Loading the same run again writes nothing; an employment already
held by another load is 409 ytd_loaded. The opening figures count only once the owner signs."""
if not hr_migration_signed:
raise Refused(409, "hr's migration must be signed first: the people and their payroll IDs come from hr", gate="upstream_signed", fix="sign it, then load")
year, ref, cc, sched = hub_year(tax_year), paye_ref.strip().upper(), hub._company(company_code), schedule_code.strip().upper()
cutover = hub._date(cutover_date, "cutover_date")
try:
got = readers.ports.get().read(source_ref)
except readers.YtdUnavailable as down:
raise Refused(down.http, down.reason) from None
with hub.repository() as repo, hub._decide():
have = repo.ytd.run(run_id)
if have is not None:
return {**_run_view(repo, have), "replay": True} # the same load again writes nothing
if got.get("tax_year") != year or str(got.get("paye_ref", "")).strip().upper() != ref:
raise RuleBroken(f"the reader answered {got.get('paye_ref')} {got.get('tax_year')}, not {ref} {year}", gate="source_matches")
report = got.get("report", {}).get("employees") or []
fps = {e["payroll_id"]: e for e in (got.get("fps", {}).get("employees") or [])}
if not report:
raise RuleBroken("the provider's report holds no employees", gate="source_matches")
rows, seen = [], set()
for e in report:
pid = e["payroll_id"]
_person(pid)
if pid in seen:
raise RuleBroken(f"{pid} is twice in the provider's report", gate="source_matches")
seen.add(pid)
ni = e.get("ni") or {}
ee_ni = sum((_amount(v.get("employee"), f"{pid} employee NI {k}") for k, v in sorted(ni.items())), ZERO)
ytd_ref = f"ytd:{pid}:{year}"
rows.append({"ytd_ref": ytd_ref, "payroll_id": pid, "tax_year": year, "paye_ref": ref, "run_id": run_id, "tax_code": (e.get("tax_code") or "").strip().upper() or None,
"basis": e.get("basis") or "cumulative", "gross_amount": _amount(e["gross"], f"{pid} gross") if e.get("gross") is not None else None,
"taxable_pay_amount": _amount(e.get("taxable"), f"{pid} taxable pay", True), "tax_amount": _amount(e.get("tax"), f"{pid} tax", True),
"ee_ni_amount": money(ee_ni), "er_ni_amount": _amount(e.get("er_ni") if e.get("er_ni") is not None else
sum((_amount(v.get("employer"), f"{pid} employer NI {k}") for k, v in sorted(ni.items())), ZERO),
f"{pid} employer NI"),
"ni_by_letter_json": json.dumps(ni, sort_keys=True), "statutory_json": json.dumps({k: e[k] for k in ("ssp", "smp", "spp", "student_loan") if e.get(k) is not None},
sort_keys=True), "pension_amount": _amount(e.get("pension"), f"{pid} pension"),
"source_file_key": got.get("source_file_key")})
held = repo.ytd.held([r["ytd_ref"] for r in rows])
if held:
first = sorted(held.values(), key=lambda r: r["ytd_ref"])[0]
raise RuleBroken(f"{first['ytd_ref']} is already held by load {first['run_id']}: discard it before loading again", gate="ytd_loaded")
fps_json = {"tax_month": got.get("tax_month"), "employees": {pid: {k: f"{_amount(v.get(k), f'{pid} FPS {k}', True):.2f}" for k in FIELDS} for pid, v in sorted(fps.items())},
"totals": {k: f"{_amount(v, f'FPS total {k}', True):.2f}" for k, v in (got.get("fps", {}).get("totals") or {}).items()}}
repo.ytd.add_run({"run_id": run_id, "tax_year": year, "paye_ref": ref, "company_code": cc, "schedule_code": sched, "cutover_date": cutover, "source_ref": source_ref,
"status": "loaded", "fps_json": json.dumps(fps_json, sort_keys=True), "checks_json": "[]", "loaded_by": hub._who(caller)}, hub._now())
repo.ytd.add_openings(rows, hub._now())
with hub.repository() as repo:
return {**_run_view(repo, _get(repo, run_id)), "replay": False}
def hub_year(v: str) -> str:
from brick_payroll.application.filing import _year
return _year(v.strip())
def checks_for(rows: list, fps: dict) -> list:
"""The reconciliation: per employee against the FPS's figures (and who is in only one of them), then the totals against the FPS's totals - each check says
the provider's figure, ours and whether they are equal to the penny."""
out = []
ours = {r["payroll_id"]: r for r in rows}
col = {"taxable": "taxable_pay_amount", "tax": "tax_amount", "ee_ni": "ee_ni_amount", "er_ni": "er_ni_amount"}
missing = sorted(set(fps["employees"]) - set(ours))
extra = sorted(set(ours) - set(fps["employees"]))
out.append({"check": "employees in the FPS and in the report", "previous": str(len(fps["employees"])), "ours": str(len(ours)), "equal": not missing and not extra, "must": "equal",
**({"differs": [*(f"{m} only in the FPS" for m in missing), *(f"{e} only in the report" for e in extra)]} if missing or extra else {})})
for pid in sorted(set(fps["employees"]) & set(ours)):
for k in FIELDS:
prev, mine = Decimal(fps["employees"][pid][k]), money(ours[pid][col[k]] or 0)
out.append({"check": f"{pid} {LABELS[k]}", "previous": f"{prev:.2f}", "ours": f"{mine:.2f}", "equal": prev == mine, "must": "equal"})
for k in FIELDS:
prev = Decimal(fps["totals"][k]) if k in fps.get("totals", {}) else sum((Decimal(v[k]) for v in fps["employees"].values()), ZERO)
mine = sum((money(r[col[k]] or 0) for r in rows), ZERO)
out.append({"check": f"total {LABELS[k]}", "previous": f"{prev:.2f}", "ours": f"{mine:.2f}", "equal": prev == mine, "must": "equal"})
return out
@PRL.use_case(kind="command", route="POST /migration/ytd/{run_id}/reconcile", permission=MIG_RECONCILE, idempotent=True, tool=False)
def reconcile_ytd(caller: dict, run_id: Code) -> dict:
"""Reconcile a loaded year to date against the provider's last FPS (loaded -> reconciled): every employee's taxable pay, tax and NI, and the totals, equal
to the penny or the check says what differs. Never edits the figures: a difference is cured by discarding the load and loading again. Machine permission."""
with hub.repository() as repo, hub._decide():
run = _get(repo, run_id)
if run["status"] == "reconciled":
return _run_view(repo, run)
run_status = pr.move(run["status"], "reconcile_prl_migration", YTD_MOVES)
checks = checks_for(repo.ytd.of_run(run_id), json.loads(run["fps_json"]))
repo.ytd.save_run({**run, "status": run_status, "checks_json": json.dumps(checks, sort_keys=True)}, hub._now())
run = repo.ytd.run(run_id)
return _run_view(repo, run)
@PRL.use_case(kind="command", route="POST /migration/ytd/{run_id}/sign", permission=MIG_SIGN, idempotent=True)
def sign_ytd(caller: dict, run_id: Code) -> dict:
"""The owner signs the reconciliation (reconciled -> signed): only when every total is equal to the penny. From then the opening figures count in cumulative
tax and on the P60, and each person becomes a payroll employee (active) with the provider's tax code dated from the cut-over, source manual, the provider
named - unless they already have one. Signed is final: a correction is a later load. Needs prl:migration:sign."""
with hub.repository() as repo, hub._decide():
run = _get(repo, run_id)
if run["status"] == "signed":
return _run_view(repo, run)
pr.move(run["status"], "sign_prl_migration", YTD_MOVES)
checks = json.loads(run["checks_json"] or "[]")
bad = [c for c in checks if not c["equal"]]
if bad or not checks:
raise Refused(409, "Cannot sign: every_total_equal: " + ("; ".join(f"{c['check']}: {c['previous']} against {c['ours']}" for c in bad[:10]) or "not reconciled"),
gate="every_total_equal", fix="discard the load and load again")
repo.claims.once("ytd_sign", run_id, hub._now(), hub._who(caller))
rows = repo.ytd.of_run(run_id)
ids = [r["payroll_id"] for r in rows]
have, facts = repo.employees.many(ids), repo.employees.facts(ids, run["cutover_date"])
me, now = hub._me(caller), hub._now()
new_emps, codes, nis = [], [], []
for r in rows:
pid = r["payroll_id"]
person, n = _person(pid)
emp = have.get(pid)
if emp is None:
new_emps.append({"payroll_id": pid, "person_id": person, "n": n, "company_code": run["company_code"], "schedule_code": run["schedule_code"], "status": "active",
"starter_declaration": None, "details_by": me.email, "details_sub": me.sub})
elif emp["status"] == "pending_details":
emp["status"] = pr.move(emp["status"], "complete_payroll_details", pr.EMPLOYEE_MOVES)
emp.update({"details_by": me.email, "details_sub": me.sub})
repo.employees.save(emp, now)
if (facts.get(pid) or {}).get("tax_code") is None and r["tax_code"]:
codes.append({"payroll_id": pid, "valid_from": run["cutover_date"], "tax_code": r["tax_code"], "basis": r["basis"] or "cumulative", "source": "manual",
"notice_ref": f"year to date load {run_id} from {run['source_ref']}", "set_by": me.email, "set_sub": me.sub})
letters = sorted(json.loads(r["ni_by_letter_json"] or "{}"))
if (facts.get(pid) or {}).get("ni") is None and letters:
nis.append({"payroll_id": pid, "valid_from": run["cutover_date"], "category": letters[0], "reason": "migration", "set_by": me.email})
repo.employees.add(new_emps, now)
for table, rs in ((repo.employees.add_tax_code, codes), (repo.employees.add_ni_category, nis)):
for x in rs:
table(x, now)
repo.ytd.save_run({**run, "status": "signed", "signed_by": me.email, "signed_sub": me.sub, "signed_on": hub._today(repo)}, now)
return _run_view(repo, repo.ytd.run(run_id) or run)
@PRL.use_case(kind="command", route="POST /migration/ytd/{run_id}/discard", permission=MIG_SIGN, destructive=True)
def discard_ytd(caller: dict, run_id: Code, reason: Reason) -> dict:
"""Discard a load that is not signed (loaded or reconciled -> discarded, with a reason): its opening rows are deleted and nothing counted. A signed load is never
discarded - a correction is a later load."""
if not reason.strip():
raise hub._bounds("a reason is required to discard a load")
with hub.repository() as repo, hub._decide():
run = _get(repo, run_id)
status = pr.move(run["status"], "discard_prl_migration", YTD_MOVES)
repo.claims.once("ytd_discard", run_id, hub._now(), hub._who(caller))
repo.ytd.delete_of_run(run_id)
repo.ytd.save_run({**run, "status": status, "discarded_by": hub._who(caller), "discard_reason": reason.strip()}, hub._now())
return _run_view(repo, repo.ytd.run(run_id) or run)
@PRL.use_case(kind="query", route="GET /migration/ytd/{run_id}", permission=MIG_READ)
def get_ytd_load(caller: dict, run_id: Code) -> dict:
"""One year-to-date load: its status, totals, every reconciliation check (the provider's figure, ours, equal or not), what signing still needs, and the
opening rows."""
with hub.repository() as repo:
return _run_view(repo, _get(repo, run_id), with_rows=True)
@PRL.use_case(kind="query", route="GET /migration/ytd", permission=MIG_READ)
def list_ytd_loads(caller: dict) -> dict:
"""Every year-to-date load, oldest first."""
with hub.repository() as repo:
return {"loads": [_run_view(repo, r) for r in repo.ytd.runs()]}
# the workflow's arrows (module.yaml prl_ytd_run), the only ways a load moves; tests/test_rules.py checks them against process.yaml
YTD_MOVES = {"load_prl_ytd": (("none",), "loaded"), "reconcile_prl_migration": (("loaded",), "reconciled"), "sign_prl_migration": (("reconciled",), "signed"),
"discard_prl_migration": (("loaded", "reconciled"), "discarded")}