Tutorial

How to Build a TikTok UGC Collection Pipeline

Build an automated pipeline to discover and collect user-generated content about your brand on TikTok. Track mentions, hashtags, and relevant videos.

User-generated content on TikTok is one of the most authentic forms of social proof for brands. Collecting UGC at scale -- brand mentions, product reviews, unboxing videos, and hashtag challenge entries -- requires monitoring multiple search vectors simultaneously. This tutorial builds an automated UGC collection pipeline using the Scavio TikTok API that searches for brand videos, monitors hashtags, and catalogs creator content. Each API call costs 1 credit ($0.005), and a daily collection run uses 5-10 credits.

Prerequisites

  • Python 3.9+ installed
  • requests library installed
  • A Scavio API key from scavio.dev
  • Brand name and relevant hashtags to monitor

Walkthrough

Step 1: Search for brand-related videos

Search TikTok for videos that mention your brand or product. This catches organic mentions that do not use your branded hashtag.

Python
import requests, os

API_KEY = os.environ['SCAVIO_API_KEY']
TIKTOK_URL = 'https://api.scavio.dev/api/v1/tiktok'

def search_brand_videos(brand: str, count: int = 30) -> list:
    resp = requests.post(f'{TIKTOK_URL}/search/videos',
        headers={'Authorization': f'Bearer {API_KEY}',
                 'Content-Type': 'application/json'},
        json={'keyword': brand, 'count': count, 'cursor': 0})
    resp.raise_for_status()
    videos = resp.json().get('data', {}).get('videos', [])
    return [{
        'id': v.get('id', ''),
        'author': v.get('author', {}).get('uniqueId', ''),
        'author_followers': v.get('author', {}).get('stats', {}).get('followerCount', 0),
        'desc': v.get('desc', ''),
        'plays': v.get('stats', {}).get('playCount', 0),
        'likes': v.get('stats', {}).get('diggCount', 0),
        'comments': v.get('stats', {}).get('commentCount', 0),
        'shares': v.get('stats', {}).get('shareCount', 0),
        'create_time': v.get('createTime', 0),
        'source': 'brand_search'
    } for v in videos]

videos = search_brand_videos('YourBrand')
print(f'Found {len(videos)} brand mention videos')

Step 2: Collect hashtag campaign entries

Monitor your branded hashtags to collect campaign-specific UGC. Multiple hashtags can be tracked in a single run.

Python
def collect_hashtag_ugc(hashtags: list, count_per_tag: int = 20) -> list:
    all_videos = []
    for hashtag in hashtags:
        resp = requests.post(f'{TIKTOK_URL}/hashtag/posts',
            headers={'Authorization': f'Bearer {API_KEY}',
                     'Content-Type': 'application/json'},
            json={'hashtag': hashtag, 'count': count_per_tag, 'cursor': 0})
        videos = resp.json().get('data', {}).get('videos', [])
        for v in videos:
            all_videos.append({
                'id': v.get('id', ''),
                'author': v.get('author', {}).get('uniqueId', ''),
                'author_followers': v.get('author', {}).get('stats', {}).get('followerCount', 0),
                'desc': v.get('desc', ''),
                'plays': v.get('stats', {}).get('playCount', 0),
                'likes': v.get('stats', {}).get('diggCount', 0),
                'hashtag': hashtag,
                'source': 'hashtag_search'
            })
    return all_videos

hashtag_videos = collect_hashtag_ugc(['yourbrand', 'yourbrandchallenge'])
print(f'Found {len(hashtag_videos)} hashtag videos')

Step 3: Deduplicate and classify UGC

Merge results from brand search and hashtag search, remove duplicates, and classify each video by content type: review, unboxing, tutorial, or general mention.

