#!/usr/bin/env python3 """ SENDGRID-OPS-008 audience source generator. Regenerates sendgrid-ops/linderlake_audience_source.txt from the Aug 27 SendGrid backup + current MySQL state, following the OPS-007 §Phase 4 "bucket A" rule (historical-LINDERLAKE-confirmed subset of current UNKNOWN contacts). Reproducibility contract ------------------------ This script must produce a byte-identical output file given the same Aug 27 backup + current MySQL UNKNOWN set. If either input changes, the output SHA256 will change and the auditor must revisit the change. Bucket-A rule (verbatim from OPS-007 §Phase 4): A = HISTORICAL_LINDERLAKE_CONFIRMED = current_UNKNOWN ∩ ( historical_LL_membership_aug27 ∪ historical_SG_export_aug27 ) − historical_BCS_membership_aug27 where "historical" = the Aug 27 13:34 SendGrid backup: ~/sendgrid_backups/20260827_133428/ Priority is enforced by set difference: any email with historical BCS evidence is excluded even if it also has historical LL evidence (that case is D_CONFLICTING per OPS-007 and is intentionally kept out of A). Inputs ------ - Aug 27 backup: ~/sendgrid_backups/20260827_133428/{sendgrid,mysql}/* - Current MySQL: contacts + contact_list_memberships - Env: MYSQL_HOST, MYSQL_USER, MYSQL_PASS, MYSQL_DB Output ------ - ./linderlake_audience_source.txt (or arg[1] destination) - Report: SHA256, line count, provenance stamp Usage ----- Intended to run on the cPanel host where the backup and MySQL live: source ./0-config_var_setup.sh python3 sendgrid-ops/generators/generate_linderlake_audience_source.py \\ sendgrid-ops/linderlake_audience_source.txt The script is READ-ONLY: it never writes to MySQL and never touches SendGrid. It only writes the one destination file. """ import csv import hashlib import os import subprocess import sys from pathlib import Path BACKUP_ROOT = Path(os.environ.get( "OPS008_BACKUP_ROOT", "/home/linder8/sendgrid_backups/20260827_133428", )) SG_EXPORT_CSV = BACKUP_ROOT / "sendgrid" / "contacts_export.csv" BACKUP_MEMBERS = BACKUP_ROOT / "mysql" / "contact_list_memberships.tsv" BACKUP_CONTACTS = BACKUP_ROOT / "mysql" / "contacts.tsv" DEFAULT_OUT = Path("sendgrid-ops/linderlake_audience_source.txt") def die(msg: str, code: int = 1) -> None: print(f"ERROR: {msg}", file=sys.stderr) sys.exit(code) def read_sg_export_emails(path: Path) -> set: """Extract LOWER(EMAIL) from the SendGrid contacts_export.csv.""" if not path.is_file(): die(f"missing SG export: {path}") out = set() with path.open(newline="", encoding="utf-8-sig") as f: r = csv.DictReader(f) for row in r: e = (row.get("EMAIL") or "").strip().lower() if e: out.add(e) return out def read_backup_membership_emails(members_tsv: Path, contacts_tsv: Path, company: str) -> set: """Join Aug-27 memberships (company_name=$company) to Aug-27 contacts.tsv (email is column 2). TSV columns match production schema `SHOW CREATE TABLE contacts` at time of backup.""" if not members_tsv.is_file(): die(f"missing memberships: {members_tsv}") if not contacts_tsv.is_file(): die(f"missing contacts: {contacts_tsv}") # 1. Set of contact_ids for the requested company_name. want_ids = set() with members_tsv.open(newline="", encoding="utf-8", errors="replace") as f: r = csv.reader(f, delimiter="\t", quoting=csv.QUOTE_NONE) hdr = next(r) cid_col = hdr.index("contact_id") co_col = hdr.index("company_name") for row in r: if len(row) <= max(cid_col, co_col): continue if (row[co_col] or "").strip() != company: continue try: want_ids.add(int(row[cid_col])) except ValueError: continue # 2. Look up emails. out = set() with contacts_tsv.open(newline="", encoding="utf-8", errors="replace") as f: r = csv.reader(f, delimiter="\t", quoting=csv.QUOTE_NONE) hdr = next(r) id_col = hdr.index("id") em_col = hdr.index("email") for row in r: if len(row) <= max(id_col, em_col): continue try: cid = int(row[id_col]) except ValueError: continue if cid not in want_ids: continue e = (row[em_col] or "").strip().lower() if e: out.add(e) return out def mysql_select(sql: str) -> list: """Run a read-only SELECT via the mysql client, return list of rows as tab-split lists. Requires MYSQL_HOST/USER/PASS/DB in env.""" for k in ("MYSQL_HOST", "MYSQL_USER", "MYSQL_PASS", "MYSQL_DB"): if k not in os.environ: die(f"missing env {k} — source 0-config_var_setup.sh first") cmd = [ "mysql", "-h", os.environ["MYSQL_HOST"], "-u", os.environ["MYSQL_USER"], f"-p{os.environ['MYSQL_PASS']}", os.environ["MYSQL_DB"], "-B", "-N", "-e", sql, ] r = subprocess.run(cmd, capture_output=True, text=True) if r.returncode != 0: die(f"mysql failed: {r.stderr.strip()}") return [line.split("\t") for line in r.stdout.splitlines() if line] def current_unknown_emails() -> set: """Contacts with zero rows in contact_list_memberships today.""" rows = mysql_select( "SELECT LOWER(c.email) FROM contacts c " "LEFT JOIN contact_list_memberships m ON m.contact_id=c.id " "WHERE m.contact_id IS NULL" ) return {r[0] for r in rows if r and r[0]} def main() -> int: dest = Path(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_OUT dest.parent.mkdir(parents=True, exist_ok=True) print(f"Aug 27 backup root: {BACKUP_ROOT}") if not BACKUP_ROOT.is_dir(): die(f"backup root not found: {BACKUP_ROOT}") print("Loading Aug 27 SendGrid contacts export ...") sg_aug27 = read_sg_export_emails(SG_EXPORT_CSV) print(f" → {len(sg_aug27):,} unique emails") print("Loading Aug 27 LINDERLAKE-tagged memberships ...") ll_aug27 = read_backup_membership_emails( BACKUP_MEMBERS, BACKUP_CONTACTS, "LINDERLAKE", ) print(f" → {len(ll_aug27):,} unique emails") print("Loading Aug 27 BCS-tagged memberships ...") bcs_aug27 = read_backup_membership_emails( BACKUP_MEMBERS, BACKUP_CONTACTS, "BCS", ) print(f" → {len(bcs_aug27):,} unique emails") print("Loading current MySQL UNKNOWN set (no memberships) ...") unk_now = current_unknown_emails() print(f" → {len(unk_now):,} unique emails") ll_evidence = ll_aug27 | sg_aug27 bucket_a = (unk_now & ll_evidence) - bcs_aug27 print(f"Bucket A (OPS-007 §Phase 4): {len(bucket_a):,}") # Canonical sort: GNU `sort -u` under LC_ALL=C so the output is stable # across systems regardless of Python's default sort key. tmp = dest.with_suffix(".unsorted") with tmp.open("w", encoding="utf-8") as f: for e in bucket_a: f.write(e + "\n") env = dict(os.environ, LC_ALL="C") with dest.open("wb") as out: r = subprocess.run(["sort", "-u", str(tmp)], env=env, stdout=out) tmp.unlink() if r.returncode != 0: die("sort failed while canonicalising output") h = hashlib.sha256(dest.read_bytes()).hexdigest() n = sum(1 for _ in dest.open()) print() print(f"Wrote: {dest}") print(f"Lines: {n}") print(f"SHA256: {h}") print() print("Compare against the SHA256 recorded in") print("sendgrid-ops/docs/OPS-008-audience-source.md") return 0 if __name__ == "__main__": sys.exit(main())