Email Sequence

#!/usr/bin/env python3
"""
Email Sequence Performance Analyzer

Analyzes email sequence metrics against industry benchmarks,
identifies bottlenecks, and recommends optimizations.

Usage:
    python performance_analyzer.py metrics.json
    python performance_analyzer.py metrics.json --json
    python performance_analyzer.py --sample

Input JSON format:
{
    "sequence_name": "Welcome Onboarding",
    "sequence_type": "welcome",
    "emails": [
        {
            "position": 1, "subject": "Welcome", "day": 0,
            "sent": 1000, "delivered": 980, "opened": 620,
            "clicked": 180, "converted": 45, "unsubscribed": 3
        }
    ]
}
"""

import argparse
import json
import sys
from pathlib import Path

BENCHMARKS = {
    "welcome": {"open": (50, 70), "click": (10, 20), "conversion": (5, 15), "unsub": 0.5},
    "nurture": {"open": (25, 40), "click": (3, 8), "conversion": (1, 3), "unsub": 0.3},
    "re_engagement": {"open": (15, 25), "click": (2, 5), "conversion": (3, 8), "unsub": 3.0},
    "trial_expiration": {"open": (40, 60), "click": (8, 15), "conversion": (10, 25), "unsub": 0.5},
    "post_purchase": {"open": (40, 55), "click": (5, 12), "conversion": (2, 8), "unsub": 0.3},
}

SAMPLE = {
    "sequence_name": "SaaS Welcome Onboarding",
    "sequence_type": "welcome",
    "emails": [
        {"position": 1, "subject": "Welcome to Acme", "day": 0, "sent": 1000, "delivered": 985, "opened": 620, "clicked": 180, "converted": 45, "unsubscribed": 3},
        {"position": 2, "subject": "The one feature", "day": 1, "sent": 982, "delivered": 975, "opened": 480, "clicked": 120, "converted": 30, "unsubscribed": 2},
        {"position": 3, "subject": "How Stripe did it", "day": 3, "sent": 950, "delivered": 945, "opened": 350, "clicked": 75, "converted": 15, "unsubscribed": 4},
        {"position": 4, "subject": "Quick question", "day": 5, "sent": 931, "delivered": 925, "opened": 300, "clicked": 50, "converted": 8, "unsubscribed": 3},
        {"position": 5, "subject": "Things you missed", "day": 7, "sent": 920, "delivered": 915, "opened": 280, "clicked": 60, "converted": 12, "unsubscribed": 5},
        {"position": 6, "subject": "Trial halfway done", "day": 10, "sent": 903, "delivered": 898, "opened": 380, "clicked": 95, "converted": 25, "unsubscribed": 3},
        {"position": 7, "subject": "Last day", "day": 13, "sent": 870, "delivered": 865, "opened": 420, "clicked": 110, "converted": 35, "unsubscribed": 8},
    ],
}


def pct(num, den):
    return round(num / den * 100, 2) if den > 0 else 0.0


