tools/youtube_video_tool.py

from smolagents import Tool
from openai import OpenAI
from tools.speech_recognition_tool import SpeechRecognitionTool
from io import BytesIO
import yt_dlp
import av
import torchaudio
import subprocess
import requests
import base64
import tempfile
import re
import os

class YouTubeVideoTool(Tool):
    name = 'youtube_video'
    description = '''Process the video and return the requested information from it.'''
    inputs = {
        'url': {
            'type': 'string',
            'description': 'The URL of the YouTube video.',
        },
        'query': {
            'type': 'string',
            'description': 'The question to answer',
        },
    }
    output_type = 'string'

    def __init__(
            self,
            video_quality: int = 360,
            frames_interval: int | float | None = 2,
            chunk_duration: int | float | None = 2,
            speech_recognition_tool: SpeechRecognitionTool | None = None,
            client: OpenAI | None = None,
            model_id: str = 'gpt-4.1-mini',
            debug: bool = False,
            **kwargs,
    ):
        self.video_quality = video_quality
        self.speech_recognition_tool = speech_recognition_tool
        self.frames_interval = frames_interval
        self.chunk_duration = chunk_duration

        self.client = client or OpenAI()
        self.model_id = model_id

        self.debug = debug

        super().__init__(**kwargs)

    def forward(self, url: str, query: str):
        '''
        Process the video and return the requested information.
        Args:
            url(str): The URL of the YouTube video.
            query(str): The question to answer.
        Returns:
            str: Answer to the query
        '''
        answer = ''
        for chunk in self._split_video_into_chunks(url):
            prompt = self._prompt(
                chunk,
                query,
                answer,
            )
            response = self.client.responses.create(
                model='gpt-4.1-mini',
                input=[
                    {
                        'role': 'user',
                        'content': [
                            {
                                'type': 'input_text',
                                'text': prompt,
                            },
                        ],
                    },
                ],
            )
            answer = response.output_text
            if self.debug:
                print(
                    f"CHUNK {chunk['start']} - {chunk['end']}:\n\n{prompt}\n\nANSWER:\n{answer}"
                )
        if answer.strip() == 'I need to keep watching.':
            answer = ''
        return answer
    def _prompt(self, chunk, query, aggregated_answer):
        prompt = [
            f"""\
These are some frames of a YouTube video.
I will ask a question about the entire video, but you'll only see part of it at a time.
Aggregate answer about the entire video, use information about previous parts but do not reference the previous parts in the answer directly.

Ground your answer based on video title, description, captions, vide frames or answer from previous parts.
If no evidences presented just say "I need to keep watching".

VIDEO TITLE:
{chunk["title"]}

VIDEO DESCRIPTION:
{chunk["description"]}

FRAMES SUBTITLES:
{chunk["captions"]}"""
        ]

        if aggregated_answer:
            prompt.append(f"""\
Here is the answer to the same question based on the previous video parts:
                          
BASED ON PREVIOUS PARTS:
{aggregated_answer}""")
        prompt.append(f"""\
QUESTION:
{query}""")
        
        return "\n\n".join(prompt)
    
    def _split_video_into_chunks(
            self, url: str, with_captions: bool = True, with_frames: bool = True
    ):
        video = self._process_video(
            url, with_captions=with_captions, with_frames=with_frames
        )
        video_duration = video['duration']
        chunk_duration = self.chunk_duration or video_duration

        chunk_start = 0.0
        while chunk_start < video_duration:
            chunk_end = min(chunk_start + chunk_duration, video_duration)
            chunk = self._get_video_chunk(video, chunk_start, chunk_end)
            yield chunk
            chunk_start += chunk_duration
        
    def _get_video_chunk(self, video, start, end):
        chunk_captions = [
            c for c in video['captions'] if c['start'] <= end and c['end'] >=  start
        ]
        chunk_frames = [
            f
            for f in video['frames']
            if f['timestamp'] >= start and f['timestamp'] <= end
        ]

        return {
            'title': video['title'],
            'description': video['description'],
            'start': start,
            'end': end,
            'captions': '\n'.join([c['text'] for c in chunk_captions]),
            'frames': chunk_frames,
        }
    
    def _process_video(
            self, url: str, with_captions: bool = True, with_frames: bool = True
    ):
        lang = 'en'
        info = self._get_video_info(url, lang)

        if with_captions:
            captions = self._extract_captions(
                lang, info.get('subtitles', {}), info.get('automatic_captions', {})
            )
            if not captions and self.speech_recognition_tool:
                audio_url = self._select_audio_format(info['formats'])
                audio = self._capture_audio(audio_url)
                with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp:
                    tmp.write(audio.read())
                    tmp.flush()
                    tmp_path = tmp.name  # Save the path before closing

                # Now the file is closed, safe to use
                try:
                    transcription = self.speech_recognition_tool(audio=tmp_path, with_time_markers=True)
                    captions = []
                    pattern = re.compile(r'\[(\d+\.\d+)\]\n(.+?)\n\[(\d+\.\d+)\]', re.DOTALL)
                    for match in pattern.finditer(transcription):
                        start, text, end = match.groups()
                        captions.append({
                            'start': float(start),
                            'end': float(end),
                            'text': text.strip(),
                        })
                finally:
                    os.remove(tmp_path)  # Clean up the temp file
        else:
            captions = []

        if with_frames:
            video_url = self._select_video_format(info['formats'], 360)['url']
            frames = self._capture_video_frames(video_url, self.frames_interval)
        else:
            frames = []
        
        return {
            'id': info['id'],
            'title': info['title'],
            'description': info['description'],
            'duration': info['duration'],
            'captions': captions,
            'frames': frames,
        }
    
    def _get_video_info(self, url: str, lang: str):
        ydl_opts = {
            'quiet': True,
            'skip_download': True,
            'format': 'bestvideo[ext=mp4][height<=360]+bestaudio[ext=m4a]/best[height<=360]',
            "forceurl": True,
            "noplaylist": True,
            "writesubtitles": True,
            "writeautomaticsub": True,
            "subtitlesformat": "vtt",
            "subtitleslangs": [lang],
        }
        
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            info = ydl.extract_info(url, download=False)

        return info
    
    def _extract_captions(self, lang, subtitles, auto_captions):
        caption_tracks = subtitles.get(lang) or auto_captions.get(lang) or []

        structured_captions = []

        srt_track = next(
            (track for track in caption_tracks if track['ext'] == 'srt'), None
        )
        vtt_track = next(
            (track for track in caption_tracks if track['ext'] == 'vtt'), None
        )

        if srt_track:
            import pysrt
            response = requests.get(srt_track['url'])
            response.raise_for_status()
            srt_data = response.content.decode('utf-8')

            def to_sec(t):
                return (
                    t.hours * 3600 + t.minutes * 60 + t.seconds + t.milliseconds / 1000
                )
            
            structured_captions = [
                {
                    "start": to_sec(sub.start),
                    "end": to_sec(sub.end),
                    "text": sub.text.strip(),
                }
                for sub in pysrt.from_string(srt_data)
            ]
        if vtt_track:
            import webvtt
            from io import StringIO

            response = requests.get(vtt_track['url'])
            response.raise_for_status()
            vtt_data = response.text

            vtt_file =  StringIO(vtt_data)

            def to_sec(t):
                """Convert 'HH:MM:SS.mmm' to float seconds"""
                h, m, s = t.split(":")
                s, ms = s.split(".")
                return int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000
            
            for caption in webvtt.read_buffer(vtt_file):
                structured_captions.append(
                    {
                        'start': to_sec(caption.start),
                        "end": to_sec(caption.end),
                        "text": caption.text.strip(),
                    }
                )
        return structured_captions

    def _select_video_format(self, formats, video_quality):
        video_format = next(
            f
            for f in formats
            if f.get('vcodec') != 'none' and f.get('height') == video_quality
        )
        return video_format
    
    def _capture_video_frames(self, video_url, capture_interval_sec=None):
        import tempfile
        import os

        with tempfile.NamedTemporaryFile(suffix='.mkv', delete=False) as tmp:
            tmp_path = tmp.name

        try:
            ffmpeg_cmd = [
                "ffmpeg",
                "-y",  # Overwrite output file if needed
                "-i", video_url,
                "-f", "matroska",
                tmp_path,
            ]
            result = subprocess.run(ffmpeg_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            if result.returncode != 0 or not os.path.exists(tmp_path) or os.path.getsize(tmp_path) == 0:
                raise RuntimeError(f"ffmpeg failed to create a valid video file for {video_url}")

            container = av.open(tmp_path)
            stream = container.streams.video[0]
            time_base = stream.time_base

            frames = []
            next_capture_time = 0
            for frame in container.decode(stream):
                if frame.pts is None:
                    continue
                timestamp = float(frame.pts * time_base)
                if capture_interval_sec is None or timestamp >= next_capture_time:
                    frames.append(
                        {
                            'timestamp': timestamp,
                            'image': frame.to_image(),  # PIL image
                        }
                    )
                    if capture_interval_sec is not None:
                        next_capture_time += capture_interval_sec
            container.close()
            return frames
        finally:
            os.remove(tmp_path)
    
    def _base64_frames(self, frames):
        base64_frames = []
        for f in frames:
            buffered = BytesIO()
            f['image'].save(buffered, format='JPEG')
            encoded = base64.b64encode(buffered.getvalue()).decode('utf-8')
            base64_frames.append(encoded)
        return base64_frames
    
    def _select_audio_format(self, formats):
        audio_formats = [
            f
            for f in formats
            if f.get('vcodec') == 'none'
            and f.get('acodec')
            and f.get('acodec') != 'none'
        ]

        if not audio_formats:
            raise ValueError('No valid audio-only formats found')
        
        # Prefer m4a > webm, highest abr first
        preferred_exts = ['m4a', 'webm']

        def sort_key(f):
            ext_score = (
                preferred_exts.index(f['ext']) if f['ext'] in preferred_exts else 99
            )
            abr =  f.get('abr') or 0
            return (ext_score, -abr)
        
        audio_formats.sort(key=sort_key)
        return audio_formats[0]['url']
    
    def _capture_audio(self, audio_url) -> BytesIO:
        audio_buffer = BytesIO()
        ffmpeg_audio_cmd = [
            "ffmpeg",
            "-i",
            audio_url,
            "-f",
            "wav",
            "-acodec",
            "pcm_s16le",  # Whisper prefers PCM
            "-ac",
            "1",  # Mono
            "-ar",
            "16000",  # 16kHz for Whisper
            "-",
        ]

        result = subprocess.run(
            ffmpeg_audio_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )
        if result.returncode != 0:
            raise RuntimeError('ffmpeg failed:\n' + result.stderr.decode())
        
        audio_buffer = BytesIO(result.stdout)
        audio_buffer.seek(0)
        return audio_buffer