new-site/scripts/workers/schedule_fueltax_newsletter_rollout.py

#!/usr/bin/env python3
"""
schedule_fueltax_newsletter_rollout.py — Pace the reefer/PTO fuel-tax newsletter
across the warm trucking list over ~2 weeks.

Audience : the 46,598 ENABLED trucking subscribers in Listmonk (attrib
           dot_number, status=enabled). This is the warm, previously-mailed,
           non-bounced/non-blocklisted pool -- the right list for an
           engagement-positive newsletter (warms domains, dilutes the
           sales-pitch ratio). NOT the 1.3M cold FMCSA universe.
Pacing   : deterministic `subscribers.id % N` split into N daily batches, each a
           private list populated via Listmonk's query-based bulk add
           (PUT /api/subscribers/query/lists), each its own campaign cloned from
           the reviewed DRAFT (default id 750) and scheduled on successive days.
           Spreading over 2 weeks also lets the invaluement /24 delist propagate
           before bulk volume.
Rate note: Listmonk enforces a GLOBAL 500/email-per-hour sliding window shared
           with the daily drip. A ~3,330 batch drains ~7h of that window, so we
           schedule each batch EARLY (default 14:00 UTC / 9am CT) and let the
           scheduler bleed it through the day; the drip (8am builder) goes first.

Idempotent-ish: creating lists/campaigns is a POST (re-running makes duplicates),
so this prints a plan and requires --commit to actually create+schedule.

    # preview the plan only (no writes):
    python3 scripts/workers/schedule_fueltax_newsletter_rollout.py
    # create the lists+campaigns and schedule them:
    python3 scripts/workers/schedule_fueltax_newsletter_rollout.py --commit
"""
import argparse
import datetime as dt
import os
import sys
import time

import requests

_REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
if _REPO not in sys.path:
    sys.path.insert(0, _REPO)

from scripts.workers.campaign_helpers import LISTMONK_URL, AUTH  # noqa: E402

# ── Config ───────────────────────────────────────────────────────────────────
SOURCE_DRAFT_ID = int(os.getenv("FUELTAX_SOURCE_DRAFT", "750"))
N_BATCHES = int(os.getenv("FUELTAX_BATCHES", "5"))            # one per day
# The SQL predicate that defines the whole CLEAN audience. Each batch ANDs an
# id-modulo slice onto this so the union == the whole audience, no overlaps.
#
# IMPORTANT: only smtp_valid (verified real mailbox) carriers. The first attempt
# (2026-06-28) targeted ALL enabled trucking subs and hit a 34% bounce rate
# because ~80% are catch_all_domain addresses (accept the MX probe, bounce at the
# real mailbox). pw_smtp_valid_stage is a staging table in the listmonk DB holding
# the lowercased smtp_valid/send_confirmed emails from fmcsa_carriers; joining to
# it restricts the audience to the same clean pool the sales drip uses.
AUDIENCE_SQL = (
    "subscribers.attribs ? 'dot_number' AND subscribers.status = 'enabled' "
    "AND EXISTS (SELECT 1 FROM pw_smtp_valid_stage v WHERE v.email = lower(subscribers.email)) "
    # Exclude anyone already mailed by the first (aborted) rollout: list 563 (batch
    # 1, fully sent) and 564 (batch 2, partially sent). Skipping the few hundred in
    # 564 who didn't actually receive it is the safe tradeoff vs. a double-send.
    "AND NOT EXISTS (SELECT 1 FROM subscriber_lists sl WHERE sl.subscriber_id = subscribers.id AND sl.list_id IN (563, 564))"
)
# First send day (default: tomorrow). Sent at SEND_HOUR_UTC each day.
SEND_HOUR_UTC = int(os.getenv("FUELTAX_SEND_HOUR_UTC", "14"))  # 14:00 UTC = 9am CT
LIST_TAGS = ["trucking", "newsletter", "fueltax", "auto"]


def lm(path, payload=None, method="GET"):
    s = requests.Session()
    s.auth = AUTH
    url = f"{LISTMONK_URL}/api{path}"
    r = s.request(method, url, json=payload, timeout=60)
    if not r.ok:
        raise RuntimeError(f"{method} {path} -> HTTP {r.status_code}: {r.text[:300]}")
    return r.json()


