YouTube Channel Monitor to Email

import requests
import json
import smtplib
from email.mime.text import MIMEText
from pathlib import Path

API_KEY = "your_scavio_api_key"
EMAIL_TO = "team@yourdomain.com"
SMTP_HOST = "smtp.yourdomain.com"

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

def send_digest(videos: list[dict]):
    lines = ["<h2>New YouTube Uploads</h2>"]
    for v in videos:
        lines.append(
            f'<p><a href="{v["link"]}">{v["title"]}</a>'
            f' - {v.get("views", "N/A")} views</p>'
        )
    html = "\n".join(lines)
    msg = MIMEText(html, "html")
    msg["Subject"] = f"YouTube Digest: {len(videos)} new videos"
    msg["To"] = EMAIL_TO
    msg["From"] = "alerts@yourdomain.com"
    with smtplib.SMTP(SMTP_HOST) as smtp:
        smtp.send_message(msg)

def run():
    channels = ["@competitor-channel", "@industry-leader"]
    seen_path = Path("seen_youtube.json")
    seen = set(json.loads(seen_path.read_text())) if seen_path.exists() else set()
    new_videos = []

    for ch in channels:
        results = search_channel(ch)
        for v in results:
            vid_id = v.get("link", "")
            if vid_id and vid_id not in seen:
                v["channel"] = ch
                new_videos.append(v)
                seen.add(vid_id)

    if new_videos:
        send_digest(new_videos)
        print(f"Sent digest with {len(new_videos)} new videos")

    seen_path.write_text(json.dumps(list(seen)))

if __name__ == "__main__":
    run()

const API_KEY = "your_scavio_api_key";

async function searchChannel(channelName) {
  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: channelName }),
  });
  if (!res.ok) throw new Error(`scavio ${res.status}`);
  const data = await res.json();
  return data.organic ?? [];
}

async function sendDigest(videos) {
  // Replace with your email service (SendGrid, SES, Resend, etc.)
  const lines = videos.map(
    (v) => `- ${v.title} (${v.views ?? "N/A"} views): ${v.link}`
  );
  console.log("YouTube Digest:\n" + lines.join("\n"));
}

async function run() {
  const fs = await import("fs/promises");
  const channels = ["@competitor-channel", "@industry-leader"];
  let seen = new Set();
  try {
    seen = new Set(JSON.parse(await fs.readFile("seen_youtube.json", "utf8")));
  } catch {}

  const newVideos = [];

  for (const ch of channels) {
    const results = await searchChannel(ch);
    for (const v of results) {
      if (v.link && !seen.has(v.link)) {
        v.channel = ch;
        newVideos.push(v);
        seen.add(v.link);
      }
    }
  }

  if (newVideos.length) {
    await sendDigest(newVideos);
    console.log(`Sent digest with ${newVideos.length} new videos`);
  }

  await fs.writeFile("seen_youtube.json", JSON.stringify([...seen]));
}

run();