X Twitter Growth

# tweets.csv — 50 sample tweets with engagement metrics for the X/Twitter growth analyzer
# Fields: tweet_id, date, type, content, likes, retweets, replies, impressions, link_clicks, profile_visits
# Types: text, thread, image, video, poll, link
# Account: @AcmeSaaS (B2B SaaS product)
tweet_id,date,type,content,likes,retweets,replies,impressions,link_clicks,profile_visits
T001,2026-01-05,text,"Hot take: Most SaaS companies are over-investing in acquisition and under-investing in activation.",142,38,22,8420,0,65
T002,2026-01-07,link,"We just published our 2025 Year in Review. Here's what we learned scaling to $10M ARR.",45,12,8,3200,280,32
T003,2026-01-09,image,"Our product roadmap for Q1 2026. What feature are you most excited about?",89,15,41,5100,0,28
T004,2026-01-11,text,"Unpopular opinion: Your onboarding flow matters more than your pricing page.",231,67,35,12800,0,95
T005,2026-01-13,thread,"Thread: 7 lessons from our first 1000 customers (1/8)",312,89,48,18500,0,142
T006,2026-01-15,video,"Quick demo: How to set up automated alerts in under 2 minutes.",67,11,14,4800,185,22
T007,2026-01-17,text,"Shipped: Dark mode is finally here. You asked we delivered.",198,42,63,9200,0,41
T008,2026-01-19,poll,"What's the #1 thing holding back your team's productivity? A) Meetings B) Context switching C) Tool sprawl D) Unclear priorities",156,28,32,7600,0,38
T009,2026-01-21,link,"New blog: Why we ditched our freemium model (and what happened next)",78,22,18,4100,340,45
T010,2026-01-23,text,"Friendly reminder that your best feature is reliability.",92,31,11,5800,0,19
T011,2026-01-25,image,"Before/after: Our dashboard redesign. Which do you prefer?",134,19,52,6900,0,33
T012,2026-01-27,text,"We don't do annual contracts. Here's why.",167,45,29,9800,0,72
T013,2026-01-29,thread,"How we reduced churn by 40% in 6 months (a thread) 1/6",278,82,41,15200,0,128
T014,2026-01-31,text,"Just realized our support team resolved 847 tickets this month with a 98.2% CSAT. Shoutout to the team.",112,18,24,5400,0,15
T015,2026-02-02,link,"Case study: How Globex Corp saved 120 hours/month with our platform",34,8,5,2100,190,18
T016,2026-02-04,text,"The best marketing channel is a product people actually talk about.",205,58,19,11400,0,88
T017,2026-02-06,video,"Building in public: Watch us ship a feature from idea to production in 48 hours.",88,24,31,6200,142,35
T018,2026-02-08,text,"gm. Shipping code.",12,2,3,980,0,4
T019,2026-02-10,image,"Our team at SaaStr Annual. Great conversations about the future of B2B.",45,6,8,2800,0,11
T020,2026-02-12,text,"If your error messages don't help users fix the problem they're just frustration messages.",276,78,22,14600,0,102
T021,2026-02-14,link,"We open-sourced our design system. 400+ components free to use.",189,62,28,10800,520,78
T022,2026-02-16,poll,"How many SaaS tools does your team use daily? A) 1-5 B) 6-10 C) 11-20 D) 20+",98,14,18,5200,0,22
T023,2026-02-18,text,"Stop building features nobody asked for. Start fixing bugs everybody complains about.",342,95,38,19200,0,155
T024,2026-02-20,thread,"The real cost of technical debt (thread) 1/5",198,54,27,11600,0,89
T025,2026-02-22,text,"Hiring: We're looking for a senior backend engineer. Remote US/EU. DM for details.",56,22,14,3400,0,48
T026,2026-02-24,video,"Customer spotlight: How Initech uses our API to automate their entire workflow.",42,9,6,2600,98,14
T027,2026-02-26,text,"v3.2 shipped. 14 bug fixes 3 new features 0 breaking changes.",87,15,19,4200,0,16
T028,2026-02-28,image,"Milestone: 5000 customers. Thank you for building with us.",224,41,56,12200,0,67
T029,2026-03-01,text,"Your pricing page is a product. Treat it like one.",118,33,15,6800,0,42
T030,2026-03-03,link,"Deep dive: How we handle 50K requests/second with a 12-person eng team",156,48,22,8900,410,62
T031,2026-03-05,text,"Meetings that could have been a Loom that could have been a Slack message that could have been nothing.",389,112,44,22400,0,178
T032,2026-03-07,thread,"Pricing lessons from experimenting with 6 different models (thread) 1/7",245,71,33,14800,0,112
T033,2026-03-09,text,"We just passed $1M in monthly revenue. Wild.",167,38,72,9600,0,54
T034,2026-03-11,image,"New feature: Collaborative dashboards. Build reports together in real-time.",78,12,16,4400,0,21
T035,2026-03-13,text,"Controversial: Self-serve is overrated. Most B2B buyers want to talk to a human.",198,52,48,11200,0,83
T036,2026-03-15,link,"Benchmark report: B2B SaaS metrics for 2026. Free download.",92,28,11,5600,380,44
T037,2026-03-17,video,"AMA recap: Our CEO answers your top 10 questions about our roadmap.",38,7,9,2200,65,12
T038,2026-03-19,text,"Every feature you add is a feature you have to maintain forever. Choose wisely.",154,44,16,8800,0,58
T039,2026-03-21,text,"Friday deploy? Bold strategy. Let's see how it plays out.",72,18,25,4000,0,13
T040,2026-03-23,poll,"Best time to post on X/Twitter for B2B? A) Early morning B) Lunch C) Late afternoon D) It doesn't matter",82,11,22,4600,0,18
T041,2026-03-24,image,"Infographic: The SaaS metrics that actually matter (not vanity metrics).",134,36,14,7200,0,42
T042,2026-03-25,text,"We spent $0 on paid ads last quarter. Our best acquisition channel is our product.",267,74,31,15400,0,118
T043,2026-03-26,thread,"What I wish I knew before starting a SaaS company (thread) 1/9",356,98,52,21000,0,165
T044,2026-03-27,link,"Tutorial: Build a custom integration with our API in 15 minutes",48,10,7,2800,220,16
T045,2026-03-28,text,"Documentation is a feature.",128,42,12,7400,0,35
T046,2026-03-28,text,"Replying to everyone asking about dark mode pricing: it's free. All features are included in every plan.",95,14,8,4800,0,18
T047,2026-03-29,video,"60-second tip: How to use keyboard shortcuts to 10x your speed.",62,11,9,3600,88,15
T048,2026-03-30,text,"Just had our best week ever for signups. Something is clicking.",84,16,21,4400,0,26
T049,2026-03-31,image,"March recap: 312 new customers 99.99% uptime 23 features shipped.",145,28,18,7800,0,38
T050,2026-03-31,text,"Q1 is done. Time to reflect plan and ship even harder in Q2. Let's go.",68,12,14,3200,0,19

