fix: Refactor polling configuration to use a list of dictionaries for accurate retry attempts; add 'innertubex' to requirements

gemini_transcript.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import re
 import sys
 import json
@@ -18,7 +19,6 @@ from youtube_transcript_api import (
     YouTubeTranscriptApi,
     TranscriptsDisabled,
     NoTranscriptFound,
-    VideoUnavailable,
 )
 
 
@@ -26,7 +26,11 @@ from youtube_transcript_api import (
 # CONFIG
 # ============================================================================
 
-GEMINI_API_KEY = "AIzaSyCNz5wQAyJ65kNRkwr0-1A-_Z6-lQzdcyc"
+GEMINI_KEY = "AIzaSyCNz5wQAyJ65kNRkwr0-1A-_Z6-lQzdcyc"
+
+# ── API Keys ────────────────────────────────────────────────────────────────
+GEMINI_API_KEY  = os.getenv("GEMINI_API_KEY",  GEMINI_KEY)
+YT_API_KEY      = os.getenv("YT_API_KEY",      "AIzaSyASnhRpV-YQQb4xvoggWIEm8nvrujerEos")
 
 GEMINI_MODELS = [
     "gemini-2.5-flash",
@@ -34,22 +38,28 @@ GEMINI_MODELS = [
     "gemini-2.5-pro",
 ]
 
-POLLING_CONFIG = {
-    "attempt_1": {"wait_before": 0,    "description": "Immediate attempt on trigger"},
-    "attempt_2": {"wait_before": 300,  "description": "Retry after 5 minutes"},
-    "attempt_3": {"wait_before": 900,  "description": "Retry after 15 minutes "},
-    "attempt_4": {"wait_before": 900, "description": "Retry after 15 minutes 30 minutes completed"},
-    "attempt_5": {"wait_before": 900, "description": "Retry after 15 minutes 45 minutes completed"},
-    "attempt_6": {"wait_before": 900, "description": "Retry after 15 minutes 1 hour completed"},
-    "attempt_7": {"wait_before": 900, "description": "Retry after 15 minutes 1 hour 15 minutes completed"},
-    "attempt_8": {"wait_before": 900, "description": "Retry after 15 minutes 1 hour 30 minutes completed"},
-    "attempt_3": {"wait_before": 900,  "description": "Retry after 15 minutes 1 hour 45 minutes completed"},
-    "attempt_4": {"wait_before": 900, "description": "Retry after 15 minutes 2 hours completed"},
-    "attempt_5": {"wait_before": 900, "description": "Retry after 15 minutes 2 hours 15 minutes completed"},
-    "attempt_6": {"wait_before": 900, "description": "Retry after 15 minutes 2 hours 30 minutes completed"},
-    "attempt_7": {"wait_before": 900, "description": "Retry after 15 minutes 2 hours 45 minutes completed"},
-    "attempt_8": {"wait_before": 900, "description": "Retry after 15 minutes 3 hours completed"},
-}
+# ── FIX: Use a LIST of dicts, not a dict.
+#    A plain dict with duplicate keys like "attempt_3" silently drops all
+#    but the last definition, collapsing 13 intended attempts down to 4.
+#    A list preserves every entry in order.
+def _polling_attempt(wait_before: int, description: str) -> dict:
+    return {"wait_before": wait_before, "description": description}
+
+POLLING_CONFIG: list[dict] = [
+    _polling_attempt(0,    "Immediate attempt on trigger"),
+    _polling_attempt(300,  "Retry after 5 minutes"),
+    _polling_attempt(900,  "Retry after 15 minutes (30 min total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (45 min total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (1 hr total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (1 hr 15 min total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (1 hr 30 min total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (1 hr 45 min total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (2 hr total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (2 hr 15 min total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (2 hr 30 min total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (2 hr 45 min total)"),
+    _polling_attempt(900,  "Retry after 15 minutes (3 hr total)"),
+]
 
 SYSTEM_PROMPT = """
 You are an expert content summarizer and educator.
@@ -148,30 +158,21 @@ def _parse_vtt(content: str) -> str:
 
     for line in lines:
         stripped = line.strip()
-        # Skip empty lines
         if not stripped:
             continue
-        # Skip VTT header
         if stripped.startswith("WEBVTT"):
             continue
-        # Skip metadata lines (Kind:, Language:, Style, NOTE, etc.)
         if re.match(r"^(Kind:|Language:|Style|NOTE)", stripped, re.IGNORECASE):
             continue
-        # Skip timestamp lines (00:00:00.000 --> 00:00:05.000)
         if re.match(r"^\d{2}:\d{2}[:\.]\d{2}[\.:]\d{3}\s*-->\s*\d{2}:\d{2}", stripped):
             continue
-        # Skip position/alignment metadata
         if re.match(r"^(position:|align:|line:|size:)", stripped, re.IGNORECASE):
             continue
-        # Skip sequence numbers (pure digits)
         if stripped.isdigit():
             continue
-        # Strip inline tags like <c>, </c>, <00:00:01.234>
-        cleaned = re.sub(r"<[^>]+>", "", stripped)
-        cleaned = cleaned.strip()
+        cleaned = re.sub(r"<[^>]+>", "", stripped).strip()
         if not cleaned:
             continue
-        # Deduplicate consecutive identical lines
         if cleaned != prev_line:
             text_lines.append(cleaned)
             prev_line = cleaned
@@ -191,13 +192,10 @@ def _parse_srt(content: str) -> str:
         stripped = line.strip()
         if not stripped:
             continue
-        # Skip sequence numbers
         if stripped.isdigit():
             continue
-        # Skip timing lines
         if re.match(r"^\d{2}:\d{2}:\d{2}[,.]\d{3}\s*-->\s*\d{2}:\d{2}", stripped):
             continue
-        # Strip HTML-style tags
         cleaned = re.sub(r"<[^>]+>", "", stripped).strip()
         if cleaned:
             text_lines.append(cleaned)
@@ -240,12 +238,12 @@ class YouTubeTranscriptFetcher:
         self,
         youtube_url: str,
         languages: Optional[List[str]] = None,
-        polling_config: dict           = None,
+        polling_config: list[dict]     = None,   # ← list, not dict
         google_creds                   = None,
     ):
         self.youtube_url    = youtube_url
         self.languages      = languages or ["en", "en-US", "en-GB"]
-        self.polling_config = polling_config or POLLING_CONFIG
+        self.polling_config = polling_config if polling_config is not None else POLLING_CONFIG
         self.video_id       = self._extract_video_id(youtube_url)
         self.api            = YouTubeTranscriptApi()
         self.google_creds   = google_creds
@@ -274,7 +272,7 @@ class YouTubeTranscriptFetcher:
             first_transcript = next(iter(transcript_list))
             logger.info("[Tier 1] Falling back to language: %s", first_transcript.language_code)
             transcript = first_transcript.fetch()
-            
+
         return " ".join(item.text for item in transcript)
 
     def _try_all_tiers(self) -> tuple[str, str]:
@@ -291,8 +289,6 @@ class YouTubeTranscriptFetcher:
             logger.info("[Tier 1] ✅ youtube_transcript_api succeeded — %d chars", len(text))
             return text, "youtube_transcript_api"
         except TranscriptsDisabled as e:
-            # Still try yt-dlp — sometimes auto-subs exist even when
-            # the captions toggle is "disabled" for the transcript API
             errors.append(f"Tier1(TranscriptsDisabled): {e}")
             logger.warning("[Tier 1] Transcripts disabled, trying fallbacks...")
         except Exception as e:
@@ -336,33 +332,35 @@ class YouTubeTranscriptFetcher:
         On each polling attempt, all tiers are tried before waiting.
         Returns (transcript_text, extraction_method).
         """
-        logger.info("Video ID          : %s", self.video_id)
-        logger.info("Polling attempts  : %d", len(self.polling_config))
+        # ── FIX: polling_config is now a list, so len() and enumeration work correctly.
+        attempts = self.polling_config
+        total    = len(attempts)
 
-        attempts = list(self.polling_config.items())
+        logger.info("Video ID          : %s", self.video_id)
+        logger.info("Polling attempts  : %d", total)
 
-        for idx, (attempt_key, config) in enumerate(attempts, start=1):
+        for idx, config in enumerate(attempts, start=1):
             wait_before = config["wait_before"]
             description = config["description"]
 
             if wait_before > 0:
                 logger.info(
                     "[%d/%d] %s — waiting %s before retry...",
-                    idx, len(attempts), description,
+                    idx, total, description,
                     _format_duration(wait_before),
                 )
                 time.sleep(wait_before)
 
             logger.info(
                 "[%d/%d] %s — trying all transcript tiers...",
-                idx, len(attempts), description,
+                idx, total, description,
             )
 
             try:
                 text, method = self._try_all_tiers()
                 logger.info(
                     "[%d/%d] ✅ Transcript fetched via %s — %d characters",
-                    idx, len(attempts), method, len(text),
+                    idx, total, method, len(text),
                 )
                 return text, method
 
@@ -371,21 +369,21 @@ class YouTubeTranscriptFetcher:
                 raise
 
             except Exception as e:
-                logger.warning("[%d/%d] All tiers failed: %s", idx, len(attempts), e)
+                logger.warning("[%d/%d] All tiers failed: %s", idx, total, e)
 
-            if idx < len(attempts):
-                next_cfg = attempts[idx][1]
+            if idx < total:
+                next_cfg = attempts[idx]   # idx is 1-based, list is 0-based → next item
                 logger.info(
                     "[%d/%d] Will retry in %s (%s)",
-                    idx, len(attempts),
+                    idx, total,
                     _format_duration(next_cfg["wait_before"]),
                     next_cfg["description"],
                 )
             else:
-                logger.error("All %d polling attempts exhausted.", len(attempts))
+                logger.error("All %d polling attempts exhausted.", total)
 
         raise RuntimeError(
-            f"Transcript not available after {len(attempts)} attempts (~5 hours). "
+            f"Transcript not available after {total} attempts (~3 hours). "
             f"Video ID: {self.video_id}"
         )
 
@@ -409,38 +407,27 @@ class YtDlpTranscriptFetcher:
         self.languages = languages or ["en", "en-US", "en-GB"]
 
     def _find_subtitle_url(self, manual_subs: dict, auto_subs: dict) -> tuple[str, str]:
-        """
-        Search manual subtitles first, then auto-generated, for a
-        matching language + preferred format.
-        If requested languages are not available, fallback to any available language.
-        Returns (url, format_ext).
-        """
-        # 1. Try preferred languages
         for subs_dict in (manual_subs, auto_subs):
             if not subs_dict:
                 continue
             for lang in self.languages:
                 if lang not in subs_dict:
                     continue
-                tracks = subs_dict[lang]  # list of {ext, url, ...}
+                tracks = subs_dict[lang]
                 if not tracks:
                     continue
-                # Try preferred formats in order
                 for fmt in self.PREFERRED_FORMATS:
                     for track in tracks:
                         if track.get("ext") == fmt and track.get("url"):
                             return track["url"], fmt
-                # No preferred format matched — use first available with URL
                 for track in tracks:
                     if track.get("url"):
                         return track["url"], track.get("ext", "vtt")
 
-        # 2. Fallback to ANY available language
         logger.info("[yt-dlp] Preferred languages %s not found. Falling back to any available language.", self.languages)
         for subs_dict in (manual_subs, auto_subs):
             if not subs_dict:
                 continue
-            # Try preferred formats across all languages
             for fmt in self.PREFERRED_FORMATS:
                 for lang, tracks in subs_dict.items():
                     if not tracks:
@@ -449,7 +436,6 @@ class YtDlpTranscriptFetcher:
                         if track.get("ext") == fmt and track.get("url"):
                             logger.info("[yt-dlp] Falling back to language: %s", lang)
                             return track["url"], fmt
-            # No preferred format matched — use first available with URL across all languages
             for lang, tracks in subs_dict.items():
                 if not tracks:
                     continue
@@ -463,11 +449,6 @@ class YtDlpTranscriptFetcher:
         )
 
     def fetch(self) -> str:
-        """
-        Extract subtitle URL from video metadata, fetch content
-        in-memory via HTTP, parse and return as plain text.
-        No files are written to disk.
-        """
         import yt_dlp
         import requests as _requests
 
@@ -504,7 +485,6 @@ class YtDlpTranscriptFetcher:
 
         logger.info("[yt-dlp] Fetching subtitle content (format=%s)", sub_fmt)
 
-        # Fetch subtitle content in-memory
         try:
             resp = _requests.get(sub_url, timeout=30)
             resp.raise_for_status()
@@ -515,13 +495,11 @@ class YtDlpTranscriptFetcher:
         if not raw_content.strip():
             raise RuntimeError("Subtitle URL returned empty content.")
 
-        # Parse based on format
         if sub_fmt in ("vtt",):
             text = _parse_vtt(raw_content)
         elif sub_fmt in ("srt",):
             text = _parse_srt(raw_content)
         else:
-            # For srv1/srv2/srv3/ttml — strip all XML/HTML tags as fallback
             text = re.sub(r"<[^>]+>", "", raw_content)
             text = re.sub(r"\s+", " ", text).strip()
 
@@ -554,10 +532,6 @@ class YouTubeApiTranscriptFetcher:
         self.languages   = languages or ["en", "en-US", "en-GB"]
 
     def fetch(self) -> str:
-        """
-        List caption tracks, find a matching language, and download.
-        Returns plain text transcript.
-        """
         if self.credentials is None:
             raise RuntimeError("No OAuth credentials provided for YouTube API.")
 
@@ -572,7 +546,6 @@ class YouTubeApiTranscriptFetcher:
                 cache_discovery=False,
             )
 
