#!/usr/bin/env python3 """Money referee. Usage: python3 money-referee.py categories.csv EXPORT.csv [EXPORT.csv ...] Re-reads every bank export on its own and checks categories.csv against it: * every export row appears in categories.csv exactly once (matched by source_file and row, row = data row number starting at 1) * date, description and amount are identical to the export * category is nonempty * categories.csv has no rows that do not exist in the exports * category totals add up to the export totals to the cent All money is handled as integer cents. Prints every mismatch. Exit status: 0 when everything matches, 1 on any mismatch, 2 on usage error. """ import csv import os import sys from decimal import Decimal, InvalidOperation EXPORT_HEADER = ["date", "description", "amount"] CATEGORY_HEADER = ["source_file", "row", "date", "description", "amount", "category"] def to_cents(text): """Return integer cents for an amount string, or None if it is not valid.""" try: value = Decimal(text.strip()) except (InvalidOperation, AttributeError): return None if not value.is_finite(): return None cents = value * 100 if cents != cents.to_integral_value(): return None return int(cents) def fmt(cents): sign = "-" if cents < 0 else "" cents = abs(cents) return f"{sign}{cents // 100}.{cents % 100:02d}" def read_rows(path): with open(path, newline="", encoding="utf-8-sig") as handle: return list(csv.reader(handle)) def main(argv): if len(argv) < 3: print("usage: python3 money-referee.py categories.csv EXPORT.csv [EXPORT.csv ...]") return 2 categories_path = argv[1] export_paths = argv[2:] problems = [] # 1. Read the exports independently. exports = {} # (source_file, row) -> (date, description, amount_text, cents) export_total = 0 file_totals = {} seen_names = set() for path in export_paths: name = os.path.basename(path) if name in seen_names: problems.append(f"export {name}: given more than once on the command line") continue seen_names.add(name) try: rows = read_rows(path) except OSError as err: problems.append(f"export {name}: cannot read ({err})") continue if not rows or [c.strip() for c in rows[0]] != EXPORT_HEADER: problems.append(f"export {name}: header is {rows[0] if rows else 'missing'}, expected {EXPORT_HEADER}") continue file_totals[name] = 0 for number, fields in enumerate(rows[1:], start=1): if len(fields) != 3: problems.append(f"export {name} row {number}: expected 3 fields, found {len(fields)}: {fields}") continue date, description, amount = fields cents = to_cents(amount) if cents is None: problems.append(f"export {name} row {number}: amount {amount!r} is not a valid amount in cents") continue exports[(name, number)] = (date, description, amount, cents) export_total += cents file_totals[name] += cents # 2. Read categories.csv. try: cat_rows = read_rows(categories_path) except OSError as err: problems.append(f"{categories_path}: cannot read ({err})") cat_rows = [] if cat_rows and [c.strip() for c in cat_rows[0]] != CATEGORY_HEADER: problems.append(f"{categories_path}: header is {cat_rows[0]}, expected {CATEGORY_HEADER}") cat_rows = [] elif not cat_rows: problems.append(f"{categories_path}: empty or unreadable") matched = {} # key -> categories.csv line number category_totals = {} category_file_totals = {} checked_files = set(file_totals) for line_no, fields in enumerate(cat_rows[1:], start=2): where = f"{categories_path} line {line_no}" if len(fields) != 6: problems.append(f"{where}: expected 6 fields, found {len(fields)}: {fields}") continue source, row_text, date, description, amount, category = fields try: row = int(row_text) except ValueError: problems.append(f"{where}: row {row_text!r} is not a whole number") continue key = (source, row) if key in matched: problems.append(f"{where}: {source} row {row} already listed on line {matched[key]} (appears more than once)") continue if source not in checked_files: problems.append(f"{where}: source_file {source!r} is not one of the export files given") continue if key not in exports: problems.append(f"{where}: {source} row {row} does not exist in the export") continue matched[key] = line_no e_date, e_desc, e_amount, e_cents = exports[key] if date != e_date: problems.append(f"{where}: {source} row {row} date {date!r} but export has {e_date!r}") if description != e_desc: problems.append(f"{where}: {source} row {row} description {description!r} but export has {e_desc!r}") cents = to_cents(amount) if cents is None: problems.append(f"{where}: {source} row {row} amount {amount!r} is not a valid amount in cents") continue if cents != e_cents: problems.append(f"{where}: {source} row {row} amount {amount!r} but export has {e_amount!r}") if not category.strip(): problems.append(f"{where}: {source} row {row} has an empty category") label = category.strip() or "(empty)" category_totals[label] = category_totals.get(label, 0) + cents category_file_totals[source] = category_file_totals.get(source, 0) + cents for key in sorted(exports): if key not in matched: e_date, e_desc, e_amount, _ = exports[key] problems.append(f"missing: {key[0]} row {key[1]} ({e_date}, {e_desc}, {e_amount}) is not in {categories_path}") # 3. Totals to the cent. category_sum = sum(category_totals.values()) for name in sorted(file_totals): got = category_file_totals.get(name, 0) if got != file_totals[name]: problems.append(f"total for {name}: categories.csv sums to {fmt(got)} but export sums to {fmt(file_totals[name])}") if category_sum != export_total: problems.append(f"grand total: category totals sum to {fmt(category_sum)} but exports sum to {fmt(export_total)}") print("Category totals:") for label in sorted(category_totals): print(f" {label}: {fmt(category_totals[label])}") print(f"Sum of category totals: {fmt(category_sum)}") print(f"Sum of export files: {fmt(export_total)}") print(f"Export rows: {len(exports)} categories.csv rows matched: {len(matched)}") if problems: print(f"\nMISMATCHES ({len(problems)}):") for problem in problems: print(f" {problem}") return 1 print("\nOK: every export row appears exactly once and totals match to the cent.") return 0 if __name__ == "__main__": sys.exit(main(sys.argv))