How to Scrape YouTube Search Results: Complete Guide

# Create and activate a virtual environment
python -m venv youtube-scraper
source youtube-scraper/bin/activate   # macOS/Linux
# OR
.\youtube-scraper\Scripts\activate.bat  # Windows CMD
.\youtube-scraper\Scripts\Activate.ps1  # Windows PowerShell

# Install yt-dlp
pip install yt-dlp

import json
import sys
from typing import Any, Dict, List, Optional

from yt_dlp import YoutubeDL
from yt_dlp.utils import DownloadError

# Configuration constants - modify these to change search behavior
SEARCH_QUERY = "what is mcp"
SEARCH_RESULTS_LIMIT = 10

# yt-dlp configuration (suppresses output and returns JSON)
YDL_OPTS = {
    "quiet": True,
    "no_warnings": True,
    "dump_single_json": True,
}


def normalize_video_info(info: Dict[str, Any]) -> Dict[str, Any]:
    """Convert YouTube's API response to a standardized format."""
    # Convert duration from seconds to MM:SS format
    duration = info.get("duration", 0)
    formatted_duration = f"{duration // 60}:{duration % 60:02d}" if duration else "N/A"

    # Parse YouTube's upload date (YYYYMMDD) to standard date format
    upload_date = info.get("upload_date")
    create_time = None
    if upload_date:
        try:
            create_time = f"{upload_date[:4]}-{upload_date[4:6]}-{upload_date[6:8]}"
        except (IndexError, TypeError):
            create_time = upload_date

    return {
        # Core video metadata
        "id": info.get("id"),
        "url": info.get("webpage_url"),
        "title": info.get("title"),
        "description": info.get("description"),
        # Engagement metrics
        "view_count": info.get("view_count"),
        "like_count": info.get("like_count"),
        "comment_count": info.get("comment_count"),
        # Channel information
        "username": info.get("uploader"),
        "user_id": info.get("channel_id"),
        "follower_count": info.get("channel_follower_count"),
        "is_verified": info.get("channel_is_verified"),
        # Temporal data
        "create_time": create_time,
        "duration": duration,
        "duration_formatted": formatted_duration,
        # Content classification
        "hashtag_names": info.get("tags", []),
        "language": info.get("language"),
        # Thumbnail should be last as it's often the longest value
        "cover_image_url": info.get("thumbnail"),
    }


def search_youtube(query: str, limit: Optional[int] = None) -> List[Dict[str, Any]]:
    """Search YouTube using yt-dlp's internal search functionality."""
    # ytsearchX: prefix tells yt-dlp to return X search results
    search_url = f"ytsearch{limit if limit is not None else ''}:{query}"

    with YoutubeDL(YDL_OPTS) as ydl:
        try:
            results = ydl.extract_info(search_url, download=False)
            return (
                [
                    normalize_video_info(entry)
                    for entry in results.get("entries", [])
                    if entry  # Skip None/empty entries
                ]
                if results
                else []
            )
        except DownloadError as e:
            print(f"Search error for query '{query}': {e}", file=sys.stderr)
            return []


def save_results_to_json(data: List[Dict[str, Any]], filename: str) -> None:
    """Save data to JSON file with proper error handling."""
    if not data:
        print("No data to save.", file=sys.stderr)
        return

    try:
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=4, ensure_ascii=False)
        print(f"Successfully saved {len(data)} results to {filename}")
    except IOError as e:
        print(f"Error saving results to {filename}: {e}", file=sys.stderr)


def main() -> None:
    """Main execution flow: search -> process -> save results."""
    print(f"Searching YouTube for '{SEARCH_QUERY}'...")

    search_results = search_youtube(SEARCH_QUERY, SEARCH_RESULTS_LIMIT)

    if search_results:
        filename = f"youtube_search.json"
        save_results_to_json(search_results, filename)

        # Print summary of first 5 results
        print(f"\nFound {len(search_results)} results:")
        for i, video in enumerate(search_results[:5], 1):
            print(f"{i}. {video['title']} - {video['username']}")

        if len(search_results) > 5:
            print(f"... and {len(search_results) - 5} more results")
    else:
        print("No results found.")


if __name__ == "__main__":
    main()

