Spaces:
Paused
Paused
| """Transcript adapter: YouTube captions first, yt-dlp + faster-whisper fallback. | |
| Used for two things: style few-shot exemplars and the clone flow. Both fail | |
| soft — a missing transcript degrades the feature, never crashes a run. | |
| """ | |
| from __future__ import annotations | |
| import tempfile | |
| from pathlib import Path | |
| from typing import Optional | |
| from core.adapters.base import TranscriptAdapter, register | |
| from core.utils import log | |
| class YouTubeTranscript(TranscriptAdapter): | |
| def __init__(self, config): | |
| self.config = config | |
| self.proxy = getattr(config, "yt_proxy", "") or "" | |
| def _proxy_config(self): | |
| """GenericProxyConfig for youtube-transcript-api, or None (direct).""" | |
| if not self.proxy: | |
| return None | |
| try: | |
| from youtube_transcript_api.proxies import GenericProxyConfig | |
| return GenericProxyConfig(http_url=self.proxy, https_url=self.proxy) | |
| except Exception as e: | |
| log.warning("YT_PROXY set but proxy config unavailable (%s) — going direct", e) | |
| return None | |
| def fetch(self, video_id: str) -> Optional[str]: | |
| text = self._via_captions(video_id) | |
| if text: | |
| return text | |
| if not getattr(self.config, "yt_audio_fallback", True): | |
| log.info("No captions for %s — audio fallback disabled, skipping", video_id) | |
| return None | |
| log.info("No captions for %s — trying audio transcription fallback", video_id) | |
| return self._via_whisper(video_id) | |
| def fetch_timed(self, video_id: str) -> Optional[list[tuple[float, str]]]: | |
| """Like ``fetch`` but keeps each line's start time: ``[(start_sec, text)]``. | |
| Captions carry timestamps; the auto-edit uses them to land a momentary | |
| effect on the exciting beat. Captions first, whisper segments as a | |
| fallback, then None — fail soft like ``fetch``. | |
| """ | |
| timed = self._via_captions_timed(video_id) | |
| if timed: | |
| return timed | |
| if not getattr(self.config, "yt_audio_fallback", True): | |
| return None | |
| return self._via_whisper_timed(video_id) | |
| # -- caption API --------------------------------------------------------- | |
| def _via_captions(self, video_id: str) -> Optional[str]: | |
| try: | |
| from youtube_transcript_api import YouTubeTranscriptApi | |
| api = YouTubeTranscriptApi(proxy_config=self._proxy_config()) | |
| fetched = api.fetch(video_id, languages=["en", "en-US", "en-GB"]) | |
| return " ".join(snippet.text.strip() for snippet in fetched if snippet.text.strip()) | |
| except Exception as e: | |
| log.info("youtube-transcript-api failed for %s: %s", video_id, e) | |
| return None | |
| def _via_captions_timed(self, video_id: str) -> Optional[list[tuple[float, str]]]: | |
| try: | |
| from youtube_transcript_api import YouTubeTranscriptApi | |
| api = YouTubeTranscriptApi(proxy_config=self._proxy_config()) | |
| fetched = api.fetch(video_id, languages=["en", "en-US", "en-GB"]) | |
| return [(float(s.start), s.text.strip()) for s in fetched if s.text.strip()] | |
| except Exception as e: | |
| log.info("youtube-transcript-api (timed) failed for %s: %s", video_id, e) | |
| return None | |
| # -- yt-dlp + faster-whisper fallback ------------------------------------- | |
| def _via_whisper(self, video_id: str) -> Optional[str]: | |
| try: | |
| import yt_dlp | |
| from faster_whisper import WhisperModel # optional dep | |
| except ImportError as e: | |
| log.info("Transcription fallback unavailable (%s) — skipping %s", e, video_id) | |
| return None | |
| try: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| out = str(Path(tmp) / "audio.%(ext)s") | |
| ydl_opts = { | |
| "format": "bestaudio/best", | |
| "outtmpl": out, | |
| "quiet": True, | |
| "no_warnings": True, | |
| } | |
| if self.proxy: | |
| ydl_opts["proxy"] = self.proxy | |
| with yt_dlp.YoutubeDL(ydl_opts) as ydl: | |
| ydl.download([f"https://www.youtube.com/watch?v={video_id}"]) | |
| audio = next(Path(tmp).glob("audio.*")) | |
| model = WhisperModel("tiny", device="cpu", compute_type="int8") | |
| segments, _ = model.transcribe(str(audio)) | |
| return " ".join(seg.text.strip() for seg in segments) or None | |
| except Exception as e: | |
| log.warning("Whisper transcription failed for %s: %s", video_id, e) | |
| return None | |
| def _via_whisper_timed(self, video_id: str) -> Optional[list[tuple[float, str]]]: | |
| try: | |
| import yt_dlp | |
| from faster_whisper import WhisperModel # optional dep | |
| except ImportError as e: | |
| log.info("Transcription fallback unavailable (%s) — skipping %s", e, video_id) | |
| return None | |
| try: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| out = str(Path(tmp) / "audio.%(ext)s") | |
| ydl_opts = {"format": "bestaudio/best", "outtmpl": out, | |
| "quiet": True, "no_warnings": True} | |
| if self.proxy: | |
| ydl_opts["proxy"] = self.proxy | |
| with yt_dlp.YoutubeDL(ydl_opts) as ydl: | |
| ydl.download([f"https://www.youtube.com/watch?v={video_id}"]) | |
| audio = next(Path(tmp).glob("audio.*")) | |
| model = WhisperModel("tiny", device="cpu", compute_type="int8") | |
| segments, _ = model.transcribe(str(audio)) | |
| timed = [(float(seg.start), seg.text.strip()) for seg in segments if seg.text.strip()] | |
| return timed or None | |
| except Exception as e: | |
| log.warning("Whisper transcription (timed) failed for %s: %s", video_id, e) | |
| return None | |
| register("transcript", "youtube", YouTubeTranscript) | |