def analyze(data: dict) -> dict:
    seq_type = data.get("sequence_type", "welcome")
    bench = BENCHMARKS.get(seq_type, BENCHMARKS["welcome"])
    emails = data.get("emails", [])

    result = {
        "sequence_name": data.get("sequence_name", "Unnamed"),
        "sequence_type": seq_type,
        "email_count": len(emails),
        "email_metrics": [],
        "aggregate": {},
        "bottlenecks": [],
        "health_signals": [],
        "recommendations": [],
    }

    totals = {"sent": 0, "delivered": 0, "opened": 0, "clicked": 0, "converted": 0, "unsub": 0}

    for email in emails:
        sent = email.get("sent", 0)
        delivered = email.get("delivered", sent)
        opened = email.get("opened", 0)
        clicked = email.get("clicked", 0)
        converted = email.get("converted", 0)
        unsub = email.get("unsubscribed", 0)

        open_r = pct(opened, delivered)
        click_r = pct(clicked, delivered)
        conv_r = pct(converted, delivered)
        unsub_r = pct(unsub, delivered)
        ctr = pct(clicked, opened) if opened > 0 else 0

        em = {
            "position": email.get("position"),
            "subject": email.get("subject", ""),
            "day": email.get("day", 0),
            "open_rate": open_r,
            "click_rate": click_r,
            "conversion_rate": conv_r,
            "unsubscribe_rate": unsub_r,
            "click_to_open": ctr,
            "status": "healthy",
        }

        # Benchmark comparison
        if open_r < bench["open"][0]:
            em["status"] = "underperforming"
            em["issue"] = "Open rate below benchmark"
        if click_r < bench["click"][0]:
            em["status"] = "underperforming"
            em["issue"] = em.get("issue", "") + " | Click rate below benchmark"
        if unsub_r > bench["unsub"]:
            em["status"] = "warning"
            em["issue"] = em.get("issue", "") + " | High unsubscribe rate"

        result["email_metrics"].append(em)

        totals["sent"] += sent
        totals["delivered"] += delivered
        totals["opened"] += opened
        totals["clicked"] += clicked
        totals["converted"] += converted
        totals["unsub"] += unsub

    # Aggregate
    result["aggregate"] = {
        "total_sent": totals["sent"],
        "total_converted": totals["converted"],
        "overall_open_rate": pct(totals["opened"], totals["delivered"]),
        "overall_click_rate": pct(totals["clicked"], totals["delivered"]),
        "overall_conversion_rate": pct(totals["converted"], totals["delivered"]),
        "overall_unsubscribe_rate": pct(totals["unsub"], totals["delivered"]),
        "sequence_conversion_rate": pct(totals["converted"], emails[0].get("sent", 1)) if emails else 0,
    }

    # Bottleneck detection
    metrics = result["email_metrics"]
    if len(metrics) >= 2:
        # Find biggest open rate drop
        max_open_drop = 0
        max_open_drop_pos = 0
        for i in range(1, len(metrics)):
            drop = metrics[i-1]["open_rate"] - metrics[i]["open_rate"]
            if drop > max_open_drop:
                max_open_drop = drop
                max_open_drop_pos = metrics[i]["position"]

        if max_open_drop > 10:
            result["bottlenecks"].append({
                "type": "open_rate_drop",
                "at_email": max_open_drop_pos,
                "drop": round(max_open_drop, 1),
                "fix": "Improve subject line or check if previous email fatigued recipients",
            })

        # Find high-open low-click emails
        for m in metrics:
            if m["open_rate"] > 30 and m["click_to_open"] < 10:
                result["bottlenecks"].append({
                    "type": "high_open_low_click",
                    "at_email": m["position"],
                    "open_rate": m["open_rate"],
                    "cto": m["click_to_open"],
                    "fix": "Subject line works but body/CTA is weak. Rewrite body copy and simplify CTA.",
                })

    # Health signals
    agg = result["aggregate"]
    if agg["overall_open_rate"] >= bench["open"][1]:
        result["health_signals"].append({"signal": "Open rates above benchmark", "status": "positive"})
    elif agg["overall_open_rate"] < bench["open"][0]:
        result["health_signals"].append({"signal": "Open rates below benchmark", "status": "negative"})
        result["recommendations"].append("Improve subject lines. Test 3+ variants per email position.")

    if agg["overall_unsubscribe_rate"] > bench["unsub"] * 2:
        result["health_signals"].append({"signal": "High unsubscribe rate", "status": "negative"})
        result["recommendations"].append("Reduce sending frequency or improve content relevance.")

    # Decay analysis
    if len(metrics) >= 3:
        early_open = sum(m["open_rate"] for m in metrics[:2]) / 2
        late_open = sum(m["open_rate"] for m in metrics[-2:]) / 2
        decay = round(early_open - late_open, 1)
        if decay > 15:
            result["recommendations"].append(f"Open rate decays by {decay}pp across sequence. Add variety in angles and formats.")

    if not result["recommendations"]:
        result["recommendations"].append("Sequence is performing well. Continue A/B testing subject lines and CTAs.")

    return result