{
    "id": "eur8dUO9mvE",
    "url": "https://www.youtube.com/watch?v=eur8dUO9mvE",
    "title": "What is MCP? Integrate AI Agents with Databases & APIs",
    "description": "Ready to become a certified Architect on Cloud Pak? Register now… Dive into the world of Model Context Protocol and learn how to seamlessly connect AI agents to databases, APIs, and more. Roy Derks breaks down its components, from hosts to servers, and showcases real-world applications. Gain the knowledge to revolutionize your AI projects…",
    "view_count": 205371,
    "like_count": 4914,
    "comment_count": 137,
    "username": "IBM Technology",
    "user_id": "UCKWaEZ-_VweaEx1j62do_vQ",
    "follower_count": 1240000,
    "is_verified": true,
    "create_time": "2025-02-19",
    "duration": 226,
    "duration_formatted": "3:46",
    "hashtag_names": ["IBM", "IBM Cloud"],
    "language": null,
    "cover_image_url": "https://i.ytimg.com/vi/eur8dUO9mvE/maxresdefault.jpg",
},

{
  "context": {
    "client": {
      "clientName": "WEB",
      "clientVersion": "2.20250620.01.00",
      "hl": "en",
      "gl": "US"
    }
  },
  "query": "your search term"
}

import json
import requests


class YouTubeSearcher:
    def __init__(self):
        # YouTube's internal API endpoint
        self.url = "https://www.youtube.com/youtubei/v1/search"
        # Required context to make requests look like they come from a web browser
        self.context = {
            "client": {"clientName": "WEB", "clientVersion": "2.20250620.01.00"}
        }

    def search(self, query, max_videos=20):
        """Search YouTube and return video data with pagination support."""
        videos = []
        continuation = None  # Token for getting the next page of results
        seen_ids = set()  # Track video IDs to prevent duplicates

        # Keep requesting pages until we have enough videos
        while len(videos) < max_videos:
            # Build the request payload
            payload = {"context": self.context, "query": query}
            if continuation:
                payload["continuation"] = continuation

            try:
                # Make API request with 10-second timeout
                data = requests.post(self.url, json=payload, timeout=10).json()
                new_videos = self._get_videos(data)

                # Filter out duplicate videos
                unique_videos = []
                for video in new_videos:
                    if video["id"] not in seen_ids:
                        seen_ids.add(video["id"])
                        unique_videos.append(video)

                # Add new unique videos to our collection
                videos.extend(unique_videos)
                continuation = self._get_continuation(data)

                # Stop if no more pages or no new videos found
                if not continuation or not unique_videos:
                    break
            except Exception as e:
                print(f"Request failed: {e}")
                break

        if not videos:
            print("No videos found for this query")

        return videos[:max_videos]

    def _get_videos(self, data):
        """Extract all videos from API response."""
        videos = []
        self._find_videos(data, videos)
        return videos

    def _find_videos(self, obj, videos):
        """Recursively search through nested YouTube response data to find video objects."""
        if isinstance(obj, dict):
            if "videoRenderer" in obj:
                # Found a video - parse it
                video = self._parse_video(obj["videoRenderer"])
                if video:
                    videos.append(video)
            else:
                # Keep searching in nested objects
                for v in obj.values():
                    self._find_videos(v, videos)
        elif isinstance(obj, list):
            # Search through list items
            for item in obj:
                self._find_videos(item, videos)

    def _parse_video(self, r):
        """Extract video information from YouTube's video renderer object."""
        try:

            def text(obj):
                """Helper to extract text from YouTube's text objects."""
                if not obj:
                    return ""
                if "simpleText" in obj:
                    return obj["simpleText"]
                if "runs" in obj and obj["runs"]:
                    return obj["runs"][0].get("text", "")
                return ""

            # Extract basic video information
            video_id = r.get("videoId", "")
            title = text(r.get("title"))
            channel = text(r.get("longBylineText"))
            views = text(r.get("viewCountText"))
            duration = text(r.get("lengthText"))
            published = text(r.get("publishedTimeText"))

            # Get the highest quality thumbnail
            thumbnail = ""
            if "thumbnail" in r and "thumbnails" in r["thumbnail"]:
                thumbnails = r["thumbnail"]["thumbnails"]
                if thumbnails:
                    thumbnail = thumbnails[-1].get(
                        "url", ""
                    )  # Last thumbnail in the array is the highest resolution

            # Extract description snippet
            description = ""
            if "detailedMetadataSnippets" in r and r["detailedMetadataSnippets"]:
                snippet = r["detailedMetadataSnippets"][0]
                if "snippetText" in snippet and "runs" in snippet["snippetText"]:
                    desc_parts = [
                        run.get("text", "") for run in snippet["snippetText"]["runs"]
                    ]
                    description = "".join(desc_parts)[:200]  # Limit to 200 characters

            return {
                "id": video_id,
                "title": title or "No title",
                "url": f"https://www.youtube.com/watch?v={video_id}",
                "channel": channel or "Unknown",
                "views": views or "No views",
                "duration": duration or "Unknown",
                "published": published or "Unknown",
                "thumbnail": thumbnail,
                "description": description,
            }
        except Exception as e:
            print(f"Failed to parse video: {e}")
            return None

    def _get_continuation(self, obj):
        """Find the continuation token for loading the next page of results."""
        if isinstance(obj, dict):
            if "continuationCommand" in obj and "token" in obj["continuationCommand"]:
                return obj["continuationCommand"]["token"]
            # Search recursively in nested objects
            for v in obj.values():
                result = self._get_continuation(v)
                if result:
                    return result
        elif isinstance(obj, list):
            # Search through list items
            for item in obj:
                result = self._get_continuation(item)
                if result:
                    return result
        return None

    def save(self, videos, filename="youtube_search.json"):
        """Save video results to a JSON file with proper Unicode support."""
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(videos, f, indent=2, ensure_ascii=False)
        print(f"Saved {len(videos)} videos")