-            # Step 1: List caption tracks
             captions_response = youtube.captions().list(
                 part="snippet",
                 videoId=self.video_id,
@@ -584,13 +557,11 @@ class YouTubeApiTranscriptFetcher:
                     f"No caption tracks found for video {self.video_id}"
                 )
 
-            # Step 2: Find best matching caption track
             caption_id = None
             for lang in self.languages:
                 for item in items:
                     snippet = item.get("snippet", {})
                     if snippet.get("language", "") == lang:
-                        # Prefer non-auto-generated (manual) captions
                         if snippet.get("trackKind") != "ASR":
                             caption_id = item["id"]
                             logger.info(
@@ -601,7 +572,6 @@ class YouTubeApiTranscriptFetcher:
                 if caption_id:
                     break
 
-            # Fallback: accept any track in preferred languages
             if not caption_id:
                 for lang in self.languages:
                     for item in items:
@@ -627,13 +597,11 @@ class YouTubeApiTranscriptFetcher:
                         f"No caption tracks found for video {self.video_id}"
                     )
 
-            # Step 3: Download caption content as SRT
             caption_content = youtube.captions().download(
                 id=caption_id,
                 tfmt="srt",
             ).execute()
 
-            # Response may be bytes or string
             if isinstance(caption_content, bytes):
                 caption_content = caption_content.decode("utf-8")
 
