#!/usr/bin/env python3 """R3 - invoice reminders: an invoices CSV in, numbered payment-reminder DRAFTS out. What this is ------------ One recipe from the Back-Office Autopilot pack. You point it at the CSV your invoicing or bookkeeping tool exports (columns ``id, customer, email, amount, due_date, status``, or your own column names mapped with ``--map``). For each invoice still open it works out, from an "as of" date, whether a reminder is due - the first 1 day after the due date, the second 7 days after it - and writes the wording it *would* use into a numbered ``.txt`` file in an outbox folder. Re-running it drafts nothing twice: what has already been drafted is readable in the outbox folder itself. What this is not, and why that is the design -------------------------------------------- It does not let any of those drafts out. Not by itself, not by a flag, not by a setting. That is the central product decision of this recipe and it is deliberate, not a missing feature. A payment reminder is money language written in *your* name to a person who owes you money. It is the one message in this pack where an unreviewed automated send does the most damage. It can be wrong about the amount (the CSV can be a day stale, or can hold the invoice total when part was already paid), wrong about the date, or - worst of all - it can go out after the customer paid, which is the kind of thing that ends a relationship you are being paid to keep. The fact that would make such a message safe to let go - "is this genuinely still unpaid today, in my books?" - is not in the CSV and is not on this machine. Only you know it. So: drafts only, a human approves every one, and ``APPROVAL_CHECKLIST.md`` is the short list of checks that approval takes. There is no code path in this file that bypasses that, and the file says so in a way you can check rather than a way you have to trust. ``assert_drafts_only()`` reads this file's own source with ``ast`` each time the module is imported and each time it runs, and stops with a ``DraftsOnlyViolation`` if a mail, network or shell import, an import of ``subprocess``, or a function whose name means sending has been added to it. The only mode switch on the command line is ``--dry-run-is-all-there-is``, and what that prints is an explanation of why there is no other mode. Read-date: 2026-11-10. Stdlib only (python 3.12): argparse, ast, csv, dataclasses, datetime, pathlib, re, sys. Runs with no network and no service. """ from __future__ import annotations import argparse import ast import csv import re import sys from dataclasses import dataclass, field from datetime import date, timedelta from pathlib import Path READ_DATE = "2026-11-10" RECIPE_NAME = "R3 invoice reminders -> numbered drafts" #: The columns this recipe reads. All six must be found in the input, either by #: name or through ``--map``; two of them may be declared absent with ``-``. REQUIRED_COLUMNS = ("id", "customer", "email", "amount", "due_date", "status") #: Columns a run can do without. A missing customer falls back to the address; #: a missing status is treated as "not known to be paid" and flagged. COLUMNS_THAT_MAY_BE_ABSENT = ("customer", "status") #: The default schedule: reminder 1 one day past the due date, reminder 2 seven. DEFAULT_REMIND_DAYS = (1, 7) #: What ``--dry-run-is-all-there-is`` answers with, and the honest version of the #: same answer wherever a buyer asks for it. DRY_RUN_EXPLANATION = """\ Dry run is all there is, and that is the design, not a missing feature. This recipe prepares payment-reminder text in your name to people who owe you money. The one fact that makes such a message safe to let go - whether the invoice is genuinely still unpaid today, in your books - is not in the CSV this reads, so nothing here can decide it. Getting this wrong is not a wasted minute: a wrong amount, a wrong date, or a chase after someone already paid damages a customer relationship, and it does it under your signature. So there is no send mode to switch on. This file writes numbered .txt drafts into an outbox folder, prints what it drafted, what it held back and why, and stops. You read each draft against APPROVAL_CHECKLIST.md and press send yourself, which is about a minute for a normal run. If you genuinely want automated sending, write it in a file of your own: read the outbox, do your own approval step, and let that file carry the risk. That is allowed and expected - the promise in this file is about THIS file, not about your machine. What this file will not do is quietly become something else. """ #: A status that means "do not write to this person about this invoice". #: Compared after folding (lower-case, non-alphanumerics removed), so #: "Paid in full" and "paid-in-full" both land on ``paidinfull``. NEVER_REMIND_STATUSES = { "paid", "paidinfull", "paidfull", "settled", "cleared", "closed", "void", "voided", "cancelled", "canceled", "duplicate", "duplicated", "refunded", "writtenoff", "waived", "credit", "creditnote", "disputed", "indispute", "underdispute", "donotremind", "noremind", "onhold", "hold", } #: A status that still means money is owed, but says the figure may not be the #: one to quote. Drafted, and flagged for a human either way. CAUTION_STATUSES = { "partial", "partiallypaid", "partpaid", "partialpayment", "promised", "promisedtopay", "paymentplanned", "paymentpromised", } #: Statuses that plainly mean "still open, chase it". Anything in none of these #: three sets is drafted and reported as a status this recipe does not know, #: because guessing at a status in money language is the wrong kind of clever. KNOWN_OPEN_STATUSES = { "open", "unpaid", "due", "overdue", "unsent", "sent", "issued", "new", "pending", "awaiting", "awaitingpayment", "notpaid", } DRAFT_BANNER = "DRAFT - WRITTEN, NOT SENT. This recipe cannot send anything." #: Default wording, chosen by which reminder in the sequence it is. Placeholders #: are listed in SETUP.md. ``first_name`` is the first word of the customer. DEFAULT_TEMPLATES = { 1: """Hi {first_name}, Just a note that invoice {invoice_id}, for {amount}, was due on {due_date} and is \ {days_past_due} day{days_s} past it as of {as_of}. If it is already on its way or has been paid, reply to this and I will leave it \ alone. Otherwise {pay_note} Thanks, {signature}""", 2: """Hi {first_name}, Following up on invoice {invoice_id} for {amount}, due {due_date} and now \ {days_past_due} days past it as of {as_of}. This is the second reminder I have \ prepared for it. If something is wrong with this invoice - the amount, the date, or a payment I \ have not recorded - tell me and I will put it right before anything else goes \ out about it. {pay_note} Thanks, {signature}""", } #: Wording for a third or later step, for a buyer who configures more than two. DEFAULT_LATER_TEMPLATE = """Hi {first_name}, Third time of asking on invoice {invoice_id} for {amount}, due {due_date} and \ now {days_past_due} days past it as of {as_of}. I would rather sort this out than keep writing. If the invoice is wrong, or \ paid, say so in one line. Otherwise {pay_note} Thanks, {signature}""" DEFAULT_PAY_NOTE = "the payment details are on the invoice." DEFAULT_SIGNATURE = "(sign it in your own name)" #: Names that would mean something in this file puts a message on a wire or runs #: another program. ``mail`` is in the list, so nothing here is named after the #: address column: it is ``address`` everywhere in this file. SENDISH_NAME = re.compile( r"(?:send|smtp|mail|http|urlopen|urlread|fetch|curl|wget|post|socket|" r"publish|broadcast|relay|transmit|deliver|dispatch|push|notify|submit|" r"upload|exec|spawn|shell|system|popen|fork)", re.IGNORECASE, ) #: Imports that would break this file's promise. The real gate is the allow-list #: below; this list exists so the error can say what it found in plain words. FORBIDDEN_IMPORTS = { "smtplib": "sends mail", "aiosmtplib": "sends mail", "smtpd": "talks SMTP", "ssl": "opens network connections", "socket": "opens network connections", "http": "makes web requests", "httplib": "makes web requests", "urllib": "makes web requests", "urllib2": "makes web requests", "requests": "makes web requests", "httpx": "makes web requests", "aiohttp": "makes web requests", "ftplib": "uploads over a network", "telnetlib": "talks to a remote machine", "imaplib": "reads a mail server", "poplib": "reads a mail server", "email": "builds a sendable message", "sendgrid": "sends mail", "mailgun": "sends mail", "postmark": "sends mail", "brevo": "sends mail", "twilio": "sends messages", "boto3": "can call any AWS service", "botocore": "can call any AWS service", "subprocess": "runs another program, and a program can be a mailer", "os": "carries os.system and os.popen, so it is a shell", "multiprocessing": "runs other programs", "webbrowser": "opens things outside this program", "ctypes": "can call into the operating system", } #: Every module this file may import. Anything else stops the run. ALLOWED_IMPORTS = { "__future__", "argparse", "ast", "csv", "dataclasses", "datetime", "pathlib", "re", "sys", } #: This module's own tripwire, named plainly because the check below proves this #: function exists in this file and is still called when the module is imported. SELF_CHECK = "assert_drafts_only" #: Builtins that could rebuild a send path out of strings. FORBIDDEN_BUILTINS = {"exec", "eval", "compile", "__import__", "breakpoint"} MONEY_RE = re.compile( r"^[A-Za-z$£€¥]{0,3}\s*" r"(-?(?:\d{1,3}(?:,\d{3})+(?:\.\d{1,2})?|\d+(?:\.\d{1,2})?))" r"\s*[A-Za-z$£€¥]{0,3}$" ) ADDRESS_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]{2,}$") #: ``0001-INV-1001-reminder1.txt``: the number, the invoice, which reminder. DRAFT_NAME_RE = re.compile(r"^(\d{4,})-(.+)-reminder([1-9][0-9]*)\.txt$") #: Past this many routine holds the summary groups them instead of printing one #: line per invoice, so a 400-row ledger still reads. HOLDS_SHOWN_WHOLE = 25 class RecipeError(Exception): """A file-level problem. Named in the message, and nothing is written.""" class InputFileError(RecipeError): """The invoices CSV cannot be used at all, so the run stops before anything.""" class OutboxError(RecipeError): """The outbox path exists but is not a folder for drafts to go in.""" class InvoiceSkipped(Exception): """One row is not drafted. ``routine`` separates "this invoice is correctly not being chased" (paid, void, cancelled) from "your export has something wrong with it" (a date that will not parse, an amount that is not a number, no address). A buyer reads the first kind in a grouped line and the second kind as a CHECK line naming the file, the line and the column, so mixing them up wastes somebody's evening. """ def __init__(self, message: str, routine: bool = False): super().__init__(message) self.routine = routine class DraftsOnlyViolation(RuntimeError): """Something has been added to this file that would put a message on a wire. This is not a complaint about your data. It means the claim at the top of this file - drafts only, a human approves - has stopped being true, so the program refuses to run rather than run while saying something it no longer does. """ @dataclass class Invoice: """One usable row from the invoices CSV, already parsed and validated.""" invoice_id: str customer: str address: str amount: float due_date: date status: str origin: str caution: str = "" @dataclass class Draft: """One reminder that was written out to the outbox.""" path: Path invoice: Invoice which: int days_past_due: int days_after_due: int @dataclass class Report: """What a run did, in the order a buyer reads it.""" as_of: date input_path: Path outbox: Path remind_days: tuple[int, ...] currency: str rows_seen: int = 0 drafted: list[Draft] = field(default_factory=list) holds: list[tuple[str, str]] = field(default_factory=list) issues: list[tuple[str, str]] = field(default_factory=list) cautions: list[str] = field(default_factory=list) @property def drafted_count(self) -> int: return len(self.drafted) def summary(self) -> str: steps = ", ".join( f"reminder {which} at +{days} day(s)" for which, days in enumerate(self.remind_days, start=1) ) lines = [ f"{RECIPE_NAME} - read-date {READ_DATE}", f" input : {self.input_path}", f" outbox : {self.outbox}", f" as of : {self.as_of.isoformat()}", f" schedule : {steps}", f" rows read : {self.rows_seen}", f" drafted : {self.drafted_count} draft file(s) written " f"(drafts only - nothing was sent)", f" held back : {len(self.holds) + len(self.issues)} row(s) not drafted " f"({len(self.issues)} of them need you to look at the file)", ] for draft in self.drafted: lines.append( f" DRAFT : {draft.path.name} {draft.invoice.customer} " f"{draft.invoice.address} {format_amount(draft.invoice.amount, self.currency)} " f"{draft.days_past_due} day(s) past due [reminder {draft.which}]" ) for note in self.cautions: lines.append(f" CHECK : {note}") for who, reason in self.issues: lines.append(f" CHECK : {who} {reason}") lines.extend(_group_holds(self.holds)) lines.append("") lines.append( "Nothing left this program. Check each draft against " "APPROVAL_CHECKLIST.md, then send it yourself or delete it." ) return "\n".join(lines) def _group_holds(holds: list[tuple[str, str]]) -> list[str]: """Routine holds (paid, not yet due, already drafted) read better grouped.""" if not holds: return [] if len(holds) <= HOLDS_SHOWN_WHOLE: return [f" held : {who} {reason}" for who, reason in holds] counts: dict[str, int] = {} for _, reason in holds: counts[reason] = counts.get(reason, 0) + 1 ordered = sorted(counts.items(), key=lambda pair: (-pair[1], pair[0])) return [ f" held : {count} row(s) - {reason}" for reason, count in ordered ] def fold(value: object) -> str: """Fold a header or a status: lower-case, non-alphanumerics removed.""" return re.sub(r"[^a-z0-9]", "", str(value).lower()) def as_text(value: object) -> str: """Turn any cell value into trimmed text; an empty cell is the empty string.""" if value is None: return "" return str(value).strip() def format_amount(value: float, currency: str) -> str: """Money the way it belongs in a letter: thousands and two decimals.""" return f"{currency}{value:,.2f}" def id_token(invoice_id: str) -> str: """The file-name-safe form of an invoice id. Documented limit: two ids that differ only in punctuation fold to the same token, and then a reminder drafted for one counts as drafted for the other. Real invoice ids are file-name-safe already; if yours are not, use ``--map`` to pick a column that is. """ return re.sub(r"[^A-Za-z0-9._-]", "-", invoice_id).strip("-.") or "invoice" def parse_amount(raw: str, origin: str, column: str) -> float: """Parse an amount. Strict on purpose: this number goes out in a letter. Accepted: ``480``, ``480.00``, ``1,250.00``, ``$320.50``, ``EUR 60``. A comma separates thousands and nothing else, so a European decimal comma (``1 200,00``) is refused by name rather than read as 120000 or as 1.2. """ match = MONEY_RE.match(raw) if not match: raise InvoiceSkipped( f"column '{column}' value {raw!r} is not an amount this recipe can read. " f"Accepted shapes: 480, 480.00, 1,250.00, $320.50 (a comma separates " f"thousands; a decimal comma is not read). [{origin}] " f"Fix the value in '{column}' in your export." ) return float(match.group(1).replace(",", "")) def parse_due_date(raw: str, origin: str, column: str) -> date: """Parse an ISO due date, and name the value when it is not one.""" head = re.split(r"[T ]", raw, maxsplit=1)[0] try: return date.fromisoformat(head) except ValueError: raise InvoiceSkipped( f"column '{column}' value {raw!r} is not a date this recipe can read. " f"Dates must be YYYY-MM-DD, so 2026-11-08. [{origin}] " f"Set the export to ISO dates rather than let this program guess whether " f"03/04/2026 is a March or an April." ) def classify_status(status: str) -> tuple[str, str]: """(class, note) for one status: stop, caution, or open.""" folded = fold(status) if not folded: return "caution", "the status column is empty, so nothing says this is unpaid" if folded in NEVER_REMIND_STATUSES: return "stop", f"status is '{status}', which says this invoice is not to be reminded" if folded in CAUTION_STATUSES: return "caution", ( f"status is '{status}': money may still be owed, but the amount column may " f"hold the invoice total and not the balance - check it against your books" ) if folded in KNOWN_OPEN_STATUSES: return "open", "" return "caution", ( f"status is '{status}', which this recipe does not know. It drafted the " f"reminder rather than drop a chase on a guess - check the status and the " f"amount before you send" ) def parse_remind_days(raw: str) -> tuple[int, ...]: """Read ``--remind-days``: two or more whole days, comma separated, ascending.""" parts = [piece.strip() for piece in raw.split(",") if piece.strip()] if not parts: raise SystemExit( "ERROR: --remind-days needs at least one day count, for example " f"{DEFAULT_REMIND_DAYS[0]},{DEFAULT_REMIND_DAYS[1]}." ) steps: list[int] = [] for piece in parts: if not piece.isdigit() or int(piece) < 1: raise SystemExit( f"ERROR: --remind-days part {piece!r} is not a whole number of days of " f"at least 1. A reminder written on or before the due date is a " f"different thing (a courtesy note), not a payment chase." ) steps.append(int(piece)) return tuple(sorted(set(steps))) def parse_maps(pairs: list[str]) -> dict[str, str]: """Read repeated ``--map column=TheirHeader`` arguments.""" mapping: dict[str, str] = {} for pair in pairs: if "=" not in pair: raise SystemExit( f"ERROR: --map {pair!r} is not in the form column=TheirHeader, for " f"example --map due_date=\"Invoice Due Date\"." ) column, header = (piece.strip() for piece in pair.split("=", 1)) if column not in REQUIRED_COLUMNS: raise SystemExit( f"ERROR: --map names column {column!r}, which this recipe does not " f"read. Its columns are {', '.join(REQUIRED_COLUMNS)}." ) mapping[column] = header return mapping @dataclass class ColumnMap: """Which header in the buyer's file means which column this recipe reads.""" wanted: dict[str, str] = field(default_factory=dict) present: list[str] = field(default_factory=list) def declared_absent(self, column: str) -> bool: """True when the buyer said, to my face, that this column does not exist.""" return self.wanted.get(column) == "-" def index(self, column: str) -> int | None: """Position of one column's header in the file, or None if it is not there.""" if self.declared_absent(column): return None folded = fold(self.wanted.get(column, column)) for position, header in enumerate(self.present): if fold(header) == folded: return position return None def check(self, path: Path) -> None: """Stop the run, naming the file and the columns, if anything is missing.""" missing = [column for column in REQUIRED_COLUMNS if self.index(column) is None and not (self.declared_absent(column) and column in COLUMNS_THAT_MAY_BE_ABSENT)] if not missing: return refused = [column for column in missing if self.declared_absent(column)] mapped_wrong = [ f"{column} was mapped to {self.wanted[column]!r}, which is not one of its " f"headers" for column in missing if column in self.wanted ] raise InputFileError( f"{path}: no column for {', '.join(missing)}. Its header row is " f"{self.present}." + (f" {'; '.join(mapped_wrong)}." if mapped_wrong else "") + (f" You declared {', '.join(refused)} absent with '-', and this recipe " f"cannot write a reminder without it." if refused else "") + f" Either name the columns in your export, or map them: " f"{_map_hint(missing, self)}. Only {', '.join(COLUMNS_THAT_MAY_BE_ABSENT)} " f"may be declared absent with '-': without an id, an amount, a due date " f"and an address there is nothing to chase and nobody to chase it about." ) def _map_hint(missing: list[str], column_map: ColumnMap) -> str: """Suggest a ``--map`` for each missing column, from the file's own headers.""" hints = [] for column in missing: needle = fold(column)[:4] best = next( (header for header in column_map.present if needle in fold(header)), "TheirHeader", ) hints.append(f"--map {column}=\"{best}\"") return " or ".join(hints) def parse_row(cells: dict[str, str], file_name: str, origin: str) -> Invoice: """Turn one row into an Invoice, or raise InvoiceSkipped saying why it cannot.""" invoice_id = cells["id"] if not invoice_id: raise InvoiceSkipped( f"column 'id' is empty, so this row has no invoice number to put in a " f"letter. [{origin}]" ) kind, note = classify_status(cells["status"]) if kind == "stop": raise InvoiceSkipped(note, routine=True) due_date = parse_due_date(cells["due_date"], origin, "due_date") if not cells["amount"]: raise InvoiceSkipped( f"column 'amount' is empty, so there is no figure to state. [{origin}]" ) amount = parse_amount(cells["amount"], origin, "amount") address = cells["email"] if not ADDRESS_RE.match(address): raise InvoiceSkipped( f"column 'email' value {address!r} is not an address, so there is nobody " f"to write to. [{origin}] Fix the value in 'email' in {file_name}." ) return Invoice( invoice_id=invoice_id, customer=cells["customer"] or address.split("@")[0], address=address, amount=amount, due_date=due_date, status=cells["status"], origin=origin, caution=note if kind == "caution" else "", ) def read_invoices( path: Path, column_map: ColumnMap ) -> tuple[list[Invoice], list[tuple[str, str]], list[tuple[str, str]]]: """Phase 1: read and validate every row. Writes nothing. Returns (usable invoices, rows with something wrong with them, rows held for a routine reason). A header problem stops the run. A row problem skips that row and names the file, the line number, the column and the value, so a person can fix their export in one pass instead of one argument at a time. """ if not path.is_file(): raise InputFileError( f"{path}: no such file. Pass --input the CSV your invoicing tool exports." ) with path.open(encoding="utf-8-sig", newline="") as handle: rows = list(csv.reader(handle)) if not rows or not any(cell.strip() for cell in rows[0]): raise InputFileError( f"{path}: has no header row, so no column can be named. Its first line " f"must name the columns, or the ones that are there must be mapped with " f"--map." ) column_map.present = [cell.strip() for cell in rows[0]] column_map.check(path) positions = {column: column_map.index(column) for column in REQUIRED_COLUMNS} invoices: list[Invoice] = [] issues: list[tuple[str, str]] = [] routine: list[tuple[str, str]] = [] for line_number, row in enumerate(rows[1:], start=2): if not any(cell.strip() for cell in row): continue origin = f"{path.name} line {line_number}" cells = { column: as_text(row[position]) if position is not None and position < len(row) else "" for column, position in positions.items() } label = cells["id"] or "(no id)" try: invoices.append(parse_row(cells, path.name, origin)) except InvoiceSkipped as exc: if exc.routine: routine.append((f"{label} [{origin}]", str(exc))) else: issues.append((label, str(exc))) return invoices, issues, routine def drafted_index(outbox: Path) -> dict[str, dict[int, str]]: """Which reminders already exist, read off the outbox folder's own file names. This is the recipe's whole idea of state: a folder of files a human can read, not a hidden database. Delete a draft file and that reminder falls due again, which is on purpose. """ index: dict[str, dict[int, str]] = {} if not outbox.is_dir(): return index for path in sorted(outbox.glob("*.txt")): match = DRAFT_NAME_RE.match(path.name) if not match: continue index.setdefault(match.group(2), {})[int(match.group(3))] = path.name return index def next_sequence(outbox: Path) -> int: """The next draft number, so a new run never overwrites an old draft.""" highest = 0 if outbox.is_dir(): for path in outbox.glob("*.txt"): match = DRAFT_NAME_RE.match(path.name) if match: highest = max(highest, int(match.group(1))) return highest + 1 def check_outbox(outbox: Path) -> None: """A path that exists but is not a folder is a named error, not a surprise.""" if outbox.exists() and not outbox.is_dir(): raise OutboxError( f"{outbox}: this is the folder drafts are written to, but a file is " f"sitting there. Move it aside, or pass --outbox somewhere else." ) def open_by_address(invoices: list[Invoice]) -> dict[str, list[Invoice]]: """Open invoices grouped by address, so one person can be sent one letter.""" grouped: dict[str, list[Invoice]] = {} for invoice in invoices: grouped.setdefault(invoice.address.lower(), []).append(invoice) return grouped def wording_for(which: int, template: str) -> str: """The buyer's own wording if they supplied one, else the built-in one.""" if template: return template return DEFAULT_TEMPLATES.get(which, DEFAULT_LATER_TEMPLATE) def render_body( invoice: Invoice, which: int, days_past_due: int, as_of: date, template: str, pay_note: str, signature: str, currency: str, ) -> str: """Fill the wording. A placeholder this recipe does not offer is named, not skipped.""" values = { "customer": invoice.customer, "first_name": (invoice.customer.split() or ["there"])[0], "invoice_id": invoice.invoice_id, "amount": format_amount(invoice.amount, currency), "due_date": invoice.due_date.isoformat(), "as_of": as_of.isoformat(), "days_past_due": str(days_past_due), "days_s": "s" if days_past_due != 1 else "", "step": str(which), "pay_note": pay_note, "signature": signature, } try: return template.format_map(values).rstrip() + "\n" except KeyError as exc: raise RecipeError( f"the reminder wording for reminder {which} holds {{{exc.args[0]}}}, which " f"this recipe does not offer. The placeholders it does offer are: " f"{', '.join(sorted(values))}." ) except (IndexError, ValueError) as exc: raise RecipeError( f"the reminder wording for reminder {which} is not a valid template " f"({exc}). A placeholder is a word in braces, like {{invoice_id}}." ) def write_draft( outbox: Path, seq: int, invoice: Invoice, which: int, remind_days: tuple[int, ...], days_past_due: int, as_of: date, body: str, signature: str, template: str, currency: str, siblings: list[Invoice], ) -> Draft: """Write one draft file. Its name carries its number, its invoice, its step.""" outbox.mkdir(parents=True, exist_ok=True) path = outbox / f"{seq:04d}-{id_token(invoice.invoice_id)}-reminder{which}.txt" subject = ( f"Invoice {invoice.invoice_id} - {days_past_due} " f"day{'s' if days_past_due != 1 else ''} past due" + (" - second reminder" if which == 2 else "") + (" (chasing)" if which > 2 else "") ) others = [ f"{other.invoice_id} {format_amount(other.amount, currency)} due {other.due_date}" for other in siblings ] fields = [ ("file", path.name), ("as of", as_of.isoformat()), ("invoice", invoice.invoice_id), ("customer", invoice.customer), ("to", invoice.address), ("amount stated", format_amount(invoice.amount, currency)), ("due date", invoice.due_date.isoformat()), ("days past due", str(days_past_due)), ("reminder", f"{which} of {len(remind_days)} " f"(steps are +{', +'.join(str(day) for day in remind_days)} day(s))"), ("status read", invoice.status or "(empty)"), ("other open", "; ".join(others) if others else "nothing else open for this address"), ("wording from", "your --template file" if template not in DEFAULT_TEMPLATES.values() and template is not DEFAULT_LATER_TEMPLATE else "the built-in wording"), ("signed", signature), ] if invoice.caution: fields.append(("CHECK FIRST", invoice.caution)) header = [DRAFT_BANNER, f"{RECIPE_NAME} read-date {READ_DATE}", ""] header += [f"{label:<15}: {value}" for label, value in fields] header += [ "", "-- Check this against APPROVAL_CHECKLIST.md before it goes anywhere. --", "-- Nothing here is sent by the program. You press send, or you delete it. --", "", f"Subject: {subject}", f"To: {invoice.address}", "", body, ] path.write_text("\n".join(header), encoding="utf-8") return Draft( path=path, invoice=invoice, which=which, days_past_due=days_past_due, days_after_due=remind_days[which - 1], ) def load_template(path_text: str) -> str: """Read a buyer's own wording file, or return empty for the built-in one.""" if not path_text: return "" path = Path(path_text) if not path.is_file(): raise RecipeError( f"--template {path} is not a file. Point it at a plain text file holding " f"the wording, or leave it out and use the built-in wording." ) text = path.read_text(encoding="utf-8").strip() if not text: raise RecipeError(f"--template {path} is empty, so every draft would be blank.") return text def hold_reason( invoice: Invoice, as_of: date, remind_days: tuple[int, ...], done: dict[int, str] ) -> str: """Why one open invoice was not drafted this time, in one readable line.""" if not done and as_of < invoice.due_date + timedelta(days=remind_days[0]): first_due = invoice.due_date + timedelta(days=remind_days[0]) return ( f"not yet due for a reminder: reminder 1 falls {first_due.isoformat()} " f"(due {invoice.due_date.isoformat()} + {remind_days[0]} day(s))" ) outstanding = [which for which in range(1, len(remind_days) + 1) if which not in done] if not outstanding: return ( f"all {len(remind_days)} reminder(s) already drafted (" f"{', '.join(done[which] for which in sorted(done))}) - this recipe has " f"nothing more to prepare" ) due_on = invoice.due_date + timedelta(days=remind_days[outstanding[0] - 1]) return ( f"reminder {outstanding[0]} is not due until {due_on.isoformat()}" + (f" (reminder {min(done)} already drafted as {done[min(done)]})" if done else "") ) def run( input_path: Path, outbox: Path, as_of: date, remind_days: tuple[int, ...] = DEFAULT_REMIND_DAYS, header_map: dict[str, str] | None = None, template: str = "", pay_note: str = DEFAULT_PAY_NOTE, signature: str = DEFAULT_SIGNATURE, currency: str = "$", ) -> Report: """One full run: read the invoices, decide what is due, write the drafts.""" assert_drafts_only() check_outbox(outbox) column_map = ColumnMap(wanted=dict(header_map or {})) invoices, issues, routine_holds = read_invoices(input_path, column_map) report = Report( as_of=as_of, input_path=input_path, outbox=outbox, remind_days=remind_days, currency=currency, rows_seen=len(invoices) + len(issues) + len(routine_holds), issues=issues, holds=list(routine_holds), ) already = drafted_index(outbox) by_address = open_by_address(invoices) seq = next_sequence(outbox) for invoice in invoices: token = id_token(invoice.invoice_id) done = already.setdefault(token, {}) due_which = next( (which for which, days in enumerate(remind_days, start=1) if which not in done and as_of >= invoice.due_date + timedelta(days=days)), None, ) if due_which is None: report.holds.append( (f"{invoice.invoice_id} [{invoice.origin}]", hold_reason(invoice, as_of, remind_days, done)) ) continue days_past_due = (as_of - invoice.due_date).days template_used = wording_for(due_which, template) body = render_body( invoice, due_which, days_past_due, as_of, template_used, pay_note, signature, currency, ) siblings = [ other for other in by_address[invoice.address.lower()] if other.invoice_id != invoice.invoice_id ] draft = write_draft( outbox, seq, invoice, due_which, remind_days, days_past_due, as_of, body, signature, template_used, currency, siblings, ) report.drafted.append(draft) done[due_which] = draft.path.name seq += 1 if invoice.caution: report.cautions.append(f"{draft.path.name} {invoice.caution}") if siblings: report.cautions.append( f"{draft.path.name} {invoice.address} also has " f"{', '.join(other.invoice_id for other in siblings)} open - one letter " f"covering them is usually the right call, and one letter per customer " f"per step is the rule in APPROVAL_CHECKLIST.md" ) return report # ------------------------------------------------------- the drafts-only guard def _forbidden_import_nodes(tree: ast.AST) -> list[str]: """Every import in this source that this source may not have.""" found: list[str] = [] for node in ast.walk(tree): if isinstance(node, ast.Import): roots = [alias.name.split(".")[0] for alias in node.names] elif isinstance(node, ast.ImportFrom) and node.module: roots = [node.module.split(".")[0]] else: continue for root in roots: where = f"line {getattr(node, 'lineno', '?')}" if root in FORBIDDEN_IMPORTS: found.append(f"{where}: imports {root}, which {FORBIDDEN_IMPORTS[root]}") elif root not in ALLOWED_IMPORTS: found.append( f"{where}: imports {root}, which is not among the modules this file " f"is allowed to use ({', '.join(sorted(ALLOWED_IMPORTS))})" ) return found def _forbidden_callable_nodes(tree: ast.AST) -> list[str]: """Every definition or call in this source whose name means sending something out. Calls made to raise an error are skipped: raising cannot put a message on a wire, and ``SystemExit`` contains a word on the list. """ raised = { id(node.exc) for node in ast.walk(tree) if isinstance(node, ast.Raise) and node.exc is not None } found: list[str] = [] for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): if SENDISH_NAME.search(node.name): found.append( f"line {node.lineno}: defines {node.name}, and a name like that " f"means something goes out" ) elif isinstance(node, ast.Call) and id(node) not in raised: bare = isinstance(node.func, ast.Name) if bare: name = node.func.id elif isinstance(node.func, ast.Attribute): name = node.func.attr else: name = "" # The builtins are only the bare ones: ``compile(...)`` could build a # send path out of text, while ``re.compile(...)`` is the same word on # a perfectly ordinary module and has to pass. if bare and name in FORBIDDEN_BUILTINS: found.append( f"line {node.lineno}: calls {name}, which could build a send path " f"out of text" ) elif name and SENDISH_NAME.search(name): found.append( f"line {node.lineno}: calls {name}(), and a name like that means " f"something goes out" ) return found def assert_drafts_only(source: Path | None = None) -> None: """Stop the run if this file has stopped being a drafts-only file. Reads this file's own source and checks three things: every import is on the allow-list, nothing defined or called in it has a name that means putting a message on a wire or running a program, and this check is still present and still called when the module is imported. What that is: a tripwire against a quiet retrofit, so whoever edits this file later finds out in one second instead of one invoice cycle. What it is not: a security boundary. It is plain text in a file the buyer owns, and anyone determined can delete the check. What breaks when they do is the sentence at the top of this file, and the pack's promise about THIS file goes with it - which is why ``RECIPE.md`` tells a buyer who wants automated sending to write their own file and carry their own approval step. """ path = Path(source) if source is not None else Path(__file__).resolve() try: text = path.read_text(encoding="utf-8") except OSError as exc: raise DraftsOnlyViolation( f"I cannot read my own source at {path} ({exc}), so I cannot prove I am " f"still a drafts-only file. I refuse to run rather than run unproven." ) from exc try: tree = ast.parse(text) except SyntaxError as exc: raise DraftsOnlyViolation( f"{path} does not parse as python ({exc}). The drafts-only check cannot run " f"on a file that is not valid python, so nothing is being written." ) from exc problems = _forbidden_import_nodes(tree) + _forbidden_callable_nodes(tree) defined = { node.name for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) } if SELF_CHECK not in defined: problems.append( f"the function {SELF_CHECK} is no longer in this file, so nothing checks " f"this file's own promise any more" ) elif not any( isinstance(node, ast.Expr) and isinstance(node.value, ast.Call) and isinstance(node.value.func, ast.Name) and node.value.func.id == SELF_CHECK for node in tree.body ): problems.append( f"{SELF_CHECK} is still defined but is no longer called when the module is " f"imported, so it could never fire" ) if problems: raise DraftsOnlyViolation( f"{path.name} has had something added that would let a message out, so it " f"will not run while its own header says drafts only.\n" f" what the check found:\n" + "".join(f" - {problem}\n" for problem in problems) + f" why this stops the run: the whole promise of this recipe is that " f"payment reminders are written and then approved by a human. A send path in " f"this file breaks that promise, and this program will not put money language " f"in your name out into the world without one.\n" f" what to do instead: leave this file as it is and write the sending you " f"want in a file of your own, with your own approval step in it - see " f"RECIPE.md. That decision is yours to make on your machine. If you would " f"genuinely rather have a sending version of this recipe, ask for one: it is " f"a different product and it is priced and documented as one." ) assert_drafts_only() # ------------------------------------------------------------------ command line def parse_as_of(raw: str) -> date: """Resolve ``--as-of`` to a date, or say what would not parse.""" if not raw: return date.today() try: return date.fromisoformat(raw) except ValueError: raise SystemExit( f"ERROR: --as-of {raw!r} is not an ISO date. Use YYYY-MM-DD, for example " f"2026-11-10." ) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="invoice_reminders.py", description=( "R3: read an invoices CSV and write numbered payment-reminder DRAFTS to an " "outbox folder. It cannot send anything and there is no flag to make it try." ), epilog=( "Drafts only, by design: a human approves every payment reminder. See " "APPROVAL_CHECKLIST.md, and --dry-run-is-all-there-is for why there is no " "other mode." ), ) parser.add_argument("--input", default="samples/invoices_sample.csv", help="your invoices CSV (default: %(default)s)") parser.add_argument("--as-of", default="", dest="as_of", help="the date the run is measured from, YYYY-MM-DD " "(default: today)") parser.add_argument("--outbox", default="outbox", help="folder the drafts are written to (default: %(default)s)") parser.add_argument("--remind-days", default="1,7", dest="remind_days", help="days past the due date for each reminder step " "(default: %(default)s)") parser.add_argument("--map", action="append", default=[], dest="maps", metavar="COLUMN=TheirHeader", help="name a column this recipe reads after the header in your " "file, e.g. --map due_date=\"Invoice Due Date\". Repeat " "it. '-' declares the column absent (customer and status " "only).") parser.add_argument("--template", default="", help="a file of your own wording for every reminder") parser.add_argument("--pay-note", default=DEFAULT_PAY_NOTE, dest="pay_note", help="the one sentence saying how to pay (default: %(default)s)") parser.add_argument("--signature", default=DEFAULT_SIGNATURE, help="how the draft signs off (default: %(default)s)") parser.add_argument("--currency", default="$", help="what goes in front of every amount (default: %(default)s)") parser.add_argument("--quiet", action="store_true", help="write the drafts but print nothing") parser.add_argument("--dry-run-is-all-there-is", action="store_true", dest="why_no_other_mode", help="print why there is no send mode, and exit") return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) if args.why_no_other_mode: print(DRY_RUN_EXPLANATION, end="") return 0 try: report = run( input_path=Path(args.input), outbox=Path(args.outbox), as_of=parse_as_of(args.as_of), remind_days=parse_remind_days(args.remind_days), header_map=parse_maps(args.maps), template=load_template(args.template), pay_note=args.pay_note, signature=args.signature, currency=args.currency, ) except (RecipeError, DraftsOnlyViolation) as exc: print(f"ERROR: {exc}", file=sys.stderr) return 2 if not args.quiet: print(report.summary()) return 0 if __name__ == "__main__": raise SystemExit(main())