analyze_tweets.py

#!/usr/bin/env python3
"""
Twitter Archive Analytics Script

Analyzes cleaned Twitter archive data to generate insights about:
- Basic statistics (tweet counts, date ranges)
- Engagement metrics (most liked/retweeted tweets)
- Temporal patterns (posting habits by time/day/month/year)
- Content analysis (hashtags, mentions, media usage)
- Conversation patterns (replies, threads)
"""

import json
import re
from pathlib import Path
from datetime import datetime
from collections import Counter, defaultdict
from typing import Dict, List, Any, Tuple
import statistics


class TwitterAnalytics:
    """Analyze Twitter archive data and generate insights."""

    def __init__(self, cleaned_archive_path: str = 'twitter_archive_clean'):
        """Initialize analytics with path to cleaned archive."""
        self.archive_path = Path(cleaned_archive_path)
        self.data_path = self.archive_path / 'data'
        self.tweets = []
        self.analytics = {
            'basic_stats': {},
            'engagement': {},
            'temporal': {},
            'content': {},
            'conversations': {}
        }

    def load_tweets(self):
        """Load tweets from cleaned JSON file."""
        tweets_file = self.data_path / 'tweets.json'

        if not tweets_file.exists():
            raise FileNotFoundError(f"Tweets file not found: {tweets_file}")

        print("📥 Loading tweets...")
        with open(tweets_file, 'r', encoding='utf-8') as f:
            data = json.load(f)

        self.tweets = [item['tweet'] for item in data]
        print(f"  ✓ Loaded {len(self.tweets)} tweets")

    def parse_tweet_date(self, date_str: str) -> datetime:
        """Parse Twitter date format to datetime object."""
        # Format: "Fri Jun 20 18:43:40 +0000 2025"
        return datetime.strptime(date_str, "%a %b %d %H:%M:%S %z %Y")

    def analyze_basic_stats(self):
        """Calculate basic statistics about tweets."""
        print("\n📊 Analyzing basic statistics...")

        total_tweets = len(self.tweets)

        # Parse dates
        dates = [self.parse_tweet_date(t['created_at']) for t in self.tweets]
        dates.sort()

        first_tweet = dates[0]
        last_tweet = dates[-1]
        date_range_days = (last_tweet - first_tweet).days

        # Categorize tweets
        replies = sum(1 for t in self.tweets if t.get('in_reply_to_status_id'))
        retweets = sum(1 for t in self.tweets if t.get('retweeted_status') or
                      t.get('full_text', '').startswith('RT @'))
        originals = total_tweets - replies - retweets

        # Calculate averages
        avg_per_day = total_tweets / max(date_range_days, 1)
        avg_per_month = avg_per_day * 30.44
        avg_per_year = avg_per_day * 365.25

        self.analytics['basic_stats'] = {
            'total_tweets': total_tweets,
            'date_range': {
                'first_tweet': first_tweet.isoformat(),
                'last_tweet': last_tweet.isoformat(),
                'days_active': date_range_days,
                'years_active': round(date_range_days / 365.25, 2)
            },
            'tweet_types': {
                'original_tweets': originals,
                'replies': replies,
                'retweets': retweets,
                'original_percentage': round(originals / total_tweets * 100, 1),
                'reply_percentage': round(replies / total_tweets * 100, 1),
                'retweet_percentage': round(retweets / total_tweets * 100, 1)
            },
            'averages': {
                'tweets_per_day': round(avg_per_day, 2),
                'tweets_per_month': round(avg_per_month, 1),
                'tweets_per_year': round(avg_per_year, 1)
            }
        }

        print(f"  ✓ {total_tweets} total tweets from {first_tweet.date()} to {last_tweet.date()}")

    def analyze_engagement(self):
        """Analyze engagement metrics (likes, retweets)."""
        print("\n💙 Analyzing engagement...")

        # Get engagement numbers
        tweets_with_engagement = []
        for tweet in self.tweets:
            likes = int(tweet.get('favorite_count', 0))
            retweets = int(tweet.get('retweet_count', 0))
            engagement = likes + retweets

            tweets_with_engagement.append({
                'id': tweet['id_str'],
                'text': tweet['full_text'][:100],  # First 100 chars
                'created_at': tweet['created_at'],
                'likes': likes,
                'retweets': retweets,
                'engagement': engagement
            })

        # Sort by different metrics
        by_likes = sorted(tweets_with_engagement, key=lambda x: x['likes'], reverse=True)
        by_retweets = sorted(tweets_with_engagement, key=lambda x: x['retweets'], reverse=True)
        by_engagement = sorted(tweets_with_engagement, key=lambda x: x['engagement'], reverse=True)

        # Calculate averages
        total_likes = sum(t['likes'] for t in tweets_with_engagement)
        total_retweets = sum(t['retweets'] for t in tweets_with_engagement)

        avg_likes = total_likes / len(tweets_with_engagement) if tweets_with_engagement else 0
        avg_retweets = total_retweets / len(tweets_with_engagement) if tweets_with_engagement else 0

        # Calculate median
        likes_list = [t['likes'] for t in tweets_with_engagement]
        retweets_list = [t['retweets'] for t in tweets_with_engagement]

        median_likes = statistics.median(likes_list) if likes_list else 0
        median_retweets = statistics.median(retweets_list) if retweets_list else 0

        self.analytics['engagement'] = {
            'totals': {
                'total_likes': total_likes,
                'total_retweets': total_retweets,
                'total_engagement': total_likes + total_retweets
            },
            'averages': {
                'avg_likes_per_tweet': round(avg_likes, 2),
                'avg_retweets_per_tweet': round(avg_retweets, 2),
                'median_likes': median_likes,
                'median_retweets': median_retweets
            },
            'top_tweets': {
                'most_liked': by_likes[:10],
                'most_retweeted': by_retweets[:10],
                'most_engagement': by_engagement[:10]
            }
        }

        print(f"  ✓ {total_likes:,} total likes, {total_retweets:,} total retweets")

    def analyze_temporal_patterns(self):
        """Analyze posting patterns over time."""
        print("\n📅 Analyzing temporal patterns...")

        by_year = Counter()
        by_month = Counter()
        by_day_of_week = Counter()
        by_hour = Counter()

        # Also track monthly timeline
        monthly_timeline = defaultdict(int)

        for tweet in self.tweets:
            dt = self.parse_tweet_date(tweet['created_at'])

            by_year[dt.year] += 1
            by_month[dt.strftime('%B')] += 1
            by_day_of_week[dt.strftime('%A')] += 1
            by_hour[dt.hour] += 1

            # Monthly timeline (YYYY-MM format for sorting)
            monthly_timeline[dt.strftime('%Y-%m')] += 1

        # Sort day of week properly
        day_order = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
        by_day_sorted = {day: by_day_of_week[day] for day in day_order}

        # Sort months properly
        month_order = ['January', 'February', 'March', 'April', 'May', 'June',
                      'July', 'August', 'September', 'October', 'November', 'December']
        by_month_sorted = {month: by_month[month] for month in month_order if month in by_month}

        # Find peak times
        most_active_year = by_year.most_common(1)[0] if by_year else ('N/A', 0)
        most_active_day = max(by_day_sorted.items(), key=lambda x: x[1]) if by_day_sorted else ('N/A', 0)
        most_active_hour = by_hour.most_common(1)[0] if by_hour else (0, 0)

        self.analytics['temporal'] = {
            'by_year': dict(sorted(by_year.items())),
            'by_month': by_month_sorted,
            'by_day_of_week': by_day_sorted,
            'by_hour': dict(sorted(by_hour.items())),
            'monthly_timeline': dict(sorted(monthly_timeline.items())),
            'peak_times': {
                'most_active_year': most_active_year[0],
                'tweets_that_year': most_active_year[1],
                'most_active_day_of_week': most_active_day[0],
                'most_active_hour': f"{most_active_hour[0]:02d}:00",
                'tweets_that_hour': most_active_hour[1]
            }
        }

        print(f"  ✓ Most active in {most_active_year[0]} with {most_active_year[1]} tweets")

    def analyze_content(self):
        """Analyze tweet content (hashtags, mentions, media, URLs)."""
        print("\n📝 Analyzing content...")

        hashtags = Counter()
        mentions = Counter()
        urls_count = 0
        media_count = 0
        tweet_lengths = []

        for tweet in self.tweets:
            text = tweet.get('full_text', '')
            tweet_lengths.append(len(text))

            # Extract hashtags
            entities = tweet.get('entities', {})
            for hashtag in entities.get('hashtags', []):
                hashtags[f"#{hashtag['text']}"] += 1

            # Extract mentions
            for mention in entities.get('user_mentions', []):
                mentions[f"@{mention['screen_name']}"] += 1

            # Count URLs
            if entities.get('urls'):
                urls_count += 1

            # Count media
            if entities.get('media') or tweet.get('extended_entities', {}).get('media'):
                media_count += 1

        avg_length = statistics.mean(tweet_lengths) if tweet_lengths else 0
        median_length = statistics.median(tweet_lengths) if tweet_lengths else 0

        self.analytics['content'] = {
            'hashtags': {
                'total_hashtags': sum(hashtags.values()),
                'unique_hashtags': len(hashtags),
                'top_hashtags': hashtags.most_common(20)
            },
            'mentions': {
                'total_mentions': sum(mentions.values()),
                'unique_users_mentioned': len(mentions),
                'top_mentions': mentions.most_common(20)
            },
            'media': {
                'tweets_with_media': media_count,
                'media_percentage': round(media_count / len(self.tweets) * 100, 1)
            },
            'urls': {
                'tweets_with_urls': urls_count,
                'url_percentage': round(urls_count / len(self.tweets) * 100, 1)
            },
            'tweet_length': {
                'average_characters': round(avg_length, 1),
                'median_characters': median_length,
                'shortest': min(tweet_lengths) if tweet_lengths else 0,
                'longest': max(tweet_lengths) if tweet_lengths else 0
            }
        }

        print(f"  ✓ {len(hashtags)} unique hashtags, {len(mentions)} unique mentions")

    def analyze_conversations(self):
        """Analyze reply patterns and conversations."""
        print("\n💬 Analyzing conversations...")

        replies_to = Counter()
        self_replies = 0

        for tweet in self.tweets:
            if tweet.get('in_reply_to_screen_name'):
                screen_name = tweet['in_reply_to_screen_name']
                replies_to[f"@{screen_name}"] += 1

                # Check if it's a self-reply (thread)
                # We'd need to know the user's own screen name for this
                # For now, we'll skip this check

        total_replies = sum(replies_to.values())
        unique_users_replied_to = len(replies_to)

        self.analytics['conversations'] = {
            'total_replies': total_replies,
            'unique_users_replied_to': unique_users_replied_to,
            'most_replied_to': replies_to.most_common(20)
        }

        print(f"  ✓ {total_replies} replies to {unique_users_replied_to} unique users")

    def generate_text_report(self) -> str:
        """Generate a human-readable text report."""
        lines = []
        lines.append("=" * 70)
        lines.append("TWITTER ARCHIVE ANALYTICS REPORT")
        lines.append("=" * 70)
        lines.append("")

        # Basic Stats
        lines.append("📊 BASIC STATISTICS")
        lines.append("-" * 70)
        bs = self.analytics['basic_stats']
        lines.append(f"Total Tweets: {bs['total_tweets']:,}")
        lines.append(f"Date Range: {bs['date_range']['first_tweet'][:10]} to {bs['date_range']['last_tweet'][:10]}")
        lines.append(f"Years Active: {bs['date_range']['years_active']}")
        lines.append(f"Days Active: {bs['date_range']['days_active']:,}")
        lines.append("")
        lines.append("Tweet Types:")
        lines.append(f"  • Original: {bs['tweet_types']['original_tweets']:,} ({bs['tweet_types']['original_percentage']}%)")
        lines.append(f"  • Replies: {bs['tweet_types']['replies']:,} ({bs['tweet_types']['reply_percentage']}%)")
        lines.append(f"  • Retweets: {bs['tweet_types']['retweets']:,} ({bs['tweet_types']['retweet_percentage']}%)")
        lines.append("")
        lines.append("Posting Frequency:")
        lines.append(f"  • Per Day: {bs['averages']['tweets_per_day']}")
        lines.append(f"  • Per Month: {bs['averages']['tweets_per_month']}")
        lines.append(f"  • Per Year: {bs['averages']['tweets_per_year']}")
        lines.append("")

        # Engagement
        lines.append("💙 ENGAGEMENT METRICS")
        lines.append("-" * 70)
        eng = self.analytics['engagement']
        lines.append(f"Total Likes: {eng['totals']['total_likes']:,}")
        lines.append(f"Total Retweets: {eng['totals']['total_retweets']:,}")
        lines.append(f"Total Engagement: {eng['totals']['total_engagement']:,}")
        lines.append("")
        lines.append(f"Average Likes per Tweet: {eng['averages']['avg_likes_per_tweet']}")
        lines.append(f"Average Retweets per Tweet: {eng['averages']['avg_retweets_per_tweet']}")
        lines.append(f"Median Likes: {eng['averages']['median_likes']}")
        lines.append(f"Median Retweets: {eng['averages']['median_retweets']}")
        lines.append("")
        lines.append("Top 5 Most Liked Tweets:")
        for i, tweet in enumerate(eng['top_tweets']['most_liked'][:5], 1):
            lines.append(f"  {i}. [{tweet['likes']} ❤️] {tweet['text']}")
        lines.append("")

        # Temporal
        lines.append("📅 TEMPORAL PATTERNS")
        lines.append("-" * 70)
        temp = self.analytics['temporal']
        lines.append(f"Most Active Year: {temp['peak_times']['most_active_year']} ({temp['peak_times']['tweets_that_year']:,} tweets)")
        lines.append(f"Most Active Day: {temp['peak_times']['most_active_day_of_week']}")
        lines.append(f"Most Active Hour: {temp['peak_times']['most_active_hour']} ({temp['peak_times']['tweets_that_hour']} tweets)")
        lines.append("")
        lines.append("Tweets by Year:")
        for year, count in sorted(temp['by_year'].items()):
            bar = "█" * int(count / 100)
            lines.append(f"  {year}: {count:4,} {bar}")
        lines.append("")
        lines.append("Tweets by Day of Week:")
        for day, count in temp['by_day_of_week'].items():
            bar = "█" * int(count / 100)
            lines.append(f"  {day:9s}: {count:4,} {bar}")
        lines.append("")

        # Content
        lines.append("📝 CONTENT ANALYSIS")
        lines.append("-" * 70)
        cont = self.analytics['content']
        lines.append(f"Average Tweet Length: {cont['tweet_length']['average_characters']} characters")
        lines.append(f"Tweets with Media: {cont['media']['tweets_with_media']:,} ({cont['media']['media_percentage']}%)")
        lines.append(f"Tweets with URLs: {cont['urls']['tweets_with_urls']:,} ({cont['urls']['url_percentage']}%)")
        lines.append("")
        lines.append(f"Top 10 Hashtags (out of {cont['hashtags']['unique_hashtags']} unique):")
        for hashtag, count in cont['hashtags']['top_hashtags'][:10]:
            lines.append(f"  {hashtag}: {count}")
        lines.append("")
        lines.append(f"Top 10 Mentions (out of {cont['mentions']['unique_users_mentioned']} unique):")
        for mention, count in cont['mentions']['top_mentions'][:10]:
            lines.append(f"  {mention}: {count}")
        lines.append("")

        # Conversations
        lines.append("💬 CONVERSATIONS")
        lines.append("-" * 70)
        conv = self.analytics['conversations']
        lines.append(f"Total Replies: {conv['total_replies']:,}")
        lines.append(f"Unique Users Replied To: {conv['unique_users_replied_to']:,}")
        lines.append("")
        lines.append("Top 10 Users You Reply To:")
        for user, count in conv['most_replied_to'][:10]:
            lines.append(f"  {user}: {count}")
        lines.append("")

        lines.append("=" * 70)

        return "\n".join(lines)

    def run(self, output_json: bool = True, output_text: bool = True):
        """Run complete analysis."""
        print("=" * 70)
        print("🔍 TWITTER ANALYTICS")
        print("=" * 70)

        # Load data
        self.load_tweets()

        # Run analyses
        self.analyze_basic_stats()
        self.analyze_engagement()
        self.analyze_temporal_patterns()
        self.analyze_content()
        self.analyze_conversations()

        # Generate outputs
        print("\n📤 Generating reports...")

        if output_json:
            json_path = self.archive_path / 'analytics_report.json'
            with open(json_path, 'w', encoding='utf-8') as f:
                json.dump(self.analytics, f, indent=2, ensure_ascii=False)
            print(f"  ✓ JSON report: {json_path}")

        if output_text:
            text_path = self.archive_path / 'analytics_report.txt'
            report = self.generate_text_report()
            with open(text_path, 'w', encoding='utf-8') as f:
                f.write(report)
            print(f"  ✓ Text report: {text_path}")

            # Also print to console
            print("\n" + report)

        print("\n" + "=" * 70)
        print("✅ ANALYSIS COMPLETE!")
        print("=" * 70)


def main():
    """Main entry point."""
    import sys

    # Get archive path from command line or use default
    if len(sys.argv) > 1:
        archive_path = sys.argv[1]
    else:
        archive_path = 'twitter_archive_clean'

    # Run analytics
    analytics = TwitterAnalytics(archive_path)
    analytics.run()


if __name__ == '__main__':
    main()