""" Afton JE Topology Survey — READ-ONLY ==================================== For all 32 true dup pairs (24 closed-2024 + 8 open-2025), pull the data needed to design Journal Entry reversal of the duplicate revenue and duplicate cash: - Target invoice (INV-B for 21 untouched pairs; INV-C for 11 Track A pairs) * Customer * TxnDate * Line items with Income account refs * Sales tax (TxnTaxDetail) * Total amount - Dup payment * DepositToAccountRef (UF vs bank vs other) * PaymentMethodRef (CC vs check vs cash etc — informational) * UnappliedAmt (should be 0 — pmt is applied to the target invoice) Outputs: Team Inbox/Afton Electric/afton_je_topology_survey.json Team Inbox/Afton Electric/afton_je_topology_survey.md """ from __future__ import annotations import json import sys from pathlib import Path from datetime import datetime if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") SCRIPT_DIR = Path(__file__).resolve().parent QBO_CLIENT_DIR = SCRIPT_DIR.parent.parent / "Atlas" / "app" sys.path.insert(0, str(QBO_CLIENT_DIR)) import qbo_client as qbo CLIENT_SLUG = "afton" ANALYSIS_JSON = SCRIPT_DIR / "afton_payment_pairs_invoice_analysis.json" OUT_JSON = SCRIPT_DIR / "afton_je_topology_survey.json" OUT_MD = SCRIPT_DIR / "afton_je_topology_survey.md" # Track A: 11 customers whose dup invoice (INV-B) was voided yesterday, # replaced with a recovery invoice (INV-C). Map customer name -> recovery inv id. # Source: Team Inbox/_Deliverables/afton_qbo_cleanup_checkpoint.md TRACK_A_RECOVERY = { "Debra Hazle": "27036", "Ralph Marcantonio": "27040", "Mike Frontz": "27044", "Kassandra Gesse": "27045", "Amy Erwin": "27046", "Scott Ferguson": "27047", "Dan Malone": "27048", "Mike Stegall": "27049", "Linda Hughes": "27050", "Joyce Yannuzzi": "27051", "Bena Videsa": "27052", } # The 3 "originally-pending" Track A pairs that aren't in the analysis JSON # (only the recovery invoice + dup payment IDs are known — synthesizing minimal # entries so the survey covers all 35 dup pairs that need JE cleanup). EXTRA_PAIRS = [ { "batch": "B0", "customer": "Linda Hughes", "classification": "INV_DUP_PAIR", "keep_payment_id": None, "void_payment_id": "24046", "keep_payment_amt": None, "void_payment_amt": 481.0, "keep_invoice": None, "void_invoice": {"id": None, "doc": None, "amt": 481.0, "date": None, "balance": 0.0}, }, { "batch": "B0", "customer": "Joyce Yannuzzi", "classification": "INV_DUP_PAIR", "keep_payment_id": None, "void_payment_id": "25521", "keep_payment_amt": None, "void_payment_amt": 344.0, "keep_invoice": None, "void_invoice": {"id": None, "doc": None, "amt": 344.0, "date": None, "balance": 0.0}, }, { "batch": "B0", "customer": "Bena Videsa", "classification": "INV_DUP_PAIR", "keep_payment_id": None, "void_payment_id": "25569", "keep_payment_amt": None, "void_payment_amt": 180.0, "keep_invoice": None, "void_invoice": {"id": None, "doc": None, "amt": 180.0, "date": None, "balance": 0.0}, }, ] def batch_query(token: str, entity: str, ids: list[str]) -> dict[str, dict]: """Query entities in batches; returns dict keyed by Id.""" 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_SLUG, token, sql) rows = resp.get(entity, []) for r in rows: out[str(r["Id"])] = r return out def main() -> None: raw = json.load(open(ANALYSIS_JSON, encoding="utf-8")) dup_pairs = [x for x in raw if x["classification"] == "INV_DUP_PAIR"] dup_pairs.extend(EXTRA_PAIRS) print(f"Loaded {len(dup_pairs)} true dup pairs (incl. {len(EXTRA_PAIRS)} extras)", file=sys.stderr) token = qbo.refresh_access_token(CLIENT_SLUG) # Build target lists target_inv_ids: list[str] = [] dup_pmt_ids: list[str] = [] pair_meta: list[dict] = [] for pair in dup_pairs: customer = pair["customer"] is_track_a = customer in TRACK_A_RECOVERY if is_track_a: target_inv_id = TRACK_A_RECOVERY[customer] target_label = "INV_C (recovery)" else: target_inv_id = pair["void_invoice"]["id"] target_label = "INV_B (original dup)" target_inv_ids.append(target_inv_id) dup_pmt_ids.append(pair["void_payment_id"]) pair_meta.append({ "pair": pair, "customer": customer, "is_track_a": is_track_a, "target_inv_id": target_inv_id, "target_label": target_label, }) print(f"Fetching {len(target_inv_ids)} invoices...", file=sys.stderr) invoices = batch_query(token, "Invoice", target_inv_ids) print(f"Fetching {len(dup_pmt_ids)} payments...", file=sys.stderr) payments = batch_query(token, "Payment", dup_pmt_ids) survey: list[dict] = [] for meta in pair_meta: customer = meta["customer"] target_inv_id = meta["target_inv_id"] dup_pmt_id = meta["pair"]["void_payment_id"] inv = invoices.get(target_inv_id) or {"_error": "not_returned"} pmt = payments.get(dup_pmt_id) or {"_error": "not_returned"} is_track_a = meta["is_track_a"] target_label = meta["target_label"] # Extract income account refs from invoice lines income_lines = [] for line in inv.get("Line", []): if line.get("DetailType") != "SalesItemLineDetail": continue detail = line.get("SalesItemLineDetail", {}) income_lines.append({ "amount": line.get("Amount"), "description": line.get("Description"), "item_ref": detail.get("ItemRef"), "item_account_ref": detail.get("ItemAccountRef"), # rare; usually inferred from Item "tax_code_ref": detail.get("TaxCodeRef"), }) tax_detail = inv.get("TxnTaxDetail", {}) or {} total_tax = tax_detail.get("TotalTax", 0) row = { "batch": pair["batch"], "customer": customer, "track": "A" if is_track_a else "untouched", "target_invoice_id": target_inv_id, "target_invoice_label": target_label, "target_invoice_doc": inv.get("DocNumber"), "target_invoice_date": inv.get("TxnDate"), "target_invoice_total": inv.get("TotalAmt"), "target_invoice_balance": inv.get("Balance"), "target_invoice_status": "voided" if inv.get("TotalAmt") == 0 and not inv.get("Line") else "active", "income_lines": income_lines, "sales_tax": total_tax, "dup_payment_id": dup_pmt_id, "dup_payment_amt": pmt.get("TotalAmt"), "dup_payment_deposit_acct": pmt.get("DepositToAccountRef"), "dup_payment_method": pmt.get("PaymentMethodRef"), "dup_payment_unapplied": pmt.get("UnappliedAmt"), "dup_payment_linked_txns": pmt.get("LinkedTxn", []), "_inv_error": inv.get("_error"), "_pmt_error": pmt.get("_error"), } survey.append(row) # Write JSON with open(OUT_JSON, "w", encoding="utf-8") as f: json.dump( {"generated_at": datetime.utcnow().isoformat() + "Z", "rows": survey}, f, indent=2, default=str, ) print(f"\nWrote {OUT_JSON}", file=sys.stderr) # Write Markdown summary lines = [ "# Afton JE Topology Survey", f"_Generated: {datetime.utcnow().isoformat()}Z_", "", f"Surveyed {len(survey)} dup pairs (24 closed-2024 + 8 open-2025).", "", "## Account Routing Summary", "", ] # Aggregate by deposit account deposit_totals: dict[str, dict] = {} income_acct_totals: dict[str, dict] = {} tax_total = 0.0 track_a_count = 0 untouched_count = 0 for r in survey: if r["track"] == "A": track_a_count += 1 else: untouched_count += 1 dep = r.get("dup_payment_deposit_acct") or {} dep_key = f"{dep.get('value', '?')} ({dep.get('name', '?')})" amt = r.get("dup_payment_amt") or 0 if dep_key not in deposit_totals: deposit_totals[dep_key] = {"count": 0, "amount": 0.0} deposit_totals[dep_key]["count"] += 1 deposit_totals[dep_key]["amount"] += amt tax_total += r.get("sales_tax") or 0 for il in r["income_lines"]: iar = il.get("item_account_ref") or {} ir = il.get("item_ref") or {} item_key = ( f"item={ir.get('value','?')} ({ir.get('name','?')}) " f"acct={iar.get('value','?')} ({iar.get('name','?')})" ) if item_key not in income_acct_totals: income_acct_totals[item_key] = {"count": 0, "amount": 0.0} income_acct_totals[item_key]["count"] += 1 income_acct_totals[item_key]["amount"] += il.get("amount") or 0 lines.append(f"- Track A (recovery inv targets): {track_a_count}") lines.append(f"- Untouched (original dup inv targets): {untouched_count}") lines.append(f"- Total sales tax across all 32 targets: ${tax_total:,.2f}") lines.append("") lines.append("### Deposit Account (Cr leg of JEs)") lines.append("") lines.append("| Account | Pairs | $ Total |") lines.append("|---|---|---|") for k, v in sorted(deposit_totals.items(), key=lambda x: -x[1]["amount"]): lines.append(f"| {k} | {v['count']} | ${v['amount']:,.2f} |") lines.append("") lines.append("### Item / Income (Dr leg of JEs)") lines.append("") lines.append("| Item / Account | Lines | $ Total |") lines.append("|---|---|---|") for k, v in sorted(income_acct_totals.items(), key=lambda x: -x[1]["amount"]): lines.append(f"| {k} | {v['count']} | ${v['amount']:,.2f} |") lines.append("") lines.append("## Detail by Pair") lines.append("") lines.append("| # | Batch | Customer | Track | Target Inv | Date | Amt | Tax | Pmt | Deposit |") lines.append("|---|---|---|---|---|---|---|---|---|---|") for i, r in enumerate(survey, 1): dep = r.get("dup_payment_deposit_acct") or {} lines.append( f"| {i} | {r['batch']} | {r['customer']} | {r['track']} | " f"{r['target_invoice_id']} #{r['target_invoice_doc'] or '?'} | " f"{r['target_invoice_date']} | " f"${r['target_invoice_total'] or 0:,.2f} | " f"${r['sales_tax'] or 0:,.2f} | " f"{r['dup_payment_id']} | " f"{dep.get('value','?')} {dep.get('name','')} |" ) OUT_MD.write_text("\n".join(lines), encoding="utf-8") print(f"Wrote {OUT_MD}", file=sys.stderr) if __name__ == "__main__": main()