# Example usage
if __name__ == "__main__":
    searcher = YouTubeSearcher()
    videos = searcher.search("what is mcp", 10)
    searcher.save(videos)

{
    "id": "eur8dUO9mvE",
    "title": "What is MCP? Integrate AI Agents with Databases & APIs",
    "url": "https://www.youtube.com/watch?v=eur8dUO9mvE",
    "channel": "IBM Technology",
    "views": "205,576 views",
    "duration": "3:46",
    "published": "4 months ago",
    "thumbnail": "https://i.ytimg.com/vi/eur8dUO9mvE/hq720.jpg?sqp=-oaymwEXCNAFEJQDSFryq4qpAwkIARUAAIhCGAE=&rs=AOn4CLBXBwZNQuJ5lEfpsX5wQrJ9SjHPvg",
    "description": "Unlock the secrets of MCP! Dive into the world of Model Context Protocol and learn how to seamlessly connect AI agents to ...",
}

# Install the library
pip install playwright

# Download browser binaries (Chromium, Firefox, WebKit)
playwright install

The following script automates this entire process:


import asyncio # For handling the asynchronous browser operations
import json # For exporting the scraped video data
import logging # For tracking the scraping progress and debugging issues.
from playwright.async_api import async_playwright # The core web scraping library that controls the browser
from urllib.parse import quote # For safely encoding the search query in the YouTube URL
from datetime import datetime # For adding timestamps to your scraped data files..

# Configuration
SEARCH_QUERY = "what is mcp"
OUTPUT_FILE = "YT_search_results.json"
MAX_RESULTS = 3
HEADLESS = True
SCROLL_DELAY = 2.0
MAX_IDLE_SCROLLS = 3

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


async def get_element_data(element, selector, attribute=None):
    """Get text or attribute from element safely."""
    try:
        sub_element = await element.query_selector(selector)
        if not sub_element:
            return ""
        if attribute:
            result = await sub_element.get_attribute(attribute)
            return result or ""
        else:
            result = await sub_element.text_content()
            return result.strip() if result else ""
    except Exception:
        return ""


