feat: local content-mining platform for YouTube creator scraping
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).
This commit is contained in:
@@ -0,0 +1,125 @@
|
||||
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
|
||||
Reference in New Issue
Block a user