dodil/erp-brick-payroll
DODIL ERP brick: payroll
src/brick_payroll/infrastructure/rti.py
"""The files handed to the capabilities, built DETERMINISTICALLY from a committed run: the same run, identities and records give the same bytes and
the same sha256 (no clock, no random id), so a retry or a second worker hands over the identical file and a provider's replay is real.
FPS raw/payroll/<company>/rti/fps/<run>-<n>.xml what each person was paid and deducted in the period and to date (PRL-13)
EPS raw/payroll/<company>/rti/eps/<paye ref>-<year>-M<nn>-<n>.xml statutory pay recovered to date, or no payment for the month (PRL-14)
pension raw/payroll/<company>/pension/<scheme>/<run>.csv one row per member (PAPDIS-shaped, PRL-15)
The XML is this brick's own document of the facts HMRC's RTI asks for; it is NOT certified against HMRC's schema (no credential, no live link here):
the provider behind hmrc.rti_submit maps it to the wire format. Identity (HR-61) is written only into these files."""
import base64
import csv
import hashlib
import io
from datetime import date
from decimal import Decimal
from xml.sax.saxutils import escape
from brick_payroll.domain import pay_run as pr
from brick_payroll.domain.money import money
def _a(v) -> str:
return escape(str(v), {'"': """})
def _amt(v) -> str:
return f"{money(v if v is not None else 0):.2f}"
def document(object_key: str, kind: str, content: bytes, content_type: str, pay_run_id: str = "", ref: str = "", published_on=None, retain_until=None) -> dict:
return {"object_key": object_key, "kind": kind, "payroll_id": None, "person_id": None, "ref": ref, "pay_run_id": pay_run_id or None,
"content_type": content_type, "sha256": hashlib.sha256(content).hexdigest(), "size_bytes": len(content),
"content_b64": base64.b64encode(content).decode(), "published_on": published_on, "retain_until": retain_until}
def fps_key(run: pr.PayRun, n: int) -> str:
return f"raw/payroll/{run.company_code}/rti/fps/{run.pay_run_id}-{n}.xml"
def eps_key(company: str, paye_ref: str, tax_year: str, month: int, n: int) -> str:
return f"raw/payroll/{company}/rti/eps/{paye_ref.replace('/', '_')}-{tax_year}-M{month:02d}-{n}.xml"
def pension_key(run: pr.PayRun, scheme: str) -> str:
return f"raw/payroll/{run.company_code}/pension/{scheme}/{run.pay_run_id}.csv"
def _sum(slip: pr.Payslip, element: str):
return sum((ln.amount for ln in slip.lines if ln.element_code == element), Decimal(0))
def fps_xml(run: pr.PayRun, n: int, accounts_office_ref: str, slips: list, idents: dict, employees: dict) -> bytes:
"""One <Employee> per payslip, in payroll_id order: identity, code and basis, starter or leaver details, pay and deductions in the period and to date
(this employer's year to date - exactly the payslip's), statutory pay and pension in the period. `slips` carry their lines."""
o = ['<?xml version="1.0" encoding="UTF-8"?>', f'<FullPaymentSubmission schema="dodil-payroll-fps-1" run="{_a(run.pay_run_id)}" n="{n}">',
f' <Employer><PayeReference>{_a(run.paye_ref)}</PayeReference><AccountsOfficeReference>{_a(accounts_office_ref)}</AccountsOfficeReference></Employer>',
f' <TaxYear>{_a(run.tax_year)}</TaxYear><PayFrequency>{_a(run.frequency)}</PayFrequency><TaxPeriod>{run.tax_period}</TaxPeriod>'
f'<PaymentDate>{run.pay_date.isoformat()}</PaymentDate>']
for p in sorted(slips, key=lambda s: s.payroll_id):
who, emp = idents[p.payroll_id], employees.get(p.payroll_id) or {}
y = p.ytd or {}
o.append(f' <Employee payrollId="{_a(p.payroll_id)}">')
if who.get("ni_number"):
o.append(f' <NINO>{_a(who["ni_number"])}</NINO>')
else: # HMRC takes the address of someone with no NI number
a = who.get("home_address") or {}
o.append(" <Address>" + "".join(f"<{k}>{_a(a[k])}</{k}>" for k in sorted(a)) + "</Address>")
o.append(f' <Name><Forenames>{_a(who.get("legal_first_names", ""))}</Forenames><Surname>{_a(who.get("legal_last_name", ""))}</Surname></Name>'
f'<BirthDate>{_a(who.get("date_of_birth", ""))}</BirthDate>')
o.append(f' <TaxCode basis="{_a(p.basis_used)}">{_a(p.tax_code_used)}</TaxCode><NICategory>{_a(p.ni_category_used)}</NICategory>')
start = emp.get("start_date")
if start and run.period_from <= start <= run.period_to:
o.append(f' <Starter statement="{_a(emp.get("starter_declaration") or "")}" startDate="{start.isoformat()}"/>')
if p.leaving_date:
o.append(f' <Leaver leavingDate="{p.leaving_date.isoformat()}" afterLeaving="{str(emp.get("status") == "left" and p.leaving_date < run.period_from).lower()}"/>')
o.append(f' <Pay taxablePayInPeriod="{_amt(p.taxable)}" taxablePayToDate="{_amt(y.get("taxable"))}" grossInPeriod="{_amt(p.gross)}" '
f'grossToDate="{_amt(y.get("gross"))}"/>')
o.append(f' <Tax inPeriod="{_amt(p.tax)}" toDate="{_amt(y.get("tax"))}"/>')
o.append(f' <NIC employeeInPeriod="{_amt(p.ee_ni)}" employeeToDate="{_amt(y.get("ee_ni"))}" employerInPeriod="{_amt(p.er_ni)}" '
f'employerToDate="{_amt(y.get("er_ni"))}" earningsForNIC="{_amt(sum((ln.amount for ln in p.lines if ln.kind == "earning"), Decimal(0)))}"/>')
o.append(f' <Statutory ssp="{_amt(_sum(p, "SSP"))}" smp="{_amt(_sum(p, "SMP"))}"/>')
o.append(f' <Pension employeeInPeriod="{_amt(p.ee_pension)}" employerInPeriod="{_amt(p.er_pension)}"/>')
o.append(f' <NetPay>{_amt(p.net)}</NetPay>')
o.append(" </Employee>")
o += ["</FullPaymentSubmission>", ""]
return "\n".join(o).encode("utf-8")
def eps_xml(paye_ref: str, accounts_office_ref: str, tax_year: str, month: int, n: int, ytd_recovered, no_payment: bool, allowance: bool) -> bytes:
o = ['<?xml version="1.0" encoding="UTF-8"?>', f'<EmployerPaymentSummary schema="dodil-payroll-eps-1" n="{n}">',
f' <Employer><PayeReference>{_a(paye_ref)}</PayeReference><AccountsOfficeReference>{_a(accounts_office_ref)}</AccountsOfficeReference></Employer>',
f' <TaxYear>{_a(tax_year)}</TaxYear><TaxMonth>{month}</TaxMonth>']
o.append(" <NoPaymentForPeriod>true</NoPaymentForPeriod>" if no_payment else " <NoPaymentForPeriod>false</NoPaymentForPeriod>")
o.append(f' <RecoverableAmounts smpRecoveredToDate="{_amt(ytd_recovered)}" nicCompensationToDate="0.00"/>')
o.append(f" <EmploymentAllowanceIndicator>{str(allowance).lower()}</EmploymentAllowanceIndicator>")
o += ["</EmployerPaymentSummary>", ""]
return "\n".join(o).encode("utf-8")
PENSION_COLUMNS = ("payroll_id", "ni_number", "surname", "forenames", "date_of_birth", "pensionable_earnings", "employee_contribution", "employer_contribution",
"enrolment_status", "enrolment_date", "assessed_as")
def pension_csv(run: pr.PayRun, rows: list) -> bytes:
"""One row per member of the scheme in this run. rows: dicts with the PENSION_COLUMNS keys, written in payroll_id order, \\n line ends."""
out = io.StringIO()
w = csv.writer(out, lineterminator="\n")
w.writerow(PENSION_COLUMNS)
for r in sorted(rows, key=lambda r: r["payroll_id"]):
w.writerow([r[c] for c in PENSION_COLUMNS])
return out.getvalue().encode("utf-8")
def pension_due(pay_date: date) -> date:
"""The 22nd of the month after the pay date (the legal due date for the contributions to reach the scheme, electronic payment)."""
y, m = (pay_date.year, pay_date.month + 1) if pay_date.month < 12 else (pay_date.year + 1, 1)
return date(y, m, 22)