YouTube Growth Agent Pipeline

import requests
import json
from pathlib import Path
from datetime import datetime

API_KEY = "your_scavio_api_key"

NICHE_KEYWORDS = ["ai coding tools", "developer productivity", "vscode extensions 2026"]
COMPETITOR_CHANNELS = ["Fireship", "Theo", "Web Dev Simplified"]

def search_youtube(query: str) -> list[dict]:
    res = requests.post(
        "https://api.scavio.dev/api/v1/search",
        headers={"x-api-key": API_KEY},
        json={"platform": "youtube", "query": query},
        timeout=15,
    )
    res.raise_for_status()
    return res.json().get("organic", [])

def analyze_topic(keyword: str) -> dict:
    videos = search_youtube(keyword)
    if not videos:
        return {"keyword": keyword, "opportunity": "low", "videos": []}

    total_views = sum(v.get("views", 0) for v in videos[:10])
    avg_views = total_views // max(len(videos[:10]), 1)

    top_videos = [{
        "title": v.get("title", ""),
        "views": v.get("views", 0),
        "channel": v.get("channel", ""),
        "link": v.get("link", ""),
    } for v in sorted(videos, key=lambda x: x.get("views", 0), reverse=True)[:5]]

    return {
        "keyword": keyword,
        "avg_views_top_10": avg_views,
        "total_views_top_10": total_views,
        "opportunity": "high" if avg_views > 100000 else "medium" if avg_views > 20000 else "low",
        "top_videos": top_videos,
    }

def check_competitors() -> list[dict]:
    competitor_data = []
    for channel in COMPETITOR_CHANNELS:
        videos = search_youtube(channel)
        recent = [{
            "title": v.get("title", ""),
            "views": v.get("views", 0),
            "channel": v.get("channel", ""),
        } for v in videos[:5]]
        competitor_data.append({"channel": channel, "recent_videos": recent})
    return competitor_data

def generate_briefs(topics: list[dict], competitors: list[dict]) -> list[dict]:
    briefs = []
    for topic in sorted(topics, key=lambda t: t["avg_views_top_10"], reverse=True):
        if topic["opportunity"] in ("high", "medium"):
            brief = {
                "topic": topic["keyword"],
                "opportunity_level": topic["opportunity"],
                "avg_views": topic["avg_views_top_10"],
                "title_inspiration": [v["title"] for v in topic["top_videos"][:3]],
                "suggested_angle": f"Create a comprehensive guide on {topic['keyword']} targeting the same audience as top performers",
            }
            briefs.append(brief)
    return briefs[:10]

def run():
    topics = [analyze_topic(kw) for kw in NICHE_KEYWORDS]
    competitors = check_competitors()
    briefs = generate_briefs(topics, competitors)

    date = datetime.utcnow().strftime("%Y-%m-%d")
    report = {
        "date": date,
        "topics_analyzed": len(topics),
        "content_briefs": briefs,
        "competitor_activity": competitors,
    }

    Path(f"yt_growth_{date}.json").write_text(json.dumps(report, indent=2))
    print(f"Generated {len(briefs)} content briefs")
    for brief in briefs:
        print(f"  [{brief['opportunity_level']}] {brief['topic']} ({brief['avg_views']:,} avg views)")

if __name__ == "__main__":
    run()

const API_KEY = "your_scavio_api_key";

const NICHE_KEYWORDS = ["ai coding tools", "developer productivity", "vscode extensions 2026"];
const COMPETITOR_CHANNELS = ["Fireship", "Theo", "Web Dev Simplified"];

async function searchYouTube(query) {
  const res = await fetch("https://api.scavio.dev/api/v1/search", {
    method: "POST",
    headers: { "x-api-key": API_KEY, "content-type": "application/json" },
    body: JSON.stringify({ platform: "youtube", query }),
  });
  if (!res.ok) throw new Error(`scavio ${res.status}`);
  return (await res.json()).organic ?? [];
}

async function analyzeTopic(keyword) {
  const videos = await searchYouTube(keyword);
  if (!videos.length) return { keyword, opportunity: "low", videos: [] };
  const top10 = videos.slice(0, 10);
  const totalViews = top10.reduce((s, v) => s + (v.views ?? 0), 0);
  const avgViews = Math.floor(totalViews / top10.length);
  const topVideos = [...videos].sort((a, b) => (b.views ?? 0) - (a.views ?? 0)).slice(0, 5)
    .map((v) => ({ title: v.title ?? "", views: v.views ?? 0, channel: v.channel ?? "", link: v.link ?? "" }));
  return { keyword, avgViewsTop10: avgViews, totalViewsTop10: totalViews, opportunity: avgViews > 100000 ? "high" : avgViews > 20000 ? "medium" : "low", topVideos };
}

async function run() {
  const fs = await import("fs/promises");
  const topics = [];
  for (const kw of NICHE_KEYWORDS) topics.push(await analyzeTopic(kw));

  const competitors = [];
  for (const channel of COMPETITOR_CHANNELS) {
    const videos = await searchYouTube(channel);
    competitors.push({ channel, recentVideos: videos.slice(0, 5).map((v) => ({ title: v.title ?? "", views: v.views ?? 0, channel: v.channel ?? "" })) });
  }

  const briefs = topics
    .filter((t) => t.opportunity !== "low")
    .sort((a, b) => (b.avgViewsTop10 ?? 0) - (a.avgViewsTop10 ?? 0))
    .slice(0, 10)
    .map((t) => ({
      topic: t.keyword,
      opportunityLevel: t.opportunity,
      avgViews: t.avgViewsTop10,
      titleInspiration: (t.topVideos ?? []).slice(0, 3).map((v) => v.title),
      suggestedAngle: `Create a comprehensive guide on ${t.keyword} targeting the same audience as top performers`,
    }));

  const date = new Date().toISOString().slice(0, 10);
  const report = { date, topicsAnalyzed: topics.length, contentBriefs: briefs, competitorActivity: competitors };
  await fs.writeFile(`yt_growth_${date}.json`, JSON.stringify(report, null, 2));
  console.log(`Generated ${briefs.length} content briefs`);
  for (const b of briefs) console.log(`  [${b.opportunityLevel}] ${b.topic} (${b.avgViews.toLocaleString()} avg views)`);
}

run();