Deep Research Daily Pipeline

import requests
import json
from pathlib import Path
from datetime import datetime, timezone

API_KEY = "your_scavio_api_key"
BASE = "https://api.scavio.dev"

# One endpoint per product; the body key differs (YouTube takes "search").
ENDPOINTS = {
    "google": ("/api/v2/google", "query"),
    "reddit": ("/api/v1/reddit/search", "query"),
    "youtube": ("/api/v1/youtube/search", "search"),
}

TOPICS = ["AI agent search tools 2026", "SERP API pricing changes", "MCP server adoption"]

HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

def post(path: str, body: dict, timeout: int = 30) -> dict:
    res = requests.post(f"{BASE}{path}", headers=HEADERS, json=body, timeout=timeout)
    res.raise_for_status()
    return res.json()

def search_platform(query: str, platform: str) -> list[dict]:
    """Google v2 is a raw passthrough (organic_results at the top level); every
    other endpoint wraps its payload in a "data" object, results under "results"."""
    path, query_key = ENDPOINTS[platform]
    payload = post(path, {query_key: query}, timeout=15)
    if platform == "google":
        return payload.get("organic_results", [])
    return payload.get("data", {}).get("results", [])

# NOTE: Scavio has no extract or crawl endpoint -- it does not fetch arbitrary web
# pages. The Google leg gives you title/link/snippet and nothing more. Depth comes
# from the two places Scavio genuinely goes deeper than a result list.

def reddit_thread(post_id: str) -> list[dict]:
    """POST /api/v1/reddit/post/comments -> data.comments (text, author, score)."""
    payload = post("/api/v1/reddit/post/comments", {"post_id": post_id, "sort": "TOP"})
    return payload.get("data", {}).get("comments", [])

def youtube_transcript(video_id: str) -> str:
    """POST /api/v1/youtube/transcript -> data.content, the whole transcript as text."""
    payload = post("/api/v1/youtube/transcript", {"video_id": video_id, "format": "text"})
    return payload.get("data", {}).get("content", "")

def research_topic(topic: str) -> dict:
    google_results = search_platform(topic, "google")
    reddit_results = search_platform(topic, "reddit")
    youtube_results = search_platform(topic, "youtube")
    calls = 3

    # Depth pass 1: full comment threads for the top 2 Reddit hits.
    threads = []
    for r in reddit_results[:2]:
        post_id = r.get("post_id", "")
        if not post_id:
            continue
        try:
            comments = reddit_thread(post_id)
            calls += 1
        except requests.HTTPError:
            continue
        threads.append({
            "post_id": post_id,
            "title": r.get("title", ""),
            "score": r.get("score", 0),
            "top_comments": [c.get("text", "")[:500] for c in comments[:5]],
        })

    # Depth pass 2: transcript for the top YouTube hit. Not every video has one.
    transcripts = []
    for v in youtube_results[:1]:
        video_id = v.get("video_id", "")
        if not video_id:
            continue
        try:
            text = youtube_transcript(video_id)
            calls += 1
        except requests.HTTPError:
            continue
        if text:
            transcripts.append({"video_id": video_id, "title": v.get("title", ""), "preview": text[:500]})

    return {
        "topic": topic,
        "google_count": len(google_results),
        "reddit_count": len(reddit_results),
        "youtube_count": len(youtube_results),
        "api_calls": calls,
        # Google contributes links and snippets only -- see the note above.
        "top_google": [
            {"title": g.get("title", ""), "link": g.get("link", ""), "snippet": g.get("snippet", "")}
            for g in google_results[:3]
        ],
        "top_reddit": [{"title": r.get("title", ""), "score": r.get("score", 0)} for r in reddit_results[:5]],
        "top_youtube": [{"title": y.get("title", ""), "views": y.get("view_count", 0)} for y in youtube_results[:5]],
        "reddit_threads": threads,
        "transcripts": transcripts,
    }

