Whisper fallback for YouTube + Knowledge Graph test isolation fix

ai-ml/ingestion/youtube/transcript.py

@@ -14,6 +14,7 @@
 )
 
 from ingestion.youtube.cleaner import clean_youtube_text
+from ingestion.youtube.whisper_fallback import transcribe_with_whisper
 from ingestion.youtube.exceptions import (
     InvalidYouTubeURLError,
     VideoUnavailableError,
@@ -67,18 +68,28 @@ def fetch_metadata(url: str) -> dict:
         "date": info.get("upload_date"),
     }
 
-def fetch_transcript(video_id: str, languages=("en",)) -> dict:
+
+def fetch_transcript(video_id: str, url: str, languages=("en",)) -> dict:
     """
-    Returns {"text": str, "language_code": str, "is_generated": bool}.
+    Returns {"text": str, "language_code": str | None, "is_generated": bool}.
+
+    Falls back to Whisper (downloading and transcribing audio) whenever
+    no usable caption track exists, instead of raising
+    TranscriptNotAvailableError.
 
     Raises:
         VideoUnavailableError: video is private/deleted/unreachable.
-        TranscriptNotAvailableError: video exists but has no usable transcript
-            (disabled by uploader, none in any language, or empty once fetched).
         TranscriptFetchError: transient upstream failure (rate limit, IP block).
     """
     api = YouTubeTranscriptApi()
 
+    def _whisper_result() -> dict:
+        return {
+            "text": transcribe_with_whisper(url),
+            "language_code": None,
+            "is_generated": False,
+        }
+
     # Step 1: find out what transcripts actually exist for this video.
     try:
         transcript_list = api.list(video_id)
@@ -87,12 +98,8 @@ def fetch_transcript(video_id: str, languages=("en",)) -> dict:
             f"The video {video_id} is unavailable, private, or has been removed.",
             video_id=video_id,
         ) from e
-    except TranscriptsDisabled as e:
-        raise TranscriptNotAvailableError(
-            f"The uploader has disabled transcripts/captions for video {video_id}.",
-            video_id=video_id,
-            details={"reason": "transcripts_disabled"},
-        ) from e
+    except TranscriptsDisabled:
+        return _whisper_result()
     except CouldNotRetrieveTranscript as e:
         # Covers RequestBlocked/IpBlocked/PoTokenRequired/YouTubeRequestFailed etc.
         raise TranscriptFetchError(
@@ -110,12 +117,7 @@ def fetch_transcript(video_id: str, languages=("en",)) -> dict:
     ]
 
     if not available:
-        raise TranscriptNotAvailableError(
-            f"No transcript (manual or auto-generated) exists for video {video_id} "
-            f"in any language.",
-            video_id=video_id,
-            details={"available_languages": []},
-        )
+        return _whisper_result()
 
     # Step 2: prefer a manually created transcript in a requested language,
     # then an auto-generated one in a requested language, then just take
@@ -140,12 +142,7 @@ def fetch_transcript(video_id: str, languages=("en",)) -> dict:
     merged = " ".join(segment.text for segment in fetched).strip()
 
     if not merged:
-        raise TranscriptNotAvailableError(
-            f"Transcript for video {video_id} (language={transcript.language_code}) "
-            f"fetched successfully but contained no text.",
-            video_id=video_id,
-            details={"available_languages": available},
-        )
+        return _whisper_result()
 
     return {
         "text": merged,
@@ -159,7 +156,7 @@ def ingest_youtube(url: str) -> dict:
 
     metadata = fetch_metadata(url)
 
-    transcript_data = fetch_transcript(video_id)
+    transcript_data = fetch_transcript(video_id, url)
 
     cleaned_text = clean_youtube_text(transcript_data["text"])
 

ai-ml/ingestion/youtube/whisper_fallback.py

@@ -0,0 +1,113 @@
+"""
+Whisper fallback for YouTube ingestion.
+
+When a video has no available captions (transcripts disabled by the
+uploader), this downloads the audio track and transcribes it locally
+using OpenAI's Whisper model — free, runs on-device, no API key.
+
+Requires:
+    pip install openai-whisper yt-dlp
+    ffmpeg installed on the system (not just pip) — Whisper and
+    yt_dlp's audio extraction both depend on it. Check with:
+        ffmpeg -version
+    If missing: https://ffmpeg.org/download.html (or `winget install ffmpeg`
+    on Windows, `brew install ffmpeg` on Mac, `apt install ffmpeg` on Linux).
+"""
+import os
+import tempfile
+
+import whisper
+import yt_dlp
+
+# "base" balances speed vs. accuracy for a first version. Larger models
+# ("small", "medium") are more accurate but slower and require a bigger
+# one-time download (up to ~1.5GB for "medium"). Worth discussing with
+# the team if accuracy issues show up in practice.
+DEFAULT_MODEL_SIZE = os.getenv("WHISPER_MODEL_SIZE", "base")
+
+_model = None
+
+
+def get_whisper_model(size: str = None):
+    """
+    Loads the Whisper model once and reuses it across calls — same
+    singleton pattern as embedding/model.py's get_embedding_model().
+    """
+    global _model
+    if _model is None:
+        _model = whisper.load_model(size or DEFAULT_MODEL_SIZE)
+    return _model
+
+
+def download_audio(youtube_url: str, output_dir: str) -> str:
+    """
+    Downloads just the audio track (not video) to output_dir as an
+    mp3, using yt_dlp. Returns the full path to the downloaded file.
+
+    Raises RuntimeError with a clear message if the download fails
+    (e.g. private/deleted/region-locked video) — this should NOT
+    fail silently, per the objectives doc's note on silent-failure
+    cases.
+    """
+    output_template = os.path.join(output_dir, "audio.%(ext)s")
+
+    ydl_opts = {
+        "format": "bestaudio/best",
+        "outtmpl": output_template,
+        "postprocessors": [{
+            "key": "FFmpegExtractAudio",
+            "preferredcodec": "mp3",
+            "preferredquality": "128",
+        }],
+        "quiet": True,
+        "no_warnings": True,
+    }
+
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.download([youtube_url])
+    except yt_dlp.utils.DownloadError as e:
+        raise RuntimeError(
+            f"Could not download audio for {youtube_url}: {e}"
+        ) from e
+
+    for filename in os.listdir(output_dir):
+        if filename.startswith("audio"):
+            return os.path.join(output_dir, filename)
+
+    raise RuntimeError(
+        f"Audio download for {youtube_url} completed but no output file was found"
+    )
+
+
+def transcribe_with_whisper(youtube_url: str, model_size: str = None) -> str:
+    """
+    Fallback transcription path: downloads audio and transcribes it
+    with Whisper. Used when normal caption-based transcript fetching
+    fails or returns empty text.
+
+    Returns the transcribed text as a single string. Raises
+    RuntimeError (not a silent empty return) if either the download
+    or transcription step fails, so the caller can surface a clear
+    error rather than silently returning nothing — per the objectives
+    doc's Part A guidance on silent-failure cases.
+    """
+    with tempfile.TemporaryDirectory() as tmpdir:
+        audio_path = download_audio(youtube_url, tmpdir)
+
+        try:
+            model = get_whisper_model(model_size)
+            result = model.transcribe(audio_path)
+        except Exception as e:
+            raise RuntimeError(
+                f"Whisper transcription failed for {youtube_url}: {e}"
+            ) from e
+
+        text = result.get("text", "").strip()
+        if not text:
+            raise RuntimeError(
+                f"Whisper produced no transcribable speech for {youtube_url} "
+                "(video may have no spoken audio)"
+            )
+
+        return text