6fce3ec9eb
- CAMPAIGN_COUPON_AB_PCTS="20,30,40" mints one daily code per arm; each carrier is bucketed by a stable sha256(email) hash so the split is even (~33/33/33 verified over 30k) and stable across re-sends (no arm-hopping). - Each arm's code stores its own percent in discount_codes, so the advertised discount always matches what checkout applies; redemptions are countable per code (marker campaign-daily:<date>:<pct>). - Empty/unset keeps legacy single-arm behavior (COUPON_PCT, legacy marker). - coupon_attribs() now takes per-recipient pct. - Tests: scripts/tests/test_coupon_ab.py (5 pass). SpamAssassin: both main campaigns (186/188) score 0.0 HAM across all 3 arms, coupon block renders clean; harness saved for re-runs.
100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Render a trucking campaign body with coupon merge-tags filled in and score it
|
|
through local SpamAssassin (4.0.1). Reads the body HTML from a file.
|
|
|
|
Usage: sa_coupon_local.py <body.html> <subject> <pct> <code>
|
|
"""
|
|
import html
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from email.message import EmailMessage
|
|
from email.utils import formatdate, make_msgid
|
|
|
|
BODY_FILE = sys.argv[1]
|
|
SUBJECT_TPL = sys.argv[2]
|
|
PCT = sys.argv[3] if len(sys.argv) > 3 else "40"
|
|
CODE = sys.argv[4] if len(sys.argv) > 4 else "KQMTN"
|
|
|
|
body = open(BODY_FILE, encoding="utf-8").read()
|
|
|
|
attribs = {
|
|
"dot_number": "1228791",
|
|
"company": "BLUE RIDGE FREIGHT LLC",
|
|
"state": "CT",
|
|
"lp_link": f"https://performancewest.net/order/mcs150-update?code={CODE}",
|
|
"coupon_code": CODE,
|
|
"coupon_pct": PCT,
|
|
"coupon_expires": "11:59 PM ET tonight",
|
|
}
|
|
|
|
|
|
def render(tpl: str, at: dict) -> str:
|
|
pat = re.compile(
|
|
r"\{\{\s*if\s+\.Subscriber\.Attribs\.(\w+)\s*\}\}(.*?)"
|
|
r"(?:\{\{\s*else\s*\}\}(.*?))?\{\{\s*end\s*\}\}",
|
|
re.DOTALL,
|
|
)
|
|
|
|
def repl(m):
|
|
key, t_branch, f_branch = m.group(1), m.group(2), m.group(3) or ""
|
|
return t_branch if str(at.get(key, "")).strip() else f_branch
|
|
|
|
prev = None
|
|
while prev != tpl:
|
|
prev = tpl
|
|
tpl = pat.sub(repl, tpl)
|
|
|
|
def val(m):
|
|
return html.escape(str(at.get(m.group(1), "")))
|
|
|
|
tpl = re.sub(r"\{\{\s*\.Subscriber\.Attribs\.(\w+)\s*\}\}", val, tpl)
|
|
tpl = tpl.replace("{{ UnsubscribeURL }}", "https://performancewest.net/unsubscribe/abc123")
|
|
tpl = re.sub(r"\{\{[^}]*\}\}", "", tpl)
|
|
return tpl
|
|
|
|
|
|
rendered_subject = render(SUBJECT_TPL, attribs)
|
|
rendered_body = render(body, attribs)
|
|
|
|
msg = EmailMessage()
|
|
msg["From"] = "Performance West <noreply@performancewest.net>"
|
|
msg["To"] = "dispatch@blueridgefreight.com"
|
|
msg["Subject"] = rendered_subject
|
|
msg["Date"] = formatdate(localtime=True)
|
|
msg["Message-ID"] = make_msgid(domain="performancewest.net")
|
|
msg["List-Unsubscribe"] = "<https://performancewest.net/unsubscribe/abc123>, <mailto:unsub@performancewest.net>"
|
|
msg["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click"
|
|
msg["Precedence"] = "bulk"
|
|
|
|
text_alt = re.sub(r"<[^>]+>", "", rendered_body)
|
|
text_alt = html.unescape(re.sub(r"\n{3,}", "\n\n", text_alt)).strip()
|
|
msg.set_content(text_alt)
|
|
msg.add_alternative(rendered_body, subtype="html")
|
|
|
|
raw = msg.as_bytes()
|
|
|
|
proc = subprocess.run(["spamassassin", "-t", "-L"], input=raw, capture_output=True)
|
|
scored = proc.stdout.decode("utf-8", "replace")
|
|
|
|
score_line = ""
|
|
for line in scored.splitlines():
|
|
if line.startswith("X-Spam-Status:"):
|
|
score_line = line
|
|
break
|
|
|
|
m = re.search(r"score=([-0-9.]+)", score_line)
|
|
score = m.group(1) if m else "?"
|
|
verdict = "SPAM" if score_line.startswith("X-Spam-Status: Yes") else "HAM"
|
|
|
|
print(f"arm={PCT}% code={CODE} -> SCORE {score} [{verdict}] subj: {rendered_subject}")
|
|
|
|
# Detailed rule report.
|
|
rep = re.search(r"X-Spam-Report:(.*?)\n(?:[A-Za-z-]+:|\n)", scored, re.DOTALL)
|
|
if "--verbose" in sys.argv:
|
|
if rep:
|
|
print(rep.group(1))
|
|
else:
|
|
for line in scored.splitlines():
|
|
if re.match(r"^\s*[-0-9.]+\s+[A-Z0-9_]+", line):
|
|
print(" ", line.strip())
|