dodil/erp-brick-payroll

DODIL ERP brick: payroll

src/brick_payroll/infrastructure/documents.py

"""The payslip as a document: an itemised pay statement (Employment Rights Act 1996 s. 8) rendered as a one-page PDF,
deterministically (the same payslip gives the same bytes and the same sha256), with no PDF library. Kept under the spec's
object key raw/payroll/<company>/<tax year>/payslip/<person_id>/<pay_run_id>.pdf."""
import base64
import hashlib

from brick_payroll.domain import pay_run as pr
from brick_payroll.domain import paye

NAMES = {"SALARY": "Salary", "UNPAID_DEDUCTION": "Unpaid leave", "HOLIDAY_IN_LIEU": "Holiday pay in lieu", "TAX": "PAYE tax",
         "EE_NI": "National Insurance", "EE_PENSION": "Pension (net pay arrangement)", "ER_NI": "Employer National Insurance",
         "ER_PENSION": "Employer pension", "SICK_DEDUCTION": "Sickness (salary not paid)", "SSP": "Statutory sick pay",
         "MATERNITY_DEDUCTION": "Maternity leave (salary not paid)", "SMP": "Statutory maternity pay"}


def object_key(run: pr.PayRun, p: pr.Payslip) -> str:
    return f"raw/payroll/{run.company_code}/{run.tax_year}/payslip/{p.person_id}/{run.pay_run_id}.pdf"


def statement_lines(run: pr.PayRun, p: pr.Payslip) -> list:
    out = [f"Payslip {run.pay_run_id}", f"Payroll ID {p.payroll_id}    Pay date {run.pay_date.isoformat()}    Tax year {run.tax_year} {paye.PERIOD_WORD[run.frequency]} {run.tax_period}",
           f"Tax code {p.tax_code_used} ({p.basis_used})    NI category {p.ni_category_used}", ""]
    for kind, title in (("earning", "Payments"), ("deduction", "Deductions"), ("employer", "Paid by your employer")):
        out.append(title)
        for ln in p.lines:
            if ln.kind == kind:
                qty = f" {ln.quantity} x {ln.rate_amount}" if ln.quantity is not None and ln.rate_amount is not None else ""
                out.append(f"  {NAMES.get(ln.element_code, ln.element_code):<34}{qty:>22} {ln.amount:>12}")
        out.append("")
    out += [f"Gross pay {p.gross}    Taxable pay {p.taxable}    Net pay GBP {p.net}",
            "To date: " + "  ".join(f"{k} {p.ytd.get(k)}" for k in ("gross", "taxable", "tax", "ee_ni", "ee_pension"))]
    return out


def _esc(t: str) -> str:
    return t.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)").encode("ascii", "replace").decode()


def render_pdf(lines: list) -> bytes:
    text = "BT /F1 10 Tf 13 TL 50 790 Td " + " ".join(f"({_esc(t)}) Tj T*" for t in lines) + " ET"
    objs = ["<< /Type /Catalog /Pages 2 0 R >>", "<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
            "<< /Type /Page /Parent 2 0 R /MediaBox [0 0 595 842] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>",
            "<< /Type /Font /Subtype /Type1 /BaseFont /Courier >>", f"<< /Length {len(text)} >>\nstream\n{text}\nendstream"]
    out, offsets = "%PDF-1.4\n", []
    for i, o in enumerate(objs, 1):
        offsets.append(len(out))
        out += f"{i} 0 obj\n{o}\nendobj\n"
    xref = len(out)
    out += f"xref\n0 {len(objs) + 1}\n0000000000 65535 f \n" + "".join(f"{off:010d} 00000 n \n" for off in offsets)
    out += f"trailer\n<< /Size {len(objs) + 1} /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n"
    return out.encode("ascii")


def payslip_document(run: pr.PayRun, p: pr.Payslip, published_on, retain_until) -> dict:
    content = render_pdf(statement_lines(run, p))
    return {"object_key": object_key(run, p), "kind": "payslip", "payroll_id": p.payroll_id, "person_id": p.person_id, "ref": run.pay_run_id,
            "pay_run_id": run.pay_run_id, "content_type": "application/pdf", "sha256": hashlib.sha256(content).hexdigest(),
            "size_bytes": len(content), "content_b64": base64.b64encode(content).decode(), "published_on": published_on,
            "retain_until": retain_until}


def p45_object_key(run: pr.PayRun, p: pr.Payslip) -> str:
    return f"raw/payroll/{run.company_code}/{run.tax_year}/p45/{p.person_id}/{run.pay_run_id}.pdf"


def p45_lines(run: pr.PayRun, p: pr.Payslip, who: dict, pay_to_date, tax_to_date) -> list:
    """P45 Part 1A (the employee's copy): who left, when, the code they were on and the pay and tax to date in the tax year. The identity is
    hr's (HR-61) and goes only into this file."""
    name = f"{who.get('legal_first_names', '')} {who.get('legal_last_name', '')}".strip()
    return [f"P45 Part 1A - details of employee leaving work    Tax year {run.tax_year}", f"Employer PAYE reference {run.paye_ref}    Payroll ID {p.payroll_id}", "",
            f"Employee {name}    NI number {who.get('ni_number') or 'none held'}    Date of birth {who.get('date_of_birth') or ''}",
            f"Leaving date {p.leaving_date.isoformat() if p.leaving_date else ''}    Final payment {run.pay_date.isoformat()} ({run.pay_run_id})", "",
            f"Tax code at leaving {p.tax_code_used}{' Week 1 / Month 1' if p.basis_used == 'week1_month1' else ''}",
            f"Total pay to date in this tax year GBP {pay_to_date}", f"Total tax to date in this tax year GBP {tax_to_date}", "",
            "Give this part to the employee. Keep a copy for HMRC's record-keeping period."]


def p45_document(run: pr.PayRun, p: pr.Payslip, who: dict, pay_to_date, tax_to_date, published_on, retain_until) -> dict:
    content = render_pdf(p45_lines(run, p, who, pay_to_date, tax_to_date))
    return {"object_key": p45_object_key(run, p), "kind": "p45", "payroll_id": p.payroll_id, "person_id": p.person_id, "ref": run.pay_run_id,
            "pay_run_id": run.pay_run_id, "content_type": "application/pdf", "sha256": hashlib.sha256(content).hexdigest(), "size_bytes": len(content),
            "content_b64": base64.b64encode(content).decode(), "published_on": published_on, "retain_until": retain_until}