def format_human(result: dict) -> str:
    lines = ["\n" + "=" * 65, "  EMAIL SEQUENCE PERFORMANCE ANALYZER", "=" * 65]
    lines.append(f"\n  {result['sequence_name']} ({result['sequence_type']})")
    agg = result["aggregate"]
    lines.append(f"  Sent: {agg['total_sent']} | Converted: {agg['total_converted']} | Seq. Conv: {agg['sequence_conversion_rate']}%")
    lines.append(f"  Open: {agg['overall_open_rate']}% | Click: {agg['overall_click_rate']}% | Unsub: {agg['overall_unsubscribe_rate']}%")

    lines.append(f"\n  {'#':<4} {'Day':<5} {'Open%':<8} {'Click%':<8} {'Conv%':<8} {'CTO%':<7} {'Status':<15} Subject")
    lines.append(f"  {'-'*80}")
    for m in result["email_metrics"]:
        lines.append(f"  {m['position']:<4} {m['day']:<5} {m['open_rate']:<8} {m['click_rate']:<8} {m['conversion_rate']:<8} {m['click_to_open']:<7} {m['status']:<15} {m['subject'][:25]}")

    if result["bottlenecks"]:
        lines.append(f"\n  Bottlenecks:")
        for b in result["bottlenecks"]:
            lines.append(f"    Email {b['at_email']}: {b['type']} -- {b['fix']}")

    if result["recommendations"]:
        lines.append(f"\n  Recommendations:")
        for i, r in enumerate(result["recommendations"], 1):
            lines.append(f"    {i}. {r}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Analyze email sequence performance metrics.")
    parser.add_argument("file", nargs="?")
    parser.add_argument("--json", action="store_true", dest="json_output")
    parser.add_argument("--sample", action="store_true")
    args = parser.parse_args()

    if args.sample:
        data = SAMPLE
    elif args.file:
        try:
            data = json.loads(Path(args.file).read_text())
        except (FileNotFoundError, json.JSONDecodeError) as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
    else:
        parser.print_help()
        sys.exit(1)

    result = analyze(data)
    if args.json_output:
        print(json.dumps(result, indent=2))
    else:
        print(format_human(result))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Email Sequence Mapper

Generates a visual sequence map with timing, branching logic,
and exit conditions from a sequence definition file.

Usage:
    python sequence_mapper.py sequence_def.json
    python sequence_mapper.py sequence_def.json --json
    python sequence_mapper.py --sample

