How to Detect TikTok Product Trends for E-Commerce

import os, requests, time
from datetime import datetime

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
TT_URL = 'https://api.scavio.dev/api/v1/tiktok'
TT_H = {'Authorization': f'Bearer {SCAVIO_KEY}', 'Content-Type': 'application/json'}
SEARCH_URL = 'https://api.scavio.dev/api/v2/google'
SEARCH_H = {'Authorization': f'Bearer {SCAVIO_KEY}', 'Content-Type': 'application/json'}

def search_tiktok_products(query: str, count: int = 20) -> list:
    """Search TikTok for product-related videos."""
    resp = requests.post(f'{TT_URL}/search/videos', headers=TT_H,
        json={'query': query})
    videos = resp.json().get('data', {}).get('videos', [])
    results = []
    for v in videos:
        stats = v.get('stats', {})
        author = v.get('author', {})
        results.append({
            'description': v.get('desc', '')[:100],
            'author': author.get('uniqueId', ''),
            'plays': stats.get('playCount', 0),
            'likes': stats.get('diggCount', 0),
            'comments': stats.get('commentCount', 0),
            'shares': stats.get('shareCount', 0),
            'created': v.get('createTime', 0),
            'video_id': v.get('id', ''),
        })
    # Sort by plays descending
    results.sort(key=lambda x: x['plays'], reverse=True)
    return results

videos = search_tiktok_products('TikTok made me buy it')
print(f'Found {len(videos)} product videos')
for v in videos[:5]:
    print(f'  {v["plays"]:>12,} plays | @{v["author"]:15s} | {v["description"][:40]}')

def detect_trending_categories(categories: list) -> list:
    """Search TikTok for each product category and rank by engagement."""
    results = []
    for category in categories:
        videos = search_tiktok_products(category, count=10)
        if not videos:
            continue
        total_plays = sum(v['plays'] for v in videos)
        total_likes = sum(v['likes'] for v in videos)
        avg_engagement = total_likes / total_plays if total_plays else 0
        results.append({
            'category': category,
            'videos': len(videos),
            'total_plays': total_plays,
            'total_likes': total_likes,
            'avg_engagement': avg_engagement,
            'top_video': videos[0] if videos else None,
        })
        time.sleep(0.3)
    results.sort(key=lambda x: x['total_plays'], reverse=True)
    return results

categories = [
    'cleaning gadget review',
    'kitchen organization hack',
    'skincare routine product',
    'desk setup accessory',
    'pet gadget review',
]

trending = detect_trending_categories(categories)
print('TikTok Product Trend Detection')
print('=' * 60)
for t in trending:
    print(f"  {t['category']:30s} | {t['total_plays']:>12,} plays | {t['avg_engagement']:.1%} eng")
    if t['top_video']:
        print(f"    Top: {t['top_video']['description'][:50]}")

def check_hashtag_velocity(hashtag: str) -> dict:
    """Check a hashtag's total views and video volume."""
    # Get hashtag info
    resp = requests.post(f'{TT_URL}/hashtag', headers=TT_H,
        json={'hashtag': hashtag})
    hashtag_data = resp.json().get('data', {})
    time.sleep(0.3)
    # Get recent videos under this hashtag
    resp = requests.post(f'{TT_URL}/hashtag/videos', headers=TT_H,
        json={'hashtag': hashtag, 'count': 20, 'cursor': 0})
    videos = resp.json().get('data', {}).get('videos', [])
    recent_plays = sum(v.get('stats', {}).get('playCount', 0) for v in videos)
    return {
        'hashtag': hashtag,
        'total_views': hashtag_data.get('stats', {}).get('videoCount', 0),
        'recent_videos': len(videos),
        'recent_plays': recent_plays,
        'avg_plays_per_video': recent_plays // len(videos) if videos else 0,
    }

hashtags = ['cleaninghack', 'kitchengadget', 'desksetup', 'skincareproduct']
print('Hashtag Velocity Check')
print('-' * 60)
for tag in hashtags:
    data = check_hashtag_velocity(tag)
    print(f'  #{data["hashtag"]:20s} | {data["recent_plays"]:>10,} recent plays | '
          f'{data["avg_plays_per_video"]:>8,} avg/video')
    time.sleep(0.3)

def validate_trend_commercially(product: str) -> dict:
    """Cross-validate a TikTok trend with Amazon data."""
    # TikTok data
    tt_videos = search_tiktok_products(product, count=10)
    tt_plays = sum(v['plays'] for v in tt_videos)
    time.sleep(0.3)
    # Amazon data
    resp = requests.post(SEARCH_URL, headers=SEARCH_H,
        json={'query': f'site:amazon.com {product}',
 'gl': 'us'})
    amazon_results = resp.json().get('organic_results', [])
    # Scoring
    tiktok_score = min(tt_plays / 100000, 10)  # Normalize to 0-10
    amazon_score = len(amazon_results) * 2  # 0-10
    combined = (tiktok_score + amazon_score) / 2
    return {
        'product': product,
        'tiktok_videos': len(tt_videos),
        'tiktok_plays': tt_plays,
        'amazon_listings': len(amazon_results),
        'tiktok_score': round(tiktok_score, 1),
        'amazon_score': round(amazon_score, 1),
        'combined_score': round(combined, 1),
        'verdict': 'HOT' if combined > 6 else 'WARM' if combined > 3 else 'COLD',
        'cost': 0.010,  # 1 TikTok + 1 Amazon search
    }

products = ['LED sunset lamp', 'portable blender', 'cloud slides shoes']
print('\nTrend Validation Report')
print('=' * 65)
for p in products:
    result = validate_trend_commercially(p)
    print(f"[{result['verdict']:4s}] {result['product']:25s} | "
          f"TT: {result['tiktok_score']}/10 AMZ: {result['amazon_score']}/10 | "
          f"Combined: {result['combined_score']}/10")
    time.sleep(0.5)
total_cost = len(products) * 0.010
print(f'\nTotal cost: ${total_cost:.3f}')