TikTok Product Trends Meet Ecommerce Intelligence

import requests, os
from datetime import datetime

API_KEY = os.environ["SCAVIO_API_KEY"]

def tiktok_api(endpoint, payload):
    return requests.post(
        f"https://api.scavio.dev/api/v1/tiktok/{endpoint}",
        headers={"Authorization": f"Bearer {API_KEY}",
                 "Content-Type": "application/json"},
        json=payload
    ).json()["data"]

def search_api(platform, query, **kwargs):
    return requests.post(
        "https://api.scavio.dev/api/v1/search",
        headers={"x-api-key": API_KEY,
                 "Content-Type": "application/json"},
        json={"platform": platform, "query": query, **kwargs}
    ).json()["data"]

def tiktok_product_buzz(keyword, min_views=100000):
    """Search TikTok for product mentions and filter by virality."""
    data = tiktok_api("search/videos",
        {"keyword": keyword, "count": 20, "sort_type": "1"})
    trending = []
    for video in data.get("aweme_list", []):
        views = video["statistics"]["play_count"]
        if views >= min_views:
            trending.append({
                "desc": video["desc"][:120],
                "views": views,
                "likes": video["statistics"]["digg_count"],
                "comments": video["statistics"]["comment_count"],
                "shares": video["statistics"]["share_count"],
                "author": video["author"]["unique_id"],
                "created": datetime.fromtimestamp(
                    video["create_time"]).strftime("%Y-%m-%d"),
            })
    return sorted(trending, key=lambda x: x["views"], reverse=True)

def amazon_demand(product_query):
    """Check if the product is already selling on Amazon."""
    results = search_api("amazon", product_query)
    products = results.get("organic", [])[:5]
    return [{
        "title": p["title"][:80],
        "price": p.get("price", "N/A"),
        "rating": p.get("rating", "N/A"),
        "reviews": p.get("reviews", 0),
        "url": p["link"],
    } for p in products]

def shopping_prices(product_query):
    """Get current Google Shopping prices for competitive analysis."""
    results = search_api("google_shopping", product_query)
    items = results.get("shopping", results.get("organic", []))[:8]
    prices = []
    for item in items:
        price = item.get("price", "")
        if price:
            prices.append({
                "title": item.get("title", "")[:60],
                "price": price,
                "source": item.get("source", ""),
            })
    return prices

def cross_platform_trend(keyword):
    """Full cross-platform product trend analysis."""
    tiktok = tiktok_product_buzz(keyword)   # 1 credit
    amazon = amazon_demand(keyword)          # 1 credit
    shopping = shopping_prices(keyword)      # 1 credit

    early_stage = len(tiktok) >= 3 and len(amazon) == 0

    print(f"{keyword}: TikTok={len(tiktok)} viral, "
          f"Amazon={len(amazon)} listings, Shopping={len(shopping)} prices")
    print(f"  Stage: {'EARLY - source now' if early_stage else 'established'}")

    return {
        "keyword": keyword,
        "tiktok_viral_count": len(tiktok),
        "amazon_listings": len(amazon),
        "shopping_prices": len(shopping),
        "stage": "early" if early_stage else "established",
    }

# 3 credits ($0.015) per keyword
trends = [cross_platform_trend(kw) for kw in
          ["portable neck fan 2026", "mushroom coffee", "desk treadmill"]]

import json

WATCHLIST = ["ice roller skincare", "mini projector portable",
             "standing desk pad", "protein water"]

def daily_scan():
    early = [cross_platform_trend(kw) for kw in WATCHLIST]
    early = [t for t in early if t["stage"] == "early"]
    if early:
        with open("early-trends.json", "w") as f:
            json.dump(early, f, indent=2)
    print(f"{len(early)} early-stage trends detected")
    return early

# Cost: 4 keywords x 3 credits = 12/day = 360/month ($30 plan)