""" Afton UF Phantom Check — 2026-05-15 Catch-up Cluster ===================================================== Investigates the 27 UF entries created in a single bulk session on 2026-05-15 to differentiate real customer payments from phantom AR-clearing entries. For each entry, pulls: - Linked invoice details (CreateTime, age, doc number) - Customer's full Payment history (do they normally have well-tracked pmts?) - Customer's open Invoices (AR aging picture) - Memo/notes on payment and invoice Heuristic classification: LIKELY_REAL — invoice recent (within 30 days of pmt), customer has other normally-tracked pmts LIKELY_PHANTOM — invoice 90+ days old (stale AR being cleared), no recent real customer activity UNCLEAR — needs Jimmie's eyeball Output: afton_uf_phantom_check_worksheet.md """ from __future__ import annotations import json import sys from collections import defaultdict from datetime import datetime from pathlib import Path if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") SCRIPT_DIR = Path(__file__).resolve().parent QBO_APP_DIR = SCRIPT_DIR.parent.parent / "Atlas" / "app" sys.path.insert(0, str(QBO_APP_DIR)) import qbo_client as qbo # type: ignore CLIENT = "afton" FORENSICS_JSON = SCRIPT_DIR / "afton_uf_forensics.json" TARGET_CREATE_DAY = "2026-05-15" OUT_MD = SCRIPT_DIR / "afton_uf_phantom_check_worksheet.md" OUT_JSON = SCRIPT_DIR / "afton_uf_phantom_check.json" def batch_query(token: str, entity: str, ids: list[str]) -> dict[str, dict]: out: dict[str, dict] = {} BATCH = 30 for i in range(0, len(ids), BATCH): chunk = ids[i : i + BATCH] in_clause = ", ".join(f"'{x}'" for x in chunk) sql = f"SELECT * FROM {entity} WHERE Id IN ({in_clause})" resp = qbo.query(CLIENT, token, sql) for r in resp.get(entity, []): out[str(r["Id"])] = r return out def query_customer_history(token: str, customer_id: str) -> dict: """Pull all payments and open invoices for a customer.""" pmts_resp = qbo.query( CLIENT, token, f"SELECT * FROM Payment WHERE CustomerRef = '{customer_id}' MAXRESULTS 100" ) invs_resp = qbo.query( CLIENT, token, f"SELECT * FROM Invoice WHERE CustomerRef = '{customer_id}' AND Balance > '0' MAXRESULTS 100" ) return { "payments": pmts_resp.get("Payment", []), "open_invoices": invs_resp.get("Invoice", []), } def classify(pmt: dict, invoice: dict, customer_history: dict, target_pmt_id: str) -> tuple[str, list[str]]: """Return (classification, evidence_list).""" evidence = [] if not invoice: return ("UNCLEAR", ["No linked invoice found"]) # Invoice age at time of payment inv_create = invoice.get("MetaData", {}).get("CreateTime", "") pmt_txn_date = pmt.get("TxnDate", "") try: inv_dt = datetime.fromisoformat(inv_create.replace("Z", "+00:00")).date() pmt_dt = datetime.fromisoformat(pmt_txn_date).date() inv_age_days = (pmt_dt - inv_dt).days evidence.append(f"Invoice created {inv_dt} ({inv_age_days}d before pmt date)") except Exception: inv_age_days = None evidence.append(f"Invoice create date unclear: {inv_create}") # Customer's other payments — how many normally have payment methods? other_pmts = [p for p in customer_history["payments"] if str(p["Id"]) != target_pmt_id] n_other = len(other_pmts) n_with_method = sum(1 for p in other_pmts if (p.get("PaymentMethodRef") or {}).get("value")) n_deposited = sum( 1 for p in other_pmts if any(lt.get("TxnType") == "Deposit" for lt in (p.get("LinkedTxn") or [])) ) evidence.append( f"Customer has {n_other} other pmts; {n_with_method} with method specified; " f"{n_deposited} linked to a Deposit" ) # Customer's currently-open AR open_total = sum(i.get("Balance", 0) or 0 for i in customer_history["open_invoices"]) if open_total > 0: evidence.append(f"Customer has ${open_total:,.2f} in OPEN invoices currently") else: evidence.append("Customer has $0 open AR") # Invoice memo if invoice.get("PrivateNote"): evidence.append(f"Invoice private note: {invoice['PrivateNote'][:100]}") if invoice.get("CustomerMemo"): evidence.append(f"Invoice customer memo: {invoice['CustomerMemo'].get('value','')[:100]}") # Payment memo if pmt.get("PrivateNote"): evidence.append(f"Payment private note: {pmt['PrivateNote'][:100]}") if pmt.get("PaymentRefNum"): evidence.append(f"Payment RefNum: {pmt['PaymentRefNum']}") # Classification rules if inv_age_days is not None and inv_age_days >= 90: return ("LIKELY_PHANTOM", evidence + [f"Strong signal: invoice was {inv_age_days}d old when 'paid'"]) if inv_age_days is not None and inv_age_days <= 30 and n_deposited > 0: return ("LIKELY_REAL", evidence + ["Strong signal: recent invoice + customer has deposited pmts elsewhere"]) if inv_age_days is not None and inv_age_days <= 7: return ("LIKELY_REAL", evidence + ["Strong signal: invoice created within a week of payment"]) return ("UNCLEAR", evidence) def main() -> None: forensics = json.load(open(FORENSICS_JSON, encoding="utf-8")) targets = [r for r in forensics["rows"] if r.get("create_day") == TARGET_CREATE_DAY] print(f"Loaded {len(targets)} entries from {TARGET_CREATE_DAY} cluster") token = qbo.refresh_access_token(CLIENT) # Pull payments pmt_ids = [r["id"] for r in targets if r["entity"] == "Payment"] print(f"Pulling {len(pmt_ids)} payments...") pmts = batch_query(token, "Payment", pmt_ids) # Get linked invoice ids invoice_ids = set() pmt_to_invoice: dict[str, str] = {} customer_ids = set() for pid, p in pmts.items(): cust = (p.get("CustomerRef") or {}).get("value") if cust: customer_ids.add(cust) for line in p.get("Line", []): for lt in line.get("LinkedTxn", []) or []: if lt.get("TxnType") == "Invoice": invoice_ids.add(str(lt["TxnId"])) pmt_to_invoice[pid] = str(lt["TxnId"]) break if pid in pmt_to_invoice: break print(f"Pulling {len(invoice_ids)} linked invoices...") invoices = batch_query(token, "Invoice", list(invoice_ids)) # Pull customer history (per customer, separate queries) print(f"Pulling history for {len(customer_ids)} customers (sequential — be patient)...") customer_history = {} for i, cid in enumerate(customer_ids, 1): print(f" [{i}/{len(customer_ids)}] customer {cid}") customer_history[cid] = query_customer_history(token, cid) # Classify each results = [] for r in targets: pid = r["id"] pmt = pmts.get(pid, {}) inv_id = pmt_to_invoice.get(pid) inv = invoices.get(inv_id) if inv_id else None cust_id = (pmt.get("CustomerRef") or {}).get("value") hist = customer_history.get(cust_id, {"payments": [], "open_invoices": []}) classification, evidence = classify(pmt, inv, hist, pid) results.append({ "pmt_id": pid, "customer": r["customer"], "customer_id": cust_id, "amount": r["amount"], "txn_date": r["txn_date"], "invoice_id": inv_id, "invoice_doc": inv.get("DocNumber") if inv else None, "classification": classification, "evidence": evidence, }) # Tally counts = defaultdict(int) totals = defaultdict(float) for r in results: counts[r["classification"]] += 1 totals[r["classification"]] += r["amount"] print(f"\n=== CLASSIFICATION ===") for k in ("LIKELY_PHANTOM", "LIKELY_REAL", "UNCLEAR"): print(f" {k:20s} {counts[k]:3d} entries ${totals[k]:>10,.2f}") # Write worksheet lines = [ "# Afton UF Phantom Check — 2026-05-15 Catch-up Cluster", f"_Generated: {datetime.utcnow().isoformat()}Z_", "", f"Investigated {len(targets)} UF entries created on 2026-05-15.", "", "## Classification summary", "", "| Class | Count | $ Total |", "|---|---|---|", f"| LIKELY_PHANTOM | {counts['LIKELY_PHANTOM']} | ${totals['LIKELY_PHANTOM']:,.2f} |", f"| LIKELY_REAL | {counts['LIKELY_REAL']} | ${totals['LIKELY_REAL']:,.2f} |", f"| UNCLEAR | {counts['UNCLEAR']} | ${totals['UNCLEAR']:,.2f} |", "", "**For LIKELY_PHANTOM**: void the payment. This reopens the invoice for AR aging/collections.", "**For LIKELY_REAL**: include in a future JE plug (later batch).", "**For UNCLEAR**: spot-check individually.", "", ] for label in ("LIKELY_PHANTOM", "UNCLEAR", "LIKELY_REAL"): rows = [r for r in results if r["classification"] == label] if not rows: continue lines.append(f"## {label} ({len(rows)} entries)") lines.append("") for r in sorted(rows, key=lambda x: -x["amount"]): lines.append(f"### {r['customer']} — ${r['amount']:,.2f}") lines.append(f"- Pmt Id: {r['pmt_id']}, Invoice Id: {r['invoice_id']} #{r['invoice_doc']}, Pmt TxnDate: {r['txn_date']}") for ev in r["evidence"]: lines.append(f"- {ev}") lines.append("") OUT_MD.write_text("\n".join(lines), encoding="utf-8") OUT_JSON.write_text(json.dumps({ "generated_at": datetime.utcnow().isoformat() + "Z", "results": results, "counts": dict(counts), "totals": {k: v for k, v in totals.items()}, }, indent=2, default=str), encoding="utf-8") print(f"\nWrote {OUT_MD}") print(f"Wrote {OUT_JSON}") if __name__ == "__main__": main()