Python
def classify_ugc(video: dict) -> str:
    desc = video.get('desc', '').lower()
    if any(w in desc for w in ['review', 'honest', 'rating', 'worth it']):
        return 'review'
    if any(w in desc for w in ['unbox', 'unboxing', 'first look', 'opening']):
        return 'unboxing'
    if any(w in desc for w in ['tutorial', 'how to', 'tip', 'hack']):
        return 'tutorial'
    if any(w in desc for w in ['haul', 'shopping', 'bought']):
        return 'haul'
    return 'mention'

def dedupe_and_classify(videos: list) -> list:
    seen_ids = set()
    unique = []
    for v in videos:
        if v['id'] not in seen_ids:
            seen_ids.add(v['id'])
            v['content_type'] = classify_ugc(v)
            unique.append(v)
    return unique

all_videos = search_brand_videos('YourBrand') + hashtag_videos
ugc = dedupe_and_classify(all_videos)
print(f'{len(ugc)} unique UGC videos')
from collections import Counter
types = Counter(v['content_type'] for v in ugc)
for t, count in types.most_common():
    print(f'  {t}: {count}')

Step 4: Score and rank UGC by repost value

Score each piece of UGC based on engagement, creator follower count, and content type. Higher scores mean better candidates for reposting or featuring.

Python
def score_ugc(video: dict) -> float:
    score = 0
    # Engagement score (0-40)
    engagement = video['likes'] + video.get('comments', 0) + video.get('shares', 0)
    if engagement > 10000: score += 40
    elif engagement > 1000: score += 30
    elif engagement > 100: score += 20
    else: score += 10
    # Creator reach (0-30)
    followers = video.get('author_followers', 0)
    if followers > 100000: score += 30
    elif followers > 10000: score += 20
    elif followers > 1000: score += 10
    # Content type bonus (0-30)
    type_bonus = {'review': 30, 'unboxing': 25, 'tutorial': 20, 'haul': 15, 'mention': 10}
    score += type_bonus.get(video.get('content_type', 'mention'), 10)
    return score

def rank_ugc(videos: list) -> list:
    for v in videos:
        v['ugc_score'] = score_ugc(v)
    ranked = sorted(videos, key=lambda v: v['ugc_score'], reverse=True)
    return ranked

ranked = rank_ugc(ugc)
print('Top UGC candidates:')
for v in ranked[:5]:
    print(f'  [{v["ugc_score"]}] @{v["author"]} ({v["content_type"]}): '
          f'{v["plays"]:,} plays, {v["likes"]:,} likes')

Step 5: Save collection with daily history

Store each collection run with timestamps so you can track new UGC appearing over time and never miss fresh content.

Python
import json
from datetime import date, datetime

def save_collection(videos: list, brand: str) -> str:
    filename = f'ugc_{brand}_{date.today()}.json'
    collection = {
        'brand': brand,
        'collected_at': datetime.now().isoformat(),
        'total_videos': len(videos),
        'by_type': dict(Counter(v.get('content_type', 'unknown') for v in videos)),
        'top_10': [{
            'id': v['id'],
            'author': v['author'],
            'content_type': v.get('content_type'),
            'ugc_score': v.get('ugc_score', 0),
            'plays': v['plays'],
            'likes': v['likes'],
            'desc': v['desc'][:100]
        } for v in videos[:10]],
        'credits_used': 3,  # 1 brand search + 2 hashtag searches
        'cost': '$0.015'
    }
    with open(filename, 'w') as f:
        json.dump(collection, f, indent=2)
    print(f'Saved {len(videos)} UGC videos to {filename}')
    print(f'Credits: {collection["credits_used"]} (${collection["credits_used"] * 0.005:.3f})')
    return filename

save_collection(ranked, 'YourBrand')

Python Example

Python
import os, requests, json
from collections import Counter
from datetime import date

API_KEY = os.environ['SCAVIO_API_KEY']
TT = 'https://api.scavio.dev/api/v1/tiktok'

