phidata/phi/document/reader/youtube_reader.py

from typing import List
from phi.document.base import Document
from phi.document.reader.base import Reader
from phi.utils.log import logger


class YouTubeReader(Reader):
    """Reader for YouTube video transcripts"""

    def read(self, video_url: str) -> List[Document]:
        if not video_url:
            raise ValueError("No video URL provided")

        try:
            from youtube_transcript_api import YouTubeTranscriptApi
        except ImportError:
            raise ImportError("`youtube_transcript_api` not installed")

        try:
            # Extract video ID from URL
            video_id = video_url.split("v=")[-1].split("&")[0]
            logger.info(f"Reading transcript for video: {video_id}")

            # Get transcript
            transcript_list = YouTubeTranscriptApi.get_transcript(video_id)

            # Combine transcript segments into full text
            transcript_text = ""
            for segment in transcript_list:
                transcript_text += f"{segment['text']} "

            documents = [
                Document(
                    name=f"youtube_{video_id}",
                    id=f"youtube_{video_id}",
                    meta_data={"video_url": video_url, "video_id": video_id},
                    content=transcript_text.strip(),
                )
            ]

            if self.chunk:
                chunked_documents = []
                for document in documents:
                    chunked_documents.extend(self.chunk_document(document))
                return chunked_documents
            return documents

        except Exception as e:
            logger.error(f"Error reading transcript for {video_url}: {e}")
            return []