TikTok 上的用户生成内容是品牌最真实的社交证明形式之一。大规模收集 UGC（品牌提及、产品评论、拆箱视频和主题标签挑战条目）需要同时监控多个搜索向量。本教程使用 Scavio TikTok API 构建自动化 UGC 收集管道，用于搜索品牌视频、监控主题标签和目录创建者内容。每个 API 调用需要 1 个积分（0.005 美元），每日收集运行需要 5-10 个积分。
前置条件
- 已安装 Python 3.9+
- 请求已安装库
- 来自 scavio.dev 的 Scavio API 密钥
- 要监控的品牌名称和相关主题标签
操作指南
步骤 1: 搜索品牌相关视频
在 TikTok 中搜索提及您的品牌或产品的视频。这会捕获不使用您的品牌主题标签的有机提及。
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')步骤 2: 收集主题标签活动条目
监控您的品牌主题标签以收集特定于活动的 UGC。可以在一次运行中跟踪多个主题标签。
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')步骤 3: 对 UGC 进行重复数据删除和分类
合并品牌搜索和主题标签搜索的结果，删除重复项，并按内容类型对每个视频进行分类：评论、拆箱、教程或一般提及。
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}')步骤 4: 按转发值对 UGC 进行评分和排名
根据参与度、创作者关注者数量和内容类型对每个 UGC 进行评分。分数越高意味着转发或推荐的候选人就越好。
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')步骤 5: 保存带有每日历史记录的集合
使用时间戳存储每个集合运行，以便您可以跟踪随着时间的推移出现的新 UGC，并且永远不会错过新内容。
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 示例
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 示例
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']);预期输出
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)