| 1 | """YouTube search and transcript extraction via yt-dlp for the v3.0.0 pipeline. |
| 2 | |
| 3 | Uses yt-dlp (https://github.com/yt-dlp/yt-dlp) for both YouTube search and |
| 4 | transcript extraction. No API keys needed — just have yt-dlp installed. |
| 5 | |
| 6 | Inspired by Peter Steinberger's toolchain approach (yt-dlp + summarize CLI). |
| 7 | """ |
| 8 | |
| 9 | import copy |
| 10 | import json |
| 11 | import math |
| 12 | import os |
| 13 | import re |
| 14 | import shlex |
| 15 | import shutil |
| 16 | import sys |
| 17 | import tempfile |
| 18 | import threading |
| 19 | import time |
| 20 | import urllib.error |
| 21 | import urllib.request |
| 22 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 23 | from pathlib import Path |
| 24 | from typing import Any, Dict, List, Optional, Set, Tuple |
| 25 | |
| 26 | # Depth configurations: how many videos to search / transcribe |
| 27 | DEPTH_CONFIG = { |
| 28 | "quick": 6, |
| 29 | "default": 8, |
| 30 | "deep": 40, |
| 31 | } |
| 32 | |
| 33 | TRANSCRIPT_LIMITS = { |
| 34 | "quick": 0, |
| 35 | "default": 2, |
| 36 | "deep": 8, |
| 37 | } |
| 38 | |
| 39 | # Cumulative yt-dlp transcript-fetch stats for the current process. The final |
| 40 | # report only sees post-pruning items, so it can't distinguish "fetches failed |
| 41 | # (stale binary)" from "fetches succeeded but the videos were pruned later". |
| 42 | # quality_nudge reads these via last30days.py to suppress the stale-yt-dlp |
| 43 | # nudge when every attempted fetch actually succeeded. yt-dlp path only: the |
| 44 | # nudge diagnoses the local binary, not the ScrapeCreators API. |
| 45 | _TRANSCRIPT_FETCH_STATS = {"attempts": 0, "failures": 0} |
| 46 | |
| 47 | |
| 48 | def get_transcript_fetch_stats() -> Dict[str, int]: |
| 49 | """Return cumulative transcript-fetch stats for this process.""" |
| 50 | return dict(_TRANSCRIPT_FETCH_STATS) |
| 51 | |
| 52 | |
| 53 | def reset_transcript_fetch_stats() -> None: |
| 54 | """Reset cumulative transcript-fetch stats (used by tests).""" |
| 55 | _TRANSCRIPT_FETCH_STATS["attempts"] = 0 |
| 56 | _TRANSCRIPT_FETCH_STATS["failures"] = 0 |
| 57 | |
| 58 | # Max words to keep from each transcript |
| 59 | TRANSCRIPT_MAX_WORDS = 5000 |
| 60 | |
| 61 | from . import dates, health, http, log, subproc |
| 62 | from .query import infer_query_intent |
| 63 | |
| 64 | from .relevance import token_overlap_relevance as _compute_relevance |
| 65 | |
| 66 | # yt-dlp transcript-fetch resilience. A non-zero yt-dlp exit means a real fetch |
| 67 | # error (rate-limit / bot-check / network), NOT "no captions" — yt-dlp exits 0 |
| 68 | # with no file for a video that genuinely lacks the requested captions. So we |
| 69 | # capture the returncode, log a classified reason instead of failing silently, |
| 70 | # and retry transient errors a couple of times with a small per-video staggered |
| 71 | # backoff. |
| 72 | _TRANSCRIPT_MAX_RETRIES = 2 |
| 73 | _TRANSCRIPT_BACKOFF_BASE = 2.0 # seconds; multiplied by (attempt + 1) |
| 74 | _TRANSCRIPT_TIMEOUT = 30 # seconds per yt-dlp attempt (keyless: no fallback to fail over to) |
| 75 | _TRANSCRIPT_FAST_TIMEOUT = 12 # seconds per attempt when a ScrapeCreators fallback exists |
| 76 | _SEARCH_TIMEOUT = 120 # seconds per ytsearch metadata extraction |
| 77 | # Comparison-mode fan-out (and nested transcript/comment pools) can stampede the |
| 78 | # same throttled YouTube IP. Cap concurrent yt-dlp processes process-wide. |
| 79 | _YTDLP_MAX_CONCURRENT = 2 |
| 80 | _ytdlp_slots = threading.Semaphore(_YTDLP_MAX_CONCURRENT) |
| 81 | # In-run search cache: comparison mode re-issues identical ytsearch queries from |
| 82 | # every entity sub-run; cache hits avoid the redundant expensive --dump-json work. |
| 83 | # Inflight coalescing prevents N concurrent identical searches from all missing |
| 84 | # the cache and stampeding YouTube together. |
| 85 | _search_cache: Dict[Tuple[str, int, str], Dict[str, Any]] = {} |
| 86 | _search_inflight: Dict[Tuple[str, int, str], tuple[threading.Event, list]] = {} |
| 87 | _search_cache_lock = threading.Lock() |
| 88 | # Comments are enrichment, not core evidence: keep the budget tight so a slow |
| 89 | # comment API can never dominate a run's wall clock (bounded to 3 videos). |
| 90 | _COMMENT_TIMEOUT = 20 |
| 91 | _SC_LOW_CREDIT_THRESHOLD = 50 # warn once ScrapeCreators credits drop below this |
| 92 | # Transient = worth retrying (and definitely not "no captions"). |
| 93 | _TRANSIENT_RE = re.compile( |
| 94 | r"429|too many requests|sign in to confirm|not a bot|rate.?limit" |
| 95 | r"|temporarily|try again|timed out|timeout|connection|unable to (extract|download)" |
| 96 | r"|failed to (extract|download)|got error|read error", |
| 97 | re.IGNORECASE, |
| 98 | ) |
| 99 | # A genuine no-captions signal — treat as no captions, never retry/surface. |
| 100 | _NO_CAPTION_RE = re.compile( |
| 101 | r"no subtitles|requested (format|language)|there'?s no .*subtitles", |
| 102 | re.IGNORECASE, |
| 103 | ) |
| 104 | |
| 105 | |
| 106 | def extract_transcript_highlights(transcript: str, topic: str, limit: int = 5) -> list[str]: |
| 107 | """Extract quotable highlights from a YouTube transcript. |
| 108 | |
| 109 | Filters filler (subscribe, welcome back, etc.), scores sentences by |
| 110 | specificity (numbers, proper nouns, topic relevance), and returns |
| 111 | the top highlights. |
| 112 | """ |
| 113 | if not transcript: |
| 114 | return [] |
| 115 | |
| 116 | sentences = re.split(r'(?<=[.!?])\s+', transcript) |
| 117 | |
| 118 | # Fallback for punctuation-free transcripts (common with auto-captions): |
| 119 | # chunk into ~20-word segments so they pass the 8-50 word filter. |
| 120 | if len(sentences) <= 1 and len(transcript.split()) > 50: |
| 121 | words = transcript.split() |
| 122 | sentences = [' '.join(words[i:i+20]) for i in range(0, len(words), 20)] |
| 123 | |
| 124 | filler = [ |
| 125 | r"^(hey |hi |what's up|welcome back|in today's video|don't forget to)", |
| 126 | r"(subscribe|like and comment|hit the bell|check out the link|down below)", |
| 127 | r"^(so |and |but |okay |alright |um |uh )", |
| 128 | r"(thanks for watching|see you (next|in the)|bye)", |
| 129 | ] |
| 130 | |
| 131 | topic_words = [w.lower() for w in topic.lower().split() if len(w) > 2] |
| 132 | |
| 133 | candidates = [] |
| 134 | for sent in sentences: |
| 135 | sent = sent.strip() |
| 136 | words = sent.split() |
| 137 | if len(words) < 8 or len(words) > 50: |
| 138 | continue |
| 139 | if any(re.search(p, sent, re.IGNORECASE) for p in filler): |
| 140 | continue |
| 141 | |
| 142 | score = 0 |
| 143 | if re.search(r'\d', sent): |
| 144 | score += 2 |
| 145 | if re.search(r'[A-Z][a-z]+', sent): |
| 146 | score += 1 |
| 147 | if '?' in sent: |
| 148 | score += 1 |
| 149 | sent_lower = sent.lower() |
| 150 | if any(w in sent_lower for w in topic_words): |
| 151 | score += 2 |
| 152 | |
| 153 | candidates.append((score, sent)) |
| 154 | |
| 155 | candidates.sort(key=lambda x: -x[0]) |
| 156 | return [sent for _, sent in candidates[:limit]] |
| 157 | |
| 158 | |
| 159 | def _log(msg: str): |
| 160 | log.source_log("YouTube", msg, tty_only=False) |
| 161 | |
| 162 | |
| 163 | def reset_search_cache() -> None: |
| 164 | """Clear the in-run ytsearch cache. |
| 165 | |
| 166 | Call at the start of each top-level research run so a long-lived process |
| 167 | (agent host, REPL, test suite) does not reuse results across runs. Within |
| 168 | one comparison fan-out the cache stays hot so identical queries coalesce. |
| 169 | """ |
| 170 | with _search_cache_lock: |
| 171 | _search_cache.clear() |
| 172 | _search_inflight.clear() |
| 173 | |
| 174 | |
| 175 | def _env_positive_float(name: str, default: float) -> float: |
| 176 | """Read a positive finite float from the environment, else ``default``.""" |
| 177 | raw = os.environ.get(name, "").strip() |
| 178 | try: |
| 179 | value = float(raw) if raw else float(default) |
| 180 | except ValueError: |
| 181 | return float(default) |
| 182 | if not math.isfinite(value) or value <= 0: |
| 183 | return float(default) |
| 184 | return value |
| 185 | |
| 186 | |
| 187 | def _search_timeout() -> float: |
| 188 | """Return the ytsearch timeout, preserving the 120s default.""" |
| 189 | return _env_positive_float("LAST30DAYS_YT_SEARCH_TIMEOUT", float(_SEARCH_TIMEOUT)) |
| 190 | |
| 191 | |
| 192 | def _run_ytdlp(cmd: List[str], *, timeout: float) -> subproc.SubprocResult: |
| 193 | """Run a yt-dlp (or SSH-wrapped) command under the process-wide concurrency gate.""" |
| 194 | with _ytdlp_slots: |
| 195 | return subproc.run_with_timeout(cmd, timeout=timeout) |
| 196 | |
| 197 | |
| 198 | def _claim_search_slot( |
| 199 | cache_key: Tuple[str, int, str], |
| 200 | ) -> tuple[Optional[Dict[str, Any]], Optional[threading.Event], Optional[list], bool]: |
| 201 | """Return ``(cached, event, slot, is_leader)`` for search coalesce. |
| 202 | |
| 203 | - Cache hit: ``(payload, None, None, False)`` — caller returns ``payload``. |
| 204 | - Waiter: ``(None, event, slot, False)`` — caller awaits ``slot`` via ``event``. |
| 205 | - Leader: ``(None, event, slot, True)`` — caller runs yt-dlp and finishes the slot. |
| 206 | """ |
| 207 | with _search_cache_lock: |
| 208 | cached = _search_cache.get(cache_key) |
| 209 | if cached is not None: |
| 210 | return copy.deepcopy(cached), None, None, False |
| 211 | existing = _search_inflight.get(cache_key) |
| 212 | if existing is not None: |
| 213 | return None, existing[0], existing[1], False |
| 214 | event = threading.Event() |
| 215 | slot: list = [None] |
| 216 | _search_inflight[cache_key] = (event, slot) |
| 217 | return None, event, slot, True |
| 218 | |
| 219 | |
| 220 | def _finish_search_slot( |
| 221 | cache_key: Tuple[str, int, str], |
| 222 | payload: Dict[str, Any], |
| 223 | *, |
| 224 | event: threading.Event, |
| 225 | slot: list, |
| 226 | ) -> Dict[str, Any]: |
| 227 | """Publish a search result to waiters; cache only clean (non-error) payloads. |
| 228 | |
| 229 | Ownership is by slot identity: after ``reset_search_cache()`` clears the |
| 230 | registry, a stale leader must still wake its own waiters but must not pop |
| 231 | or overwrite a newer run's registration for the same key. |
| 232 | """ |
| 233 | shared = copy.deepcopy(payload) |
| 234 | with _search_cache_lock: |
| 235 | if slot[0] is not None: |
| 236 | # Idempotent re-finish of this slot (e.g. finally after return). |
| 237 | event.set() |
| 238 | return payload |
| 239 | slot[0] = shared |
| 240 | current = _search_inflight.get(cache_key) |
| 241 | if current is not None and current[1] is slot: |
| 242 | if not payload.get("error"): |
| 243 | _search_cache[cache_key] = shared |
| 244 | _search_inflight.pop(cache_key, None) |
| 245 | # else: stale leader after a reset — wake local waiters only. |
| 246 | event.set() |
| 247 | return payload |
| 248 | |
| 249 | |
| 250 | def _await_search_slot( |
| 251 | event: threading.Event, |
| 252 | slot: list, |
| 253 | ) -> Dict[str, Any]: |
| 254 | """Wait for a leader search to publish. |
| 255 | |
| 256 | Waiters block until the leader finishes (success or failure). The leader |
| 257 | path always publishes via ``_finish_search_slot``, including on unexpected |
| 258 | exceptions, so a timed wait would only invent a false timeout while the |
| 259 | leader was still queued behind other yt-dlp work. |
| 260 | """ |
| 261 | event.wait() |
| 262 | shared = slot[0] |
| 263 | if isinstance(shared, dict): |
| 264 | return copy.deepcopy(shared) |
| 265 | return {"items": [], "error": "YouTube search failed"} |
| 266 | |
| 267 | |
| 268 | def classify_run_failure(detail: str) -> str: |
| 269 | """Map yt-dlp's text-only throttling and bot-gate errors.""" |
| 270 | text = detail.lower() |
| 271 | if any(marker in text for marker in ("yt-dlp not installed", "yt-dlp not found")): |
| 272 | return health.SKIPPED_UNCONFIGURED |
| 273 | if any(marker in text for marker in ("timed out", "timeout")): |
| 274 | return health.TIMEOUT |
| 275 | if any( |
| 276 | marker in text |
| 277 | for marker in ("http error 429", "confirm you're not a bot", "confirm you’re not a bot", "bot-gate") |
| 278 | ): |
| 279 | return health.RATE_LIMITED |
| 280 | if any(marker in text for marker in ("sign in", "login required", "cookies are no longer valid")): |
| 281 | return health.AUTH_FAILED |
| 282 | return http.classify_failure(message=detail) |
| 283 | |
| 284 | |
| 285 | def is_ytdlp_installed() -> bool: |
| 286 | """Check if yt-dlp is available locally, or if SSH routing is configured. |
| 287 | |
| 288 | When LAST30DAYS_YOUTUBE_SSH_HOST is set, returns True without a local check — |
| 289 | yt-dlp lives on the remote host. Failures surface naturally on first use. |
| 290 | """ |
| 291 | if _ytdlp_ssh_host(): |
| 292 | return True |
| 293 | return shutil.which("yt-dlp") is not None |
| 294 | |
| 295 | |
| 296 | # Host aliases must be plain hostnames / SSH config aliases — no flags, no |
| 297 | # shell metacharacters. Rejects any value that could be reinterpreted by ssh |
| 298 | # (or the surrounding shell) as something other than a destination. |
| 299 | _SSH_HOST_ALIAS_RE = re.compile(r"^[a-zA-Z0-9._-]+$") |
| 300 | |
| 301 | |
| 302 | def _ytdlp_ssh_host() -> Optional[str]: |
| 303 | """Return SSH host alias if yt-dlp should be routed via SSH, else None. |
| 304 | |
| 305 | Set LAST30DAYS_YOUTUBE_SSH_HOST=<ssh-alias> (e.g. 'macmini') in the environment |
| 306 | to route yt-dlp through SSH for residential IP egress. This bypasses |
| 307 | YouTube's bot-wall on datacenter IPs (Hetzner, DigitalOcean, AWS, etc.) |
| 308 | where ytsearch returns 0 results regardless of cookies. |
| 309 | |
| 310 | The remote host must have yt-dlp installed and reachable via the named |
| 311 | SSH alias (configured in ~/.ssh/config). On macOS hosts with Homebrew, |
| 312 | add brew shellenv to ~/.zshenv (not just ~/.zprofile) so non-login SSH |
| 313 | shells find yt-dlp on PATH. |
| 314 | |
| 315 | Validation: host value must match ``[A-Za-z0-9._-]+``. Anything starting |
| 316 | with ``-`` or containing shell/SSH metacharacters is rejected with a |
| 317 | stderr warning and treated as unset, so a misconfigured or attacker- |
| 318 | controlled value can't slip through as an SSH option flag or proxy command. |
| 319 | The ``--`` option terminator in ``_wrap_ytdlp_cmd`` is a second line of |
| 320 | defense; this regex closes the door on the env var ever reaching ssh |
| 321 | in the first place. |
| 322 | |
| 323 | To use a value from ~/.config/last30days/.env, export it into the |
| 324 | environment before invoking the engine, e.g. in a wrapper: |
| 325 | set -a; source ~/.config/last30days/.env; set +a |
| 326 | python3 last30days.py "..." |
| 327 | """ |
| 328 | host = os.environ.get("LAST30DAYS_YOUTUBE_SSH_HOST", "").strip() |
| 329 | if not host: |
| 330 | return None |
| 331 | if not _SSH_HOST_ALIAS_RE.match(host): |
| 332 | sys.stderr.write( |
| 333 | f"[youtube_yt] WARNING: LAST30DAYS_YOUTUBE_SSH_HOST={host!r} " |
| 334 | "does not look like a plain hostname/alias; ignoring. " |
| 335 | "Expected pattern: letters, digits, dot, underscore, hyphen.\n" |
| 336 | ) |
| 337 | return None |
| 338 | return host |
| 339 | |
| 340 | |
| 341 | _PLAYER_CLIENT_RE = re.compile(r"^[A-Za-z0-9_-]+$") |
| 342 | |
| 343 | |
| 344 | def _ytdlp_player_client() -> Optional[str]: |
| 345 | """Return the yt-dlp YouTube player_client, or None to leave the default. |
| 346 | |
| 347 | Default is ``android``, which bypasses the web bot-gate without cookies. |
| 348 | Set ``LAST30DAYS_YT_PLAYER_CLIENT`` empty to disable; any other value is |
| 349 | passed through when it is a safe extractor token. |
| 350 | """ |
| 351 | if "LAST30DAYS_YT_PLAYER_CLIENT" in os.environ: |
| 352 | raw = os.environ.get("LAST30DAYS_YT_PLAYER_CLIENT", "").strip() |
| 353 | if not raw: |
| 354 | return None |
| 355 | else: |
| 356 | raw = "android" |
| 357 | if not _PLAYER_CLIENT_RE.match(raw): |
| 358 | sys.stderr.write( |
| 359 | f"[youtube_yt] WARNING: LAST30DAYS_YT_PLAYER_CLIENT={raw!r} " |
| 360 | "is not a plain player-client token; ignoring.\n" |
| 361 | ) |
| 362 | return None |
| 363 | return raw |
| 364 | |
| 365 | |
| 366 | def _ytdlp_cmd_needs_player_client(cmd: List[str]) -> bool: |
| 367 | blob = " ".join(cmd) |
| 368 | return any( |
| 369 | marker in blob |
| 370 | for marker in ( |
| 371 | "ytsearch", |
| 372 | "youtube.com", |
| 373 | "--write-comments", |
| 374 | "--write-auto-subs", |
| 375 | ) |
| 376 | ) |
| 377 | |
| 378 | |
| 379 | def _inject_youtube_player_client(cmd: List[str]) -> List[str]: |
| 380 | """Merge player_client into a single youtube --extractor-args (#1052). |
| 381 | |
| 382 | yt-dlp does not merge two ``--extractor-args`` for the same extractor; |
| 383 | the last one wins. Always fold into an existing ``youtube:`` spec. |
| 384 | """ |
| 385 | client = _ytdlp_player_client() |
| 386 | if not client or not _ytdlp_cmd_needs_player_client(cmd): |
| 387 | return list(cmd) |
| 388 | out = list(cmd) |
| 389 | needle = f"player_client={client}" |
| 390 | for i, arg in enumerate(out): |
| 391 | if arg == "--extractor-args" and i + 1 < len(out): |
| 392 | spec = out[i + 1] |
| 393 | if spec.startswith("youtube:"): |
| 394 | if "player_client=" in spec: |
| 395 | return out |
| 396 | out[i + 1] = f"{spec};{needle}" |
| 397 | return out |
| 398 | out.extend(["--extractor-args", f"youtube:{needle}"]) |
| 399 | return out |
| 400 | |
| 401 | |
| 402 | def _wrap_ytdlp_cmd(cmd: List[str]) -> List[str]: |
| 403 | """Wrap a yt-dlp command list with `ssh <host>` when SSH routing is set. |
| 404 | |
| 405 | Args are shell-quoted to survive the remote shell. Uses BatchMode=yes so |
| 406 | a misconfigured key fails fast instead of hanging on a password prompt. |
| 407 | The `--` option terminator prevents an SSH option-injection if |
| 408 | LAST30DAYS_YOUTUBE_SSH_HOST were ever set to a value starting with `-`. |
| 409 | """ |
| 410 | cmd = _inject_youtube_player_client(cmd) |
| 411 | host = _ytdlp_ssh_host() |
| 412 | if not host: |
| 413 | return cmd |
| 414 | remote_cmd = " ".join(shlex.quote(a) for a in cmd) |
| 415 | return ["ssh", "-o", "BatchMode=yes", "--", host, remote_cmd] |
| 416 | |
| 417 | |
| 418 | def _extract_core_subject(topic: str) -> str: |
| 419 | """Extract core subject from verbose query for YouTube search. |
| 420 | |
| 421 | NOTE: 'tips', 'tricks', 'tutorial', 'guide', 'review', 'reviews' |
| 422 | are intentionally KEPT — they're YouTube content types that improve search. |
| 423 | """ |
| 424 | from .query import VIRAL_NOISE, extract_core_subject |
| 425 | # YouTube extends VIRAL_NOISE with temporal/meta words the planner emits |
| 426 | # that don't appear in YouTube titles (months, recent year tokens, etc.). |
| 427 | _YT_EXTRA = frozenset({ |
| 428 | 'last', 'days', 'recent', 'recently', 'month', 'week', |
| 429 | 'january', 'february', 'march', 'april', 'may', 'june', |
| 430 | 'july', 'august', 'september', 'october', 'november', 'december', |
| 431 | '2025', '2026', '2027', |
| 432 | 'music', 'public', 'appearances', 'developments', 'discussions', 'coverage', |
| 433 | }) |
| 434 | return extract_core_subject(topic, noise=VIRAL_NOISE | _YT_EXTRA) |
| 435 | |
| 436 | |
| 437 | def expand_youtube_queries(topic: str, depth: str) -> List[str]: |
| 438 | """Generate multiple YouTube search queries from a topic. |
| 439 | |
| 440 | Mirrors reddit.py's expand_reddit_queries() pattern: |
| 441 | 1. Extract core subject (strip noise words) |
| 442 | 2. Include original topic if different from core |
| 443 | 3. Add intent-specific OR-joined content-type variants |
| 444 | 4. Cap by depth: 1 for quick, 2 for default, 3 for deep |
| 445 | |
| 446 | Returns 1-3 query strings depending on depth. |
| 447 | """ |
| 448 | core = _extract_core_subject(topic) |
| 449 | queries = [core] |
| 450 | |
| 451 | # Include cleaned original topic as variant if different from core |
| 452 | original_clean = topic.strip().rstrip('?!.') |
| 453 | if core.lower() != original_clean.lower() and len(original_clean.split()) <= 8: |
| 454 | queries.append(original_clean) |
| 455 | |
| 456 | qtype = infer_query_intent(topic) |
| 457 | |
| 458 | # Intent-specific YouTube content-type variants |
| 459 | if qtype == "opinion": |
| 460 | queries.append(f"{core} review OR reaction OR breakdown") |
| 461 | elif qtype == "product": |
| 462 | queries.append(f"{core} review OR comparison OR unboxing") |
| 463 | elif qtype == "comparison": |
| 464 | queries.append(f"{core} vs OR compared OR head to head") |
| 465 | elif qtype == "how_to": |
| 466 | queries.append(f"{core} tutorial OR guide OR explained") |
| 467 | else: |
| 468 | # breaking_news / general — YouTube content types |
| 469 | queries.append(f"{core} review OR reaction OR breakdown") |
| 470 | |
| 471 | # Deep depth: add full-length content variant |
| 472 | if depth == "deep": |
| 473 | queries.append(f"{core} full OR complete OR official") |
| 474 | |
| 475 | # Cap by depth budget |
| 476 | caps = {"quick": 1, "default": 2, "deep": 3} |
| 477 | cap = caps.get(depth, 2) |
| 478 | return queries[:cap] |
| 479 | |
| 480 | |
| 481 | def search_youtube( |
| 482 | topic: str, |
| 483 | from_date: str, |
| 484 | to_date: str, |
| 485 | depth: str = "default", |
| 486 | ) -> Dict[str, Any]: |
| 487 | """Search YouTube via yt-dlp. No API key needed. |
| 488 | |
| 489 | Args: |
| 490 | topic: Search topic |
| 491 | from_date: Start date (YYYY-MM-DD) |
| 492 | to_date: End date (YYYY-MM-DD) |
| 493 | depth: 'quick', 'default', or 'deep' |
| 494 | |
| 495 | Returns: |
| 496 | Dict with 'items' list of video metadata dicts. |
| 497 | """ |
| 498 | if not is_ytdlp_installed(): |
| 499 | return {"items": [], "error": "yt-dlp not installed"} |
| 500 | |
| 501 | count = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 502 | core_topic = _extract_core_subject(topic) |
| 503 | cache_key = (core_topic, count, from_date) |
| 504 | timeout = _search_timeout() |
| 505 | |
| 506 | cached, event, slot, is_leader = _claim_search_slot(cache_key) |
| 507 | if cached is not None: |
| 508 | _log(f"YouTube search cache hit for '{core_topic}' (count={count})") |
| 509 | return cached |
| 510 | assert event is not None and slot is not None |
| 511 | if not is_leader: |
| 512 | _log(f"YouTube search awaiting in-flight query for '{core_topic}'") |
| 513 | return _await_search_slot(event, slot) |
| 514 | |
| 515 | def _publish(payload: Dict[str, Any]) -> Dict[str, Any]: |
| 516 | return _finish_search_slot(cache_key, payload, event=event, slot=slot) |
| 517 | |
| 518 | _log(f"Searching YouTube for '{core_topic}' (since {from_date}, count={count})") |
| 519 | |
| 520 | # yt-dlp search with full metadata (no --flat-playlist so dates are real). |
| 521 | # NOTE: --dateafter intentionally omitted — YouTube search returns |
| 522 | # relevance-sorted results and strict date filtering returns 0 for |
| 523 | # evergreen topics. Python soft filter (below) handles date filtering. |
| 524 | cmd = [ |
| 525 | "yt-dlp", |
| 526 | "--ignore-config", |
| 527 | "--no-cookies-from-browser", |
| 528 | f"ytsearch{count}:{core_topic}", |
| 529 | "--dump-json", |
| 530 | "--no-warnings", |
| 531 | "--no-download", |
| 532 | ] |
| 533 | cmd = _wrap_ytdlp_cmd(cmd) |
| 534 | ssh_host = _ytdlp_ssh_host() |
| 535 | |
| 536 | published: Dict[str, Any] | None = None |
| 537 | try: |
| 538 | try: |
| 539 | result = _run_ytdlp(cmd, timeout=timeout) |
| 540 | except subproc.SubprocTimeout: |
| 541 | _log(f"YouTube search timed out ({timeout:g}s)") |
| 542 | published = _publish( |
| 543 | {"items": [], "error": f"Search timed out after {timeout:g}s"} |
| 544 | ) |
| 545 | return published |
| 546 | except FileNotFoundError: |
| 547 | published = _publish({"items": [], "error": "yt-dlp not found"}) |
| 548 | return published |
| 549 | |
| 550 | stdout = result.stdout |
| 551 | if ssh_host and result.returncode != 0 and not stdout.strip(): |
| 552 | stderr_first = (result.stderr or "").strip().splitlines() |
| 553 | first_line = stderr_first[0] if stderr_first else "(no stderr)" |
| 554 | _log( |
| 555 | f"YouTube search via SSH host {ssh_host!r} failed " |
| 556 | f"(rc={result.returncode}): {first_line}" |
| 557 | ) |
| 558 | published = _publish( |
| 559 | {"items": [], "error": f"SSH routing to {ssh_host!r} failed: {first_line}"}, |
| 560 | ) |
| 561 | return published |
| 562 | if not stdout.strip(): |
| 563 | _log("YouTube search returned 0 results") |
| 564 | published = _publish({"items": []}) |
| 565 | return published |
| 566 | |
| 567 | # Parse JSON-per-line output |
| 568 | items = [] |
| 569 | for line in stdout.strip().split("\n"): |
| 570 | line = line.strip() |
| 571 | if not line: |
| 572 | continue |
| 573 | try: |
| 574 | video = json.loads(line) |
| 575 | except json.JSONDecodeError: |
| 576 | continue |
| 577 | |
| 578 | video_id = video.get("id", "") |
| 579 | view_count = video.get("view_count") if video.get("view_count") is not None else 0 |
| 580 | like_count = video.get("like_count") if video.get("like_count") is not None else 0 |
| 581 | comment_count = video.get("comment_count") if video.get("comment_count") is not None else 0 |
| 582 | upload_date = video.get("upload_date", "") # YYYYMMDD |
| 583 | |
| 584 | # Convert YYYYMMDD to YYYY-MM-DD |
| 585 | date_str = None |
| 586 | if upload_date and len(upload_date) == 8: |
| 587 | date_str = f"{upload_date[:4]}-{upload_date[4:6]}-{upload_date[6:8]}" |
| 588 | |
| 589 | description = str(video.get("description", ""))[:500] |
| 590 | items.append({ |
| 591 | "video_id": video_id, |
| 592 | "title": video.get("title", ""), |
| 593 | "url": f"https://www.youtube.com/watch?v={video_id}", |
| 594 | "channel_name": video.get("channel", video.get("uploader", "")), |
| 595 | "date": date_str, |
| 596 | "engagement": { |
| 597 | "views": view_count, |
| 598 | "likes": like_count, |
| 599 | "comments": comment_count, |
| 600 | }, |
| 601 | "duration": video.get("duration"), |
| 602 | "relevance": _compute_relevance(core_topic, f"{video.get('title', '')} {description}"), |
| 603 | "why_relevant": f"YouTube: {video.get('title', core_topic)[:60]}", |
| 604 | "description": description, |
| 605 | }) |
| 606 | |
| 607 | # Soft date filter: prefer recent items but fall back to all if too few |
| 608 | recent = [i for i in items if i["date"] and i["date"] >= from_date] |
| 609 | if len(recent) >= 3: |
| 610 | items = recent |
| 611 | _log(f"Found {len(items)} videos within date range") |
| 612 | else: |
| 613 | _log(f"Found {len(items)} videos ({len(recent)} within date range, keeping all)") |
| 614 | |
| 615 | # Sort by views descending |
| 616 | items.sort(key=lambda x: x["engagement"]["views"], reverse=True) |
| 617 | published = _publish({"items": items}) |
| 618 | return published |
| 619 | except Exception as exc: |
| 620 | # Post-subprocess failures (parse/relevance/sort) must still unblock |
| 621 | # coalesced waiters — otherwise the inflight key orphans forever. |
| 622 | published = _publish({"items": [], "error": str(exc)}) |
| 623 | return published |
| 624 | finally: |
| 625 | if published is None: |
| 626 | _publish({"items": [], "error": "YouTube search failed"}) |
| 627 | |
| 628 | |
| 629 | def _clean_vtt(vtt_text: str) -> str: |
| 630 | """Convert VTT subtitle format to clean plaintext.""" |
| 631 | # Strip VTT header |
| 632 | text = re.sub(r'^WEBVTT.*?\n\n', '', vtt_text, flags=re.DOTALL) |
| 633 | # Strip timestamps |
| 634 | text = re.sub(r'\d{2}:\d{2}:\d{2}\.\d{3}\s*-->\s*\d{2}:\d{2}:\d{2}\.\d{3}.*\n', '', text) |
| 635 | # Strip position/alignment tags |
| 636 | text = re.sub(r'<[^>]+>', '', text) |
| 637 | # Strip cue numbers |
| 638 | text = re.sub(r'^\d+\s*$', '', text, flags=re.MULTILINE) |
| 639 | # Deduplicate overlapping lines |
| 640 | lines = text.strip().split('\n') |
| 641 | seen = set() |
| 642 | unique = [] |
| 643 | for line in lines: |
| 644 | stripped = line.strip() |
| 645 | if stripped and stripped not in seen: |
| 646 | seen.add(stripped) |
| 647 | unique.append(stripped) |
| 648 | return re.sub(r'\s+', ' ', ' '.join(unique)).strip() |
| 649 | |
| 650 | |
| 651 | _YT_USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" |
| 652 | |
| 653 | |
| 654 | def _fetch_transcript_direct( |
| 655 | video_id: str, |
| 656 | timeout: int = 30, |
| 657 | status: Optional[Dict[str, Any]] = None, |
| 658 | ) -> Optional[str]: |
| 659 | """Fetch YouTube transcript via direct HTTP without yt-dlp. |
| 660 | |
| 661 | Scrapes the watch page HTML for the captions track URL in |
| 662 | ytInitialPlayerResponse, then fetches the VTT subtitle file. |
| 663 | |
| 664 | Args: |
| 665 | video_id: YouTube video ID |
| 666 | timeout: HTTP request timeout in seconds |
| 667 | status: Optional dict mutated to record per-video signals. Sets |
| 668 | ``status["no_caption_tracks"] = True`` when the player response |
| 669 | confirms the uploader has no caption tracks (vs. fetch failure). |
| 670 | |
| 671 | Returns: |
| 672 | Raw VTT text, or None if captions are unavailable. |
| 673 | """ |
| 674 | watch_url = f"https://www.youtube.com/watch?v={video_id}" |
| 675 | headers = { |
| 676 | "User-Agent": _YT_USER_AGENT, |
| 677 | "Accept-Language": "en-US,en;q=0.9", |
| 678 | } |
| 679 | |
| 680 | # Step 1: Fetch the watch page HTML |
| 681 | req = urllib.request.Request(watch_url, headers=headers) |
| 682 | try: |
| 683 | with urllib.request.urlopen(req, timeout=timeout) as resp: |
| 684 | html = resp.read().decode("utf-8", errors="replace") |
| 685 | except (urllib.error.URLError, urllib.error.HTTPError, OSError, TimeoutError) as exc: |
| 686 | _log(f"Direct transcript: failed to fetch watch page for {video_id}: {exc}") |
| 687 | return None |
| 688 | |
| 689 | # Step 2: Extract captions URL from ytInitialPlayerResponse |
| 690 | # YouTube embeds this as a JS variable in the page HTML |
| 691 | match = re.search( |
| 692 | r'ytInitialPlayerResponse\s*=\s*(\{.+?\})\s*;(?:\s*var\s|\s*<\/script>)', |
| 693 | html, |
| 694 | ) |
| 695 | if not match: |
| 696 | # Fallback: try the JSON embedded in the script tag |
| 697 | match = re.search( |
| 698 | r'var\s+ytInitialPlayerResponse\s*=\s*(\{.+?\})\s*;', |
| 699 | html, |
| 700 | ) |
| 701 | if not match: |
| 702 | _log(f"Direct transcript: no ytInitialPlayerResponse found for {video_id}") |
| 703 | return None |
| 704 | |
| 705 | try: |
| 706 | player_response = json.loads(match.group(1)) |
| 707 | except json.JSONDecodeError: |
| 708 | _log(f"Direct transcript: failed to parse ytInitialPlayerResponse for {video_id}") |
| 709 | return None |
| 710 | |
| 711 | # Navigate to caption tracks |
| 712 | captions = player_response.get("captions", {}) |
| 713 | renderer = captions.get("playerCaptionsTracklistRenderer", {}) |
| 714 | caption_tracks = renderer.get("captionTracks", []) |
| 715 | |
| 716 | if not caption_tracks: |
| 717 | _log(f"Direct transcript: no caption tracks for {video_id}") |
| 718 | if status is not None: |
| 719 | status["no_caption_tracks"] = True |
| 720 | return None |
| 721 | |
| 722 | # Find English track (prefer exact 'en', then any en variant, then first track) |
| 723 | base_url = None |
| 724 | for track in caption_tracks: |
| 725 | lang = track.get("languageCode", "") |
| 726 | if lang == "en": |
| 727 | base_url = track.get("baseUrl") |
| 728 | break |
| 729 | if not base_url: |
| 730 | for track in caption_tracks: |
| 731 | lang = track.get("languageCode", "") |
| 732 | if lang.startswith("en"): |
| 733 | base_url = track.get("baseUrl") |
| 734 | break |
| 735 | if not base_url: |
| 736 | # Fall back to first available track |
| 737 | base_url = caption_tracks[0].get("baseUrl") |
| 738 | if not base_url: |
| 739 | _log(f"Direct transcript: no baseUrl in caption tracks for {video_id}") |
| 740 | return None |
| 741 | |
| 742 | # Step 3: Fetch the VTT subtitle file |
| 743 | sep = "&" if "?" in base_url else "?" |
| 744 | vtt_url = f"{base_url}{sep}fmt=vtt" |
| 745 | vtt_req = urllib.request.Request(vtt_url, headers=headers) |
| 746 | try: |
| 747 | with urllib.request.urlopen(vtt_req, timeout=timeout) as resp: |
| 748 | vtt_text = resp.read().decode("utf-8", errors="replace") |
| 749 | except (urllib.error.URLError, urllib.error.HTTPError, OSError, TimeoutError) as exc: |
| 750 | _log(f"Direct transcript: failed to fetch VTT for {video_id}: {exc}") |
| 751 | return None |
| 752 | |
| 753 | if not vtt_text or not vtt_text.strip(): |
| 754 | return None |
| 755 | |
| 756 | return vtt_text |
| 757 | |
| 758 | |
| 759 | def _fetch_transcript_ytdlp_via_ssh(video_id: str, ssh_host: str) -> Optional[str]: |
| 760 | """Fetch transcript via yt-dlp on a remote SSH host (mktemp + cat pipeline).""" |
| 761 | if not _SSH_HOST_ALIAS_RE.match(ssh_host): |
| 762 | return None |
| 763 | url = f"https://www.youtube.com/watch?v={video_id}" |
| 764 | quoted_url = shlex.quote(url) |
| 765 | sub_langs = shlex.quote(_ytdlp_sub_langs()) |
| 766 | client = _ytdlp_player_client() |
| 767 | extractor = ( |
| 768 | f"--extractor-args {shlex.quote(f'youtube:player_client={client}')} " |
| 769 | if client |
| 770 | else "" |
| 771 | ) |
| 772 | remote_script = ( |
| 773 | "set -e; " |
| 774 | "TMPD=$(mktemp -d); " |
| 775 | "yt-dlp --ignore-config --no-cookies-from-browser " |
| 776 | f"{extractor}" |
| 777 | f"--write-auto-subs --sub-lang {sub_langs} --sub-format vtt " |
| 778 | "--skip-download --no-warnings " |
| 779 | f'-o "$TMPD/%(id)s" {quoted_url} >/dev/null 2>&1 || true; ' |
| 780 | 'VTT=$(find "$TMPD" -maxdepth 1 -name "*.vtt" 2>/dev/null | head -1); ' |
| 781 | '[ -n "$VTT" ] && cat "$VTT"; ' |
| 782 | 'rm -rf "$TMPD"' |
| 783 | ) |
| 784 | cmd = ["ssh", "-o", "BatchMode=yes", "--", ssh_host, remote_script] |
| 785 | try: |
| 786 | result = _run_ytdlp(cmd, timeout=45) |
| 787 | except subproc.SubprocTimeout: |
| 788 | _log(f"SSH yt-dlp transcript timed out for {video_id} via {ssh_host!r}") |
| 789 | return None |
| 790 | except FileNotFoundError: |
| 791 | _log("ssh executable not found; cannot route transcript fetch") |
| 792 | return None |
| 793 | out = result.stdout or "" |
| 794 | if not out.strip().startswith("WEBVTT"): |
| 795 | if result.returncode != 0 and result.stderr: |
| 796 | first_line = result.stderr.strip().splitlines()[0] |
| 797 | _log( |
| 798 | f"SSH yt-dlp transcript via {ssh_host!r} failed for " |
| 799 | f"{video_id} (rc={result.returncode}): {first_line}" |
| 800 | ) |
| 801 | return None |
| 802 | return out |
| 803 | |
| 804 | |
| 805 | def _ytdlp_sub_langs() -> str: |
| 806 | """Caption languages to try, from LAST30DAYS_YT_SUB_LANGS (default en,es,pt).""" |
| 807 | raw = os.environ.get("LAST30DAYS_YT_SUB_LANGS", "").strip() |
| 808 | if not raw: |
| 809 | return "en,es,pt" |
| 810 | return ",".join(code.strip().lower() for code in raw.split(",") if code.strip()) or "en,es,pt" |
| 811 | |
| 812 | |
| 813 | def _transcript_fast_timeout() -> float: |
| 814 | """Return the keyed-run yt-dlp timeout, preserving the 12s default.""" |
| 815 | return _env_positive_float( |
| 816 | "LAST30DAYS_YT_TRANSCRIPT_FAST_TIMEOUT", |
| 817 | float(_TRANSCRIPT_FAST_TIMEOUT), |
| 818 | ) |
| 819 | |
| 820 | def _pick_ytdlp_vtt(video_id: str, temp_dir: str, priority: List[str]) -> Optional[Path]: |
| 821 | """Return the best on-disk VTT match for video_id, preferring priority order.""" |
| 822 | matches = list(Path(temp_dir).glob(f"{video_id}*.vtt")) |
| 823 | if not matches: |
| 824 | return None |
| 825 | priority_index = {code: i for i, code in enumerate(priority)} |
| 826 | |
| 827 | def rank(p: Path) -> int: |
| 828 | stem = p.stem |
| 829 | suffix = stem[len(video_id) + 1:] if stem.startswith(video_id + ".") else "" |
| 830 | code = suffix.split("-")[0].split(".")[0] |
| 831 | return priority_index.get(code, len(priority_index)) |
| 832 | |
| 833 | return sorted(matches, key=rank)[0] |
| 834 | |
| 835 | |
| 836 | def _transcript_backoff(video_id: str, attempt: int) -> float: |
| 837 | """Backoff seconds before a transcript retry. |
| 838 | |
| 839 | Staggered per-video (a sub-second offset derived from the id) so parallel |
| 840 | workers don't retry in lockstep and re-trip YouTube's limiter. |
| 841 | """ |
| 842 | offset = (sum(ord(c) for c in video_id) % 1000) / 1000.0 # 0.0–1.0s |
| 843 | return _TRANSCRIPT_BACKOFF_BASE * (attempt + 1) + offset |
| 844 | |
| 845 | |
| 846 | def _read_vtt(video_id: str, temp_dir: str) -> Optional[str]: |
| 847 | """Return the VTT text yt-dlp wrote for ``video_id``, or None if absent.""" |
| 848 | vtt_path = _pick_ytdlp_vtt(video_id, temp_dir, _ytdlp_sub_langs().split(",")) |
| 849 | if vtt_path is None: |
| 850 | return None |
| 851 | |
| 852 | try: |
| 853 | return vtt_path.read_text(encoding="utf-8", errors="replace") |
| 854 | except OSError: |
| 855 | return None |
| 856 | |
| 857 | |
| 858 | def _fetch_transcript_ytdlp( |
| 859 | video_id: str, |
| 860 | temp_dir: str, |
| 861 | status: Optional[Dict[str, Any]] = None, |
| 862 | fast_fail: bool = False, |
| 863 | ) -> Optional[str]: |
| 864 | """Fetch transcript using yt-dlp (original implementation). |
| 865 | |
| 866 | Args: |
| 867 | video_id: YouTube video ID |
| 868 | temp_dir: Temporary directory for subtitle files |
| 869 | status: Optional dict mutated to record a yt-dlp failure reason |
| 870 | (``status["ytdlp_error"]``) so the caller can tell a real fetch |
| 871 | error (rate-limit / bot-check / network / timeout) apart from a |
| 872 | video that genuinely has no captions, and skip the misleading |
| 873 | "no captions found" log + the YouTube-blocked HTTP fallback. |
| 874 | fast_fail: When True, a ScrapeCreators fallback is available, so a |
| 875 | transient failure (429 / bot-gate) should fail over fast rather |
| 876 | than retry into the same rate limit. Collapses to a single attempt |
| 877 | with a shorter per-attempt timeout. The slow thing in a run is |
| 878 | yt-dlp retrying, not the fast SC fetch, so this is what keeps |
| 879 | "yt-dlp first" from reintroducing the multi-minute hang. |
| 880 | |
| 881 | Returns: |
| 882 | Raw VTT text, or None if no captions are available or the fetch failed. |
| 883 | On a hard (non-no-caption) failure, sets ``status["ytdlp_error"]``. |
| 884 | """ |
| 885 | cmd = [ |
| 886 | "yt-dlp", |
| 887 | "--ignore-config", |
| 888 | "--no-cookies-from-browser", |
| 889 | "--write-auto-subs", |
| 890 | "--sub-lang", _ytdlp_sub_langs(), |
| 891 | "--sub-format", "vtt", |
| 892 | "--skip-download", |
| 893 | "--no-warnings", |
| 894 | "-o", f"{temp_dir}/%(id)s", |
| 895 | f"https://www.youtube.com/watch?v={video_id}", |
| 896 | ] |
| 897 | cmd = _inject_youtube_player_client(cmd) |
| 898 | |
| 899 | timeout = _transcript_fast_timeout() if fast_fail else _TRANSCRIPT_TIMEOUT |
| 900 | attempts = 1 if fast_fail else _TRANSCRIPT_MAX_RETRIES + 1 |
| 901 | last_reason: Optional[str] = None |
| 902 | for attempt in range(attempts): |
| 903 | try: |
| 904 | result = _run_ytdlp(cmd, timeout=timeout) |
| 905 | except subproc.SubprocTimeout: |
| 906 | last_reason = f"timed out after {timeout}s" |
| 907 | _log(f"yt-dlp transcript timed out after {timeout}s for {video_id} " |
| 908 | f"(attempt {attempt + 1}/{attempts})") |
| 909 | # yt-dlp downloads requested languages sequentially. A timeout can |
| 910 | # therefore leave a complete first-choice VTT on disk; keep it |
| 911 | # instead of spending a ScrapeCreators fallback credit. |
| 912 | partial_vtt = _read_vtt(video_id, temp_dir) |
| 913 | if partial_vtt is not None: |
| 914 | return partial_vtt |
| 915 | if attempt < attempts - 1: |
| 916 | time.sleep(_transcript_backoff(video_id, attempt)) |
| 917 | continue |
| 918 | break |
| 919 | except FileNotFoundError: |
| 920 | # yt-dlp binary missing — not transient, not retryable. |
| 921 | if status is not None: |
| 922 | status["ytdlp_error"] = "yt-dlp not found" |
| 923 | return None |
| 924 | |
| 925 | if result.returncode == 0: |
| 926 | vtt = _read_vtt(video_id, temp_dir) |
| 927 | if vtt is not None: |
| 928 | return vtt |
| 929 | # Exit 0 with no file == the uploader has no matching captions. |
| 930 | # Genuine no-captions: return quietly (caller may still try direct). |
| 931 | return None |
| 932 | |
| 933 | # Non-zero exit, but yt-dlp may have written a usable VTT before the |
| 934 | # failing language errored. With the default `--sub-lang en,es,pt`, an |
| 935 | # English video fetches `en` fine, then `es`/`pt` hit a 429 and yt-dlp |
| 936 | # exits non-zero — yet the `en` track is already on disk. A partial |
| 937 | # success is still a real transcript, so salvage any VTT before |
| 938 | # classifying this as an error (and, worse, retrying straight back into |
| 939 | # the same rate limit). This is the root cause of the 0/N transcript |
| 940 | # runs reported when every video had captions. |
| 941 | partial_vtt = _read_vtt(video_id, temp_dir) |
| 942 | if partial_vtt is not None: |
| 943 | return partial_vtt |
| 944 | |
| 945 | # Non-zero exit == a real error worth classifying & surfacing. |
| 946 | stderr = (result.stderr or "").strip() |
| 947 | snippet = (stderr.splitlines()[-1][:200] if stderr |
| 948 | else f"exit {result.returncode}") |
| 949 | if _NO_CAPTION_RE.search(stderr): |
| 950 | # yt-dlp can exit non-zero when the requested language is absent. |
| 951 | # Treat as genuine no-captions, not an error worth retrying. |
| 952 | return None |
| 953 | last_reason = snippet |
| 954 | if _TRANSIENT_RE.search(stderr) and attempt < attempts - 1: |
| 955 | _log(f"yt-dlp transcript transient failure for {video_id} " |
| 956 | f"(attempt {attempt + 1}/{attempts}): {snippet}") |
| 957 | time.sleep(_transcript_backoff(video_id, attempt)) |
| 958 | continue |
| 959 | # Non-transient, or retries exhausted — surface the real reason. |
| 960 | _log(f"yt-dlp transcript failed for {video_id} " |
| 961 | f"(exit {result.returncode}): {snippet}") |
| 962 | break |
| 963 | |
| 964 | if status is not None and last_reason is not None: |
| 965 | status["ytdlp_error"] = last_reason |
| 966 | return None |
| 967 | |
| 968 | |
| 969 | def _should_try_sc_transcript(status: Optional[Dict[str, Any]]) -> bool: |
| 970 | """Whether to spend a ScrapeCreators credit after the keyless cascade failed. |
| 971 | |
| 972 | Skip when the keyless path *proved* the uploader has no caption track |
| 973 | (``no_caption_tracks``): SC would also return nothing, so a credit would be |
| 974 | wasted. A transient hard failure (``ytdlp_error``: 429 / bot-gate / timeout) |
| 975 | is a false negative, so SC is worth trying. |
| 976 | """ |
| 977 | st = status or {} |
| 978 | return not st.get("no_caption_tracks") |
| 979 | |
| 980 | |
| 981 | def fetch_transcript( |
| 982 | video_id: str, |
| 983 | temp_dir: str, |
| 984 | status: Optional[Dict[str, Any]] = None, |
| 985 | token: Optional[str] = None, |
| 986 | ) -> Optional[str]: |
| 987 | """Fetch auto-generated transcript for a YouTube video. |
| 988 | |
| 989 | Uses yt-dlp when available (preferred, more robust). Falls back to |
| 990 | direct HTTP transcript fetching when yt-dlp is not installed, and finally |
| 991 | to the ScrapeCreators transcript endpoint when a key is present and the |
| 992 | keyless cascade comes back empty. |
| 993 | |
| 994 | Args: |
| 995 | video_id: YouTube video ID |
| 996 | temp_dir: Temporary directory for subtitle files |
| 997 | status: Optional dict mutated by the direct-HTTP path to record |
| 998 | per-video signals like ``no_caption_tracks``. Used to surface a |
| 999 | captions-disabled count so the quality nudge avoids false-positive |
| 1000 | "stale yt-dlp" flags. |
| 1001 | token: Optional ScrapeCreators API key. When present, yt-dlp fails over |
| 1002 | fast (see ``_fetch_transcript_ytdlp`` ``fast_fail``) and a true hard |
| 1003 | failure falls back to the SC transcript endpoint. A credit is only |
| 1004 | spent on a genuine yt-dlp failure, never on success and never on a |
| 1005 | video proven to have no captions. None preserves keyless behavior. |
| 1006 | |
| 1007 | Returns: |
| 1008 | Plaintext transcript string, or None if no captions available. |
| 1009 | """ |
| 1010 | raw_vtt = None |
| 1011 | ssh_host = _ytdlp_ssh_host() |
| 1012 | if ssh_host and is_ytdlp_installed(): |
| 1013 | raw_vtt = _fetch_transcript_ytdlp_via_ssh(video_id, ssh_host) |
| 1014 | if not raw_vtt: |
| 1015 | _log(f"SSH yt-dlp transcript failed for {video_id}, trying direct HTTP fallback") |
| 1016 | raw_vtt = _fetch_transcript_direct(video_id, status=status) |
| 1017 | elif is_ytdlp_installed(): |
| 1018 | raw_vtt = _fetch_transcript_ytdlp( |
| 1019 | video_id, temp_dir, status=status, fast_fail=bool(token), |
| 1020 | ) |
| 1021 | if not raw_vtt: |
| 1022 | ytdlp_error = (status or {}).get("ytdlp_error") |
| 1023 | if ytdlp_error: |
| 1024 | # Hard failure (429 / bot-gate / timeout). The direct-HTTP |
| 1025 | # fallback is also YouTube-blocked, so skip it and let the |
| 1026 | # ScrapeCreators fallback below handle it when a key is present. |
| 1027 | _log(f"Transcript fetch failed for {video_id}: {ytdlp_error}") |
| 1028 | else: |
| 1029 | _log(f"yt-dlp found no captions for {video_id}, trying direct HTTP fallback") |
| 1030 | raw_vtt = _fetch_transcript_direct(video_id, status=status) |
| 1031 | else: |
| 1032 | _log("yt-dlp not installed, using direct HTTP transcript fetch") |
| 1033 | raw_vtt = _fetch_transcript_direct(video_id, status=status) |
| 1034 | |
| 1035 | if raw_vtt: |
| 1036 | transcript = _clean_vtt(raw_vtt) |
| 1037 | # Truncate to max words |
| 1038 | words = transcript.split() |
| 1039 | if len(words) > TRANSCRIPT_MAX_WORDS: |
| 1040 | transcript = ' '.join(words[:TRANSCRIPT_MAX_WORDS]) + '...' |
| 1041 | return transcript if transcript else None |
| 1042 | |
| 1043 | # Keyless cascade produced nothing. When a ScrapeCreators key is present and |
| 1044 | # the video was not proven caption-less, fall back to the SC transcript |
| 1045 | # endpoint (fetched server-side: no 429, cookies, or PO tokens). Returns |
| 1046 | # already-cleaned, word-capped plaintext. |
| 1047 | if token and _should_try_sc_transcript(status): |
| 1048 | sc_transcript = _sc_fetch_transcript(video_id, token) |
| 1049 | if sc_transcript: |
| 1050 | # The keyless cascade (yt-dlp / direct HTTP) already logged its |
| 1051 | # failure above. Without this line that failure is the last thing |
| 1052 | # printed for this video, and the batch summary in |
| 1053 | # fetch_transcripts_parallel() counts it as a plain success — |
| 1054 | # making a rate-limited/bot-gated run look like nothing went |
| 1055 | # wrong. Log the rescue and flag it in `status` so the summary |
| 1056 | # can report it explicitly instead of masking it (#831). |
| 1057 | _log(f"ScrapeCreators transcript fallback rescued {video_id} " |
| 1058 | f"after the keyless fetch cascade failed") |
| 1059 | if status is not None: |
| 1060 | status["sc_rescued"] = True |
| 1061 | return sc_transcript |
| 1062 | |
| 1063 | _log(f"No transcript available for {video_id}") |
| 1064 | return None |
| 1065 | |
| 1066 | |
| 1067 | def fetch_transcripts_parallel( |
| 1068 | video_ids: List[str], |
| 1069 | max_workers: int = 5, |
| 1070 | out_captions_disabled: Optional[Set[str]] = None, |
| 1071 | token: Optional[str] = None, |
| 1072 | ) -> Dict[str, Optional[str]]: |
| 1073 | """Fetch transcripts for multiple videos in parallel. |
| 1074 | |
| 1075 | Args: |
| 1076 | video_ids: List of YouTube video IDs |
| 1077 | max_workers: Max parallel fetches |
| 1078 | out_captions_disabled: Optional set mutated to record video_ids whose |
| 1079 | uploader confirmed no caption tracks (vs. transient fetch failures). |
| 1080 | Backward-compatible: callers that don't care can omit. |
| 1081 | token: Optional ScrapeCreators API key, threaded to each |
| 1082 | ``fetch_transcript`` so the per-video SC fallback activates on |
| 1083 | yt-dlp failure. None preserves keyless behavior. |
| 1084 | |
| 1085 | Returns: |
| 1086 | Dict mapping video_id to transcript text (or None). |
| 1087 | """ |
| 1088 | if not video_ids: |
| 1089 | return {} |
| 1090 | |
| 1091 | _log(f"Fetching transcripts for {len(video_ids)} videos") |
| 1092 | |
| 1093 | results = {} |
| 1094 | statuses: Dict[str, Dict[str, Any]] = {vid: {} for vid in video_ids} |
| 1095 | with tempfile.TemporaryDirectory() as temp_dir: |
| 1096 | with ThreadPoolExecutor(max_workers=max_workers) as executor: |
| 1097 | futures = { |
| 1098 | http.submit_with_context( |
| 1099 | executor, fetch_transcript, vid, temp_dir, statuses[vid], token, |
| 1100 | ): vid |
| 1101 | for vid in video_ids |
| 1102 | } |
| 1103 | for future in as_completed(futures): |
| 1104 | vid = futures[future] |
| 1105 | try: |
| 1106 | results[vid] = future.result() |
| 1107 | except OSError as exc: |
| 1108 | _log(f"Transcript fetch error for {vid}: {exc}") |
| 1109 | results[vid] = None |
| 1110 | except Exception as exc: |
| 1111 | _log(f"Unexpected transcript error for {vid}: {type(exc).__name__}: {exc}") |
| 1112 | results[vid] = None |
| 1113 | |
| 1114 | if out_captions_disabled is not None: |
| 1115 | for vid, st in statuses.items(): |
| 1116 | if st.get("no_caption_tracks"): |
| 1117 | out_captions_disabled.add(vid) |
| 1118 | |
| 1119 | got = sum(1 for v in results.values() if v) |
| 1120 | errors = sum(1 for v in results.values() if v is None) |
| 1121 | # `got` includes videos that only succeeded because the ScrapeCreators |
| 1122 | # fallback rescued a failed keyless fetch — yt-dlp when available, or the |
| 1123 | # direct HTTP path alone (see fetch_transcript()). Folding |
| 1124 | # those into a bare "M failed" count previously made a fully rate-limited |
| 1125 | # yt-dlp run — every fetch failing, silently saved by the fallback — read |
| 1126 | # as "0 failed", with no trace of the fallback ever having fired (#831). |
| 1127 | # Surface the split so the summary can't misrepresent a masked failure |
| 1128 | # as a clean success. |
| 1129 | sc_rescued = sum(1 for st in statuses.values() if st.get("sc_rescued")) |
| 1130 | if sc_rescued: |
| 1131 | _log(f"Got transcripts for {got}/{len(video_ids)} videos " |
| 1132 | f"({errors} failed, {sc_rescued} rescued via ScrapeCreators fallback)") |
| 1133 | else: |
| 1134 | _log(f"Got transcripts for {got}/{len(video_ids)} videos ({errors} failed)") |
| 1135 | return results |
| 1136 | |
| 1137 | |
| 1138 | def backfill_transcripts( |
| 1139 | items: List[Any], topic: str = "", depth: str = "default", |
| 1140 | token: Optional[str] = None, |
| 1141 | ) -> None: |
| 1142 | """Second-pass transcript fetch for finalized items that lack one (#542). |
| 1143 | |
| 1144 | ``token`` is the optional ScrapeCreators key, threaded to |
| 1145 | ``fetch_transcripts_parallel`` so the SC fallback covers backfill survivors |
| 1146 | that yt-dlp can't fetch. None preserves keyless behavior. |
| 1147 | """ |
| 1148 | limit = TRANSCRIPT_LIMITS.get(depth, TRANSCRIPT_LIMITS["default"]) |
| 1149 | if limit <= 0 or not items or not is_ytdlp_installed(): |
| 1150 | return |
| 1151 | have = sum( |
| 1152 | 1 for it in items |
| 1153 | if it.metadata.get("transcript_highlights") or it.metadata.get("transcript_snippet") |
| 1154 | ) |
| 1155 | need = limit - have |
| 1156 | if need <= 0: |
| 1157 | return |
| 1158 | missing = [ |
| 1159 | it for it in items |
| 1160 | if it.item_id |
| 1161 | and not it.metadata.get("transcript_highlights") |
| 1162 | and not it.metadata.get("transcript_snippet") |
| 1163 | and not it.metadata.get("captions_disabled") |
| 1164 | ] |
| 1165 | attempts = missing[: need * 3] |
| 1166 | if not attempts: |
| 1167 | return |
| 1168 | _log(f"Backfilling transcripts for {len(attempts)} finalized videos (target: {need})") |
| 1169 | captions_disabled: Set[str] = set() |
| 1170 | transcripts = fetch_transcripts_parallel( |
| 1171 | [it.item_id for it in attempts], |
| 1172 | out_captions_disabled=captions_disabled, |
| 1173 | token=token, |
| 1174 | ) |
| 1175 | for it in attempts: |
| 1176 | if it.item_id in captions_disabled: |
| 1177 | it.metadata["captions_disabled"] = True |
| 1178 | continue |
| 1179 | transcript = transcripts.get(it.item_id) |
| 1180 | if not transcript: |
| 1181 | continue |
| 1182 | it.metadata["transcript_snippet"] = transcript |
| 1183 | highlights = extract_transcript_highlights(transcript, topic) |
| 1184 | if highlights: |
| 1185 | it.metadata["transcript_highlights"] = highlights |
| 1186 | if not it.snippet: |
| 1187 | it.snippet = " ".join(transcript.split()[:80]) |
| 1188 | |
| 1189 | |
| 1190 | def _transcript_candidate_sort_key(item: dict) -> tuple: |
| 1191 | """Sort key for transcript candidate selection. |
| 1192 | |
| 1193 | Combines views with recency so that recent videos (which survive |
| 1194 | strict_recent freshness pruning) are prioritised over old high-view |
| 1195 | videos whose transcripts would be discarded downstream. |
| 1196 | """ |
| 1197 | views = item.get("engagement", {}).get("views", 0) or 0 |
| 1198 | recency = dates.recency_score(item.get("date", "")) |
| 1199 | return (views, recency) |
| 1200 | |
| 1201 | |
| 1202 | def _prefer_search_error(current: Optional[str], new: str) -> str: |
| 1203 | """Keep the most actionable search failure across multi-query merges.""" |
| 1204 | if current is None: |
| 1205 | return new |
| 1206 | priority = ("timed out", "timeout", "429", "bot") |
| 1207 | |
| 1208 | def _rank(text: str) -> int: |
| 1209 | lower = text.lower() |
| 1210 | for index, marker in enumerate(priority): |
| 1211 | if marker in lower: |
| 1212 | return index |
| 1213 | return len(priority) |
| 1214 | |
| 1215 | return new if _rank(new) < _rank(current) else current |
| 1216 | |
| 1217 | |
| 1218 | def search_and_transcribe( |
| 1219 | topic: str, |
| 1220 | from_date: str, |
| 1221 | to_date: str, |
| 1222 | depth: str = "default", |
| 1223 | token: Optional[str] = None, |
| 1224 | ) -> Dict[str, Any]: |
| 1225 | """Full YouTube search: find videos, then fetch transcripts for top results. |
| 1226 | |
| 1227 | Uses expand_youtube_queries() to generate multiple search queries, |
| 1228 | runs yt-dlp for each, and merges/deduplicates results by video ID. |
| 1229 | |
| 1230 | Args: |
| 1231 | topic: Search topic |
| 1232 | from_date: Start date (YYYY-MM-DD) |
| 1233 | to_date: End date (YYYY-MM-DD) |
| 1234 | depth: 'quick', 'default', or 'deep' |
| 1235 | token: Optional ScrapeCreators key for the per-video transcript |
| 1236 | fallback (threaded to fetch_transcripts_parallel). |
| 1237 | |
| 1238 | Returns: |
| 1239 | Dict with 'items' list. Each item has a 'transcript_snippet' field. |
| 1240 | """ |
| 1241 | # Step 1: Multi-query search — run yt-dlp for each expanded query |
| 1242 | queries = expand_youtube_queries(topic, depth) |
| 1243 | seen_ids: Set[str] = set() |
| 1244 | items: List[Dict[str, Any]] = [] |
| 1245 | search_error: Optional[str] = None |
| 1246 | for q in queries: |
| 1247 | search_result = search_youtube(q, from_date, to_date, depth) |
| 1248 | err = search_result.get("error") |
| 1249 | if err: |
| 1250 | search_error = _prefer_search_error(search_error, str(err)) |
| 1251 | for item in search_result.get("items", []): |
| 1252 | vid = item.get("video_id", "") |
| 1253 | if vid and vid not in seen_ids: |
| 1254 | seen_ids.add(vid) |
| 1255 | items.append(item) |
| 1256 | |
| 1257 | # Sort merged results by views descending |
| 1258 | items.sort(key=lambda x: x.get("engagement", {}).get("views") or 0, reverse=True) |
| 1259 | |
| 1260 | if not items: |
| 1261 | return {"items": [], **({"error": search_error} if search_error else {})} |
| 1262 | |
| 1263 | # Step 2: Fetch transcripts for top videos. |
| 1264 | # Sort candidates by a combination of views and recency so that recent |
| 1265 | # videos (which survive strict_recent pruning) are not starved of |
| 1266 | # transcript budget by older high-view-count outliers. |
| 1267 | # Try more candidates than the limit because some videos (music videos, |
| 1268 | # short clips) lack captions. Attempt up to 3x the limit so we have a |
| 1269 | # good chance of reaching the target number of successful transcripts. |
| 1270 | transcript_limit = TRANSCRIPT_LIMITS.get(depth, TRANSCRIPT_LIMITS["default"]) |
| 1271 | transcripts: Dict[str, Optional[str]] = {} |
| 1272 | captions_disabled_ids: Set[str] = set() |
| 1273 | if transcript_limit > 0: |
| 1274 | attempt_count = min(len(items), transcript_limit * 3) |
| 1275 | transcript_candidates = sorted( |
| 1276 | items, key=_transcript_candidate_sort_key, reverse=True, |
| 1277 | ) |
| 1278 | candidate_ids = [item["video_id"] for item in transcript_candidates[:attempt_count]] |
| 1279 | _log(f"Fetching transcripts for up to {attempt_count} videos (target: {transcript_limit}): {candidate_ids}") |
| 1280 | transcripts = fetch_transcripts_parallel( |
| 1281 | candidate_ids, out_captions_disabled=captions_disabled_ids, |
| 1282 | token=token, |
| 1283 | ) |
| 1284 | # Record fetch outcomes (captions-disabled videos can never succeed, |
| 1285 | # so they don't count as failures) for the stale-yt-dlp nudge. |
| 1286 | _TRANSCRIPT_FETCH_STATS["attempts"] += len(candidate_ids) |
| 1287 | _TRANSCRIPT_FETCH_STATS["failures"] += sum( |
| 1288 | 1 for vid in candidate_ids |
| 1289 | if not transcripts.get(vid) and vid not in captions_disabled_ids |
| 1290 | ) |
| 1291 | else: |
| 1292 | _log(f"Transcript limit is 0 for depth={depth}, skipping transcript fetch") |
| 1293 | |
| 1294 | # Step 3: Attach transcripts and extract highlights. Mark captions_disabled |
| 1295 | # so quality_nudge can subtract those videos from the degraded-ratio |
| 1296 | # denominator (uploader-disabled captions can never produce a transcript; |
| 1297 | # counting them was producing false-positive stale-yt-dlp nudges). |
| 1298 | core_topic = _extract_core_subject(topic) |
| 1299 | for item in items: |
| 1300 | vid = item["video_id"] |
| 1301 | transcript = transcripts.get(vid) |
| 1302 | item["transcript_snippet"] = transcript or "" |
| 1303 | item["transcript_highlights"] = extract_transcript_highlights( |
| 1304 | transcript or "", core_topic, |
| 1305 | ) |
| 1306 | item["captions_disabled"] = vid in captions_disabled_ids |
| 1307 | |
| 1308 | result: Dict[str, Any] = {"items": items} |
| 1309 | if search_error: |
| 1310 | # Partial coverage: some queries succeeded; keep the failure visible so |
| 1311 | # source_status becomes partial/timeout rather than a quiet OK. |
| 1312 | result["error"] = search_error |
| 1313 | return result |
| 1314 | |
| 1315 | |
| 1316 | def parse_youtube_response(response: Dict[str, Any]) -> List[Dict[str, Any]]: |
| 1317 | """Parse YouTube search response to normalized format. |
| 1318 | |
| 1319 | Returns: |
| 1320 | List of item dicts ready for normalization. |
| 1321 | """ |
| 1322 | return response.get("items", []) |
| 1323 | |
| 1324 | |
| 1325 | # --------------------------------------------------------------------------- |
| 1326 | # ScrapeCreators YouTube API support |
| 1327 | # --------------------------------------------------------------------------- |
| 1328 | |
| 1329 | SCRAPECREATORS_YT_BASE = "https://api.scrapecreators.com/v1/youtube" |
| 1330 | |
| 1331 | |
| 1332 | def _total_engagement(item: Dict[str, Any]) -> int: |
| 1333 | """Combined engagement score for ranking which videos to enrich.""" |
| 1334 | eng = item.get("engagement", {}) |
| 1335 | views = eng.get("views", 0) or 0 |
| 1336 | likes = eng.get("likes", 0) or 0 |
| 1337 | comments = eng.get("comments", 0) or 0 |
| 1338 | return views + likes + comments |
| 1339 | |
| 1340 | |
| 1341 | def enrich_with_comments( |
| 1342 | items: List[Dict[str, Any]], |
| 1343 | token: str, |
| 1344 | max_videos: int = 3, |
| 1345 | max_comments: int = 5, |
| 1346 | ) -> List[Dict[str, Any]]: |
| 1347 | """Enrich top YouTube videos with comment data from ScrapeCreators. |
| 1348 | |
| 1349 | For the top N videos by engagement, fetches comments via the SC API |
| 1350 | and attaches them as a ``top_comments`` field on each item. |
| 1351 | |
| 1352 | Args: |
| 1353 | items: YouTube items from search_and_transcribe() or search_youtube_sc() |
| 1354 | token: ScrapeCreators API key |
| 1355 | max_videos: How many videos to enrich with comments |
| 1356 | max_comments: Max comments to keep per video |
| 1357 | |
| 1358 | Returns: |
| 1359 | Items list (mutated in place) with top_comments added to enriched items. |
| 1360 | """ |
| 1361 | if not items or max_videos <= 0: |
| 1362 | return items |
| 1363 | # yt-dlp needs no key, so an empty token is only fatal when it is absent too. |
| 1364 | if not token and not is_ytdlp_installed(): |
| 1365 | return items |
| 1366 | |
| 1367 | ranked = sorted(items, key=_total_engagement, reverse=True) |
| 1368 | top_items = ranked[:max_videos] |
| 1369 | _log(f"Enriching comments for {len(top_items)} YouTube videos") |
| 1370 | |
| 1371 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 1372 | |
| 1373 | def _enrich_one(item: dict) -> bool: |
| 1374 | video_id = item.get("video_id", "") |
| 1375 | if not video_id: |
| 1376 | return False |
| 1377 | try: |
| 1378 | comments = _fetch_video_comments(video_id, token, max_comments) |
| 1379 | if comments: |
| 1380 | item["top_comments"] = comments |
| 1381 | return True |
| 1382 | except Exception as exc: |
| 1383 | _log(f"Comment enrichment failed for {video_id}: {exc}") |
| 1384 | return False |
| 1385 | |
| 1386 | enriched_count = 0 |
| 1387 | with ThreadPoolExecutor(max_workers=min(4, len(top_items))) as executor: |
| 1388 | futures = {http.submit_with_context(executor, _enrich_one, item): item for item in top_items} |
| 1389 | for future in as_completed(futures): |
| 1390 | if future.result(): |
| 1391 | enriched_count += 1 |
| 1392 | |
| 1393 | _log(f"Enriched {enriched_count}/{len(top_items)} videos with comments") |
| 1394 | return items |
| 1395 | |
| 1396 | |
| 1397 | def _ytdlp_comments_result( |
| 1398 | video_id: str, |
| 1399 | max_comments: int = 5, |
| 1400 | ) -> tuple[List[Dict[str, Any]], bool]: |
| 1401 | """Fetch top comments via yt-dlp, returning ``(comments, ran_cleanly)``. |
| 1402 | |
| 1403 | The bool distinguishes "yt-dlp succeeded, this video simply has no |
| 1404 | comments" (True, []) from "yt-dlp was absent or errored" (False, []), so |
| 1405 | the caller only spends a ScrapeCreators credit on a genuine failure — not |
| 1406 | on a video that legitimately has zero comments. Mirrors the transcript |
| 1407 | path, which is likewise careful not to bill SC for a caption-less video. |
| 1408 | |
| 1409 | Comments are sorted by top so a low ``max_comments`` still returns the |
| 1410 | highest-voted ones rather than an arbitrary slice. |
| 1411 | """ |
| 1412 | if not is_ytdlp_installed(): |
| 1413 | return [], False |
| 1414 | |
| 1415 | cmd = _wrap_ytdlp_cmd([ |
| 1416 | "yt-dlp", |
| 1417 | "--write-comments", |
| 1418 | "--skip-download", |
| 1419 | "--dump-single-json", |
| 1420 | "--no-warnings", |
| 1421 | "--ignore-config", |
| 1422 | "--extractor-args", |
| 1423 | f"youtube:comment_sort=top;max_comments={max_comments},all,{max_comments}", |
| 1424 | f"https://www.youtube.com/watch?v={video_id}", |
| 1425 | ]) |
| 1426 | |
| 1427 | try: |
| 1428 | result = _run_ytdlp(cmd, timeout=_COMMENT_TIMEOUT) |
| 1429 | except Exception as exc: |
| 1430 | _log(f"yt-dlp comment fetch failed for {video_id}: {exc}") |
| 1431 | return [], False |
| 1432 | |
| 1433 | if result.returncode != 0 or not result.stdout: |
| 1434 | _log(f"yt-dlp comment fetch failed for {video_id} (exit {result.returncode})") |
| 1435 | return [], False |
| 1436 | |
| 1437 | try: |
| 1438 | payload = json.loads(result.stdout) |
| 1439 | except (ValueError, TypeError) as exc: |
| 1440 | _log(f"yt-dlp comment JSON parse failed for {video_id}: {exc}") |
| 1441 | return [], False |
| 1442 | |
| 1443 | comments = [] |
| 1444 | for c in (payload.get("comments") or [])[:max_comments]: |
| 1445 | text = c.get("text") or "" |
| 1446 | if not text: |
| 1447 | continue |
| 1448 | comments.append({ |
| 1449 | "author": c.get("author") or "", |
| 1450 | "text": text[:400], |
| 1451 | "likes": c.get("like_count") or 0, |
| 1452 | "date": c.get("_time_text") or "", |
| 1453 | }) |
| 1454 | return comments, True |
| 1455 | |
| 1456 | |
| 1457 | def _fetch_video_comments_ytdlp( |
| 1458 | video_id: str, |
| 1459 | max_comments: int = 5, |
| 1460 | ) -> List[Dict[str, Any]]: |
| 1461 | """Comments for a video via yt-dlp (free, keyless), or [] on any failure. |
| 1462 | |
| 1463 | Thin list-returning wrapper over ``_ytdlp_comments_result`` for callers |
| 1464 | that don't need to tell a clean empty result from a failure. |
| 1465 | """ |
| 1466 | return _ytdlp_comments_result(video_id, max_comments)[0] |
| 1467 | |
| 1468 | |
| 1469 | def _fetch_video_comments( |
| 1470 | video_id: str, |
| 1471 | token: str, |
| 1472 | max_comments: int = 5, |
| 1473 | ) -> List[Dict[str, Any]]: |
| 1474 | """Fetch comments for one video, preferring the free yt-dlp path. |
| 1475 | |
| 1476 | yt-dlp is tried first because it is keyless and costs nothing. |
| 1477 | ScrapeCreators stays as the backstop for when yt-dlp is absent or gets |
| 1478 | throttled, and is only called when a token is actually configured. |
| 1479 | |
| 1480 | Args: |
| 1481 | video_id: YouTube video ID |
| 1482 | token: ScrapeCreators API key (may be empty — yt-dlp needs none) |
| 1483 | max_comments: Maximum comments to return |
| 1484 | |
| 1485 | Returns: |
| 1486 | List of comment dicts with author, text, likes, date. |
| 1487 | """ |
| 1488 | ytdlp_comments, ran_cleanly = _ytdlp_comments_result(video_id, max_comments) |
| 1489 | if ytdlp_comments: |
| 1490 | return ytdlp_comments |
| 1491 | # Clean run with no comments -> the video simply has none. Don't spend an |
| 1492 | # SC credit chasing comments that aren't there; only fall back on failure. |
| 1493 | if ran_cleanly: |
| 1494 | return [] |
| 1495 | |
| 1496 | if not token: |
| 1497 | return [] |
| 1498 | |
| 1499 | video_url = f"https://www.youtube.com/watch?v={video_id}" |
| 1500 | try: |
| 1501 | data = http.get( |
| 1502 | f"{SCRAPECREATORS_YT_BASE}/video/comments", |
| 1503 | params={"url": video_url}, |
| 1504 | headers=http.scrapecreators_headers(token), |
| 1505 | timeout=30, |
| 1506 | retries=2, |
| 1507 | ) |
| 1508 | except Exception as exc: |
| 1509 | _log(f"Comment fetch error for {video_id}: {exc}") |
| 1510 | return [] |
| 1511 | |
| 1512 | raw_comments = data.get("comments", data.get("data", [])) |
| 1513 | comments = [] |
| 1514 | for c in raw_comments[:max_comments]: |
| 1515 | text = c.get("text") or c.get("body") or c.get("content", "") |
| 1516 | if not text: |
| 1517 | continue |
| 1518 | |
| 1519 | # SC returns author as {"name": "@handle", ...}; legacy mocks may pass a string. |
| 1520 | author = c.get("author") or c.get("author_name", "") |
| 1521 | if isinstance(author, dict): |
| 1522 | author = author.get("name") or author.get("handle") or "" |
| 1523 | |
| 1524 | # SC nests likes under engagement.likes; legacy shapes used top-level keys. |
| 1525 | engagement = c.get("engagement") or {} |
| 1526 | likes = c.get("likes") |
| 1527 | if likes is None: |
| 1528 | likes = engagement.get("likes", 0) if isinstance(engagement, dict) else 0 |
| 1529 | if not likes: |
| 1530 | likes = c.get("vote_count", 0) |
| 1531 | |
| 1532 | date = ( |
| 1533 | c.get("date") |
| 1534 | or c.get("published_at") |
| 1535 | or c.get("publishedTime") |
| 1536 | or c.get("publishedTimeText", "") |
| 1537 | ) |
| 1538 | |
| 1539 | comments.append({ |
| 1540 | "author": author, |
| 1541 | "text": text[:400], |
| 1542 | "likes": likes, |
| 1543 | "date": date, |
| 1544 | }) |
| 1545 | |
| 1546 | return comments |
| 1547 | |
| 1548 | |
| 1549 | def search_youtube_sc( |
| 1550 | topic: str, |
| 1551 | from_date: str, |
| 1552 | to_date: str, |
| 1553 | depth: str = "default", |
| 1554 | token: str = None, |
| 1555 | ) -> Dict[str, Any]: |
| 1556 | """Search YouTube via ScrapeCreators API (fallback when yt-dlp is unavailable). |
| 1557 | |
| 1558 | Uses SC keyword search to find videos and SC transcript endpoint to |
| 1559 | fetch transcripts. Called by pipeline.py when yt-dlp fails. |
| 1560 | |
| 1561 | Args: |
| 1562 | topic: Search topic |
| 1563 | from_date: Start date (YYYY-MM-DD) |
| 1564 | to_date: End date (YYYY-MM-DD) |
| 1565 | depth: 'quick', 'default', or 'deep' |
| 1566 | token: ScrapeCreators API key |
| 1567 | |
| 1568 | Returns: |
| 1569 | Dict with 'items' list of video metadata dicts. |
| 1570 | """ |
| 1571 | if not token: |
| 1572 | return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"} |
| 1573 | |
| 1574 | count = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 1575 | core_topic = _extract_core_subject(topic) |
| 1576 | _log(f"Searching YouTube via ScrapeCreators for '{core_topic}' (depth={depth})") |
| 1577 | |
| 1578 | # Step 1: Search |
| 1579 | raw_items = _sc_youtube_search(core_topic, token) |
| 1580 | if not raw_items: |
| 1581 | _log("SC YouTube search returned 0 results") |
| 1582 | return {"items": []} |
| 1583 | |
| 1584 | # Parse into normalized items |
| 1585 | items = [] |
| 1586 | for i, raw in enumerate(raw_items[:count]): |
| 1587 | video_id = ( |
| 1588 | raw.get("id") or raw.get("video_id") or raw.get("videoId") or "" |
| 1589 | ) |
| 1590 | title = raw.get("title", "") |
| 1591 | channel = raw.get("channel") or raw.get("channel_name") or raw.get("uploader", "") |
| 1592 | description = str(raw.get("description", ""))[:500] |
| 1593 | view_count = raw.get("view_count") or raw.get("views", 0) |
| 1594 | like_count = raw.get("like_count") or raw.get("likes", 0) |
| 1595 | comment_count = raw.get("comment_count") or raw.get("comments", 0) |
| 1596 | |
| 1597 | # Date: try multiple field names |
| 1598 | date_str = raw.get("upload_date") or raw.get("date") or raw.get("published_at", "") |
| 1599 | if date_str and len(date_str) == 8 and date_str.isdigit(): |
| 1600 | date_str = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:8]}" |
| 1601 | elif date_str and "T" in date_str: |
| 1602 | date_str = date_str[:10] |
| 1603 | |
| 1604 | url = raw.get("url", "") |
| 1605 | if not url and video_id: |
| 1606 | url = f"https://www.youtube.com/watch?v={video_id}" |
| 1607 | |
| 1608 | items.append({ |
| 1609 | "video_id": video_id, |
| 1610 | "title": title, |
| 1611 | "url": url, |
| 1612 | "channel_name": channel, |
| 1613 | "date": date_str if date_str else None, |
| 1614 | "engagement": { |
| 1615 | "views": view_count or 0, |
| 1616 | "likes": like_count or 0, |
| 1617 | "comments": comment_count or 0, |
| 1618 | }, |
| 1619 | "duration": raw.get("duration"), |
| 1620 | "relevance": _compute_relevance(core_topic, f"{title} {description}"), |
| 1621 | "why_relevant": f"YouTube: {title[:60]}" if title else f"YouTube: {core_topic}", |
| 1622 | "description": description, |
| 1623 | }) |
| 1624 | |
| 1625 | # Soft date filter |
| 1626 | recent = [i for i in items if i["date"] and i["date"] >= from_date] |
| 1627 | if len(recent) >= 3: |
| 1628 | items = recent |
| 1629 | _log(f"Found {len(items)} videos within date range") |
| 1630 | else: |
| 1631 | _log(f"Found {len(items)} videos ({len(recent)} within date range, keeping all)") |
| 1632 | |
| 1633 | # Sort by views |
| 1634 | items.sort(key=lambda x: x["engagement"]["views"], reverse=True) |
| 1635 | |
| 1636 | # Step 2: Fetch transcripts for top videos |
| 1637 | transcript_limit = TRANSCRIPT_LIMITS.get(depth, TRANSCRIPT_LIMITS["default"]) |
| 1638 | if transcript_limit > 0 and items: |
| 1639 | attempt_count = min(len(items), transcript_limit * 3) |
| 1640 | # Same in-window-first ordering as search_and_transcribe(): don't let |
| 1641 | # an out-of-window back-catalog (kept by the soft date filter above) |
| 1642 | # consume the transcript budget of videos the freshness scorer keeps. |
| 1643 | in_window = [i for i in items if i.get("date") and i["date"] >= from_date] |
| 1644 | out_of_window = [i for i in items if not (i.get("date") and i["date"] >= from_date)] |
| 1645 | _log(f"Fetching SC transcripts for up to {attempt_count} videos (target: {transcript_limit})") |
| 1646 | for item in (in_window + out_of_window)[:attempt_count]: |
| 1647 | vid = item["video_id"] |
| 1648 | if not vid: |
| 1649 | continue |
| 1650 | transcript = _sc_fetch_transcript(vid, token) |
| 1651 | item["transcript_snippet"] = transcript or "" |
| 1652 | item["transcript_highlights"] = extract_transcript_highlights( |
| 1653 | transcript or "", core_topic, |
| 1654 | ) |
| 1655 | else: |
| 1656 | for item in items: |
| 1657 | item["transcript_snippet"] = "" |
| 1658 | item["transcript_highlights"] = [] |
| 1659 | |
| 1660 | _log(f"SC YouTube: {len(items)} videos returned") |
| 1661 | return {"items": items} |
| 1662 | |
| 1663 | |
| 1664 | def _sc_youtube_search(keyword: str, token: str) -> List[Dict[str, Any]]: |
| 1665 | """Call ScrapeCreators YouTube search endpoint. |
| 1666 | |
| 1667 | Args: |
| 1668 | keyword: Search keyword |
| 1669 | token: ScrapeCreators API key |
| 1670 | |
| 1671 | Returns: |
| 1672 | List of raw video dicts from the API. |
| 1673 | """ |
| 1674 | try: |
| 1675 | # SC's /v1/youtube/search rejects ?keyword= with HTTP 400; the canonical |
| 1676 | # parameter for that endpoint is `query`. Other SC endpoints use their |
| 1677 | # own per-endpoint param names so this was the lone outlier. |
| 1678 | data = http.get( |
| 1679 | f"{SCRAPECREATORS_YT_BASE}/search", |
| 1680 | params={"query": keyword}, |
| 1681 | headers=http.scrapecreators_headers(token), |
| 1682 | timeout=30, |
| 1683 | retries=2, |
| 1684 | ) |
| 1685 | return data.get("videos", data.get("data", data.get("items", []))) |
| 1686 | except Exception as exc: |
| 1687 | _log(f"SC YouTube search error: {exc}") |
| 1688 | return [] |
| 1689 | |
| 1690 | |
| 1691 | def _sc_segment_text(seg: Any) -> str: |
| 1692 | """Extract caption text from a ScrapeCreators transcript segment. |
| 1693 | |
| 1694 | The transcript endpoint returns a list of segment dicts |
| 1695 | (``{text, startMs, endMs}``); older/simpler shapes return plain strings. |
| 1696 | Pull the ``text`` field for dicts so segment metadata is not stringified |
| 1697 | into the output (``{'text': ...}`` garbage). |
| 1698 | """ |
| 1699 | if isinstance(seg, dict): |
| 1700 | # `or ""` (not a get default): a present-but-null `text` returns None, |
| 1701 | # which would stringify to the literal "None" for silent/music segments. |
| 1702 | return str(seg.get("text") or "") |
| 1703 | return str(seg) |
| 1704 | |
| 1705 | |
| 1706 | def _warn_low_sc_credits(data: Dict[str, Any]) -> None: |
| 1707 | """Surface a low-credit warning from a ScrapeCreators response, if present.""" |
| 1708 | credits = data.get("credits_remaining") |
| 1709 | if isinstance(credits, (int, float)) and not isinstance(credits, bool): |
| 1710 | if credits < _SC_LOW_CREDIT_THRESHOLD: |
| 1711 | _log(f"ScrapeCreators credits low: {int(credits)} remaining " |
| 1712 | f"(below {_SC_LOW_CREDIT_THRESHOLD})") |
| 1713 | |
| 1714 | |
| 1715 | def _sc_fetch_transcript(video_id: str, token: str) -> Optional[str]: |
| 1716 | """Fetch transcript for a YouTube video via ScrapeCreators. |
| 1717 | |
| 1718 | Args: |
| 1719 | video_id: YouTube video ID |
| 1720 | token: ScrapeCreators API key |
| 1721 | |
| 1722 | Returns: |
| 1723 | Plaintext transcript string, or None if unavailable. |
| 1724 | """ |
| 1725 | video_url = f"https://www.youtube.com/watch?v={video_id}" |
| 1726 | try: |
| 1727 | # Isolate SC transcript fetch errors from the pipeline-level |
| 1728 | # capture_failures() context. |
| 1729 | with http.capture_failures() as _tf: |
| 1730 | data = http.get( |
| 1731 | f"{SCRAPECREATORS_YT_BASE}/video/transcript", |
| 1732 | params={"url": video_url}, |
| 1733 | headers=http.scrapecreators_headers(token), |
| 1734 | timeout=30, |
| 1735 | retries=1, |
| 1736 | ) |
| 1737 | except Exception as exc: |
| 1738 | _log(f"SC transcript error for {video_id}: {exc}") |
| 1739 | return None |
| 1740 | |
| 1741 | _warn_low_sc_credits(data) |
| 1742 | |
| 1743 | transcript = data.get("transcript") |
| 1744 | if not transcript: |
| 1745 | return None |
| 1746 | |
| 1747 | if isinstance(transcript, list): |
| 1748 | transcript = " ".join(_sc_segment_text(seg) for seg in transcript).strip() |
| 1749 | |
| 1750 | # Clean VTT formatting if present |
| 1751 | transcript = _clean_vtt(transcript) |
| 1752 | |
| 1753 | # Truncate to max words |
| 1754 | words = transcript.split() |
| 1755 | if len(words) > TRANSCRIPT_MAX_WORDS: |
| 1756 | transcript = " ".join(words[:TRANSCRIPT_MAX_WORDS]) + "..." |
| 1757 | |
| 1758 | return transcript if transcript else None |
| 1759 |