def parse_number_with_suffix(text):
    """Parse views (1.2M) or duration (5:30) into numbers."""
    if not text:
        return 0

    text = text.lower().strip()

    # Handle duration (5:30 -> seconds)
    if ":" in text:
        clean_text = "".join(c for c in text if c.isdigit() or c == ":")
        if ":" in clean_text:
            try:
                parts = [int(p) for p in clean_text.split(":") if p]
                return sum(part * (60**i) for i, part in enumerate(reversed(parts)))
            except Exception:
                return 0

    # Handle view counts (1.2M -> 1200000)
    text = text.replace("views", "").strip()
    multipliers = {"k": 1_000, "m": 1_000_000, "b": 1_000_000_000}

    for suffix, multiplier in multipliers.items():
        if suffix in text:
            try:
                return int(float(text.replace(suffix, "")) * multiplier)
            except Exception:
                return 0

    try:
        return int(float(text))
    except Exception:
        return 0


async def extract_video_data(element):
    """Extract all video data from element."""
    title = await get_element_data(element, "a#video-title")
    url_path = await get_element_data(element, "a#video-title", "href")

    metadata = await element.query_selector_all("#metadata-line .inline-metadata-item")
    views_text = upload_time = ""

    try:
        for item in metadata:
            text = (await item.text_content()).strip()
            if "view" in text.lower():
                views_text = text
            elif text:
                upload_time = text
    except Exception:
        pass

    return {
        "title": title,
        "url": (
            f"https://www.youtube.com{url_path.split('&pp=')[0]}" if url_path else ""
        ),
        "views": parse_number_with_suffix(views_text),
        "upload_time": upload_time,
        "duration_seconds": parse_number_with_suffix(
            await get_element_data(
                element, "ytd-thumbnail-overlay-time-status-renderer span"
            )
        ),
        "channel": await get_element_data(element, "#channel-name a"),
        "channel_url": (
            f"https://www.youtube.com{channel_path}"
            if (
                channel_path := await get_element_data(
                    element, "#channel-name a", "href"
                )
            )
            else ""
        ),
        "thumbnail": await get_element_data(element, "yt-image img", "src"),
        "verified": await element.query_selector(".badge-style-type-verified")
        is not None,
    }


async def load_videos(page, max_results):
    """Scroll and collect video elements."""
    logger.info("Loading videos...")
    videos = []
    idle_count = 0

    while True:
        current_videos = await page.query_selector_all("ytd-video-renderer")

        if len(current_videos) > len(videos):
            videos = current_videos
            idle_count = 0
            logger.info(f"Found {len(videos)} videos")
        else:
            idle_count += 1
            if idle_count >= MAX_IDLE_SCROLLS:
                break

        if max_results and len(videos) >= max_results:
            videos = videos[:max_results]
            break

        await page.evaluate("window.scrollTo(0, document.documentElement.scrollHeight)")
        await page.wait_for_timeout(SCROLL_DELAY * 1000)

    return videos


async def scrape_youtube(query, max_results):
    """Main scraping function."""
    url = f"https://www.youtube.com/results?search_query={quote(query)}"

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=HEADLESS)
        page = await browser.new_page()

        try:
            logger.info(f"Scraping: {query}")
            await page.goto(url, wait_until="domcontentloaded", timeout=60000)

            # Accept cookies if needed
            cookie_btn = page.locator(
                'button:has-text("Accept all"), button:has-text("I agree")'
            )
            if await cookie_btn.count() > 0:
                await cookie_btn.first.click()

            await page.wait_for_selector("ytd-video-renderer", timeout=15000)

            video_elements = await load_videos(page, max_results)
            logger.info(f"Extracting data from {len(video_elements)} videos...")

            videos_data = await asyncio.gather(
                *[extract_video_data(v) for v in video_elements]
            )

            result = {
                "search_query": query,
                "total_videos": len(videos_data),
                "timestamp": datetime.now().isoformat(),
                "videos": videos_data,
            }

            with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
                json.dump(result, f, indent=2, ensure_ascii=False)

            logger.info(f"Saved {len(videos_data)} videos to {OUTPUT_FILE}")
            return result

        except Exception as e:
            logger.error(f"Error: {e}")
            await page.screenshot(path="error.png")
            return None
        finally:
            await browser.close()


async def main():
    result = await scrape_youtube(SEARCH_QUERY, MAX_RESULTS)
    if not result:
        logger.warning("Scraping failed")


if __name__ == "__main__":
    asyncio.run(main())