← back to AbramsEgo

scripts/import-affiliate-commissions.mjs

182 lines

#!/usr/bin/env node
// import-affiliate-commissions.mjs — book affiliate payout CSVs into AbramsEgo's P&L.
//
// WHY: all affiliate commissions go into AbramsEgo (Steve, 2026-08-03). Payouts
// arrive out-of-band from each program's dashboard (PartnerStack / CJ / etc.).
// This maps a payout-export CSV → engine:"affiliate" rows in revenue-ledger.jsonl,
// which the 30s snapshot folds into the today/week/month self-funding P&L.
//
// SAFE BY DESIGN:
//   • DRY-RUN by default — prints what WOULD be booked + totals. Add --apply to write.
//   • REAL payout date preserved (accurate P&L windowing), not stamped "now".
//   • IDEMPOTENT — every booked row is hashed into data/affiliate-commissions-imported.json;
//     re-running the same CSV books nothing twice (the ledger has no server-side idempotency).
//   • Appends directly to the ledger (no auth, no junk refresh row); dashboard updates ≤30s.
//
// USAGE:
//   node scripts/import-affiliate-commissions.mjs <payouts.csv> --program elevenlabs [--network partnerstack] [--apply]
//   node scripts/import-affiliate-commissions.mjs <payouts.csv> --network cj --apply
//
//   --program  <slug>     affiliate registry slug, tags the ledger source (recommended)
//   --network  <name>     partnerstack | cj | generic (column-map preset; default: generic)
//   --apply               actually write rows (omit = dry-run preview only)
//   --date-col/--amount-col/--desc-col/--id-col  override auto-detected column names

import fs from 'fs';
import path from 'path';
import crypto from 'crypto';
import { fileURLToPath } from 'url';

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const ROOT = path.resolve(__dirname, '..');
// paths env-overridable so tests can target a scratch ledger (never the real P&L)
const LEDGER = process.env.ABRAMSEGO_LEDGER || path.join(ROOT, 'data', 'revenue-ledger.jsonl');
const IMPORTED = process.env.ABRAMSEGO_IMPORTED || path.join(ROOT, 'data', 'affiliate-commissions-imported.json');

// ── column-map presets (case-insensitive "header contains" candidates) ──────────
const PRESETS = {
  partnerstack: {
    date:   ['transaction date', 'created', 'date'],
    amount: ['partner commission', 'commission', 'amount', 'payout'],
    desc:   ['customer', 'product', 'deal', 'group'],
    id:     ['transaction id', 'transaction key', 'id'],
  },
  cj: {
    date:   ['event date', 'posting date', 'transaction date', 'date'],
    amount: ['publisher commission', 'commission', 'pub commission', 'amount'],
    desc:   ['advertiser name', 'website name', 'advertiser', 'action name'],
    id:     ['commission id', 'action id', 'order id', 'original action id'],
  },
  generic: {
    date:   ['date', 'created', 'posted', 'period'],
    amount: ['commission', 'amount', 'payout', 'earning', 'earnings', 'total'],
    desc:   ['description', 'customer', 'product', 'advertiser', 'program', 'note'],
    id:     ['id', 'transaction', 'reference', 'ref'],
  },
};

// ── args ────────────────────────────────────────────────────────────────────
const argv = process.argv.slice(2);
const VALUE_FLAGS = ['--program', '--network', '--date-col', '--amount-col', '--desc-col', '--id-col'];
const flag = (name) => { const i = argv.indexOf(name); return i >= 0 ? argv[i + 1] : null; };
const has  = (name) => argv.includes(name);
// positional args = anything not a flag and not the value consumed by a value-flag
const positionals = argv.filter((a, i) => !a.startsWith('-') && !VALUE_FLAGS.includes(argv[i - 1]));
const csvPath = positionals[0];
const APPLY = has('--apply');
const program = flag('--program') || '';
const network = (flag('--network') || 'generic').toLowerCase();

if (!csvPath || has('--help') || has('-h')) {
  console.log('usage: node scripts/import-affiliate-commissions.mjs <payouts.csv> --program <slug> [--network partnerstack|cj|generic] [--apply]');
  process.exit(csvPath ? 0 : 1);
}
if (!fs.existsSync(csvPath)) { console.error(`✗ CSV not found: ${csvPath}`); process.exit(1); }
const preset = PRESETS[network] || PRESETS.generic;

// ── minimal RFC-4180-ish CSV parser (quoted fields, embedded commas/quotes) ───
function parseCsv(text) {
  const rows = []; let row = [], field = '', inQ = false;
  for (let i = 0; i < text.length; i++) {
    const c = text[i];
    if (inQ) {
      if (c === '"') { if (text[i + 1] === '"') { field += '"'; i++; } else inQ = false; }
      else field += c;
    } else if (c === '"') inQ = true;
    else if (c === ',') { row.push(field); field = ''; }
    else if (c === '\n' || c === '\r') {
      if (c === '\r' && text[i + 1] === '\n') i++;
      if (field !== '' || row.length) { row.push(field); rows.push(row); row = []; field = ''; }
    } else field += c;
  }
  if (field !== '' || row.length) { row.push(field); rows.push(row); }
  return rows;
}

function resolveCol(headers, override, candidates) {
  const lower = headers.map((h) => (h || '').trim().toLowerCase());
  if (override) { const i = lower.indexOf(override.toLowerCase()); if (i >= 0) return i; }
  for (const cand of candidates) { const i = lower.findIndex((h) => h.includes(cand)); if (i >= 0) return i; }
  return -1;
}