@@ -673,12 +641,10 @@ class GeminiSummarizer:
     """
 
     MAX_RETRIES = 5
-    BASE_WAIT   = 10    # seconds
-    MAX_WAIT    = 120   # seconds cap
+    BASE_WAIT   = 10
+    MAX_WAIT    = 120
 
-    # Errors → retry same model with exponential backoff
     RETRYABLE    = ["503", "502", "500", "UNAVAILABLE", "SERVICE_UNAVAILABLE"]
-    # Errors → skip to next model immediately
     SKIP_TO_NEXT = ["429", "RESOURCE_EXHAUSTED", "quota", "404", "NOT_FOUND"]
 
     def __init__(
@@ -690,10 +656,6 @@ class GeminiSummarizer:
         self.models = models or GEMINI_MODELS
 
     def _call_api(self, transcript: str) -> tuple[str, str]:
-        """
-        Try each model in order with per-model retry + backoff.
-        Returns (full_response_text, model_used).
-        """
         overall_last_error = None
 
         for model in self.models:
@@ -764,10 +726,6 @@ class GeminiSummarizer:
         return full_text.strip(), ""
 
     def run(self, transcript: str) -> tuple[str, str, str]:
-        """
-        Summarize transcript.
-        Returns (summary, qa, model_used) — nothing is written to disk.
-        """
         full, model_used = self._call_api(transcript)
         summary, qa      = self._split(full)
         logger.info("✅ Summarization complete — model: %s", model_used)
@@ -789,7 +747,7 @@ class TranscriptSummaryPipeline:
         self,
         youtube_url: str,
         languages: Optional[List[str]] = None,
-        polling_config: dict           = None,
+        polling_config: list[dict]     = None,   # ← list, not dict
         google_creds                   = None,
     ):
         self.youtube_url = youtube_url

requirements.txt

@@ -7,4 +7,5 @@ requests
 youtube_transcript_api
 google-generativeai
 google-genai
-yt-dlp
+yt-dlp
+innertubex