def get_source_draft():
    d = lm(f"/campaigns/{SOURCE_DRAFT_ID}")["data"]
    if not d.get("body"):
        raise SystemExit(f"source draft {SOURCE_DRAFT_ID} has no body")
    return d


def make_list(name):
    return lm("/lists", {
        "name": name, "type": "private", "optin": "single", "tags": LIST_TAGS,
    }, "POST")["data"]["id"]


def fill_list_by_query(list_id, batch_idx, n_batches):
    """Add the id-modulo slice of the warm audience to list_id in one call."""
    query = f"({AUDIENCE_SQL}) AND (subscribers.id % {n_batches} = {batch_idx})"
    lm("/subscribers/query/lists", {
        "query": query,
        "action": "add",
        "target_list_ids": [list_id],
        "status": "confirmed",
    }, "PUT")


def list_count(list_id):
    # subscriber_count is async in Listmonk; query directly for an exact count.
    res = lm("/subscribers?" + "list_id=%d&per_page=1" % list_id)
    return res.get("data", {}).get("total", 0)


def create_scheduled_campaign(src, list_id, name, send_at_utc):
    payload = {
        "name": name,
        "subject": src["subject"],
        "lists": [list_id],
        "from_email": src.get("from_email") or "Performance West <noreply@send.performancewest.net>",
        "type": "regular",
        "content_type": src["content_type"],
        "body": src["body"],
        "altbody": src.get("altbody"),
        "messenger": src.get("messenger") or "email",
        "tags": (src.get("tags") or []) + ["fueltax-newsletter"],
        "send_at": send_at_utc.strftime("%Y-%m-%dT%H:%M:%S+00:00"),
    }
    if src.get("template_id"):
        payload["template_id"] = src["template_id"]
    if src.get("headers"):
        payload["headers"] = src["headers"]
    cid = lm("/campaigns", payload, "POST")["data"]["id"]
    lm(f"/campaigns/{cid}/status", {"status": "scheduled"}, "PUT")
    return cid


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--commit", action="store_true", help="actually create lists+campaigns and schedule")
    ap.add_argument("--start-date", help="first send date YYYY-MM-DD (default: tomorrow, UTC)")
    ap.add_argument("--batches", type=int, default=N_BATCHES)
    args = ap.parse_args()

    n_batches = args.batches

    src = get_source_draft()
    print(f"Source draft #{SOURCE_DRAFT_ID}: {src['name']!r}")
    print(f"  subject: {src['subject']!r}")
    print(f"  body: {len(src['body']):,} chars  altbody: {len(src.get('altbody') or ''):,} chars")

    today = dt.datetime.now(dt.timezone.utc).date()
    start = (dt.date.fromisoformat(args.start_date) if args.start_date
             else today + dt.timedelta(days=1))

    print(f"\nPlan: {n_batches} daily batches starting {start} at {SEND_HOUR_UTC:02d}:00 UTC")
    print(f"  audience: {AUDIENCE_SQL}")
    print(f"  split:    subscribers.id %% {n_batches}\n")

    for i in range(n_batches):
        day = start + dt.timedelta(days=i)
        send_at = dt.datetime(day.year, day.month, day.day, SEND_HOUR_UTC, 0, 0,
                              tzinfo=dt.timezone.utc)
        label = day.isoformat()
        list_name = f"Trucking Newsletter — Reefer/PTO Fuel-Tax — {label} (batch {i+1}/{n_batches})"
        camp_name = f"Trucking Newsletter — Reefer/PTO Fuel-Tax — {label}"
        if not args.commit:
            print(f"  [{i+1:2d}/{n_batches}] {label} {SEND_HOUR_UTC:02d}:00Z  id%{n_batches}=={i}  (dry-run)")
            continue
        lid = make_list(list_name)
        fill_list_by_query(lid, i, n_batches)
        time.sleep(2)  # let the bulk add land before we read the count
        cnt = list_count(lid)
        cid = create_scheduled_campaign(src, lid, camp_name, send_at)
        print(f"  [{i+1:2d}/{n_batches}] {label} {SEND_HOUR_UTC:02d}:00Z  list={lid} subs={cnt:,} campaign={cid} SCHEDULED")

    if not args.commit:
        print("\n(dry-run) re-run with --commit to create + schedule.")
    else:
        print("\nDone. All batches scheduled. Review in Listmonk; un-schedule any to pause.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())