#!/usr/bin/env python3 """chat_invoice_review.py - daily second look at QuickBooks invoices drafted or sent from an AI chat. Agents Finance (agentsfinance.io), 10 November 2026. Licence: MIT-0 (use, copy and change it freely, no attribution needed). Standard library only (Python 3.9+). It never connects to QuickBooks: you feed it two CSV exports. Input 1 - invoices CSV (one row per invoice), columns: number, date (YYYY-MM-DD), customer, total, send_to (email the invoice went or will go to), status (draft | sent | paid | ...), source (chat | ui | other; fill it from the Intuit Intelligence audit log or your own notes - QuickBooks exports do not add this column for you) Input 2 - customers CSV, columns: customer, emails (separate several with ;), created (YYYY-MM-DD) Rules (each flag is one row in the output): R1 DUPLICATE same customer and same total within --dup-days days (default 31) R2 NEW_CUSTOMER customer record created --new-days days or fewer before the invoice (default 7) R3 EMAIL send_to is empty or not one of the customer's emails on file R4 AMOUNT total >= --max-total, or > --ratio x the customer's median of earlier invoices (needs 3+) R5 BULK more than --bulk invoices from source=chat on the same date (default 5) R6 UNKNOWN customer is not in the customers file Severity: high for R1, R3, R6 and R4 on sent invoices; medium otherwise. Exit code: 0 no high flags, 1 at least one high flag, 2 bad input. Usage: python3 chat_invoice_review.py invoices.csv customers.csv [--out flags.csv] [--max-total 10000] python3 chat_invoice_review.py --self-test """ import argparse import csv import io import statistics import sys from collections import defaultdict from datetime import date INV_COLS = ["number", "date", "customer", "total", "send_to", "status", "source"] CUS_COLS = ["customer", "emails", "created"] def _d(s, field, row): try: return date.fromisoformat(s.strip()) except Exception: raise ValueError(f"bad {field} '{s}' in row {row}") def load(fh, cols): rd = csv.DictReader(fh) missing = [c for c in cols if c not in (rd.fieldnames or [])] if missing: raise ValueError(f"missing columns: {', '.join(missing)}") return [{k: (v or "").strip() for k, v in r.items()} for r in rd] def review(invoices, customers, dup_days=31, new_days=7, max_total=10000.0, ratio=3.0, bulk=5): cust = {} for i, c in enumerate(customers, 2): cust[c["customer"].lower()] = { "emails": {e.strip().lower() for e in c["emails"].split(";") if e.strip()}, "created": _d(c["created"], "created", i), } rows = [] for i, r in enumerate(invoices, 2): rows.append({**r, "_date": _d(r["date"], "date", i), "_total": float(r["total"].replace(",", "")), "_line": i}) rows.sort(key=lambda r: (r["_date"], r["number"])) flags = [] def flag(r, rule, sev, msg): flags.append({"number": r["number"], "date": r["date"], "customer": r["customer"], "total": f"{r['_total']:.2f}", "source": r["source"], "status": r["status"], "rule": rule, "severity": sev, "detail": msg}) history = defaultdict(list) chat_per_day = defaultdict(list) for r in rows: key = r["customer"].lower() c = cust.get(key) sent = r["status"].lower() in ("sent", "paid") if c is None: flag(r, "R6_UNKNOWN", "high", "customer not in customers file") else: age = (r["_date"] - c["created"]).days if 0 <= age <= new_days: flag(r, "R2_NEW_CUSTOMER", "medium", f"customer created {age} day(s) before the invoice") to = r["send_to"].lower() if not to or to not in c["emails"]: flag(r, "R3_EMAIL", "high", f"send_to '{r['send_to'] or '(empty)'}' is not on the customer record") for p in history[key]: if abs(p["_total"] - r["_total"]) < 0.005 and (r["_date"] - p["_date"]).days <= dup_days: flag(r, "R1_DUPLICATE", "high", f"same total as {p['number']} ({p['date']})") break prior = [p["_total"] for p in history[key]] if r["_total"] >= max_total: flag(r, "R4_AMOUNT", "high" if sent else "medium", f"total >= {max_total:.2f}") elif len(prior) >= 3 and r["_total"] > ratio * statistics.median(prior): flag(r, "R4_AMOUNT", "high" if sent else "medium", f"total > {ratio:g}x median of {len(prior)} earlier invoices ({statistics.median(prior):.2f})") history[key].append(r) if r["source"].lower() == "chat": chat_per_day[r["date"]].append(r) for day, lst in sorted(chat_per_day.items()): if len(lst) > bulk: for r in lst: flag(r, "R5_BULK", "medium", f"{len(lst)} chat invoices on {day} (limit {bulk})") return flags def write(flags, fh): w = csv.DictWriter(fh, fieldnames=["number", "date", "customer", "total", "source", "status", "rule", "severity", "detail"]) w.writeheader() w.writerows(flags) FIX_CUS = """customer,emails,created Harbor Dental,ap@harbordental.example;owner@harbordental.example,2025-02-01 Pine Street Cafe,billing@pinestreet.example,2025-06-10 Nova Fitness,accounts@novafit.example,2026-11-03 """ FIX_INV = """number,date,customer,total,send_to,status,source 1001,2026-09-05,Harbor Dental,1200.00,ap@harbordental.example,paid,ui 1002,2026-10-05,Harbor Dental,1250.00,ap@harbordental.example,paid,ui 1003,2026-10-20,Harbor Dental,1180.00,ap@harbordental.example,paid,chat 1004,2026-11-04,Harbor Dental,1180.00,ap@harbordental.example,draft,chat 1005,2026-11-04,Nova Fitness,640.00,accounts@novafit.example,draft,chat 1006,2026-11-05,Pine Street Cafe,310.00,pinestreet.billing@gmail.example,sent,chat 1007,2026-11-06,Harbor Dental,4900.00,ap@harbordental.example,draft,chat 1008,2026-11-06,Lakeview Storage,800.00,info@lakeview.example,draft,chat 1009,2026-11-07,Pine Street Cafe,295.00,billing@pinestreet.example,draft,ui """ def self_test(): flags = review(load(io.StringIO(FIX_INV), INV_COLS), load(io.StringIO(FIX_CUS), CUS_COLS)) got = sorted((f["number"], f["rule"], f["severity"]) for f in flags) want = sorted([ ("1004", "R1_DUPLICATE", "high"), ("1005", "R2_NEW_CUSTOMER", "medium"), ("1006", "R3_EMAIL", "high"), ("1007", "R4_AMOUNT", "medium"), ("1008", "R6_UNKNOWN", "high"), ]) ok = got == want print("expected:", want) print("got: ", got) # bulk rule: 6 chat invoices on one day -> 6 R5 flags many = "number,date,customer,total,send_to,status,source\n" + "".join( f"2{i},2026-11-08,Pine Street Cafe,{100 + i}.00,billing@pinestreet.example,draft,chat\n" for i in range(6)) bulk = [f for f in review(load(io.StringIO(many), INV_COLS), load(io.StringIO(FIX_CUS), CUS_COLS)) if f["rule"] == "R5_BULK"] ok_bulk = len(bulk) == 6 print("bulk flags:", len(bulk), "(expected 6)") print("SELF-TEST", "PASSED" if ok and ok_bulk else "FAILED") return 0 if ok and ok_bulk else 1 def main(argv=None): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("invoices", nargs="?") ap.add_argument("customers", nargs="?") ap.add_argument("--out", help="write flags to this CSV (default: stdout)") ap.add_argument("--dup-days", type=int, default=31) ap.add_argument("--new-days", type=int, default=7) ap.add_argument("--max-total", type=float, default=10000.0) ap.add_argument("--ratio", type=float, default=3.0) ap.add_argument("--bulk", type=int, default=5) ap.add_argument("--self-test", action="store_true") a = ap.parse_args(argv) if a.self_test: return self_test() if not (a.invoices and a.customers): ap.error("give invoices.csv and customers.csv, or --self-test") try: with open(a.invoices, newline="", encoding="utf-8-sig") as f: inv = load(f, INV_COLS) with open(a.customers, newline="", encoding="utf-8-sig") as f: cus = load(f, CUS_COLS) flags = review(inv, cus, a.dup_days, a.new_days, a.max_total, a.ratio, a.bulk) except (ValueError, OSError) as e: print(f"error: {e}", file=sys.stderr) return 2 if a.out: with open(a.out, "w", newline="", encoding="utf-8") as f: write(flags, f) else: write(flags, sys.stdout) high = sum(f["severity"] == "high" for f in flags) print(f"{len(inv)} invoices, {len(flags)} flags, {high} high", file=sys.stderr) return 1 if high else 0 if __name__ == "__main__": sys.exit(main())