#!/usr/bin/env python3 """Build the IFTA quarterly-return reminder campaign. IFTA returns are due on a FIXED calendar (no per-carrier data needed): Q1 (Jan-Mar) -> Apr 30 Q2 (Apr-Jun) -> Jul 31 Q3 (Jul-Sep) -> Oct 31 Q4 (Oct-Dec) -> Jan 31 Every interstate carrier (FMCSA operation code 'A') files IFTA 4x/year, forever, so this is pure recurring revenue. We email the interstate population ~3 weeks before each deadline with the ifta-quarterly filing CTA + the same-day coupon. Reuses the listmonk + coupon plumbing from build_trucking_campaigns so there is one source of truth for sending, suppression, warmup caps and the daily coupon. Cron (run daily; it self-gates to the reminder window): python3 scripts/build_ifta_quarterly_campaign.py --start-campaign Manual / preview: python3 scripts/build_ifta_quarterly_campaign.py --preview python3 scripts/build_ifta_quarterly_campaign.py --date 2026-04-09 --dry-run """ from __future__ import annotations import argparse import os import sys from datetime import date, datetime, timedelta, timezone ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if ROOT not in sys.path: sys.path.insert(0, ROOT) import psycopg2 # noqa: E402 # Reuse the trucking sender plumbing (single source of truth). import scripts.build_trucking_campaigns as tc # noqa: E402 LOG = tc.LOG IFTA_SVC_SLUG = "ifta-quarterly" LP_LINK = f"{tc.SITE_DOMAIN}/order/{IFTA_SVC_SLUG}" # Multi-touch cadence: days-before-deadline for each escalating touch. # Send late enough that DIY feels stressful (we sell convenience under pressure) # but early enough that we can still file in time (need ~3-5d + their e-signed # authorization). Each touch escalates tone: soft -> urgent -> last-chance. TOUCHES = [ # (days_before, touch_no, subject_suffix, headline, urgency_blurb) (10, 1, "is due in about a week", "Your IFTA quarterly return is due {due}.", "It is coming up fast. We can file it for you so it is one less thing on your plate."), (7, 2, "is due in one week - still time for us to file", "One week left: your IFTA return is due {due}.", "If you would rather not dig through your mileage and fuel records yourself, there is still time for us to handle it - but the window is closing."), (4, 3, "is due in days - last chance for us to file it for you", "Almost out of time: your IFTA return is due {due}.", "This is about the last point we can still file your return in time. After this you would be filing it yourself, fast, or risking a late penalty. Let us take it off your plate today."), ] # Only send within this many days of a touch day (so a daily cron fires once each). REMINDER_WINDOW_DAYS = int(os.getenv("IFTA_REMINDER_WINDOW_DAYS", "1")) # Listmonk source campaign holding the IFTA reminder body/subject/template. SOURCE_ENV = "CAMPAIGN_IFTA_QUARTERLY_ID" def ifta_deadlines(year: int) -> list[tuple[str, date, str]]: """(quarter_label, due_date, period_label) for a given year.""" return [ ("Q4", date(year, 1, 31), f"Q4 {year-1} (Oct-Dec)"), ("Q1", date(year, 4, 30), f"Q1 {year} (Jan-Mar)"), ("Q2", date(year, 7, 31), f"Q2 {year} (Apr-Jun)"), ("Q3", date(year, 10, 31), f"Q3 {year} (Jul-Sep)"), ] def next_deadline(today: date) -> tuple[str, date, str]: """The soonest upcoming IFTA deadline (this year or next).""" cands = ifta_deadlines(today.year) + ifta_deadlines(today.year + 1) upcoming = [d for d in cands if d[1] >= today] return min(upcoming, key=lambda d: d[1]) def _is_business_day(d: date) -> bool: # Mon-Fri. (Federal holidays are not adjusted for here; the +/-window and the # 3-touch cadence give enough slack that a holiday only shifts a touch by a day.) return d.weekday() < 5 def minus_business_days(d: date, n: int) -> date: """Return the date that is `n` business days before `d`.""" cur = d while n > 0: cur -= timedelta(days=1) if _is_business_day(cur): n -= 1 return cur def due_touch_today(today: date) -> tuple[tuple, tuple[str, date, str]] | tuple[None, tuple[str, date, str]]: """Return (touch_tuple, (quarter, due, period)) if a touch fires today, else (None, ...). Each touch fires on its business-day-before-deadline date (within a small window so the daily cron catches it even if a day is skipped). Only fires on business days. """ q, due, period = next_deadline(today) if not _is_business_day(today): return None, (q, due, period) for touch in TOUCHES: bdays = touch[0] send_on = minus_business_days(due, bdays) if abs((today - send_on).days) <= REMINDER_WINDOW_DAYS: return touch, (q, due, period) return None, (q, due, period) def _reset_cycle_if_new(conn, quarter: str, due: date) -> None: """Clear ifta_reminded_at once per quarter so each cycle reminds the full pool. A tiny marker table records the last cycle key (quarter + due year). When the builder enters a new quarter's window for the first time, it nulls the column. """ cycle_key = f"{quarter}-{due.year}" cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS ifta_reminder_cycle ( id boolean PRIMARY KEY DEFAULT true, cycle_key text NOT NULL, reset_at timestamptz NOT NULL DEFAULT now(), CONSTRAINT ifta_reminder_cycle_singleton CHECK (id) ) """) cur.execute("SELECT cycle_key FROM ifta_reminder_cycle WHERE id = true") row = cur.fetchone() if row and row[0] == cycle_key: conn.commit() return # already reset for this cycle cur.execute(""" UPDATE fmcsa_carriers SET ifta_reminded_at = NULL, ifta_touch_no = NULL, ifta_self_filed_at = NULL WHERE ifta_touch_no IS NOT NULL OR ifta_self_filed_at IS NOT NULL """) cleared = cur.rowcount cur.execute(""" INSERT INTO ifta_reminder_cycle (id, cycle_key, reset_at) VALUES (true, %s, now()) ON CONFLICT (id) DO UPDATE SET cycle_key = EXCLUDED.cycle_key, reset_at = now() """, (cycle_key,)) conn.commit() LOG.info("[ifta] new cycle %s -- cleared %d prior ifta_reminded_at marks", cycle_key, cleared) SELECT_SQL = f""" SELECT dot_number, email_address, legal_name, phy_state FROM fmcsa_carriers WHERE carrier_operation = 'A' -- interstate => files IFTA AND email_address IS NOT NULL AND email_address <> '' AND {tc.USABLE_FILTER} AND lower(split_part(email_address, '@', 2)) <> ALL(%s) AND ifta_self_filed_at IS NULL -- clicked "I already filed it" AND COALESCE(ifta_touch_no, 0) < %s -- not yet sent THIS touch ORDER BY dot_number LIMIT %s """ def _ifta_filed_token(dot: str) -> str: """Must match api/src/routes/ifta.ts iftaFiledToken(): HMAC-SHA256, first 24 hex.""" import hmac import hashlib secret = (os.getenv("ADMIN_JWT_SECRET") or os.getenv("CUSTOMER_JWT_SECRET") or os.getenv("APPROVE_FILE_TOKEN") or "pw-ifta-filed-fallback-secret") return hmac.new(secret.encode(), f"ifta-filed:{dot}".encode(), hashlib.sha256).hexdigest()[:24] def _filed_link(dot: str) -> str: # The "I already filed" handler is an API route, so it must hit the API host # (api.performancewest.net), NOT the Astro site host (performancewest.net). api_base = os.getenv("PUBLIC_API_URL", "https://api.performancewest.net").rstrip("/") return f"{api_base}/api/v1/ifta/filed?dot={dot}&t={_ifta_filed_token(dot)}" def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--date", help="override 'today' (YYYY-MM-DD) for testing") ap.add_argument("--start-campaign", action="store_true", help="schedule/start the campaign (default builds a draft)") ap.add_argument("--preview", action="store_true", help="send only to the owner test address with sample attribs") ap.add_argument("--dry-run", action="store_true") ap.add_argument("--limit", type=int, default=int(os.getenv("IFTA_DAILY_CAP", "2000")), help="max recipients this run (warmup-safe default 2000)") ap.add_argument("--ignore-window", action="store_true", help="build even if not in the reminder window (manual sends)") args = ap.parse_args() logging_level = os.getenv("LOG_LEVEL", "INFO") import logging logging.basicConfig(level=logging_level, format="%(asctime)s %(levelname)s %(message)s") today = datetime.strptime(args.date, "%Y-%m-%d").date() if args.date else date.today() touch, (q, due, period) = due_touch_today(today) if touch is None and not args.ignore_window and not args.preview: LOG.info("[ifta] no touch fires today (next %s due %s); exiting", q, due) return 0 if touch is None: touch = TOUCHES[0] # preview/manual default to the first touch days_before, touch_no, subj_suffix, headline_t, urgency = touch src = os.getenv(SOURCE_ENV) if not src: LOG.error("[ifta] %s not set -- create the IFTA source campaign first", SOURCE_ENV) return 2 base = tc.get_base_campaign(int(src)) conn = psycopg2.connect(tc.DB_URL) # New-cycle reset: at the start of each quarter, clear prior touch/self-filed # marks so this quarter starts fresh. Runs once per cycle. if args.start_campaign and not args.preview and not args.dry_run: _reset_cycle_if_new(conn, q, due) coupon = None if args.start_campaign and not args.preview and not args.dry_run: try: coupon = tc.get_or_create_daily_coupon(conn, today) except Exception as exc: # noqa: BLE001 LOG.warning("[ifta] coupon mint failed: %s (sending without)", exc) cur = conn.cursor() cur.execute(SELECT_SQL, [list(tc.BLOCKED_EMAIL_DOMAINS), touch_no, args.limit]) rows = cur.fetchall() LOG.info("[ifta] %s due %s | touch %d (%d biz-days before) | %d candidate carriers", q, due, touch_no, days_before, len(rows)) if not rows and not args.preview: LOG.info("[ifta] no candidates for touch %d; exiting", touch_no) return 0 due_human = due.strftime("%B %-d, %Y") headline = headline_t.format(due=due_human) def attribs(dot, company, state): lp = f"{LP_LINK}?code={coupon}" if coupon else LP_LINK a = { "dot_number": dot or "", "company": company or "", "state": state or "", "lp_link": lp, "ifta_due_date": due_human, "ifta_quarter": period, "ifta_headline": headline, "ifta_urgency": urgency, "ifta_filed_link": _filed_link(dot) if dot else "", } a.update(tc.coupon_attribs(coupon)) return a if args.preview: subs = [{ "email": tc.TEST_EMAIL, "name": "Sample Carrier", "attribs": attribs("0000000", "Sample Carrier LLC", "TX"), }] else: subs = [{ "email": r[1], "name": r[2] or r[1], "attribs": attribs(r[0], r[2], r[3]), } for r in rows] send_date = today.isoformat() list_name = f"IFTA {q} {due.year} T{touch_no} - {send_date}" campaign_name = f"IFTA Reminder {q} {due.year} - Touch {touch_no} ({days_before}bd) - {today.strftime('%b %d %Y')}" if args.dry_run: LOG.info("[ifta] DRY RUN -- touch %d would send to %d carriers (coupon=%s, due=%s)", touch_no, len(subs), coupon, due_human) return 0 # Per-touch subject override on the cloned campaign. base = dict(base) base["subject"] = f"DOT# {{{{ .Subscriber.Attribs.dot_number }}}} - Your IFTA quarterly return {subj_suffix}" list_id = tc.create_list(list_name) added = tc.import_subscribers(list_id, subs) LOG.info("[ifta] list %d: %d/%d subscribers added", list_id, added, len(subs)) if added == 0: LOG.error("[ifta] 0 subscribers added; aborting") return 1 send_at = datetime.now(timezone.utc) + timedelta(minutes=5) cid = tc.create_and_schedule_campaign( base, list_id, campaign_name, send_at, schedule=args.start_campaign and not args.preview) LOG.info("[ifta] campaign %d created (%s)", cid, "scheduled" if args.start_campaign and not args.preview else "draft") # Record the touch so the next touch only hits carriers who have not yet # reached this touch level (and never re-hits within a touch). if args.start_campaign and not args.preview and rows: cur.execute(""" UPDATE fmcsa_carriers SET ifta_reminded_at = now(), ifta_touch_no = %s WHERE dot_number = ANY(%s::text[]) """, (touch_no, [r[0] for r in rows])) conn.commit() LOG.info("[ifta] marked %d carriers at touch %d", len(rows), touch_no) conn.close() return 0 if __name__ == "__main__": raise SystemExit(main())