2 min read

YouTube Viewbot Detector (Stream Monitor)

This utility monitors a YouTube live stream to detect potential viewbotting behavior. It tracks the Viewer Count against Chat Activity and Likes. A common sign of viewbotting is a high viewer count with disproportionately low chat engagement or likes.

Modules Used:

  • google-api-python-client: To interact with the YouTube Data API.
  • python-dotenv: To securely manage the API key.
  • argparse: To handle command-line arguments.

Prerequisites

  1. Google Cloud Console: Create a project and enable the YouTube Data API v3.
  2. API Key: Create an API Key.

Installation

pip install google-api-python-client python-dotenv

Setup

Create a .env file in your project directory:

YOUTUBE_API_KEY=your_api_key_here

The Code

Save this as viewbot_check.py.

import os
import time
import argparse
from googleapiclient.discovery import build
from dotenv import load_dotenv
from datetime import datetime

# Load Config
load_dotenv()
API_KEY = os.getenv('YOUTUBE_API_KEY')

def get_service():
    return build('youtube', 'v3', developerKey=API_KEY)

def get_stream_details(youtube, video_id):
    """Fetches viewer count, likes, and chat ID."""
    try:
        request = youtube.videos().list(
            part="liveStreamingDetails,statistics,snippet",
            id=video_id
        )
        response = request.execute()

        if not response['items']:
            print("Error: Video not found.")
            return None

        item = response['items'][0]

        # Check if live
        if item['snippet']['liveBroadcastContent'] == 'none':
            print("Error: Video is not live.")
            return None

        stats = {
            'title': item['snippet']['title'],
            'viewers': int(item['liveStreamingDetails'].get('concurrentViewers', 0)),
            'likes': int(item['statistics'].get('likeCount', 0)),
            'chat_id': item['liveStreamingDetails'].get('activeLiveChatId')
        }
        return stats
    except Exception as e:
        print(f"API Error: {e}")
        return None

def count_chat_messages(youtube, chat_id, page_token=None):
    """Fetches chat messages and returns count + next token."""
    if not chat_id:
        return 0, None, 10

    try:
        request = youtube.liveChatMessages().list(
            liveChatId=chat_id,
            part="id",
            pageToken=page_token
        )
        response = request.execute()

        count = len(response.get('items', []))
        next_token = response.get('nextPageToken')
        # API tells us how long to wait before next poll
        polling_interval = response.get('pollingIntervalMillis', 10000) / 1000

        return count, next_token, polling_interval
    except Exception as e:
        print(f"Chat API Error: {e}")
        return 0, page_token, 10

def monitor_stream(video_id, duration_minutes):
    youtube = get_service()

    print(f"Initializing monitor for video ID: {video_id}...")
    initial_stats = get_stream_details(youtube, video_id)

    if not initial_stats:
        return

    print(f"Stream: {initial_stats['title']}")
    print(f"Initial Viewers: {initial_stats['viewers']}")
    print("-" * 60)
    print(f"{'Time':<10} | {'Viewers':<10} | {'Likes':<10} | {'Chats/min':<12} | {'Engagement %':<15}")
    print("-" * 60)

    chat_id = initial_stats['chat_id']
    next_chat_token = None

    # Get initial chat token without counting (to establish baseline)
    if chat_id:
        _, next_chat_token, _ = count_chat_messages(youtube, chat_id)

    start_time = time.time()
    end_time = start_time + (duration_minutes * 60)

    try:
        while time.time() < end_time:
            # Wait loop (approx 60 seconds for "Chats per minute" calculation)
            # We poll chat frequently but report every minute

            cycle_start = time.time()
            total_chats_in_cycle = 0

            # Poll chat for ~60 seconds
            while time.time() - cycle_start < 60:
                if chat_id:
                    count, next_chat_token, wait_ms = count_chat_messages(youtube, chat_id, next_chat_token)
                    total_chats_in_cycle += count
                    # Respect API polling interval (usually ~5-10s)
                    time.sleep(max(wait_ms, 5)) 
                else:
                    time.sleep(60)
                    break

            # Fetch updated viewer stats
            stats = get_stream_details(youtube, video_id)
            if not stats: break

            viewers = stats['viewers']
            likes = stats['likes']

            # Calculate Engagement
            # Engagement = (Chats per minute / Viewers) * 100
            engagement = 0.0
            if viewers > 0:
                engagement = (total_chats_in_cycle / viewers) * 100

            timestamp = datetime.now().strftime("%H:%M")

            print(f"{timestamp:<10} | {viewers:<10} | {likes:<10} | {total_chats_in_cycle:<12} | {engagement:.2f}%")

            # Heuristic Warning
            if viewers > 100 and engagement < 0.1:
                print(f"  [!] Suspiciously low activity ({engagement:.2f}%)")

    except KeyboardInterrupt:
        print("\nMonitoring stopped.")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="YouTube Viewbot Detector")
    parser.add_argument("video_id", help="The ID of the live stream (from URL)")
    parser.add_argument("--duration", type=int, default=10, help="Duration to monitor in minutes (default: 10)")

    args = parser.parse_args()

    monitor_stream(args.video_id, args.duration)

Usage

  1. Find Video ID: If the URL is https://www.youtube.com/watch?v=dQw4w9WgXcQ, the ID is dQw4w9WgXcQ.
  2. Run Monitor:
    python viewbot_check.py dQw4w9WgXcQ --duration 30

Interpreting Results:

  • Normal Stream: Engagement usually ranges from 0.5% to 5% depending on the content (e.g., 1000 viewers might generate 10-50 messages/min).
  • Suspicious: High viewers (e.g., 10,000) with near-zero chat (e.g., 5 messages/min) results in engagement < 0.05%.

programming/python/python