#!/usr/bin/env python3
"""
Twitter/X Growth Tracker

Tracks follower growth, engagement rates, and identifies best posting times
from analytics data. Supports daily, weekly, and monthly period analysis.

Expected CSV columns: date, followers, impressions, engagements, likes,
  retweets, replies, tweets_posted, profile_visits

Usage:
    python growth_tracker.py analytics.csv
    python growth_tracker.py analytics.csv --period monthly
    python growth_tracker.py analytics.csv --format json
    python growth_tracker.py analytics.csv --compare-periods
"""

import argparse
import csv
import json
import statistics
import sys
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple


def parse_date(date_str: str) -> Optional[datetime]:
    """Parse date from common formats."""
    if not date_str or date_str.strip() == "":
        return None
    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%m/%d/%Y"):
        try:
            return datetime.strptime(date_str.strip(), fmt)
        except ValueError:
            continue
    return None


def safe_int(val: str, default: int = 0) -> int:
    try:
        return int(float(val))
    except (ValueError, TypeError):
        return default


def load_analytics(filepath: str) -> List[Dict[str, Any]]:
    """Load analytics data from CSV."""
    data = []
    with open(filepath, "r", newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            dt = parse_date(row.get("date", ""))
            if not dt:
                continue
            entry = {
                "date": dt,
                "followers": safe_int(row.get("followers", "0")),
                "impressions": safe_int(row.get("impressions", "0")),
                "engagements": safe_int(row.get("engagements", "0")),
                "likes": safe_int(row.get("likes", "0")),
                "retweets": safe_int(row.get("retweets", "0")),
                "replies": safe_int(row.get("replies", "0")),
                "tweets_posted": safe_int(row.get("tweets_posted", "0")),
                "profile_visits": safe_int(row.get("profile_visits", "0")),
            }
            data.append(entry)
    return sorted(data, key=lambda x: x["date"])


def group_by_period(data: List[Dict], period: str) -> Dict[str, List[Dict]]:
    """Group data entries by period."""
    groups = defaultdict(list)
    for entry in data:
        dt = entry["date"]
        if period == "daily":
            key = dt.strftime("%Y-%m-%d")
        elif period == "weekly":
            # Start of week (Monday)
            start = dt - timedelta(days=dt.weekday())
            key = f"{start.strftime('%Y-%m-%d')} to {(start + timedelta(days=6)).strftime('%Y-%m-%d')}"
        elif period == "monthly":
            key = dt.strftime("%Y-%m")
        else:
            key = dt.strftime("%Y-%m-%d")
        groups[key].append(entry)
    return dict(sorted(groups.items()))


def calculate_period_metrics(entries: List[Dict]) -> Dict[str, Any]:
    """Calculate aggregate metrics for a period."""
    if not entries:
        return {}

    followers_start = entries[0]["followers"]
    followers_end = entries[-1]["followers"]
    follower_growth = followers_end - followers_start

    total_impressions = sum(e["impressions"] for e in entries)
    total_engagements = sum(e["engagements"] for e in entries)
    total_likes = sum(e["likes"] for e in entries)
    total_retweets = sum(e["retweets"] for e in entries)
    total_replies = sum(e["replies"] for e in entries)
    total_tweets = sum(e["tweets_posted"] for e in entries)
    total_profile_visits = sum(e["profile_visits"] for e in entries)

    eng_rate = (total_engagements / total_impressions * 100) if total_impressions > 0 else 0

    # Per-tweet metrics
    per_tweet_impressions = total_impressions / total_tweets if total_tweets > 0 else 0
    per_tweet_engagements = total_engagements / total_tweets if total_tweets > 0 else 0

    # Follower conversion rate (profile visits to followers)
    conversion_rate = (follower_growth / total_profile_visits * 100) if total_profile_visits > 0 and follower_growth > 0 else 0

    return {
        "days": len(entries),
        "followers_start": followers_start,
        "followers_end": followers_end,
        "follower_growth": follower_growth,
        "follower_growth_pct": round(follower_growth / followers_start * 100, 2) if followers_start > 0 else 0,
        "total_impressions": total_impressions,
        "total_engagements": total_engagements,
        "engagement_rate": round(eng_rate, 2),
        "total_likes": total_likes,
        "total_retweets": total_retweets,
        "total_replies": total_replies,
        "total_tweets_posted": total_tweets,
        "total_profile_visits": total_profile_visits,
        "per_tweet_impressions": round(per_tweet_impressions, 0),
        "per_tweet_engagements": round(per_tweet_engagements, 1),
        "profile_conversion_rate": round(conversion_rate, 2),
    }


def calculate_growth_trajectory(data: List[Dict]) -> Dict[str, Any]:
    """Calculate growth trends and projections."""
    if len(data) < 2:
        return {"trend": "insufficient_data"}

    followers = [(e["date"], e["followers"]) for e in data if e["followers"] > 0]
    if len(followers) < 2:
        return {"trend": "insufficient_data"}

    # Daily growth rates
    daily_growth = []
    for i in range(1, len(followers)):
        days_diff = (followers[i][0] - followers[i-1][0]).days
        if days_diff > 0 and followers[i-1][1] > 0:
            growth = (followers[i][1] - followers[i-1][1]) / days_diff
            daily_growth.append(growth)

    if not daily_growth:
        return {"trend": "flat"}

    avg_daily = statistics.mean(daily_growth)
    recent_daily = statistics.mean(daily_growth[-7:]) if len(daily_growth) >= 7 else avg_daily

    current = followers[-1][1]
    # Projections
    days_to_next_milestone = None
    milestones = [100, 500, 1000, 5000, 10000, 25000, 50000, 100000]
    next_milestone = None
    for m in milestones:
        if current < m:
            next_milestone = m
            if recent_daily > 0:
                days_to_next_milestone = int((m - current) / recent_daily)
            break

    # Trend direction
    if len(daily_growth) >= 14:
        first_half = statistics.mean(daily_growth[:len(daily_growth)//2])
        second_half = statistics.mean(daily_growth[len(daily_growth)//2:])
        if second_half > first_half * 1.1:
            trend = "accelerating"
        elif second_half < first_half * 0.9:
            trend = "decelerating"
        else:
            trend = "steady"
    else:
        trend = "steady"

    return {
        "trend": trend,
        "avg_daily_growth": round(avg_daily, 1),
        "recent_daily_growth": round(recent_daily, 1),
        "current_followers": current,
        "next_milestone": next_milestone,
        "days_to_milestone": days_to_next_milestone,
        "projected_30d": int(current + recent_daily * 30),
        "projected_90d": int(current + recent_daily * 90),
    }


def find_best_posting_patterns(data: List[Dict]) -> Dict[str, Any]:
    """Identify best posting frequency and patterns."""
    if not data:
        return {}

    # Correlation between tweets posted and engagement
    daily_data = [(e["tweets_posted"], e["engagements"], e["impressions"]) for e in data if e["tweets_posted"] > 0]

    frequency_buckets = defaultdict(list)
    for tweets, eng, imp in daily_data:
        eng_rate = (eng / imp * 100) if imp > 0 else 0
        if tweets <= 1:
            frequency_buckets["1 tweet/day"].append(eng_rate)
        elif tweets <= 3:
            frequency_buckets["2-3 tweets/day"].append(eng_rate)
        elif tweets <= 5:
            frequency_buckets["4-5 tweets/day"].append(eng_rate)
        else:
            frequency_buckets["6+ tweets/day"].append(eng_rate)

    freq_analysis = {}
    for bucket, rates in frequency_buckets.items():
        if rates:
            freq_analysis[bucket] = {
                "days": len(rates),
                "avg_engagement_rate": round(statistics.mean(rates), 2),
            }

    # Best day of week
    dow_data = defaultdict(list)
    for e in data:
        day = e["date"].strftime("%A")
        if e["impressions"] > 0:
            eng_rate = e["engagements"] / e["impressions"] * 100
            dow_data[day].append(eng_rate)

    day_order = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
    dow_analysis = {}
    for day in day_order:
        if day in dow_data:
            dow_analysis[day] = {
                "avg_engagement_rate": round(statistics.mean(dow_data[day]), 2),
                "days_sampled": len(dow_data[day]),
            }

    return {
        "posting_frequency": freq_analysis,
        "day_of_week": dow_analysis,
    }


def format_number(n: int) -> str:
    if n >= 1_000_000:
        return f"{n/1_000_000:.1f}M"
    elif n >= 1_000:
        return f"{n/1_000:.1f}K"
    return str(n)


def print_human(period_data: Dict[str, Dict], trajectory: Dict,
                patterns: Dict, period_type: str) -> None:
    """Print growth report in human-readable format."""
    print("=" * 65)
    print(f"  X/Twitter Growth Report ({period_type})")
    print("=" * 65)

    # Trajectory
    t = trajectory
    if t.get("trend") != "insufficient_data":
        print(f"\n  --- Growth Trajectory ---")
        print(f"  Current Followers:   {format_number(t['current_followers'])}")
        print(f"  Trend:               {t['trend']}")
        print(f"  Avg Daily Growth:    {t['avg_daily_growth']:+.1f} followers/day")
        print(f"  Recent Daily Growth: {t['recent_daily_growth']:+.1f} followers/day")
        print(f"  30-Day Projection:   {format_number(t['projected_30d'])}")
        print(f"  90-Day Projection:   {format_number(t['projected_90d'])}")
        if t.get("next_milestone"):
            days = t["days_to_milestone"]
            if days and days > 0:
                print(f"  Next Milestone:      {format_number(t['next_milestone'])} (est. {days} days)")

    # Period breakdown
    print(f"\n  --- Period Breakdown ---")
    print(f"  {'Period':<22} {'Followers':>10} {'Growth':>8} {'Eng Rate':>9} {'Tweets':>7} {'Impressions':>12}")
    print(f"  {'-'*22} {'-'*10} {'-'*8} {'-'*9} {'-'*7} {'-'*12}")

    for period_key, metrics in period_data.items():
        growth_str = f"{metrics['follower_growth']:+,}"
        print(f"  {period_key:<22} {metrics['followers_end']:>10,} {growth_str:>8} "
              f"{metrics['engagement_rate']:>8.2f}% {metrics['total_tweets_posted']:>7} "
              f"{format_number(metrics['total_impressions']):>12}")

    # Posting patterns
    freq = patterns.get("posting_frequency", {})
    if freq:
        print(f"\n  --- Posting Frequency Impact ---")
        for bucket, stats in sorted(freq.items()):
            bar = "#" * max(1, int(stats["avg_engagement_rate"] * 3))
            print(f"  {bucket:<20} {stats['avg_engagement_rate']:.2f}% eng (n={stats['days']})  {bar}")

    dow = patterns.get("day_of_week", {})
    if dow:
        print(f"\n  --- Day of Week Performance ---")
        for day, stats in dow.items():
            bar = "#" * max(1, int(stats["avg_engagement_rate"] * 3))
            print(f"  {day:<12} {stats['avg_engagement_rate']:.2f}% eng  {bar}")

    # Health assessment
    print(f"\n  --- Assessment ---")
    last_period = list(period_data.values())[-1] if period_data else {}
    if last_period:
        eng = last_period.get("engagement_rate", 0)
        if eng >= 6:
            print(f"  Engagement rate is excellent ({eng:.2f}%)")
        elif eng >= 3:
            print(f"  Engagement rate is good ({eng:.2f}%)")
        elif eng >= 1:
            print(f"  Engagement rate is average ({eng:.2f}%) - focus on content quality")
        else:
            print(f"  Engagement rate is low ({eng:.2f}%) - review content strategy")

        growth_pct = last_period.get("follower_growth_pct", 0)
        if growth_pct > 10:
            print(f"  Strong follower growth ({growth_pct:.1f}%)")
        elif growth_pct > 0:
            print(f"  Positive follower growth ({growth_pct:.1f}%)")
        elif growth_pct == 0:
            print(f"  Flat follower growth - increase content volume or try new formats")
        else:
            print(f"  Follower decline ({growth_pct:.1f}%) - investigate content and engagement strategy")

    print()


def main():
    parser = argparse.ArgumentParser(
        description="Track follower growth, engagement rates, and best posting times"
    )
    parser.add_argument("file", help="CSV file with analytics data")
    parser.add_argument("--format", choices=["human", "json"], default="human", help="Output format")
    parser.add_argument("--period", choices=["daily", "weekly", "monthly"], default="weekly",
                        help="Aggregation period (default: weekly)")
    args = parser.parse_args()

    data = load_analytics(args.file)
    if not data:
        print("Error: No valid analytics data found", file=sys.stderr)
        sys.exit(1)

    grouped = group_by_period(data, args.period)
    period_metrics = {k: calculate_period_metrics(v) for k, v in grouped.items()}
    trajectory = calculate_growth_trajectory(data)
    patterns = find_best_posting_patterns(data)

    if args.format == "json":
        output = {
            "period_type": args.period,
            "periods": period_metrics,
            "trajectory": trajectory,
            "posting_patterns": patterns,
        }
        print(json.dumps(output, indent=2, default=str))
    else:
        print_human(period_metrics, trajectory, patterns, args.period)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Twitter Thread Builder

Structures long-form content into optimal Twitter threads with hooks,
transitions, and engagement triggers. Supports text and JSON input.

Input: Text file with sections (use # headers or [Section] markers) or JSON.

Usage:
    python thread_builder.py content.txt
    python thread_builder.py content.txt --target-tweets 8
    python thread_builder.py content.json --format json
    python thread_builder.py content.txt --style numbered
"""

import argparse
import json
import re
import sys
import textwrap
from typing import Any, Dict, List, Optional, Tuple


MAX_TWEET_LENGTH = 280
THREAD_CONNECTOR_WORDS = ["Here's the thing:", "But here's what happened:", "And that's where it gets interesting:",
                          "The key insight:", "What most people miss:", "Here's why this matters:"]


def load_content(filepath: str) -> Dict[str, Any]:
    """Load content from text or JSON file."""
    with open(filepath, "r", encoding="utf-8") as f:
        content = f.read().strip()

    # Try JSON first
    try:
        data = json.loads(content)
        return data
    except (json.JSONDecodeError, ValueError):
        pass

    # Parse as text with sections
    return parse_text_content(content)


def parse_text_content(text: str) -> Dict[str, Any]:
    """Parse text content into structured sections."""
    lines = text.split("\n")
    title = ""
    sections = []
    current_section = {"title": "", "content": ""}

    for line in lines:
        stripped = line.strip()

        # Detect title (first # heading)
        if stripped.startswith("# ") and not title:
            title = stripped[2:].strip()
            continue

        # Detect section headers
        if stripped.startswith("## ") or re.match(r"^\[.+\]$", stripped):
            if current_section["content"].strip():
                sections.append(current_section)
            section_title = stripped.lstrip("#").strip().strip("[]")
            current_section = {"title": section_title, "content": ""}
            continue

        current_section["content"] += line + "\n"

    if current_section["content"].strip():
        sections.append(current_section)

    # If no sections found, split by paragraphs
    if not sections:
        paragraphs = re.split(r"\n\s*\n", text)
        if paragraphs and not title:
            title = paragraphs[0].strip()[:100]
            paragraphs = paragraphs[1:]
        for i, para in enumerate(paragraphs):
            if para.strip():
                sections.append({"title": f"Point {i+1}", "content": para.strip()})

    return {"title": title, "sections": sections}


def split_text_to_tweets(text: str, max_length: int = MAX_TWEET_LENGTH) -> List[str]:
    """Split text into tweet-sized chunks, breaking at sentence boundaries."""
    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
    tweets = []
    current = ""

    for sentence in sentences:
        sentence = sentence.strip()
        if not sentence:
            continue

        # If single sentence exceeds limit, wrap it
        if len(sentence) > max_length:
            if current:
                tweets.append(current.strip())
                current = ""
            # Hard wrap long sentences
            wrapped = textwrap.wrap(sentence, width=max_length - 5)
            for chunk in wrapped:
                tweets.append(chunk.strip())
            continue

        test = f"{current} {sentence}".strip() if current else sentence
        if len(test) <= max_length:
            current = test
        else:
            if current:
                tweets.append(current.strip())
            current = sentence

    if current.strip():
        tweets.append(current.strip())

    return tweets


def build_thread(data: Dict[str, Any], target_tweets: int, style: str,
                 add_numbering: bool) -> List[Dict[str, str]]:
    """Build optimized thread from structured content."""
    title = data.get("title", "")
    sections = data.get("sections", [])

    thread = []

    # Tweet 1: Hook
    hook = data.get("hook", "")
    if not hook:
        if title:
            hook = title
            # Add engagement hook if title is short
            if len(hook) < 200:
                hook += "\n\nA thread:"
        elif sections:
            hook = sections[0]["content"].split(".")[0] + "."

    thread.append({"role": "hook", "text": hook[:MAX_TWEET_LENGTH]})

    # Build body tweets from sections
    body_tweets = []
    for section in sections:
        content = section["content"].strip()
        if not content:
            continue

        # Add section transition if titled
        if section.get("title") and section["title"] != content[:50]:
            content = f"{section['title']}:\n\n{content}"

        chunks = split_text_to_tweets(content, MAX_TWEET_LENGTH - 10)  # Leave room for numbering
        for chunk in chunks:
            body_tweets.append({"role": "body", "text": chunk})

    # If we have more tweets than target, consolidate
    if len(body_tweets) > target_tweets - 2:  # -2 for hook and closer
        consolidated = []
        i = 0
        while i < len(body_tweets):
            current = body_tweets[i]["text"]
            while i + 1 < len(body_tweets):
                combined = current + " " + body_tweets[i + 1]["text"]
                if len(combined) <= MAX_TWEET_LENGTH - 10:
                    current = combined
                    i += 1
                else:
                    break
            consolidated.append({"role": "body", "text": current})
            i += 1
        body_tweets = consolidated[:target_tweets - 2]

    thread.extend(body_tweets)

    # Final tweet: CTA / closer
    closer = data.get("cta", "")
    if not closer:
        closer = "If you found this valuable:\n\n"
        closer += "1. Follow me for more insights\n"
        closer += "2. Repost the first tweet to share\n"
        closer += "3. Drop a reply with your thoughts"

    thread.append({"role": "closer", "text": closer[:MAX_TWEET_LENGTH]})

    # Apply numbering
    if add_numbering or style == "numbered":
        total = len(thread)
        for i, tweet in enumerate(thread):
            prefix = f"{i+1}/{total} "
            if len(prefix + tweet["text"]) <= MAX_TWEET_LENGTH:
                tweet["text"] = prefix + tweet["text"]

    return thread


def analyze_thread(thread: List[Dict]) -> Dict[str, Any]:
    """Analyze thread quality metrics."""
    texts = [t["text"] for t in thread]
    lengths = [len(t) for t in texts]

    # Check for engagement triggers
    has_question = any("?" in t for t in texts[1:-1])  # Questions in body
    has_data = any(re.search(r'\d+%|\$\d+|\d+x', t) for t in texts)
    hook_length = lengths[0] if lengths else 0

    issues = []
    if hook_length < 80:
        issues.append("Hook tweet is short - consider making it more compelling")
    if hook_length > 250:
        issues.append("Hook tweet is very long - shorter hooks often perform better")
    if not has_question:
        issues.append("No questions in thread body - add one to drive replies")
    if not has_data:
        issues.append("No data points or statistics - consider adding one for credibility")

    avg_length = sum(lengths) / len(lengths) if lengths else 0
    fill_rate = avg_length / MAX_TWEET_LENGTH * 100

    return {
        "total_tweets": len(thread),
        "total_characters": sum(lengths),
        "avg_tweet_length": round(avg_length, 0),
        "character_fill_rate": round(fill_rate, 1),
        "shortest_tweet": min(lengths) if lengths else 0,
        "longest_tweet": max(lengths) if lengths else 0,
        "has_question_in_body": has_question,
        "has_data_point": has_data,
        "issues": issues,
    }


def print_human(thread: List[Dict], analysis: Dict) -> None:
    """Print thread in human-readable format."""
    print("=" * 60)
    print("  Twitter Thread Builder")
    print("=" * 60)

    for i, tweet in enumerate(thread):
        role_label = {"hook": "HOOK", "body": "BODY", "closer": "CTA"}.get(tweet["role"], "")
        chars = len(tweet["text"])
        bar = "#" * int(chars / MAX_TWEET_LENGTH * 20)
        print(f"\n  --- Tweet {i+1} [{role_label}] ({chars}/{MAX_TWEET_LENGTH} chars) {bar} ---")
        # Indent tweet text
        for line in tweet["text"].split("\n"):
            print(f"  {line}")

    # Analysis
    a = analysis
    print(f"\n  {'=' * 56}")
    print(f"  THREAD ANALYSIS")
    print(f"  {'=' * 56}")
    print(f"  Total Tweets:      {a['total_tweets']}")
    print(f"  Total Characters:  {a['total_characters']}")
    print(f"  Avg Tweet Length:  {a['avg_tweet_length']:.0f} chars ({a['character_fill_rate']:.0f}% fill)")
    print(f"  Has Questions:     {'Yes' if a['has_question_in_body'] else 'No'}")
    print(f"  Has Data Points:   {'Yes' if a['has_data_point'] else 'No'}")

    if a["issues"]:
        print(f"\n  Suggestions:")
        for issue in a["issues"]:
            print(f"  [!] {issue}")

    print()


def main():
    parser = argparse.ArgumentParser(
        description="Structure long-form content into optimal Twitter threads"
    )
    parser.add_argument("file", help="Text or JSON file with content")
    parser.add_argument("--format", choices=["human", "json"], default="human", help="Output format")
    parser.add_argument("--target-tweets", type=int, default=8,
                        help="Target number of tweets in thread (default: 8)")
    parser.add_argument("--style", choices=["numbered", "clean"], default="clean",
                        help="Thread style (default: clean)")
    parser.add_argument("--numbered", action="store_true", help="Add tweet numbering (1/N format)")
    args = parser.parse_args()

    data = load_content(args.file)
    thread = build_thread(data, args.target_tweets, args.style, args.numbered)
    analysis = analyze_thread(thread)

    if args.format == "json":
        output = {"thread": thread, "analysis": analysis}
        print(json.dumps(output, indent=2))
    else:
        print_human(thread, analysis)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Tweet Performance Analyzer

Analyzes tweet performance patterns from exported X/Twitter analytics data.
Identifies top-performing content types, optimal posting times, engagement
patterns, and actionable content insights.

Expected CSV columns: tweet_id, text, created_at, impressions, engagements,
  likes, retweets, replies, type, has_media

Usage:
    python tweet_analyzer.py tweets.csv
    python tweet_analyzer.py tweets.csv --format json
    python tweet_analyzer.py tweets.csv --min-impressions 500
    python tweet_analyzer.py tweets.csv --top 20
"""

import argparse
import csv
import json
import re
import statistics
import sys
from collections import Counter, defaultdict
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple


def parse_datetime(dt_str: str) -> Optional[datetime]:
    """Parse datetime from common formats."""
    if not dt_str or dt_str.strip() == "":
        return None
    for fmt in (
        "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M",
        "%m/%d/%Y %H:%M", "%Y-%m-%dT%H:%M:%SZ",
    ):
        try:
            return datetime.strptime(dt_str.strip(), fmt)
        except ValueError:
            continue
    return None


def safe_int(val: str, default: int = 0) -> int:
    try:
        return int(val)
    except (ValueError, TypeError):
        return default


def safe_float(val: str, default: float = 0.0) -> float:
    try:
        return float(val)
    except (ValueError, TypeError):
        return default


def load_tweets(filepath: str) -> List[Dict[str, Any]]:
    """Load tweet data from CSV."""
    tweets = []
    with open(filepath, "r", newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            dt = parse_datetime(row.get("created_at", ""))
            impressions = safe_int(row.get("impressions", "0"))
            engagements = safe_int(row.get("engagements", "0"))
            likes = safe_int(row.get("likes", "0"))
            retweets = safe_int(row.get("retweets", "0"))
            replies = safe_int(row.get("replies", "0"))

            eng_rate = (engagements / impressions * 100) if impressions > 0 else 0

            tweet = {
                "tweet_id": row.get("tweet_id", "").strip(),
                "text": row.get("text", "").strip(),
                "created_at": dt,
                "impressions": impressions,
                "engagements": engagements,
                "likes": likes,
                "retweets": retweets,
                "replies": replies,
                "engagement_rate": round(eng_rate, 2),
                "type": row.get("type", "single").strip().lower(),
                "has_media": row.get("has_media", "").strip().lower() in ("yes", "true", "1"),
                "hour": dt.hour if dt else None,
                "day_of_week": dt.strftime("%A") if dt else None,
            }
            tweets.append(tweet)
    return tweets


def analyze_overall(tweets: List[Dict]) -> Dict[str, Any]:
    """Calculate overall performance metrics."""
    impressions = [t["impressions"] for t in tweets]
    eng_rates = [t["engagement_rate"] for t in tweets]
    likes = [t["likes"] for t in tweets]
    retweets = [t["retweets"] for t in tweets]
    replies = [t["replies"] for t in tweets]

    return {
        "total_tweets": len(tweets),
        "total_impressions": sum(impressions),
        "avg_impressions": round(statistics.mean(impressions), 0) if impressions else 0,
        "median_impressions": round(statistics.median(impressions), 0) if impressions else 0,
        "avg_engagement_rate": round(statistics.mean(eng_rates), 2) if eng_rates else 0,
        "median_engagement_rate": round(statistics.median(eng_rates), 2) if eng_rates else 0,
        "avg_likes": round(statistics.mean(likes), 1) if likes else 0,
        "avg_retweets": round(statistics.mean(retweets), 1) if retweets else 0,
        "avg_replies": round(statistics.mean(replies), 1) if replies else 0,
        "total_likes": sum(likes),
        "total_retweets": sum(retweets),
        "total_replies": sum(replies),
    }


def analyze_by_type(tweets: List[Dict]) -> Dict[str, Any]:
    """Analyze performance by tweet type."""
    type_groups = defaultdict(list)
    for t in tweets:
        type_groups[t["type"]].append(t)

    result = {}
    for tweet_type, group in sorted(type_groups.items()):
        eng_rates = [t["engagement_rate"] for t in group]
        impressions = [t["impressions"] for t in group]
        result[tweet_type] = {
            "count": len(group),
            "avg_engagement_rate": round(statistics.mean(eng_rates), 2),
            "avg_impressions": round(statistics.mean(impressions), 0),
            "total_impressions": sum(impressions),
        }
    return result


def analyze_by_media(tweets: List[Dict]) -> Dict[str, Any]:
    """Analyze performance by media presence."""
    with_media = [t for t in tweets if t["has_media"]]
    without_media = [t for t in tweets if not t["has_media"]]

    def stats(group):
        if not group:
            return {"count": 0, "avg_engagement_rate": 0, "avg_impressions": 0}
        eng = [t["engagement_rate"] for t in group]
        imp = [t["impressions"] for t in group]
        return {
            "count": len(group),
            "avg_engagement_rate": round(statistics.mean(eng), 2),
            "avg_impressions": round(statistics.mean(imp), 0),
        }

    return {"with_media": stats(with_media), "without_media": stats(without_media)}


def analyze_posting_times(tweets: List[Dict]) -> Dict[str, Any]:
    """Analyze performance by posting time."""
    hour_groups = defaultdict(list)
    day_groups = defaultdict(list)

    for t in tweets:
        if t["hour"] is not None:
            hour_groups[t["hour"]].append(t["engagement_rate"])
        if t["day_of_week"]:
            day_groups[t["day_of_week"]].append(t["engagement_rate"])

    # Best hours
    hour_stats = {}
    for hour, rates in sorted(hour_groups.items()):
        hour_stats[f"{hour:02d}:00"] = {
            "count": len(rates),
            "avg_engagement_rate": round(statistics.mean(rates), 2),
        }

    # Best days
    day_order = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
    day_stats = {}
    for day in day_order:
        if day in day_groups:
            rates = day_groups[day]
            day_stats[day] = {
                "count": len(rates),
                "avg_engagement_rate": round(statistics.mean(rates), 2),
            }

    # Find best posting windows
    best_hours = sorted(hour_stats.items(), key=lambda x: x[1]["avg_engagement_rate"], reverse=True)[:5]
    best_days = sorted(day_stats.items(), key=lambda x: x[1]["avg_engagement_rate"], reverse=True)[:3]

    return {
        "by_hour": hour_stats,
        "by_day": day_stats,
        "best_hours": [{"hour": h, "avg_eng_rate": s["avg_engagement_rate"], "tweets": s["count"]}
                       for h, s in best_hours],
        "best_days": [{"day": d, "avg_eng_rate": s["avg_engagement_rate"], "tweets": s["count"]}
                      for d, s in best_days],
    }


def analyze_text_patterns(tweets: List[Dict]) -> Dict[str, Any]:
    """Analyze text content patterns correlated with performance."""
    patterns = {
        "has_question": {"pattern": r"\?", "with": [], "without": []},
        "has_number": {"pattern": r"\d+", "with": [], "without": []},
        "has_emoji": {"pattern": r"[\U0001F600-\U0001F9FF]", "with": [], "without": []},
        "starts_with_hook": {"pattern": r"^(Here's|I just|Most people|The secret|Why do|How to|What if)",
                             "with": [], "without": []},
        "has_thread_marker": {"pattern": r"(thread|🧵|\d+/)", "with": [], "without": []},
    }

    for t in tweets:
        text = t["text"]
        eng = t["engagement_rate"]
        for key, p in patterns.items():
            if re.search(p["pattern"], text, re.IGNORECASE):
                p["with"].append(eng)
            else:
                p["without"].append(eng)

    results = {}
    for key, p in patterns.items():
        with_avg = round(statistics.mean(p["with"]), 2) if p["with"] else 0
        without_avg = round(statistics.mean(p["without"]), 2) if p["without"] else 0
        lift = ((with_avg - without_avg) / without_avg * 100) if without_avg > 0 else 0
        results[key] = {
            "with_count": len(p["with"]),
            "without_count": len(p["without"]),
            "with_avg_engagement": with_avg,
            "without_avg_engagement": without_avg,
            "lift_pct": round(lift, 1),
        }

    # Tweet length analysis
    length_buckets = {"short (<100)": [], "medium (100-200)": [], "long (200-280)": []}
    for t in tweets:
        text_len = len(t["text"])
        eng = t["engagement_rate"]
        if text_len < 100:
            length_buckets["short (<100)"].append(eng)
        elif text_len < 200:
            length_buckets["medium (100-200)"].append(eng)
        else:
            length_buckets["long (200-280)"].append(eng)

    length_stats = {}
    for bucket, rates in length_buckets.items():
        length_stats[bucket] = {
            "count": len(rates),
            "avg_engagement_rate": round(statistics.mean(rates), 2) if rates else 0,
        }

    results["length_analysis"] = length_stats
    return results


def get_top_tweets(tweets: List[Dict], n: int = 10) -> List[Dict]:
    """Get top performing tweets by engagement rate."""
    return sorted(tweets, key=lambda t: t["engagement_rate"], reverse=True)[:n]


def print_human(overall: Dict, type_analysis: Dict, media_analysis: Dict,
                time_analysis: Dict, text_analysis: Dict, top_tweets: List[Dict]) -> None:
    """Print analysis in human-readable format."""
    print("=" * 70)
    print("  X/Twitter Performance Analysis")
    print("=" * 70)

    o = overall
    print(f"\n  --- Overall Metrics ---")
    print(f"  Total Tweets:        {o['total_tweets']}")
    print(f"  Total Impressions:   {o['total_impressions']:,}")
    print(f"  Avg Impressions:     {o['avg_impressions']:,.0f}")
    print(f"  Avg Engagement Rate: {o['avg_engagement_rate']:.2f}%")
    print(f"  Avg Likes:           {o['avg_likes']:.1f}")
    print(f"  Avg Retweets:        {o['avg_retweets']:.1f}")
    print(f"  Avg Replies:         {o['avg_replies']:.1f}")

    # By type
    print(f"\n  --- By Content Type ---")
    print(f"  {'Type':<20} {'Count':>6} {'Avg Eng%':>9} {'Avg Impressions':>16}")
    for t_type, stats in sorted(type_analysis.items(), key=lambda x: x[1]["avg_engagement_rate"], reverse=True):
        print(f"  {t_type:<20} {stats['count']:>6} {stats['avg_engagement_rate']:>8.2f}% {stats['avg_impressions']:>15,.0f}")

    # Media
    print(f"\n  --- Media Impact ---")
    wm = media_analysis["with_media"]
    wom = media_analysis["without_media"]
    print(f"  With media:    {wm['avg_engagement_rate']:.2f}% eng rate (n={wm['count']})")
    print(f"  Without media: {wom['avg_engagement_rate']:.2f}% eng rate (n={wom['count']})")
    if wom["avg_engagement_rate"] > 0:
        lift = (wm["avg_engagement_rate"] - wom["avg_engagement_rate"]) / wom["avg_engagement_rate"] * 100
        direction = "+" if lift > 0 else ""
        print(f"  Media lift:    {direction}{lift:.1f}%")

    # Best times
    print(f"\n  --- Best Posting Times ---")
    for h in time_analysis["best_hours"][:3]:
        print(f"  {h['hour']}: {h['avg_eng_rate']:.2f}% avg engagement (n={h['tweets']})")
    print()
    for d in time_analysis["best_days"][:3]:
        print(f"  {d['day']}: {d['avg_eng_rate']:.2f}% avg engagement (n={d['tweets']})")

    # Text patterns
    print(f"\n  --- Content Patterns ---")
    for key, stats in text_analysis.items():
        if key == "length_analysis":
            continue
        if stats["with_count"] >= 3:
            label = key.replace("_", " ").title()
            direction = "+" if stats["lift_pct"] > 0 else ""
            print(f"  {label:<25} {direction}{stats['lift_pct']:.1f}% lift "
                  f"({stats['with_avg_engagement']:.2f}% vs {stats['without_avg_engagement']:.2f}%)")

    length = text_analysis.get("length_analysis", {})
    if length:
        print(f"\n  --- Tweet Length ---")
        for bucket, stats in length.items():
            if stats["count"] > 0:
                bar = "#" * max(1, int(stats["avg_engagement_rate"] * 3))
                print(f"  {bucket:<20} {stats['avg_engagement_rate']:.2f}% (n={stats['count']})  {bar}")

    # Top tweets
    print(f"\n  --- Top Performing Tweets ---")
    for i, t in enumerate(top_tweets[:5], 1):
        text_preview = t["text"][:60] + "..." if len(t["text"]) > 60 else t["text"]
        print(f"  {i}. [{t['engagement_rate']:.2f}%] {text_preview}")
        print(f"     Impressions: {t['impressions']:,} | Likes: {t['likes']} | "
              f"RT: {t['retweets']} | Replies: {t['replies']}")

    print()


def main():
    parser = argparse.ArgumentParser(
        description="Analyze tweet performance patterns from exported X/Twitter data"
    )
    parser.add_argument("file", help="CSV file with tweet data")
    parser.add_argument("--format", choices=["human", "json"], default="human", help="Output format")
    parser.add_argument("--min-impressions", type=int, default=0,
                        help="Minimum impressions filter")
    parser.add_argument("--top", type=int, default=10, help="Number of top tweets to show")
    args = parser.parse_args()

    tweets = load_tweets(args.file)
    if not tweets:
        print("Error: No valid tweet data found", file=sys.stderr)
        sys.exit(1)

    if args.min_impressions > 0:
        tweets = [t for t in tweets if t["impressions"] >= args.min_impressions]

    overall = analyze_overall(tweets)
    type_analysis = analyze_by_type(tweets)
    media_analysis = analyze_by_media(tweets)
    time_analysis = analyze_posting_times(tweets)
    text_analysis = analyze_text_patterns(tweets)
    top_tweets = get_top_tweets(tweets, args.top)

    if args.format == "json":
        # Serialize datetimes
        for t in top_tweets:
            t["created_at"] = t["created_at"].isoformat() if t["created_at"] else None
        output = {
            "overall": overall,
            "by_type": type_analysis,
            "media_impact": media_analysis,
            "posting_times": time_analysis,
            "content_patterns": text_analysis,
            "top_tweets": top_tweets,
        }
        print(json.dumps(output, indent=2, default=str))
    else:
        print_human(overall, type_analysis, media_analysis, time_analysis, text_analysis, top_tweets)


if __name__ == "__main__":
    main()