Tiktok Research

#!/usr/bin/env python3
"""
Identify outlier TikTok videos based on engagement metrics.
Outputs JSON with outliers and metadata for report generation.
"""

import json
import argparse
import statistics
from datetime import datetime
from pathlib import Path
from collections import Counter
import re


def load_posts(input_path: str) -> list[dict]:
    """Load videos from JSON file."""
    with open(input_path, 'r') as f:
        return json.load(f)


def calculate_engagement_score(video: dict) -> float:
    """
    Calculate weighted engagement score for TikTok.
    - Comments (3x): Active engagement, hardest to get
    - Shares (2x): Strong signal of value
    - Saves/Collects (2x): Intent to revisit
    - Likes/Diggs (1x): Passive approval
    - Views/Plays (0.05x): Weighted lower due to auto-play
    """
    likes = video.get('diggCount', 0) or 0
    comments = video.get('commentCount', 0) or 0
    shares = video.get('shareCount', 0) or 0
    saves = video.get('collectCount', 0) or 0
    plays = video.get('playCount', 0) or 0
    return likes + (3 * comments) + (2 * shares) + (2 * saves) + (0.05 * plays)


def calculate_engagement_rate(video: dict) -> float:
    """Calculate engagement rate relative to follower count."""
    followers = video.get('authorFollowers', 0) or 0
    engagement = calculate_engagement_score(video)
    if followers == 0:
        return engagement
    return (engagement / followers) * 100


def identify_outliers(videos: list[dict], threshold_multiplier: float = 2.0) -> list[dict]:
    """
    Identify outlier videos with engagement rate > mean + (threshold x std_dev).
    """
    if not videos:
        return []

    for video in videos:
        video['_engagement_score'] = calculate_engagement_score(video)
        video['_engagement_rate'] = calculate_engagement_rate(video)

    rates = [v['_engagement_rate'] for v in videos]
    if len(rates) < 2:
        return videos

    mean_rate = statistics.mean(rates)
    std_dev = statistics.stdev(rates) if len(rates) > 1 else 0
    threshold = mean_rate + (threshold_multiplier * std_dev)

    outliers = [v for v in videos if v['_engagement_rate'] > threshold]
    outliers.sort(key=lambda x: x['_engagement_score'], reverse=True)
    return outliers


def extract_topics(videos: list[dict]) -> dict:
    """Extract trending hashtags, sounds, and keywords."""
    hashtags = Counter()
    sounds = Counter()
    keywords = Counter()

    stop_words = {
        'the', 'a', 'an', 'is', 'are', 'was', 'were', 'be', 'been', 'being',
        'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could',
        'should', 'may', 'might', 'can', 'this', 'that', 'these', 'those',
        'i', 'you', 'he', 'she', 'it', 'we', 'they', 'what', 'which', 'who',
        'when', 'where', 'why', 'how', 'all', 'each', 'every', 'both', 'few',
        'more', 'most', 'other', 'some', 'such', 'no', 'nor', 'not', 'only',
        'own', 'same', 'so', 'than', 'too', 'very', 'just', 'and', 'but',
        'if', 'or', 'because', 'as', 'until', 'while', 'of', 'at', 'by',
        'for', 'with', 'about', 'against', 'between', 'into', 'through',
        'during', 'before', 'after', 'above', 'below', 'to', 'from', 'up',
        'down', 'in', 'out', 'on', 'off', 'over', 'under', 'again', 'further',
        'then', 'once', 'here', 'there', 'your', 'my', 'his', 'her', 'its',
        'our', 'their', 'get', 'got', 'like', 'dont', 'im', 'ive', 'youre',
        'https', 'http', 'amp', 'link', 'bio', 'comment', 'follow', 'check',
        'fyp', 'foryou', 'foryoupage', 'viral', 'trending', 'xyzbca'
    }

    for video in videos:
        text = video.get('text', '') or ''

        # Hashtags
        video_hashtags = video.get('hashtags', []) or []
        if isinstance(video_hashtags, list):
            for h in video_hashtags:
                if isinstance(h, dict):
                    tag = h.get('name', '') or h.get('title', '')
                else:
                    tag = str(h)
                if tag and tag.lower() not in stop_words:
                    hashtags[tag.lower().lstrip('#')] += 1
        hashtags.update([h.lower() for h in re.findall(r'#(\w+)', text.lower()) if h.lower() not in stop_words])

        # Sounds/Music
        music_name = video.get('musicName')
        if music_name and music_name.strip():
            sounds[music_name.strip()] += 1

        # Keywords
        text_clean = re.sub(r'https?://\S+', '', text)
        text_clean = re.sub(r'[@#]\w+', '', text_clean)
        text_words = re.findall(r'\b[a-zA-Z]{4,}\b', text_clean.lower())
        keywords.update([w for w in text_words if w not in stop_words])

    return {
        'hashtags': hashtags.most_common(20),
        'sounds': sounds.most_common(10),
        'keywords': keywords.most_common(30)
    }


