ScavioScavio
ProductPricingDocs
Sign InGet Started
  1. Home
  2. Tutorials
  3. How to Build a TikTok UGC Collection Pipeline
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.

Get Free API KeyAPI Docs

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

  • How to Build a TikTok Influencer Scorer
  • How to Monitor a TikTok Hashtag Campaign via API
  • How to Detect TikTok Fake Followers via API

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 50 free credits on signup.

Yes. The free tier includes 50 credits on signup, 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.

Related Resources

Best Of

Best TikTok UGC Campaign Tracking APIs (May 2026)

Read more
Best Of

Best TikTok UGC Tracking Tools in 2026

Read more
Workflow

Daily TikTok UGC Monitoring Workflow

Read more
Solution

TikTok UGC Collection via API

Read more
Use Case

TikTok UGC Collection and Tracking

Read more
Use Case

TikTok UGC Campaign Tracking

Read more

Start Building

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

Get Free API KeyRead the Docs
ScavioScavio

Real-time search API for AI agents. Search every platform, not just Google.

Product

  • Features
  • Pricing
  • Dashboard
  • Affiliates

Developers

  • Documentation
  • API Reference
  • Quickstart
  • MCP Integration
  • Python SDK

Alternatives

  • Tavily Alternative
  • SerpAPI Alternative
  • Firecrawl Alternative
  • Exa Alternative

Tools

  • JSON Formatter
  • cURL to Code
  • Token Counter
  • All Tools

© 2026 Scavio. All rights reserved.

Featured on TAAFT
Terms of ServicePrivacy Policy