urieljareth
621bbc5f5c
Scraper de canales de YouTube hacia notas Markdown para base de conocimiento (Obsidian-ready), con plataforma web local. Engine + CLI (Workstream A): - Modular pipeline: discover/extract/parse/chapters/render/store + ratelimit - SQLite store con migración idempotente: FTS5 (transcript search), columnas de metadata enriquecida, tablas cookies_meta y scrape_jobs - Módulos: segments, cookies (Netscape vault), export (json/csv/srt/html), analysis (word freq/timeline/wordcloud), monitor (watch loop), pipeline - CLI Click group: search, export, audio, channels, watch, analyze, re-render - Fix del bug de scoping de cookies en cli.py Webapp local (Workstream B): - FastAPI backend: dashboard, channels, videos facetado, transcript, search FTS, analysis, scrape jobs con SSE, cookies drag-and-drop, exports, folders (abrir en OS), tools (re-render, formato) - SPA no-build (Alpine.js + Tailwind + Chart.js por CDN): 9 vistas, tema dark command-center con acento rojo→rosa, cookie vault drag-drop, consola de scrapeo con progreso live vía SSE Launcher + subagentes (Workstream C): - start-server.bat / stop-server.bat con auto port-scan + browser open - .opencode/agent/webapp-builder.md + .opencode/goals/webapp-build.md Tests: 38 pytest verdes. Sin funcionalidad de IA (enfoque data-mining).
126 lines
3.6 KiB
Python
126 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import requests
|
|
import yt_dlp
|
|
|
|
from .parse import Segment, parse_auto_dump
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class SubtitlePick:
|
|
url: str
|
|
ext: str
|
|
lang: str
|
|
source: str # "manual" | "auto"
|
|
|
|
|
|
@dataclass
|
|
class VideoData:
|
|
info: dict[str, Any]
|
|
segments: list[Segment]
|
|
subtitle: SubtitlePick | None
|
|
has_chapters: bool
|
|
|
|
|
|
def extract_video(
|
|
video_url: str,
|
|
languages: list[str],
|
|
retries: int = 10,
|
|
sleep_subrequests: float = 2.0,
|
|
prefer_manual: bool = True,
|
|
cookies_file: str | None = None,
|
|
cookies_from_browser: str | None = None,
|
|
) -> VideoData:
|
|
ydl_opts: dict[str, Any] = {
|
|
"writesubtitles": True,
|
|
"writeautomaticsub": True,
|
|
"subtitleslangs": languages,
|
|
"skip_download": True,
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
"retries": retries,
|
|
"sleep_subrequests": sleep_subrequests,
|
|
"noprogress": True,
|
|
}
|
|
if cookies_file:
|
|
ydl_opts["cookiefile"] = cookies_file
|
|
if cookies_from_browser:
|
|
ydl_opts["cookiesfrombrowser"] = (cookies_from_browser,)
|
|
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(video_url, download=False)
|
|
|
|
pick = pick_subtitle(info, languages, prefer_manual)
|
|
segments: list[Segment] = []
|
|
if pick:
|
|
raw = _download_subtitle(pick.url)
|
|
if raw:
|
|
segments = parse_auto_dump(raw)
|
|
if not segments:
|
|
log.warning("Could not parse subtitle for %s (format=%s)", video_url, pick.ext)
|
|
|
|
has_chapters = bool(info.get("chapters"))
|
|
return VideoData(info=info, segments=segments, subtitle=pick, has_chapters=has_chapters)
|
|
|
|
|
|
def pick_subtitle(info: dict[str, Any], languages: list[str], prefer_manual: bool = True) -> SubtitlePick | None:
|
|
manual = info.get("subtitles") or {}
|
|
auto = info.get("automatic_captions") or {}
|
|
|
|
ordered_sources: list[tuple[str, dict[str, Any]]]
|
|
if prefer_manual:
|
|
ordered_sources = [("manual", manual), ("auto", auto)]
|
|
else:
|
|
ordered_sources = [("auto", auto), ("manual", manual)]
|
|
|
|
for source_label, tracks in ordered_sources:
|
|
for lang in languages:
|
|
normalized = _normalize_lang(lang)
|
|
for track_lang, formats in tracks.items():
|
|
if _normalize_lang(track_lang) != normalized:
|
|
continue
|
|
if not formats:
|
|
continue
|
|
pick = _pick_best_format(formats)
|
|
if pick:
|
|
return SubtitlePick(
|
|
url=pick["url"],
|
|
ext=pick["ext"],
|
|
lang=track_lang,
|
|
source="manual" if source_label == "manual" else "auto",
|
|
)
|
|
return None
|
|
|
|
|
|
def _pick_best_format(formats: list[dict[str, Any]]) -> dict[str, Any] | None:
|
|
priority = ["json3", "srv1", "srv3", "vtt", "ttml"]
|
|
for ext in priority:
|
|
for fmt in formats:
|
|
if fmt.get("ext") == ext and fmt.get("url"):
|
|
return fmt
|
|
for fmt in formats:
|
|
if fmt.get("url"):
|
|
return fmt
|
|
return None
|
|
|
|
|
|
def _normalize_lang(code: str) -> str:
|
|
base = code.replace("_", "-").split("-")[0].lower()
|
|
return base
|
|
|
|
|
|
def _download_subtitle(url: str) -> str | None:
|
|
try:
|
|
resp = requests.get(url, timeout=15, headers={"User-Agent": "Mozilla/5.0"})
|
|
resp.raise_for_status()
|
|
return resp.text
|
|
except requests.RequestException as exc:
|
|
log.error("Failed to download subtitle from %s: %s", url, exc)
|
|
return None
|