def main():
    parser = argparse.ArgumentParser(description='Identify TikTok outliers')
    parser.add_argument('--input', '-i', required=True, help='Input JSON file')
    parser.add_argument('--output', '-o', required=True, help='Output JSON file')
    parser.add_argument('--threshold', '-t', type=float, default=2.0,
                        help='Outlier threshold multiplier (default: 2.0)')

    args = parser.parse_args()

    print(f"Loading videos from: {args.input}")
    videos = load_posts(args.input)
    print(f"Loaded {len(videos)} videos")

    print(f"Identifying outliers (threshold: {args.threshold}x std dev)...")
    outliers = identify_outliers(videos, args.threshold)
    print(f"Found {len(outliers)} outlier videos")

    print("Extracting topics...")
    topics = extract_topics(videos)

    # Build output with metadata
    output = {
        'generated': datetime.now().isoformat(),
        'total_videos': len(videos),
        'outlier_count': len(outliers),
        'threshold': args.threshold,
        'topics': topics,
        'accounts': list(set(v.get('authorUsername', '') for v in videos if v.get('authorUsername'))),
        'outliers': outliers
    }

    Path(args.output).parent.mkdir(parents=True, exist_ok=True)
    with open(args.output, 'w') as f:
        json.dump(output, f, indent=2, default=str)

    print(f"Outliers saved to: {args.output}")
    print(f"- {len(outliers)} outliers identified")
    if topics['hashtags']:
        print(f"- Top hashtag: #{topics['hashtags'][0][0]}")
    if topics['sounds']:
        print(f"- Top sound: {topics['sounds'][0][0]}")
    if topics['keywords']:
        print(f"- Top keyword: {topics['keywords'][0][0]}")


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Fetch TikTok videos from specified accounts using Apify TikTok Scraper.
Requires APIFY_TOKEN environment variable (or in .env file).
"""

import os
import sys
import json
import argparse
from datetime import datetime, timedelta
from pathlib import Path

# Load .env file if present
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass  # dotenv not installed, rely on environment variables

try:
    from apify_client import ApifyClient
except ImportError:
    print("Error: apify-client not installed. Run: pip install apify-client")
    sys.exit(1)


def parse_accounts_file(accounts_path: str) -> list[str]:
    """Parse tiktok-accounts.md and extract usernames."""
    usernames = []
    with open(accounts_path, 'r') as f:
        in_table = False
        for line in f:
            line = line.strip()
            if line.startswith('| Username') or line.startswith('| Handle'):
                in_table = True
                continue
            if line.startswith('|---'):
                continue
            if in_table and line.startswith('|'):
                parts = [p.strip() for p in line.split('|')]
                if len(parts) >= 2:
                    username = parts[1]
                    if username.startswith('@') and not username.startswith('@example'):
                        usernames.append(username.lstrip('@'))
    return usernames


def fetch_tiktok(
    usernames: list[str],
    results_limit: int = 50,
    days_back: int = 30,
    sorting: str = "latest",
    output_path: str = None
) -> list[dict]:
    """
    Fetch TikTok videos from specified usernames using Apify TikTok Scraper.

    Args:
        usernames: List of TikTok usernames (without @)
        results_limit: Maximum videos per account
        days_back: Filter to only include posts newer than this many days
        sorting: Sort order - "latest", "popular", or "oldest"
        output_path: Optional path to save raw JSON output

    Returns:
        List of video objects
    """
    token = os.environ.get('APIFY_TOKEN')
    if not token:
        print("Error: APIFY_TOKEN environment variable not set")
        sys.exit(1)

    client = ApifyClient(token)

    # Calculate date filter
    oldest_date = (datetime.now() - timedelta(days=days_back)).strftime('%Y-%m-%d')

    print(f"Fetching videos from {len(usernames)} accounts...")
    print(f"Accounts: {', '.join(usernames)}")
    print(f"Results limit per account: {results_limit}")
    print(f"Videos newer than: {oldest_date}")
    print(f"Sorting: {sorting}")

    run_input = {
        "profiles": usernames,
        "resultsPerPage": results_limit,
        "profileScrapeSections": ["videos"],
        "profileSorting": sorting,
        "oldestPostDateUnified": oldest_date,
        "excludePinnedPosts": False,
        "shouldDownloadVideos": False,
        "shouldDownloadCovers": False,
    }

    # Run the Actor (clockworks/tiktok-scraper)
    run = client.actor("GdWCkxBtKWOsKjdch").call(run_input=run_input)

    # Fetch results
    items = []
    for item in client.dataset(run["defaultDatasetId"]).iterate_items():
        # Normalize field names for consistency with analysis script
        normalized = {
            'id': item.get('id'),
            'text': item.get('text', ''),
            'createTime': item.get('createTime'),
            'createTimeISO': item.get('createTimeISO'),
            'webVideoUrl': item.get('webVideoUrl'),
            # Engagement metrics
            'diggCount': item.get('diggCount', 0),  # likes/hearts
            'shareCount': item.get('shareCount', 0),
            'playCount': item.get('playCount', 0),
            'commentCount': item.get('commentCount', 0),
            'collectCount': item.get('collectCount', 0),  # saves/bookmarks
            # Author metadata
            'authorUsername': item.get('authorMeta', {}).get('name', ''),
            'authorNickname': item.get('authorMeta', {}).get('nickName', ''),
            'authorFollowers': item.get('authorMeta', {}).get('fans', 0),
            'authorFollowing': item.get('authorMeta', {}).get('following', 0),
            'authorHearts': item.get('authorMeta', {}).get('heart', 0),
            'authorVerified': item.get('authorMeta', {}).get('verified', False),
            # Video metadata
            'videoDuration': item.get('videoMeta', {}).get('duration', 0),
            'videoHeight': item.get('videoMeta', {}).get('height'),
            'videoWidth': item.get('videoMeta', {}).get('width'),
            'coverUrl': item.get('videoMeta', {}).get('coverUrl'),
            # Content metadata
            'hashtags': item.get('hashtags', []),
            'mentions': item.get('mentions', []),
            'isPinned': item.get('isPinned', False),
            'isAd': item.get('isAd', False),
            # Music metadata
            'musicName': item.get('musicMeta', {}).get('musicName'),
            'musicAuthor': item.get('musicMeta', {}).get('musicAuthor'),
            'musicOriginal': item.get('musicMeta', {}).get('musicOriginal', False),
            # Raw item for reference
            '_raw': item
        }
        items.append(normalized)

    print(f"Fetched {len(items)} videos total")

    if output_path:
        Path(output_path).parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, 'w') as f:
            json.dump(items, f, indent=2, default=str)
        print(f"Saved raw data to: {output_path}")

    return items


def main():
    parser = argparse.ArgumentParser(description='Fetch TikTok videos from accounts')
    parser.add_argument('--accounts-file', '-a',
                        default='.claude/context/tiktok-accounts.md',
                        help='Path to accounts markdown file')
    parser.add_argument('--usernames', '-u', nargs='+',
                        help='Specific usernames to fetch (overrides accounts file)')
    parser.add_argument('--limit', '-l', type=int, default=50,
                        help='Max videos per account (default: 50)')
    parser.add_argument('--days', '-d', type=int, default=30,
                        help='Days back to search (default: 30)')
    parser.add_argument('--sorting', '-s', choices=['latest', 'popular', 'oldest'],
                        default='latest',
                        help='Sort order (default: latest)')
    parser.add_argument('--output', '-o',
                        help='Output path for raw JSON')

    args = parser.parse_args()

    if args.usernames:
        usernames = [u.lstrip('@') for u in args.usernames]
    else:
        if not os.path.exists(args.accounts_file):
            print(f"Error: Accounts file not found: {args.accounts_file}")
            sys.exit(1)
        usernames = parse_accounts_file(args.accounts_file)

    if not usernames:
        print("Error: No valid usernames found")
        sys.exit(1)

    print(f"Usernames to fetch: {', '.join(usernames)}")

    items = fetch_tiktok(
        usernames=usernames,
        results_limit=args.limit,
        days_back=args.days,
        sorting=args.sorting,
        output_path=args.output
    )

    # Output summary
    if items:
        print(f"\nFetch complete. {len(items)} videos retrieved.")
        print("Use analyze_posts.py to identify outliers and generate report.")

    return items


if __name__ == '__main__':
    main()