Input JSON format:
{
    "name": "SaaS Welcome Onboarding",
    "trigger": "New signup",
    "goal": "Activate user",
    "emails": [
        {"position": 1, "day": 0, "subject": "Welcome", "purpose": "Set expectations"},
        {"position": 2, "day": 1, "subject": "Core feature", "purpose": "Drive activation"}
    ],
    "branches": [
        {"after_email": 2, "condition": "activated core feature", "action": "skip to email 4"}
    ],
    "exit_conditions": ["Converts to paid", "Unsubscribes"]
}
"""

import argparse
import json
import sys
from pathlib import Path

TIMING_BENCHMARKS = {
    "welcome": {"ideal_gap": [0, 1, 3, 5, 7, 10, 13], "max_days": 14},
    "nurture": {"ideal_gap": [0, 3, 7, 14, 21, 28], "max_days": 30},
    "trial_expiration": {"ideal_gap": [7, 3, 1, 0, -3], "max_days": 10},
    "re_engagement": {"ideal_gap": [0, 3, 7, 14], "max_days": 14},
}

SAMPLE = {
    "name": "SaaS Welcome Onboarding",
    "type": "welcome",
    "trigger": "New user signup",
    "goal": "Activate core feature within 14 days",
    "emails": [
        {"position": 1, "day": 0, "subject": "Welcome to [Product] -- start here", "purpose": "Set expectations, one CTA to activate"},
        {"position": 2, "day": 1, "subject": "The one feature that changes everything", "purpose": "Drive to core value action"},
        {"position": 3, "day": 3, "subject": "How [Company] got [Result] in [Timeframe]", "purpose": "Social proof, case study"},
        {"position": 4, "day": 5, "subject": "Quick question", "purpose": "Check engagement, offer help"},
        {"position": 5, "day": 7, "subject": "3 things you might have missed", "purpose": "Feature discovery"},
        {"position": 6, "day": 10, "subject": "Your trial is halfway done", "purpose": "Progress report, urgency"},
        {"position": 7, "day": 13, "subject": "Last day of your trial", "purpose": "Convert or lose access"},
    ],
    "branches": [
        {"after_email": 2, "condition": "User completed core action", "action": "Skip email 3, go to email 4"},
        {"after_email": 4, "condition": "No login for 5+ days", "action": "Switch to re-engagement variant"},
        {"after_email": 6, "condition": "Visited pricing page", "action": "Send pricing-focused email 7 variant"},
    ],
    "exit_conditions": ["Converts to paid", "Unsubscribes", "Requests removal"],
}


def analyze_sequence(data: dict) -> dict:
    emails = data.get("emails", [])
    branches = data.get("branches", [])
    exits = data.get("exit_conditions", [])
    seq_type = data.get("type", "welcome")

    result = {
        "name": data.get("name", "Unnamed"),
        "type": seq_type,
        "trigger": data.get("trigger", "Unknown"),
        "goal": data.get("goal", "Unknown"),
        "email_count": len(emails),
        "total_days": max((e.get("day", 0) for e in emails), default=0),
        "timing_analysis": [],
        "flow_map": [],
        "branch_count": len(branches),
        "exit_conditions": exits,
        "warnings": [],
        "suggestions": [],
    }

    # Timing analysis
    benchmark = TIMING_BENCHMARKS.get(seq_type, TIMING_BENCHMARKS["welcome"])
    days = [e.get("day", 0) for e in emails]
    gaps = [days[i+1] - days[i] for i in range(len(days)-1)]

    for i, email in enumerate(emails):
        entry = {
            "position": email.get("position", i+1),
            "day": email.get("day", 0),
            "subject": email.get("subject", ""),
            "purpose": email.get("purpose", ""),
        }
        if i < len(gaps):
            entry["gap_to_next"] = gaps[i]
        result["timing_analysis"].append(entry)

    # Flow map (ASCII visualization)
    for i, email in enumerate(emails):
        node = f"[Email {email.get('position', i+1)}] Day {email.get('day', 0)}: {email.get('subject', '')[:40]}"
        result["flow_map"].append(node)

        # Add branches after this email
        for branch in branches:
            if branch.get("after_email") == email.get("position", i+1):
                result["flow_map"].append(f"  IF {branch['condition']} -> {branch['action']}")

    # Warnings
    if len(emails) > 7:
        result["warnings"].append(f"Sequence has {len(emails)} emails. Consider trimming to 5-7 to reduce fatigue.")
    if len(emails) < 3:
        result["warnings"].append(f"Sequence has only {len(emails)} emails. Most conversions happen in follow-ups.")

    if any(g == 0 for g in gaps):
        result["warnings"].append("Some emails are sent on the same day. Ensure they have different triggers.")

    if any(g > 10 for g in gaps):
        result["warnings"].append("Long gaps (>10 days) between emails risk losing momentum.")

    if not branches:
        result["suggestions"].append("Add branching logic for different user behaviors (active vs. inactive).")

    if not exits:
        result["suggestions"].append("Define exit conditions (conversion, unsubscribe, sequence completion).")

    if len(exits) < 2:
        result["suggestions"].append("Add more exit conditions to prevent over-emailing converted users.")

    # Suppression check
    result["suggestions"].append("Set suppression rules to prevent overlap with other active sequences.")

    return result


def format_human(result: dict) -> str:
    lines = ["\n" + "=" * 65, "  EMAIL SEQUENCE MAPPER", "=" * 65]
    lines.append(f"\n  Name: {result['name']}")
    lines.append(f"  Type: {result['type']} | Emails: {result['email_count']} | Duration: {result['total_days']} days")
    lines.append(f"  Trigger: {result['trigger']}")
    lines.append(f"  Goal: {result['goal']}")

    lines.append(f"\n  Sequence Flow:")
    for node in result["flow_map"]:
        if node.startswith("  IF"):
            lines.append(f"    {node}")
        else:
            lines.append(f"    {node}")

    lines.append(f"\n  Timing:")
    for t in result["timing_analysis"]:
        gap = f" (+{t['gap_to_next']}d)" if "gap_to_next" in t else " (final)"
        lines.append(f"    Email {t['position']}: Day {t['day']}{gap} - {t['purpose']}")

    if result["exit_conditions"]:
        lines.append(f"\n  Exit Conditions:")
        for e in result["exit_conditions"]:
            lines.append(f"    - {e}")

    if result["warnings"]:
        lines.append(f"\n  Warnings:")
        for w in result["warnings"]:
            lines.append(f"    ! {w}")

    if result["suggestions"]:
        lines.append(f"\n  Suggestions:")
        for s in result["suggestions"]:
            lines.append(f"    > {s}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Map and analyze email sequences.")
    parser.add_argument("file", nargs="?", help="Sequence definition JSON file")
    parser.add_argument("--json", action="store_true", dest="json_output")
    parser.add_argument("--sample", action="store_true", help="Run with sample data")
    args = parser.parse_args()

    if args.sample:
        data = SAMPLE
    elif args.file:
        try:
            data = json.loads(Path(args.file).read_text())
        except (FileNotFoundError, json.JSONDecodeError) as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
    else:
        parser.print_help()
        sys.exit(1)

    result = analyze_sequence(data)
    if args.json_output:
        print(json.dumps(result, indent=2))
    else:
        print(format_human(result))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Email Sequence Subject Line Scorer

Scores email sequence subject lines for open-rate potential,
deliverability risk, and A/B testing readiness. Optimized for
automated/lifecycle emails (welcome, nurture, trial expiration).

Usage:
    python subject_line_scorer.py "Your trial ends tomorrow"
    python subject_line_scorer.py --file subjects.txt --json
"""

