Implemented transcription fallback using Whisper

.gitattributes

@@ -1,3 +1,4 @@
+
 *.7z filter=lfs diff=lfs merge=lfs -text
 *.arrow filter=lfs diff=lfs merge=lfs -text
 *.bin filter=lfs diff=lfs merge=lfs -text

README.md

@@ -1,14 +1,45 @@
----
-title: VideoTranscription
-emoji: 💬
-colorFrom: yellow
-colorTo: purple
-sdk: gradio
-sdk_version: 5.0.1
-app_file: app.py
-pinned: false
-license: apache-2.0
-short_description: Transcribe video
----
-
-An example chatbot using [Gradio](https://gradio.app), [`huggingface_hub`](https://huggingface.co/docs/huggingface_hub/v0.22.2/en/index), and the [Hugging Face Inference API](https://huggingface.co/docs/api-inference/index).
+# YouTube Transcript SmolAgent Tool
+
+## Overview
+This is a tool to fetch transcripts for YouTube videos. It attempts to retrieve the official YouTube transcript first, and if that fails or if the user opts to skip it, it falls back to using Whisper ASR to transcribe the audio.
+
+## Project Structure
+- `app.py`: Main application file that sets up the Gradio interface and handles the transcript fetching logic.
+- `transcription.py`: Contains functions for downloading audio, converting it to WAV format, and transcribing it using Whisper ASR.
+- `youtube_utils.py`: Contains utility functions for extracting video IDs and fetching official YouTube transcripts.
+- `logging_config.py`: Configures logging for the application.
+- `ffmpeg_setup.py`: Ensures that `ffmpeg` is available in the system path.
+
+## Dependencies
+- Python
+- pip
+- Gradio
+- yt-dlp
+- torch
+- transformers
+- youtube_transcript_api
+- ffmpeg
+
+## How It Works
+1. **User Input**: The user provides a YouTube URL and optionally chooses to skip the official transcript check.
+2. **Transcript Fetching**:
+   - If the user opts to skip the official transcript, the tool directly uses Whisper ASR to transcribe the audio.
+   - Otherwise, it first attempts to fetch the official YouTube transcript.
+   - If the official transcript is not found, it falls back to using Whisper ASR.
+3. **Whisper ASR**:
+   - Downloads the best audio track using `yt-dlp`.
+   - Converts the audio to WAV format using `ffmpeg`.
+   - Transcribes the audio using the Whisper ASR model from the `transformers` library.
+4. **Output**: The transcript and logs are displayed in the Gradio interface.
+
+## Running the Application
+1. Ensure you have all dependencies installed:
+   ```sh
+   pip install gradio yt-dlp torch transformers youtube_transcript_api
+   ```
+2. Run the application:
+   ```sh
+   python app.py
+   ```
+3. Open the provided URL in your browser to access the Gradio interface.
+

app.py

@@ -1,45 +1,79 @@
-import re
 import gradio as gr
-from youtube_transcript_api import YouTubeTranscriptApi
-from huggingface_hub import InferenceClient
 
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
+from logging_config import logger, log_buffer
+from ffmpeg_setup import ensure_ffmpeg_in_path
+from youtube_utils import fetch_youtube_transcript
+from transcription import fallback_whisper_transcription
 
-def get_video_id(youtube_url: str) -> str:
-    """
-    Extract the video ID from a given YouTube URL.
-    """
-    # Typical patterns:
-    #   https://www.youtube.com/watch?v=VIDEO_ID
-    #   https://youtu.be/VIDEO_ID
-    #   https://www.youtube.com/shorts/VIDEO_ID
-    pattern = r"(?:v=|/shorts/|\.be/)([^&\n?#]+)"
-    match = re.search(pattern, youtube_url)
-    if not match:
-        raise ValueError("Could not extract video ID from the provided URL.")
-    return match.group(1)
-
-def fetch_transcript(youtube_url: str) -> str:
+
+def get_transcript(youtube_url: str, skip_official_transcript: bool):
     """
-    Given a YouTube URL, fetch the transcript and return it as a single string.
+        Fetch the official YouTube transcript,
+        or fall back to Whisper transcription if required
+        returns (partial_transcript, logs_so_far).
     """
-    try:
-        video_id = get_video_id(youtube_url)
-        # Fetch transcript
-        transcript_data = YouTubeTranscriptApi.get_transcript(video_id)
-        # Combine the transcript lines
-        transcript = " ".join([entry["text"] for entry in transcript_data])
-        return transcript
-    except Exception as e:
-        return f"Error fetching transcript: {str(e)}"
-
-demo = gr.Interface(
-    fn=fetch_transcript,
-    inputs=gr.Textbox(label="YouTube URL"),
-    outputs="text",
-    title="YouTube Transcript Fetcher",
-    description="Enter a YouTube link to retrieve its transcript."
-)
+    # Clear logs
+    log_buffer.seek(0)
+    log_buffer.truncate()
+
+    logger.info(f"Received YouTube URL: {youtube_url}")
+    logger.info(f"Skip official transcript check? {skip_official_transcript}")
+    logger.info("")
+    yield "", log_buffer.getvalue()
+
+    # If user wants to skip the official transcript, go straight to fallback
+    if skip_official_transcript:
+        logger.info("User selected to skip official transcript.")
+        yield "", log_buffer.getvalue()
+        for partial_transcript, current_logs in fallback_whisper_transcription(youtube_url):
+            yield partial_transcript, current_logs
+        return
+
+    # Otherwise, try official transcript first
+    transcript = fetch_youtube_transcript(youtube_url)
+    logger.info("")
+    yield "", log_buffer.getvalue()
+
+    # If official transcript wasn't found, do fallback
+    if transcript.startswith("Error"):
+        logger.info("Transcript not found. Falling back to local Whisper transcription...")
+        logger.info("")
+        yield "", log_buffer.getvalue()
+
+        for partial_transcript, current_logs in fallback_whisper_transcription(youtube_url):
+            yield partial_transcript, current_logs
+        return
+
+    # Otherwise, we succeeded with the official transcript
+    logger.info("Official transcript found successfully.")
+    logger.info("")
+    yield transcript, log_buffer.getvalue()
+
+
+def run_demo():
+    ensure_ffmpeg_in_path()
+
+    demo = gr.Interface(
+        fn=get_transcript,
+        inputs=[
+            gr.Textbox(label="YouTube URL"),
+            gr.Checkbox(label="Skip official transcript check?", value=False)
+        ],
+        outputs=[
+            gr.Textbox(label="Transcript"),
+            gr.Textbox(label="Logs (Streaming)"),
+        ],
+        title="YouTube Transcript SmolAgent Tool",
+        description=(
+            "Enter a YouTube link to retrieve its official transcript. "
+            "If that fails (or if 'Skip' is selected), we'll download the best "
+            "audio track with yt-dlp, convert it to WAV (via ffmpeg), and "
+            "then run Whisper to transcribe. Logs are displayed as it runs."
+        ),
+    )
 
-if __name__ == "__main__":
     demo.launch()
+
+
+if __name__ == "__main__":
+    run_demo()

ffmpeg_setup.py

@@ -0,0 +1,49 @@
+import os
+import stat
+import shutil
+import subprocess
+import imageio_ffmpeg
+from logging_config import logger
+
+
+def is_ffmpeg_in_path() -> bool:
+    try:
+        subprocess.run(
+            ["ffmpeg", "-version"],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            check=True
+        )
+        return True
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        return False
+
+
+def ensure_ffmpeg_in_path():
+    ffmpeg_path_original = imageio_ffmpeg.get_ffmpeg_exe()
+    ffmpeg_dir = os.path.dirname(ffmpeg_path_original)
+    binary_name = os.path.basename(ffmpeg_path_original)
+
+    logger.info(f"imageio-ffmpeg reported path: {ffmpeg_path_original}")
+    logger.info(f"Directory contents: {os.listdir(ffmpeg_dir)}")
+    logger.info(f"Binary name: {binary_name}")
+
+    expected_binary_name = "ffmpeg"
+    copied_path = os.path.join(ffmpeg_dir, expected_binary_name)
+
+    if not os.path.exists(copied_path):
+        logger.info(f"Copying {binary_name} to {expected_binary_name} in {ffmpeg_dir}.")
+        shutil.copy2(ffmpeg_path_original, copied_path)
+        st = os.stat(copied_path)
+        os.chmod(copied_path, st.st_mode | stat.S_IEXEC)
+    else:
+        logger.info(f"'{copied_path}' already exists; skipping copy.")
+
+    # Add directory to PATH
+    os.environ["PATH"] = ffmpeg_dir + os.pathsep + os.environ["PATH"]
+    logger.info(f"PATH now: {os.environ['PATH']}")
+
+    if is_ffmpeg_in_path():
+        logger.info("FFmpeg is accessible on PATH.")
+    else:
+        logger.warning("Tried appending the directory to PATH, but 'ffmpeg' is still not found.")

logging_config.py

@@ -0,0 +1,16 @@
+import logging
+import io
+
+# StringIO buffer to capture logs for streaming
+log_buffer = io.StringIO()
+
+logger = logging.getLogger("my_logger")
+logger.setLevel(logging.DEBUG)
+
+# log handler that writes to our StringIO buffer
+log_handler = logging.StreamHandler(log_buffer)
+log_handler.setLevel(logging.DEBUG)
+
+logger.addHandler(log_handler)
+# prevent duplicate logs
+logger.propagate = False

requirements.txt

@@ -1,3 +1,7 @@
 huggingface_hub==0.25.2
 gradio==5.12.0
 youtube-transcript-api==0.6.3
+yt-dlp==2025.1.15
+transformers==4.48.1
+torch==2.2.2
+imageio-ffmpeg==0.6.0

transcription.py

@@ -0,0 +1,96 @@
+import os
+import subprocess
+import tempfile
+import yt_dlp
+import torch
+from transformers import pipeline
+from logging_config import logger, log_buffer
+
+device = "cuda" if torch.cuda.is_available() else "cpu"
+
+
+def convert_audio_to_wav(input_file: str, output_file: str) -> str:
+    logger.info(f"Converting {input_file} to WAV: {output_file}")
+    cmd = [
+        "ffmpeg",
+        "-y",
+        "-i", input_file,
+        "-ar", "16000",  # sample rate
+        "-ac", "1",  # mono
+        output_file
+    ]
+    subprocess.run(cmd, check=True)
+    return output_file
+
+
+def fallback_whisper_transcription(youtube_url: str):
+    # returns (transcript, logs).
+    try:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Create temp dir
+            logger.info("")
+            logger.info(f"Created temporary directory: {tmpdir}")
+            logger.info("")
+            yield "", log_buffer.getvalue()
+
+            # Download best audio
+            logger.info("Downloading best audio via yt-dlp...")
+            logger.info("")
+            yield "", log_buffer.getvalue()
+
+            download_path = os.path.join(tmpdir, "audio.%(ext)s")
+            ydl_opts = {
+                'format': 'bestaudio/best',
+                'outtmpl': download_path,
+                'quiet': True,
+                'postprocessors': []
+            }
+
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                ydl.download([youtube_url])
+
+            logger.info("Audio downloaded. Locating the audio file in the temp folder...")
+            logger.info("")
+            yield "", log_buffer.getvalue()
+
+            # confirm audio file
+            downloaded_files = os.listdir(tmpdir)
+            if not downloaded_files:
+                raise RuntimeError("No audio file was downloaded via yt-dlp.")
+
+            audio_file_path = os.path.join(tmpdir, downloaded_files[0])
+            logger.info(f"Found audio file: {audio_file_path}")
+            logger.info("Video has downloaded!")
+            logger.info("")
+            yield "", log_buffer.getvalue()
+
+            # Convert to wav
+            wav_file_path = os.path.join(tmpdir, "audio.wav")
+            convert_audio_to_wav(audio_file_path, wav_file_path)
+            logger.info("Audio converted to WAV successfully.")
+            logger.info("")
+            yield "", log_buffer.getvalue()
+
+            # Run whisper
+            logger.info("Running Whisper ASR pipeline on the WAV file...")
+            logger.info("")
+            yield "", log_buffer.getvalue()
+
+            asr_pipeline = pipeline(
+                "automatic-speech-recognition",
+                model="openai/whisper-small",
+                return_timestamps=True,
+                device=device,
+                generate_kwargs={"task": "transcribe", "language": "<|en|>"}
+            )
+            result = asr_pipeline(inputs=wav_file_path)
+            transcription = result["text"]
+
+            logger.info("Whisper transcription completed successfully.")
+            logger.info("")
+            yield transcription, log_buffer.getvalue()
+
+    except Exception as e:
+        err_msg = f"Error in fallback transcription: {str(e)}"
+        logger.error(err_msg)
+        yield err_msg, log_buffer.getvalue()

youtube_utils.py

@@ -0,0 +1,24 @@
+import re
+from youtube_transcript_api import YouTubeTranscriptApi
+from logging_config import logger
+
+
+def get_video_id(youtube_url: str) -> str:
+    pattern = r"(?:v=|/shorts/|\.be/)([^&\n?#]+)"
+    match = re.search(pattern, youtube_url)
+    if not match:
+        raise ValueError("Could not extract video ID from the provided URL.")
+    return match.group(1)
+
+
+def fetch_youtube_transcript(youtube_url: str) -> str:
+    try:
+        video_id = get_video_id(youtube_url)
+        logger.info(f"Fetching official YouTube transcript for video ID: {video_id}")
+        transcript_data = YouTubeTranscriptApi.get_transcript(video_id)
+        transcript = " ".join([entry["text"] for entry in transcript_data])
+        return transcript
+    except Exception as e:
+        err_msg = f"Error fetching transcript: {str(e)}"
+        logger.error(err_msg)
+        return err_msg