Using AI to Automate YouTube Channel Research

name,channel_id,niche
MKBHD,UCBcRF18a7Qf58cCRy5xuWwQ,tech reviews
Linus Tech Tips,UCXuqSBlHAE6Xw-yeJA0Tunw,tech reviews
Kurzgesagt,UCsXVk37bltHxD1rDPwtNM8Q,education

import requests

def get_channel_stats(channel_ids: list[str], api_key: str) -> list[dict]:
    """Fetch stats for up to 50 channels in one API call."""
    resp = requests.get(
        "https://www.googleapis.com/youtube/v3/channels",
        params={
            "part": "snippet,statistics",
            "id": ",".join(channel_ids),
            "key": api_key
        }
    )
    resp.raise_for_status()

    results = []
    for item in resp.json().get("items", []):
        stats = item.get("statistics", {})
        results.append({
            "channel_id": item["id"],
            "name": item["snippet"]["title"],
            "subscribers": int(stats.get("subscriberCount", 0)),
            "total_views": int(stats.get("viewCount", 0)),
            "video_count": int(stats.get("videoCount", 0)),
        })
    return results

def get_recent_videos(channel_id: str, api_key: str, max_results: int = 10) -> list[dict]:
    """Get recent video titles and publish dates for a channel."""
    ch_resp = requests.get(
        "https://www.googleapis.com/youtube/v3/channels",
        params={"part": "contentDetails", "id": channel_id, "key": api_key}
    )
    uploads_playlist = ch_resp.json()["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"]

    vid_resp = requests.get(
        "https://www.googleapis.com/youtube/v3/playlistItems",
        params={
            "part": "snippet",
            "playlistId": uploads_playlist,
            "maxResults": max_results,
            "key": api_key
        }
    )

    videos = []
    for item in vid_resp.json().get("items", []):
        snip = item["snippet"]
        videos.append({
            "title": snip["title"],
            "published_at": snip["publishedAt"],
            "video_id": snip["resourceId"]["videoId"],
        })
    return videos

import anthropic

def analyze_content_gaps(your_videos: list[dict], competitor_videos: list[dict]) -> str:
    client = anthropic.Anthropic()

    your_titles = "\n".join(f"- {v['title']}" for v in your_videos)
    comp_titles = "\n".join(f"- {v['title']}" for v in competitor_videos)

    message = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""Compare these two YouTube channels' recent content and identify content gaps.

My recent videos:
{your_titles}

Competitor's recent videos:
{comp_titles}

List: topics the competitor covers that I don't, topics I cover that they don't, 
and 3 specific video ideas I should make based on this analysis."""
        }]
    )
    return message.content[0].text

def weekly_competitor_report(competitor_channels: dict, my_channel_id: str, api_key: str):
    """competitor_channels: {"Name": "UCxxxxxx", ...}"""
    all_ids = list(competitor_channels.values())
    stats = {s["channel_id"]: s for s in get_channel_stats(all_ids, api_key)}

    report_data = []
    for name, channel_id in competitor_channels.items():
        videos = get_recent_videos(channel_id, api_key, max_results=20)
        channel_stats = stats.get(channel_id, {})
        report_data.append({
            "name": name,
            "subscribers": channel_stats.get("subscribers"),
            "recent_titles": [v["title"] for v in videos[:5]],
        })

    client = anthropic.Anthropic()
    summary = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Here is this week's competitor data: {report_data}\n\nWrite a concise weekly summary: who is gaining momentum, what topics are trending, and 3 content opportunities I should act on this week."
        }]
    )
    return summary.content[0].text