import argparse
import json
import re
import sys

SPAM_TRIGGERS = [
    "free", "guarantee", "act now", "limited time", "urgent",
    "click here", "buy now", "order now", "risk-free", "no obligation",
    "100%", "amazing", "incredible", "miracle", "cash bonus",
    "earn extra", "million", "congratulations", "winner",
]

SEQUENCE_TYPE_KEYWORDS = {
    "welcome": ["welcome", "get started", "confirm", "activate", "first steps"],
    "trial_expiration": ["trial", "expires", "ending", "last day", "access", "lose"],
    "nurture": ["how", "guide", "framework", "mistake", "learn", "discover"],
    "re_engagement": ["miss", "away", "back", "update", "new", "changed"],
    "post_purchase": ["thank", "receipt", "invoice", "getting started", "next steps"],
}

MOBILE_TRUNCATION = 35
DESKTOP_TRUNCATION = 50


def detect_sequence_type(subject: str) -> str:
    lower = subject.lower()
    scores = {}
    for stype, keywords in SEQUENCE_TYPE_KEYWORDS.items():
        scores[stype] = sum(1 for kw in keywords if kw in lower)
    best = max(scores, key=scores.get)
    return best if scores[best] > 0 else "unknown"


def score_subject(subject: str) -> dict:
    result = {
        "subject_line": subject,
        "char_count": len(subject),
        "word_count": len(subject.split()),
        "detected_type": detect_sequence_type(subject),
        "scores": {},
        "issues": [],
        "tips": [],
    }

    score = 100

    # Length check
    if len(subject) <= MOBILE_TRUNCATION:
        result["scores"]["length"] = 95
        result["tips"].append("Good length: visible on mobile without truncation.")
    elif len(subject) <= DESKTOP_TRUNCATION:
        result["scores"]["length"] = 80
        result["tips"].append("May truncate on mobile. Consider shortening to under 35 chars.")
    elif len(subject) <= 70:
        result["scores"]["length"] = 60
        result["issues"].append("Subject will truncate on most devices.")
        score -= 10
    else:
        result["scores"]["length"] = 30
        result["issues"].append("Subject is too long and will be cut off.")
        score -= 20

    # Spam triggers
    lower = subject.lower()
    found = [w for w in SPAM_TRIGGERS if w in lower]
    if not found:
        result["scores"]["spam_safety"] = 100
    else:
        result["scores"]["spam_safety"] = max(0, 100 - len(found) * 15)
        result["issues"].append(f"Spam triggers found: {', '.join(found)}")
        score -= len(found) * 10

    # Personalization check
    has_personalization = bool(re.search(r"\{|\[name\]|\[company\]", subject, re.IGNORECASE))
    if has_personalization:
        result["scores"]["personalization"] = 90
        result["tips"].append("Personalization token detected -- good for open rates.")
    else:
        result["scores"]["personalization"] = 50
        result["tips"].append("Consider adding personalization ({name}, {company}).")

    # Preview text complement check
    has_colon_setup = ":" in subject or " -- " in subject or " - " in subject
    result["scores"]["structure"] = 80 if has_colon_setup else 65

    # Urgency (appropriate for trial/re-engagement)
    urgency_words = ["last", "ending", "expires", "final", "today", "tomorrow", "hours left"]
    has_urgency = any(w in lower for w in urgency_words)
    if has_urgency and result["detected_type"] in ("trial_expiration", "re_engagement"):
        result["scores"]["urgency_fit"] = 90
        result["tips"].append("Appropriate urgency for this sequence type.")
    elif has_urgency:
        result["scores"]["urgency_fit"] = 60
        result["issues"].append("Urgency language may not fit this sequence type.")
        score -= 5

    # Caps and punctuation
    if re.search(r"\b[A-Z]{4,}\b", subject):
        non_acronyms = [w for w in re.findall(r"\b[A-Z]{4,}\b", subject) if w not in ("SALE", "FREE", "RSVP")]
        if non_acronyms:
            result["issues"].append(f"ALL CAPS words: {', '.join(non_acronyms)}")
            score -= 10
    if subject.count("!") > 1:
        result["issues"].append("Multiple exclamation marks reduce deliverability.")
        score -= 10

    # A/B test readiness
    words = subject.split()
    is_testable = 3 <= len(words) <= 8 and len(subject) <= 50
    result["ab_test_ready"] = is_testable
    if is_testable:
        result["tips"].append("Good length for A/B testing (recommend 3+ variants per email).")

    result["overall_score"] = max(0, min(100, score))
    result["grade"] = "A" if score >= 85 else "B" if score >= 70 else "C" if score >= 55 else "D" if score >= 40 else "F"

    return result


