290 lines
12 KiB
Python
290 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""reconcile_bounces_to_source.py — close the bounce → source feedback loop.
|
|
|
|
WHY THIS EXISTS
|
|
---------------
|
|
The daily trucking builder and the fuel-tax newsletter both send only to the
|
|
"clean" pool: fmcsa_carriers rows with email_verify_result in
|
|
('smtp_valid','send_confirmed'). That filter is correct, BUT nothing ever
|
|
demoted a clean address after it hard-bounced on a real send. So a mailbox that
|
|
was smtp_valid at verification time (trucking addresses churn fast — brokers
|
|
fold, drivers change carriers, free mailboxes get deleted) stayed smtp_valid
|
|
forever, stayed in pw_smtp_valid_stage, and got re-mailed every cycle.
|
|
|
|
Measured 2026-07-02: 1,371 of 11,551 (12%) staged "clean" emails had ALREADY
|
|
hard-bounced in Listmonk, and the "filtered" newsletters were still running
|
|
10-15% bounce — almost entirely smtp_valid addresses. That is what keeps
|
|
tanking sender reputation even though the audience query looks clean.
|
|
|
|
WHAT IT DOES
|
|
------------
|
|
1. Reads every HARD bounce Listmonk has recorded (across the trucking DB and,
|
|
optionally, healthcare) within a lookback window.
|
|
2. Demotes the matching fmcsa_carriers row to email_verify_result='hard_bounced'
|
|
(email_verified=FALSE), the SAME terminal state burner_list_verify.py uses.
|
|
'hard_bounced' is never sendable, so the builder + newsletter both drop it.
|
|
3. Rebuilds / prunes the pw_smtp_valid_stage staging table (used by the
|
|
newsletter) so it only contains addresses that are (a) still smtp_valid/
|
|
send_confirmed in fmcsa_carriers AND (b) not blocklisted / hard-bounced in
|
|
Listmonk. This makes the "clean pool" self-healing.
|
|
|
|
It is idempotent and safe to run on a timer. Never upgrades anything; the worst
|
|
signal (a real hard bounce) always wins.
|
|
|
|
USAGE
|
|
-----
|
|
# dry-run: show how many source rows WOULD be demoted + staged pruned
|
|
python3 -m scripts.reconcile_bounces_to_source --dry-run
|
|
|
|
# apply, default 60-day bounce lookback
|
|
python3 -m scripts.reconcile_bounces_to_source
|
|
|
|
# wider lookback + include the healthcare listmonk_hc DB
|
|
python3 -m scripts.reconcile_bounces_to_source --days 120 --include-hc
|
|
|
|
Runs on the prod host (needs the api-postgres container) OR anywhere with the
|
|
same psql access. Uses `docker exec` like coupon_ab_scoreboard.py so it needs no
|
|
python driver on the host.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import uuid
|
|
|
|
PG_CONTAINER = os.getenv("PW_PG_CONTAINER", "performancewest-api-postgres-1")
|
|
PW_DB = os.getenv("PW_DB", "performancewest")
|
|
PW_USER = os.getenv("PW_DB_USER", "pw")
|
|
LISTMONK_DB = os.getenv("PW_LISTMONK_DB", "listmonk")
|
|
LISTMONK_HC_DB = os.getenv("PW_LISTMONK_HC_DB", "listmonk_hc")
|
|
|
|
# fmcsa_carriers results we are allowed to demote on a real hard bounce. We only
|
|
# touch rows currently considered SENDABLE (or unknown); we never "un-bounce" a
|
|
# row and never rewrite an already-hard_bounced row (idempotent).
|
|
DEMOTABLE = (
|
|
"smtp_valid", "send_confirmed",
|
|
"catch_all_domain", "catch_all_detected",
|
|
"mx_probe_blocked", "mx_unreachable",
|
|
"smtp_temp_error", "smtp_unknown_451", "smtp_unknown_450",
|
|
)
|
|
|
|
|
|
def psql(db: str, sql: str, user: str = PW_USER) -> str:
|
|
out = subprocess.run(
|
|
["docker", "exec", "-i", PG_CONTAINER, "psql", "-U", user, "-d", db,
|
|
"-tAF", "\t", "-c", sql],
|
|
capture_output=True, text=True, timeout=120,
|
|
)
|
|
if out.returncode != 0:
|
|
raise RuntimeError(f"psql {db} failed: {out.stderr[:400]}")
|
|
return out.stdout
|
|
|
|
|
|
def psql_script(db: str, script: str, user: str = PW_USER) -> str:
|
|
"""Run a multi-statement psql SCRIPT (may contain \\copy ... FROM a file) by
|
|
feeding it on stdin with `-f -`. \\copy is a psql meta-command and cannot
|
|
appear inside a `-c` SQL string, so we always route batch loads through here.
|
|
"""
|
|
out = subprocess.run(
|
|
["docker", "exec", "-i", PG_CONTAINER, "psql", "-U", user, "-d", db,
|
|
"-tAF", "\t", "-v", "ON_ERROR_STOP=1", "-f", "-"],
|
|
input=script, capture_output=True, text=True, timeout=300,
|
|
)
|
|
if out.returncode != 0:
|
|
raise RuntimeError(f"psql {db} failed: {out.stderr[:400]}")
|
|
return out.stdout
|
|
|
|
|
|
def load_emails_into_container(emails: set[str] | list[str]) -> str:
|
|
"""Write emails to a host temp file and `docker cp` it into the postgres
|
|
container, returning the in-container path. Loading a temp table from a real
|
|
file (indexed + ANALYZEd) is dramatically faster than a correlated per-row
|
|
probe and avoids giant inline stdin scripts. Caller removes it via
|
|
remove_container_file when done.
|
|
"""
|
|
remote = f"/tmp/pw_reconcile_{uuid.uuid4().hex}.txt"
|
|
with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as fh:
|
|
fh.write("\n".join(sorted(set(e.strip().lower() for e in emails if e.strip()))))
|
|
fh.write("\n")
|
|
local = fh.name
|
|
try:
|
|
subprocess.run(["docker", "cp", local, f"{PG_CONTAINER}:{remote}"],
|
|
check=True, capture_output=True, text=True, timeout=120)
|
|
finally:
|
|
try:
|
|
os.unlink(local)
|
|
except OSError:
|
|
pass
|
|
return remote
|
|
|
|
|
|
def remove_container_file(path: str) -> None:
|
|
subprocess.run(["docker", "exec", "-i", PG_CONTAINER, "rm", "-f", path],
|
|
capture_output=True, text=True, timeout=30)
|
|
|
|
|
|
def hard_bounced_emails(db: str, days: int) -> set[str]:
|
|
"""Distinct lowercased emails with a HARD bounce in the last `days` days."""
|
|
sql = f"""
|
|
SELECT DISTINCT lower(s.email)
|
|
FROM bounces b
|
|
JOIN subscribers s ON s.id = b.subscriber_id
|
|
WHERE b.type = 'hard'
|
|
AND b.created_at > now() - interval '{days} days'
|
|
AND s.email <> ''
|
|
"""
|
|
rows = psql(db, sql)
|
|
return {ln.strip() for ln in rows.splitlines() if ln.strip()}
|
|
|
|
|
|
def demote_source(emails: set[str], dry_run: bool) -> int:
|
|
"""Mark matching fmcsa_carriers rows hard_bounced. Returns rows changed."""
|
|
if not emails:
|
|
return 0
|
|
demotable = ", ".join(f"'{r}'" for r in DEMOTABLE)
|
|
remote = load_emails_into_container(emails)
|
|
try:
|
|
if dry_run:
|
|
script = (
|
|
"CREATE TEMP TABLE _bounced_emails(email text);\n"
|
|
f"\\copy _bounced_emails FROM '{remote}'\n"
|
|
"CREATE INDEX ON _bounced_emails(email);\n"
|
|
"ANALYZE _bounced_emails;\n"
|
|
"SELECT count(*)\n"
|
|
"FROM fmcsa_carriers f\n"
|
|
"JOIN _bounced_emails b ON lower(f.email_address) = b.email\n"
|
|
f"WHERE f.email_verify_result IN ({demotable});\n"
|
|
)
|
|
else:
|
|
script = (
|
|
"CREATE TEMP TABLE _bounced_emails(email text);\n"
|
|
f"\\copy _bounced_emails FROM '{remote}'\n"
|
|
"CREATE INDEX ON _bounced_emails(email);\n"
|
|
"ANALYZE _bounced_emails;\n"
|
|
"WITH upd AS (\n"
|
|
" UPDATE fmcsa_carriers f\n"
|
|
" SET email_verify_result = 'hard_bounced',\n"
|
|
" email_verified = FALSE\n"
|
|
" FROM _bounced_emails b\n"
|
|
" WHERE lower(f.email_address) = b.email\n"
|
|
f" AND f.email_verify_result IN ({demotable})\n"
|
|
" RETURNING 1\n"
|
|
")\n"
|
|
"SELECT count(*) FROM upd;\n"
|
|
)
|
|
out = psql_script(PW_DB, script)
|
|
finally:
|
|
remove_container_file(remote)
|
|
nums = [l for l in out.splitlines() if l.strip().isdigit()]
|
|
return int(nums[-1]) if nums else 0
|
|
|
|
|
|
def rebuild_stage(dry_run: bool) -> tuple[int, int]:
|
|
"""Rebuild pw_smtp_valid_stage (in the listmonk DB) from the now-cleaned
|
|
fmcsa_carriers pool, excluding anything blocklisted/hard-bounced in Listmonk.
|
|
|
|
Returns (old_count, new_count). The newsletter's AUDIENCE_SQL joins to this
|
|
table, so pruning it is what actually removes dead rows from the next send.
|
|
"""
|
|
old = psql(LISTMONK_DB, "SELECT count(*) FROM pw_smtp_valid_stage;")
|
|
try:
|
|
old_n = int(old.strip() or 0)
|
|
except ValueError:
|
|
old_n = 0
|
|
|
|
# The clean source emails (smtp_valid + send_confirmed) live in the
|
|
# performancewest DB; export them, then load into a fresh stage table in the
|
|
# listmonk DB and subtract listmonk-side suppressions with a set-based
|
|
# anti-join (indexed temp table -> sub-second even for tens of thousands).
|
|
clean = psql(PW_DB, """
|
|
SELECT DISTINCT lower(email_address)
|
|
FROM fmcsa_carriers
|
|
WHERE email_address <> ''
|
|
AND email_verify_result IN ('smtp_valid','send_confirmed')
|
|
""")
|
|
clean_emails = [l.strip() for l in clean.splitlines() if l.strip()]
|
|
remote = load_emails_into_container(clean_emails)
|
|
|
|
# A clean email SURVIVES unless it is blocklisted or has a hard bounce.
|
|
survive_filter = (
|
|
"WHERE NOT EXISTS (\n"
|
|
" SELECT 1 FROM subscribers s\n"
|
|
" WHERE lower(s.email) = c.email\n"
|
|
" AND (s.status = 'blocklisted'\n"
|
|
" OR EXISTS (SELECT 1 FROM bounces b\n"
|
|
" WHERE b.subscriber_id = s.id AND b.type='hard'))\n"
|
|
")\n"
|
|
)
|
|
try:
|
|
if dry_run:
|
|
script = (
|
|
"CREATE TEMP TABLE _clean_probe(email text);\n"
|
|
f"\\copy _clean_probe FROM '{remote}'\n"
|
|
"CREATE INDEX ON _clean_probe(email);\n"
|
|
"ANALYZE _clean_probe;\n"
|
|
"SELECT count(*) FROM _clean_probe c\n"
|
|
f"{survive_filter};\n"
|
|
)
|
|
out = psql_script(LISTMONK_DB, script)
|
|
nums = [l for l in out.splitlines() if l.strip().isdigit()]
|
|
return old_n, (int(nums[-1]) if nums else 0)
|
|
|
|
script = (
|
|
"CREATE TABLE IF NOT EXISTS pw_smtp_valid_stage(email text PRIMARY KEY);\n"
|
|
"CREATE TEMP TABLE _clean_new(email text);\n"
|
|
f"\\copy _clean_new FROM '{remote}'\n"
|
|
"CREATE INDEX ON _clean_new(email);\n"
|
|
"ANALYZE _clean_new;\n"
|
|
"TRUNCATE pw_smtp_valid_stage;\n"
|
|
"INSERT INTO pw_smtp_valid_stage(email)\n"
|
|
"SELECT DISTINCT c.email FROM _clean_new c\n"
|
|
f"{survive_filter}"
|
|
"ON CONFLICT DO NOTHING;\n"
|
|
"SELECT count(*) FROM pw_smtp_valid_stage;\n"
|
|
)
|
|
out = psql_script(LISTMONK_DB, script)
|
|
finally:
|
|
remove_container_file(remote)
|
|
nums = [l for l in out.splitlines() if l.strip().isdigit()]
|
|
return old_n, (int(nums[-1]) if nums else 0)
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--days", type=int, default=60,
|
|
help="hard-bounce lookback window (default 60)")
|
|
ap.add_argument("--include-hc", action="store_true",
|
|
help="also reconcile the healthcare listmonk_hc bounces")
|
|
ap.add_argument("--no-stage-rebuild", action="store_true",
|
|
help="skip rebuilding pw_smtp_valid_stage")
|
|
ap.add_argument("--dry-run", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
dbs = [LISTMONK_DB] + ([LISTMONK_HC_DB] if args.include_hc else [])
|
|
all_bounced: set[str] = set()
|
|
for db in dbs:
|
|
emails = hard_bounced_emails(db, args.days)
|
|
print(f"[{db}] hard-bounced emails in {args.days}d: {len(emails):,}")
|
|
all_bounced |= emails
|
|
print(f"total distinct hard-bounced: {len(all_bounced):,}")
|
|
|
|
changed = demote_source(all_bounced, args.dry_run)
|
|
verb = "WOULD demote" if args.dry_run else "demoted"
|
|
print(f"fmcsa_carriers {verb} to hard_bounced: {changed:,}")
|
|
|
|
if not args.no_stage_rebuild:
|
|
old_n, new_n = rebuild_stage(args.dry_run)
|
|
verb = "WOULD rebuild" if args.dry_run else "rebuilt"
|
|
print(f"pw_smtp_valid_stage {verb}: {old_n:,} -> {new_n:,} "
|
|
f"({old_n - new_n:+,} pruned)")
|
|
|
|
if args.dry_run:
|
|
print("\n(dry-run) no changes written. Re-run without --dry-run to apply.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|