def tt(endpoint, body):
    return requests.post(f'{TT}/{endpoint}',
        headers={'Authorization': f'Bearer {API_KEY}', 'Content-Type': 'application/json'},
        json=body).json()

def collect_ugc(brand, hashtags):
    # Brand search
    brand_vids = tt('search/videos', {'keyword': brand, 'count': 30, 'cursor': 0})
    videos = brand_vids.get('data', {}).get('videos', [])
    # Hashtag search
    for tag in hashtags:
        tag_vids = tt('hashtag/posts', {'hashtag': tag, 'count': 20, 'cursor': 0})
        videos.extend(tag_vids.get('data', {}).get('videos', []))
    # Dedupe
    seen = set()
    unique = [v for v in videos if v.get('id') not in seen and not seen.add(v['id'])]
    print(f'Collected {len(unique)} unique UGC videos')
    for v in sorted(unique, key=lambda x: x.get('stats', {}).get('diggCount', 0), reverse=True)[:5]:
        print(f'  @{v.get("author", {}).get("uniqueId", "")}: {v.get("stats", {}).get("playCount", 0):,} plays')

collect_ugc('YourBrand', ['yourbrand', 'yourbrandchallenge'])

JavaScript Example

JavaScript
const API_KEY = process.env.SCAVIO_API_KEY;
const TT = 'https://api.scavio.dev/api/v1/tiktok';

async function tt(endpoint, body) {
  const r = await fetch(`${TT}/${endpoint}`, {
    method: 'POST',
    headers: { 'Authorization': `Bearer ${API_KEY}`, 'Content-Type': 'application/json' },
    body: JSON.stringify(body)
  });
  return r.json();
}

async function collectUgc(brand, hashtags) {
  const brandVids = await tt('search/videos', { keyword: brand, count: 30, cursor: 0 });
  let videos = brandVids.data?.videos || [];
  for (const tag of hashtags) {
    const tagVids = await tt('hashtag/posts', { hashtag: tag, count: 20, cursor: 0 });
    videos.push(...(tagVids.data?.videos || []));
  }
  const seen = new Set();
  const unique = videos.filter(v => v.id && !seen.has(v.id) && seen.add(v.id));
  console.log(`Collected ${unique.length} unique UGC videos`);
  unique.sort((a, b) => (b.stats?.diggCount || 0) - (a.stats?.diggCount || 0))
    .slice(0, 5).forEach(v => {
      console.log(`  @${v.author?.uniqueId}: ${(v.stats?.playCount || 0).toLocaleString()} plays`);
    });
}

collectUgc('YourBrand', ['yourbrand', 'yourbrandchallenge']);

Expected Output

JSON
Found 28 brand mention videos
Found 35 hashtag videos
63 unique UGC videos
  review: 8
  unboxing: 5
  tutorial: 3
  haul: 4
  mention: 43

Top UGC candidates:
  [85] @creator1 (review): 234,000 plays, 18,500 likes
  [75] @creator2 (unboxing): 156,000 plays, 12,300 likes
  [70] @creator3 (tutorial): 89,000 plays, 7,200 likes

Saved 63 UGC videos to ugc_YourBrand_2026-05-13.json
Credits: 3 ($0.015)

Related Tutorials

Frequently Asked Questions

Most developers complete this tutorial in 15 to 30 minutes. You will need a Scavio API key (free tier works) and a working Python or JavaScript environment.

Python 3.9+ installed. requests library installed. A Scavio API key from scavio.dev. Brand name and relevant hashtags to monitor. A Scavio API key gives you 250 free credits per month.

Yes. The free tier includes 250 credits per month, which is more than enough to complete this tutorial and prototype a working solution.

Scavio has a native LangChain package (langchain-scavio), an MCP server, and a plain REST API that works with any HTTP client. This tutorial uses the raw REST API, but you can adapt to your framework of choice.

Start Building

Build an automated pipeline to discover and collect user-generated content about your brand on TikTok. Track mentions, hashtags, and relevant videos.