def format_human(results: list) -> str:
    lines = ["\n" + "=" * 55, "  EMAIL SEQUENCE SUBJECT LINE SCORER", "=" * 55]
    for r in results:
        lines.append(f"\n  \"{r['subject_line']}\"")
        lines.append(f"  Score: {r['overall_score']}/100 ({r['grade']}) | Type: {r['detected_type']} | {r['char_count']} chars")
        if r["issues"]:
            for i in r["issues"]:
                lines.append(f"    ! {i}")
        if r["tips"]:
            for t in r["tips"]:
                lines.append(f"    + {t}")
        lines.append(f"    A/B Ready: {'Yes' if r['ab_test_ready'] else 'No (adjust length)'}")
        lines.append("-" * 55)
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Score email sequence subject lines.")
    parser.add_argument("subject", nargs="?", help="Subject line to score")
    parser.add_argument("--file", "-f", help="File with one subject per line")
    parser.add_argument("--json", action="store_true", dest="json_output")
    args = parser.parse_args()

    subjects = []
    if args.file:
        with open(args.file) as f:
            subjects = [l.strip() for l in f if l.strip()]
    elif args.subject:
        subjects = [args.subject]
    else:
        parser.print_help()
        sys.exit(1)

    results = [score_subject(s) for s in subjects]
    if args.json_output:
        print(json.dumps(results if len(results) > 1 else results[0], indent=2))
    else:
        print(format_human(results))


if __name__ == "__main__":
    main()