#!/usr/bin/env python3 """ verify_email_marketing.py — the email-strategist master's real verify gate. Scores ONE deliverable file against the discipline rubric defined in growth-masters/email-marketing/MANUAL.md §8.1, for the deliverable type declared in the file's frontmatter. Exits 0 if the artifact meets the standard, 1 if it fails, printing every named check that failed. python3 stdlib only. No network, no LLM call — this is a mechanical linter, not a judgment call. It exists to catch the tells named in MANUAL.md §8.2 (open rates as success, unclosed loops, fake scarcity, multi-CTA emails, blasting the whole list, skipping Layer 0, cherry-picked numbers, cadence without freshness, corporate-blast tone). Usage: python3 verify_email_marketing.py """ import re import sys from pathlib import Path # --------------------------------------------------------------------------- # House-voice banned vocabulary (mirrors ~/.claude/skills/brand-bible/skill.md # "Absolute forbiddens" table). Case-insensitive substring match. # --------------------------------------------------------------------------- BANNED_VOCAB = [ "—", # em-dash "delve", "leverage", "leveraging", "utilize", "paradigm shift", "in today's fast-paced world", "in conclusion", "it's worth noting", "it's important to note", "whilst", "moreover", "furthermore", "nevertheless", "vibrant tapestry", "rich tapestry", "bustling", "navigate the complexities", "a plethora of", "a myriad of", "foster", "empower", "empowerment", "seamless", "streamline", "robust", "holistic", "synergy", "cutting-edge", "ever-evolving", "multifaceted", "testament to", ] URGENCY_WORDS = [ "closes", "closing", "only", "last chance", "limited", "hurry", "deadline", "act now", "ends soon", "spots left", "running out", "expires", "final hours", "don't miss", ] PRODUCT_LEDE_WORDS = [ "ziion", "passport", "buy now", "purchase", "discount", "% off", "offer", "$", "price", "sale", "cart", "checkout", "order now", ] GOAL_TAGS = {"orient", "story", "proof", "soft-pitch", "pitch"} REQUIRED_ECOM_FLOWS = { "welcome", "browse-abandon", "cart-abandon", "post-purchase", "winback" } REQUIRED_AUDIT_ITEMS = { "spf": ["spf"], "dkim": ["dkim"], "dmarc": ["dmarc"], "one-click unsubscribe": ["one-click unsubscribe", "one click unsubscribe", "rfc 8058"], "spam-rate": ["spam-rate", "spam rate", "postmaster"], "list-hygiene": ["list-hygiene", "list hygiene", "sunset"], "warm-up plan": ["warm-up", "warmup", "warm up"], } # --------------------------------------------------------------------------- # Generic helpers # --------------------------------------------------------------------------- def md_links(s): return re.findall(r'\[([^\]]+)\]\(([^)]+)\)', s or "") def distinct_link_targets(s): """Number of distinct link destinations — repeating the SAME CTA target twice in a body (reinforcement) is not a multi-CTA violation; two DIFFERENT destinations is (Fogg: one prompt, one action).""" return len({url for _, url in md_links(s)}) def word_count(s): return len(re.findall(r'\S+', s or "")) def read_frontmatter_type(text): m = re.search(r'^---\s*\n(.*?)\n---\s*\n', text, re.DOTALL | re.MULTILINE) if not m: return None block = m.group(1) tm = re.search(r'^\s*type:\s*([a-z0-9_-]+)\s*$', block, re.MULTILINE | re.IGNORECASE) return tm.group(1).lower() if tm else None def strip_frontmatter(text): return re.sub(r'^---\s*\n.*?\n---\s*\n', '', text, count=1, flags=re.DOTALL | re.MULTILINE) def split_sections(text, header_pattern): """Split text into (header_line, body_text) chunks on a heading regex.""" matches = list(re.finditer(header_pattern, text, re.MULTILINE)) sections = [] for i, m in enumerate(matches): start = m.end() end = matches[i + 1].start() if i + 1 < len(matches) else len(text) sections.append((m.group(0).strip(), text[start:end])) return sections def parse_fields(block_text): """Parse 'Key: value' lines up to a 'Body:' marker; everything after is body.""" lines = block_text.splitlines() fields = {} body_lines = [] in_body = False for line in lines: if not in_body: stripped = line.strip() if stripped.lower().startswith("body:"): in_body = True rest = stripped[5:].strip() if rest: body_lines.append(rest) continue if not stripped: continue m = re.match(r'^([A-Za-z][A-Za-z _-]*):\s*(.*)$', stripped) if m: key = m.group(1).strip().lower().replace(" ", "_").replace("-", "_") fields[key] = m.group(2).strip() else: body_lines.append(line) body = "\n".join(body_lines).strip() return fields, body def check_compliance(fields, idx, label, failures): unsub = fields.get("unsubscribe", "") postal = fields.get("postaladdress", "") or fields.get("postal_address", "") if "{{" not in unsub or "}}" not in unsub: failures.append(f"{label} {idx}: missing unsubscribe merge tag") if "{{" not in postal or "}}" not in postal: failures.append(f"{label} {idx}: missing postal-address merge tag") def has_banned_vocab(text): low = text.lower() hits = [] for w in BANNED_VOCAB: needle = w.lower() if needle == "—": if "—" in text: hits.append("em-dash") elif needle in low: hits.append(w) return hits # --------------------------------------------------------------------------- # Rubric: welcome-sequence # --------------------------------------------------------------------------- def check_welcome_sequence(text): failures = [] sections = split_sections(text, r'^##\s*Email\s+(\d+)\s*$') if not sections: return ["no '## Email N' blocks found"] n = len(sections) if not (5 <= n <= 7): failures.append(f"sequence has {n} emails; must define 5-7") loops_opened = {} loops_closed = set() pitch_count = 0 delivers_lead_magnet = False for idx, (header, block) in enumerate(sections, start=1): fields, body = parse_fields(block) for req in ("day", "subject", "goal", "cta"): if not fields.get(req): failures.append(f"email {idx}: missing required field '{req}'") subject = fields.get("subject", "") if subject and word_count(subject) > 9: failures.append(f"email {idx}: subject exceeds 9 words ('{subject}')") goal = fields.get("goal", "").lower() if goal and goal not in GOAL_TAGS: failures.append(f"email {idx}: invalid goal tag '{goal}' (must be one of {sorted(GOAL_TAGS)})") if goal == "pitch": pitch_count += 1 if idx < 3: failures.append(f"email {idx}: hard pitch before email 3") loopopen = fields.get("loopopen", "none").lower() loopclose = fields.get("loopclose", "none").lower() if loopopen not in ("", "none"): loops_opened[loopopen] = idx if loopclose not in ("", "none"): loops_closed.add(loopclose) check_compliance(fields, idx, "email", failures) n_targets = distinct_link_targets(fields.get("cta", "") + "\n" + body) if n_targets != 1: failures.append(f"email {idx}: expected exactly 1 CTA link target, found {n_targets} (multi-CTA fails)") if idx == 1: delivers = fields.get("delivers", "").lower() if "lead-magnet" in delivers or "lead_magnet" in delivers: delivers_lead_magnet = True if not delivers_lead_magnet: failures.append("email 1 does not deliver the promised lead-magnet link (missing 'Delivers: lead-magnet')") if pitch_count != 1: failures.append(f"expected exactly 1 hard-pitch email (goal: pitch), found {pitch_count}") unclosed = sorted(lid for lid in loops_opened if lid not in loops_closed) if unclosed: failures.append(f"loop(s) opened and never closed: {unclosed}") return failures # --------------------------------------------------------------------------- # Rubric: launch-sequence # --------------------------------------------------------------------------- def check_launch_sequence(text): failures = [] sections = split_sections(text, r'^##\s*Email\s+(\d+)\s*$') if not sections: return ["no '## Email N' blocks found"] counts = {"prelaunch": 0, "cart-open": 0, "proof": 0, "objection": 0, "deadline": 0, "post-close": 0} deadlines_seen = set() offers = set() for idx, (header, block) in enumerate(sections, start=1): fields, body = parse_fields(block) phase = fields.get("phase", "").lower() if phase in counts: counts[phase] += 1 else: failures.append(f"email {idx}: missing/invalid Phase field ('{phase}')") if phase == "deadline": dl = fields.get("deadline", "") if not dl: failures.append(f"email {idx}: deadline-phase email missing 'Deadline' value") else: deadlines_seen.add(dl) offer = fields.get("offer", "") if offer: offers.add(offer) combined = (body + " " + fields.get("subject", "")).lower() has_urgency_language = any(w in combined for w in URGENCY_WORDS) urgency_tag = fields.get("urgency", "").lower() if has_urgency_language: if not urgency_tag.startswith("real:") or not urgency_tag.split(":", 1)[1].strip(): failures.append(f"email {idx}: urgency language present but not tagged '[real:mechanism]'") check_compliance(fields, idx, "email", failures) n_targets = distinct_link_targets(fields.get("cta", "") + "\n" + body) if n_targets > 1: failures.append(f"email {idx}: multiple CTA link targets found ({n_targets})") if counts["prelaunch"] < 2: failures.append(f"only {counts['prelaunch']} prelaunch value email(s); need >=2") if counts["cart-open"] < 1: failures.append("missing cart-open email") if counts["proof"] < 1: failures.append("missing proof/social-proof email (need >=1)") if counts["objection"] < 1: failures.append("missing objection/FAQ email (need >=1)") if counts["deadline"] < 2: failures.append(f"only {counts['deadline']} deadline-day email(s); need >=2") if counts["post-close"] < 1: failures.append("missing post-close email") if len(deadlines_seen) > 1: failures.append(f"mismatched deadlines across deadline emails: {sorted(deadlines_seen)}") if len(offers) > 1: failures.append(f"multiple offers found ({sorted(offers)}); single offer per sequence required") return failures # --------------------------------------------------------------------------- # Rubric: ecom-flow-set # --------------------------------------------------------------------------- def check_ecom_flow_set(text): failures = [] sections = split_sections(text, r'^##\s*Flow:\s*([a-zA-Z0-9_-]+)\s*$') if not sections: return ["no '## Flow: ' blocks found"] found_flows = set() for header, block in sections: m = re.match(r'^##\s*Flow:\s*([a-zA-Z0-9_-]+)', header, re.IGNORECASE) flowname = m.group(1).lower() if m else "unknown" found_flows.add(flowname) fields, body = parse_fields(block) if not fields.get("trigger"): failures.append(f"flow '{flowname}': missing Trigger event name") if not fields.get("exit"): failures.append(f"flow '{flowname}': missing Exit/suppression condition") if not fields.get("emails"): failures.append(f"flow '{flowname}': missing Emails count/timing offsets") if not fields.get("metric"): failures.append(f"flow '{flowname}': missing success Metric") if not fields.get("benchmark"): failures.append(f"flow '{flowname}': missing Benchmark reference") missing = REQUIRED_ECOM_FLOWS - found_flows if missing: failures.append(f"missing required flow(s): {sorted(missing)}") return failures # --------------------------------------------------------------------------- # Rubric: daily-email # --------------------------------------------------------------------------- def check_daily_email(text): failures = [] sections = split_sections(text, r'^##\s*Email\s+(\d+)\s*$') if not sections: return ["no '## Email 1' block found"] if len(sections) != 1: failures.append(f"daily-email must be a single email; found {len(sections)} blocks") fields, body = parse_fields(sections[0][1]) subject = fields.get("subject", "") if not subject: failures.append("missing Subject field") elif word_count(subject) > 9: failures.append(f"subject exceeds 9 words ('{subject}')") paragraphs = [p.strip() for p in re.split(r'\n\s*\n', body) if p.strip()] first_para = paragraphs[0] if paragraphs else "" if any(w in first_para.lower() for w in PRODUCT_LEDE_WORDS): failures.append("first paragraph mentions the product/hard-sells (hard-sell-first fails; lede must be story/observation)") if not fields.get("pivot"): failures.append("missing 'Pivot:' field (the bridge line from story to point)") n_targets = distinct_link_targets(fields.get("cta", "") + "\n" + body) if n_targets != 1: failures.append(f"expected exactly 1 CTA link target, found {n_targets} (multi-CTA fails)") wc = word_count(body) if wc > 350: failures.append(f"body is {wc} words; must be <=350") hits = has_banned_vocab(subject + " " + body) if hits: failures.append(f"house-voice check failed — banned vocabulary found: {sorted(set(hits))}") check_compliance(fields, 1, "email", failures) return failures # --------------------------------------------------------------------------- # Rubric: reactivation-campaign # --------------------------------------------------------------------------- def check_reactivation_campaign(text): failures = [] header_end = re.search(r'^##\s*Email\s+1\s*$', text, re.MULTILINE) preamble = text[: header_end.start()] if header_end else text pre_fields, _ = parse_fields(preamble) segment = pre_fields.get("segment", "") if not segment: failures.append("missing top-level 'Segment:' explicit inactivity rule") elif not re.search(r'\d+\s*d(ay)?s?', segment.lower()): failures.append(f"'Segment:' rule does not state an explicit day-based inactivity window ('{segment}')") sunset = pre_fields.get("sunset", "") if not sunset: failures.append("missing top-level 'Sunset:' rule for non-responders") deliverability_note = pre_fields.get("deliverabilitynote", "") or pre_fields.get("deliverability_note", "") if not deliverability_note: failures.append("missing top-level 'DeliverabilityNote:' explaining sender-reputation protection") sections = split_sections(text, r'^##\s*Email\s+(\d+)\s*$') if not sections: return failures + ["no '## Email N' blocks found"] n = len(sections) if not (2 <= n <= 3): failures.append(f"reactivation arc has {n} emails; must define 2-3") for idx, (header, block) in enumerate(sections, start=1): fields, body = parse_fields(block) check_compliance(fields, idx, "email", failures) if idx == 1: subj_wc = word_count(fields.get("subject", "")) body_wc = word_count(body) total = subj_wc + body_wc if total > 25: failures.append(f"email 1 is {total} words (subject+body); must be <=25 (nine-word-style)") return failures # --------------------------------------------------------------------------- # Rubric: deliverability-audit # --------------------------------------------------------------------------- def check_deliverability_audit(text): failures = [] row_re = re.compile(r'^\|(.+)\|(.+)\|(.+)\|$') rows = [] for line in text.splitlines(): line = line.strip() m = row_re.match(line) if not m: continue cols = [c.strip() for c in m.groups()] # skip separator rows like | --- | --- | --- | if all(re.fullmatch(r':?-{2,}:?', c) for c in cols): continue # skip header row (labeled "Item"/"Status"/"Evidence") if cols[0].lower() in ("item", "check", "label"): continue rows.append(cols) if not rows: return ["no checklist table rows found (expected '| Item | [status] | Evidence |' rows)"] covered = set() for label, status, evidence in rows: status_m = re.search(r'\[(verified|failed|todo)\]', status.lower()) if not status_m: failures.append(f"item '{label}': status column has no [verified|failed|todo] tag") if not evidence.strip() or evidence.strip() in ("", "-", "—"): failures.append(f"item '{label}': no evidence line — items with no evidence fail the linter") low_label = label.lower() for key, aliases in REQUIRED_AUDIT_ITEMS.items(): if any(a in low_label for a in aliases): covered.add(key) missing = set(REQUIRED_AUDIT_ITEMS.keys()) - covered if missing: failures.append(f"missing required checklist item(s): {sorted(missing)}") spam_rows = [r for r in rows if "spam" in r[0].lower() or "postmaster" in r[0].lower()] if spam_rows and not any(re.search(r'0\.\d+\s*%|<\s*0\.3\s*%|<\s*0\.1\s*%', r[2]) for r in spam_rows): failures.append("spam-rate item does not state the numeric <0.3% threshold (or a reading) in its evidence") return failures DISPATCH = { "welcome-sequence": check_welcome_sequence, "launch-sequence": check_launch_sequence, "ecom-flow-set": check_ecom_flow_set, "daily-email": check_daily_email, "reactivation-campaign": check_reactivation_campaign, "deliverability-audit": check_deliverability_audit, } def main(argv): if len(argv) != 2: print("usage: python3 verify_email_marketing.py ", file=sys.stderr) return 2 path = Path(argv[1]) if not path.exists(): print(f"FAIL: file not found: {path}", file=sys.stderr) return 1 text = path.read_text(encoding="utf-8") dtype = read_frontmatter_type(text) if not dtype: print(f"FAIL: {path} — no frontmatter 'type: ' found", file=sys.stderr) return 1 checker = DISPATCH.get(dtype) if not checker: print(f"FAIL: {path} — unknown deliverable type '{dtype}' (known: {sorted(DISPATCH)})", file=sys.stderr) return 1 body_text = strip_frontmatter(text) failures = checker(body_text) if failures: print(f"FAIL [{dtype}] {path} — {len(failures)} check(s) failed:") for f in failures: print(f" - {f}") return 1 print(f"PASS [{dtype}] {path} — meets the email-marketing rubric (MANUAL.md §8.1)") return 0 if __name__ == "__main__": sys.exit(main(sys.argv))