TikTok Hashtag Campaign Tracker Workflow

import requests, os, json
from pathlib import Path
from datetime import date

API_KEY = os.environ["SCAVIO_API_KEY"]
TH = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

HASHTAGS = ["#YourBrandChallenge", "#YourBrandLaunch"]
HISTORY_FILE = Path("hashtag_campaign.json")

def search_hashtag(hashtag: str) -> list:
    resp = requests.post(
        "https://api.scavio.dev/api/v1/tiktok/search",
        headers=TH,
        json={"query": hashtag},
        timeout=15,
    )
    return resp.json().get("results", [])

def track_campaign():
    history = json.loads(HISTORY_FILE.read_text()) if HISTORY_FILE.exists() else {"days": []}
    today_data = {"date": str(date.today()), "hashtags": {}}

    for tag in HASHTAGS:
        posts = search_hashtag(tag)
        total_views = sum(p.get("views", 0) for p in posts)
        total_likes = sum(p.get("likes", 0) for p in posts)
        total_comments = sum(p.get("comments", 0) for p in posts)
        top_posts = sorted(posts, key=lambda p: p.get("likes", 0), reverse=True)[:5]

        today_data["hashtags"][tag] = {
            "post_count": len(posts),
            "total_views": total_views,
            "total_likes": total_likes,
            "total_comments": total_comments,
            "top_posts": [{"url": p.get("url", ""), "likes": p.get("likes", 0), "views": p.get("views", 0)} for p in top_posts],
        }

    # Detect trend
    recent_days = history["days"][-7:]
    for tag in HASHTAGS:
        if recent_days:
            avg_likes = sum(d["hashtags"].get(tag, {}).get("total_likes", 0) for d in recent_days) / len(recent_days)
            current = today_data["hashtags"][tag]["total_likes"]
            trend = "UP" if current > avg_likes * 1.1 else "DOWN" if current < avg_likes * 0.9 else "STABLE"
            today_data["hashtags"][tag]["trend"] = trend

    history["days"].append(today_data)
    HISTORY_FILE.write_text(json.dumps(history, indent=2))
    return today_data

report = track_campaign()
for tag, data in report["hashtags"].items():
    print(f"{tag}: {data['post_count']} posts, {data['total_views']} views, {data['total_likes']} likes, trend={data.get('trend', 'N/A')}")

const TH = {'Authorization': 'Bearer '+process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
const fs = await import('fs');

const HASHTAGS = ['#YourBrandChallenge', '#YourBrandLaunch'];
const HISTORY_FILE = 'hashtag_campaign.json';

async function searchHashtag(hashtag) {
  const r = await fetch('https://api.scavio.dev/api/v1/tiktok/search', {method:'POST', headers:TH, body:JSON.stringify({query:hashtag})});
  return (await r.json()).results || [];
}

async function trackCampaign() {
  let history = {days:[]};
  try { history = JSON.parse(fs.readFileSync(HISTORY_FILE, 'utf8')); } catch {}
  const todayData = {date:new Date().toISOString().split('T')[0], hashtags:{}};
  for (const tag of HASHTAGS) {
    const posts = await searchHashtag(tag);
    const totalViews = posts.reduce((s,p)=>s+(p.views||0),0);
    const totalLikes = posts.reduce((s,p)=>s+(p.likes||0),0);
    const totalComments = posts.reduce((s,p)=>s+(p.comments||0),0);
    const topPosts = [...posts].sort((a,b)=>(b.likes||0)-(a.likes||0)).slice(0,5).map(p=>({url:p.url||'', likes:p.likes||0, views:p.views||0}));
    todayData.hashtags[tag] = {postCount:posts.length, totalViews, totalLikes, totalComments, topPosts};
    const recentDays = history.days.slice(-7);
    if (recentDays.length) {
      const avgLikes = recentDays.reduce((s,d)=>s+((d.hashtags[tag]||{}).totalLikes||0),0)/recentDays.length;
      todayData.hashtags[tag].trend = totalLikes>avgLikes*1.1?'UP':totalLikes<avgLikes*0.9?'DOWN':'STABLE';
    }
  }
  history.days.push(todayData);
  fs.writeFileSync(HISTORY_FILE, JSON.stringify(history, null, 2));
  return todayData;
}

const report = await trackCampaign();
for (const [tag, data] of Object.entries(report.hashtags)) {
  console.log(tag+': '+data.postCount+' posts, '+data.totalViews+' views, '+data.totalLikes+' likes, trend='+(data.trend||'N/A'));
}