def run():
    date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    briefs = [research_topic(t) for t in TOPICS]
    total_calls = sum(b["api_calls"] for b in briefs)
    report = {"date": date, "topics": len(TOPICS), "api_calls": total_calls, "briefs": briefs}
    Path(f"research_{date}.json").write_text(json.dumps(report, indent=2))
    print(f"Research complete: {len(TOPICS)} topics, {total_calls} API calls")
    for brief in briefs:
        print(
            f"  {brief['topic']}: {brief['google_count']}G {brief['reddit_count']}R "
            f"{brief['youtube_count']}Y {len(brief['reddit_threads'])} threads "
            f"{len(brief['transcripts'])} transcripts"
        )

if __name__ == "__main__":
    run()

const API_KEY = "your_scavio_api_key";
const BASE = "https://api.scavio.dev";
const TOPICS = ["AI agent search tools 2026", "SERP API pricing changes", "MCP server adoption"];

// One endpoint per product; the body key differs (YouTube takes "search").
const ENDPOINTS = {
  google: ["/api/v2/google", "query"],
  reddit: ["/api/v1/reddit/search", "query"],
  youtube: ["/api/v1/youtube/search", "search"],
};

async function post(path, body) {
  const res = await fetch(BASE + path, {
    method: "POST",
    headers: { Authorization: `Bearer ${API_KEY}`, "content-type": "application/json" },
    body: JSON.stringify(body),
  });
  if (!res.ok) throw new Error(`scavio ${path} ${res.status}`);
  return res.json();
}

async function search(query, platform) {
  const [path, queryKey] = ENDPOINTS[platform];
  const payload = await post(path, { [queryKey]: query });
  // Google v2 is a raw passthrough; the rest wrap their payload in data.results.
  return platform === "google" ? payload.organic_results ?? [] : payload.data?.results ?? [];
}

// NOTE: Scavio has no extract or crawl endpoint -- it does not fetch arbitrary web
// pages. Google gives you title/link/snippet only. Depth comes from the platforms.
async function redditThread(postId) {
  const payload = await post("/api/v1/reddit/post/comments", { post_id: postId, sort: "TOP" });
  return payload.data?.comments ?? [];
}

async function youtubeTranscript(videoId) {
  const payload = await post("/api/v1/youtube/transcript", { video_id: videoId, format: "text" });
  return payload.data?.content ?? "";
}

async function run() {
  const fs = await import("fs/promises");
  const briefs = [];
  for (const topic of TOPICS) {
    const [google, reddit, youtube] = await Promise.all([
      search(topic, "google"), search(topic, "reddit"), search(topic, "youtube"),
    ]);

    const threads = [];
    for (const r of reddit.slice(0, 2)) {
      if (!r.post_id) continue;
      try {
        const comments = await redditThread(r.post_id);
        threads.push({
          postId: r.post_id,
          title: r.title ?? "",
          topComments: comments.slice(0, 5).map(c => (c.text ?? "").slice(0, 500)),
        });
      } catch { /* thread unavailable, skip */ }
    }

    const transcripts = [];
    for (const v of youtube.slice(0, 1)) {
      if (!v.video_id) continue;
      try {
        const text = await youtubeTranscript(v.video_id);
        if (text) transcripts.push({ videoId: v.video_id, title: v.title ?? "", preview: text.slice(0, 500) });
      } catch { /* no captions on this video, skip */ }
    }

    briefs.push({
      topic,
      google: google.length,
      reddit: reddit.length,
      youtube: youtube.length,
      topGoogle: google.slice(0, 3).map(g => ({ title: g.title ?? "", link: g.link ?? "", snippet: g.snippet ?? "" })),
      threads,
      transcripts,
    });
  }
  const date = new Date().toISOString().slice(0, 10);
  await fs.writeFile(`research_${date}.json`, JSON.stringify(briefs, null, 2));
  for (const b of briefs) console.log(`  ${b.topic}: ${b.google}G ${b.reddit}R ${b.youtube}Y ${b.threads.length} threads ${b.transcripts.length} transcripts`);
}

run();