function parseAmount(s) {
  const str = String(s).trim();
  const isNeg = /^\(.*\)$/.test(str); // (12.50) = refund/reversal in CJ/PartnerStack exports
  const n = Number(str.replace(/[(),]/g, '').replace(/[^0-9.\-]/g, ''));
  if (!isFinite(n)) return NaN;
  return isNeg ? -Math.abs(n) : n;
}
function parseDate(s) {
  const d = new Date(String(s).trim());
  if (isNaN(d)) return null;
  // noon UTC anchor when the source gives a bare date, so tz never bumps the day
  if (!/\d{1,2}:\d{2}/.test(String(s))) d.setUTCHours(12, 0, 0, 0);
  return d.toISOString();
}
// full 64-char digest (no truncation → no birthday collisions); `occ` disambiguates
// genuinely-distinct rows that share program/date/amount/desc and have a blank extId
const rowHash = (o, occ) => crypto.createHash('sha256').update([o.program, o.ts.slice(0, 10), o.amount.toFixed(2), o.extId, o.desc, occ].join('|')).digest('hex');

// ── load ──────────────────────────────────────────────────────────────────
const rows = parseCsv(fs.readFileSync(csvPath, 'utf8')).filter((r) => r.some((c) => (c || '').trim() !== ''));
if (rows.length < 2) { console.error('✗ CSV has no data rows'); process.exit(1); }
const headers = rows[0];
const cDate = resolveCol(headers, flag('--date-col'), preset.date);
const cAmt  = resolveCol(headers, flag('--amount-col'), preset.amount);
const cDesc = resolveCol(headers, flag('--desc-col'), preset.desc);
const cId   = resolveCol(headers, flag('--id-col'), preset.id);
if (cDate < 0 || cAmt < 0) {
  console.error(`✗ Could not locate required columns (network=${network}).`);
  console.error(`  headers: ${headers.join(' | ')}`);
  console.error('  Override with --date-col "<name>" --amount-col "<name>".');
  process.exit(1);
}

const imported = fs.existsSync(IMPORTED) ? JSON.parse(fs.readFileSync(IMPORTED, 'utf8')) : { hashes: [] };
const seen = new Set(imported.hashes);
const fileOcc = new Map(); // per-file occurrence counter for blank-extId rows

const toBook = [], skipped = [], bad = [];
for (const r of rows.slice(1)) {
  const ts = parseDate(r[cDate]); const amount = parseAmount(r[cAmt]);
  if (!ts || !isFinite(amount)) { bad.push(r); continue; }
  const desc = (cDesc >= 0 ? r[cDesc] : '').trim();
  const extId = (cId >= 0 ? r[cId] : '').trim();
  const rec = { program: program || network, ts, amount: Math.round(amount * 100) / 100, desc, extId };
  const baseKey = [rec.program, rec.ts.slice(0, 10), rec.amount.toFixed(2), rec.extId, rec.desc].join('|');
  const occ = rec.extId ? 0 : (fileOcc.get(baseKey) || 0);
  if (!rec.extId) fileOcc.set(baseKey, occ + 1);
  rec.hash = rowHash(rec, occ);
  if (seen.has(rec.hash)) { skipped.push(rec); continue; }
  seen.add(rec.hash); toBook.push(rec);
}

// ── report ──────────────────────────────────────────────────────────────────
const fmt$ = (n) => `$${n.toFixed(2)}`;
console.log(`\nAffiliate commission import — ${APPLY ? 'APPLY' : 'DRY-RUN'} (network=${network}, program=${program || '—'})`);
console.log(`source: ${csvPath}\n`);
console.log('  date        amount     program        detail');
console.log('  ' + '─'.repeat(64));
for (const r of toBook) console.log(`  ${r.ts.slice(0, 10)}  ${fmt$(r.amount).padStart(9)}  ${(r.program).padEnd(13)}  ${(r.desc || r.extId || '').slice(0, 30)}  NEW`);
const total = toBook.reduce((s, r) => s + r.amount, 0);
console.log('  ' + '─'.repeat(64));
console.log(`  ${toBook.length} new row(s) → ${fmt$(total)} booked to AbramsEgo (engine:"affiliate")`);
if (skipped.length) console.log(`  ${skipped.length} already-imported row(s) skipped (idempotent)`);
if (bad.length)     console.log(`  ⚠ ${bad.length} unparseable row(s) skipped (bad date/amount)`);

if (!APPLY) { console.log(`\n(dry-run — re-run with --apply to write these ${toBook.length} row(s) to the ledger)\n`); process.exit(0); }
if (!toBook.length) { console.log('\nNothing new to book.\n'); process.exit(0); }

// ── apply: append real-dated rows, persist dedup hashes ─────────────────────
const lines = toBook.map((r) => JSON.stringify({
  ts: r.ts, engine: 'affiliate', amount: r.amount,
  source: `${r.program}${r.desc ? ' — ' + r.desc : ''}${r.extId ? ' [' + r.extId + ']' : ''}`.slice(0, 200),
})).join('\n') + '\n';
fs.appendFileSync(LEDGER, lines);
imported.hashes = [...seen];
imported.updated = new Date().toISOString();
fs.writeFileSync(IMPORTED, JSON.stringify(imported, null, 2));
console.log(`\n✓ Booked ${toBook.length} row(s) (${fmt$(total)}) to ${path.relative(ROOT, LEDGER)}.`);
console.log('  AbramsEgo P&L reflects them within 30s (snapshot auto-refresh). $0 (local).\n');