TikTok UGC Campaign Tracking 2026

import requests, os
from datetime import date

KEY = os.environ["SCAVIO_API_KEY"]
H = {"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"}
BASE = "https://api.scavio.dev/api/v1/tiktok"

def get_campaign_videos(hashtag_name: str):
    """Get all videos under a campaign hashtag."""
    # Step 1: get hashtag ID
    hashtag = requests.post(f"{BASE}/hashtag",
        headers=H, json={"name": hashtag_name}).json()
    hashtag_id = hashtag.get("id")

    # Step 2: get videos under this hashtag
    videos = requests.post(f"{BASE}/hashtag/videos",
        headers=H, json={"hashtag_id": hashtag_id}).json()
    return [{
        "video_id": v.get("video_id"),
        "author": v.get("author", {}).get("unique_id", ""),
        "desc": v.get("desc", "")[:100],
        "views": v.get("play_count", 0),
        "likes": v.get("digg_count", 0),
        "comments": v.get("comment_count", 0),
        "shares": v.get("share_count", 0),
        "created": v.get("create_time"),
    } for v in videos.get("videos", [])]

# 2 API calls = $0.01 per check
videos = get_campaign_videos("YourBrandChallenge")
print(f"Found {len(videos)} campaign videos")

def campaign_report(videos: list):
    """Generate campaign performance report."""
    if not videos:
        return {"error": "no videos found"}
    total_views = sum(v["views"] for v in videos)
    total_engagement = sum(v["likes"] + v["comments"] + v["shares"] for v in videos)
    avg_views = total_views / len(videos)
    avg_engagement = total_engagement / len(videos)

    # Flag viral videos (10x above average)
    viral = [v for v in videos if v["views"] > avg_views * 10]

    # Top performers by engagement
    top_5 = sorted(videos, key=lambda v: v["likes"] + v["comments"], reverse=True)[:5]

    return {
        "date": date.today().isoformat(),
        "total_videos": len(videos),
        "total_views": total_views,
        "total_engagement": total_engagement,
        "avg_views": int(avg_views),
        "avg_engagement": int(avg_engagement),
        "viral_videos": len(viral),
        "top_5_creators": [v["author"] for v in top_5],
    }

report = campaign_report(videos)
print(f"Total views: {report['total_views']:,}")
print(f"Viral videos: {report['viral_videos']}")

import json

def daily_tracking(hashtag: str, history_file: str = "campaign_history.json"):
    """Run daily, compare against yesterday, alert on spikes."""
    videos = get_campaign_videos(hashtag)
    today_report = campaign_report(videos)

    # Load yesterday's data
    try:
        with open(history_file) as f:
            history = json.load(f)
        yesterday = history[-1] if history else None
    except FileNotFoundError:
        history = []
        yesterday = None

    # Detect changes
    if yesterday:
        new_videos = today_report["total_videos"] - yesterday["total_videos"]
        view_growth = today_report["total_views"] - yesterday["total_views"]
        today_report["new_videos_24h"] = new_videos
        today_report["view_growth_24h"] = view_growth

    history.append(today_report)
    with open(history_file, "w") as f:
        json.dump(history, f, indent=2)
    return today_report

# Daily cost: 2 API calls = $0.01/day = $0.30/mo
report = daily_tracking("YourBrandChallenge")