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).
158 lines
4.8 KiB
Python
158 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from yt_scraper.store import Store, VideoRef
|
|
from yt_scraper.parse import Segment
|
|
from yt_scraper import segments, export, analysis
|
|
from yt_scraper.render import build_filename_stem
|
|
|
|
|
|
SAMPLE_MD = """---
|
|
video_id: abc123
|
|
title: "Vaporwave and nostalgia"
|
|
channel: Alpha
|
|
upload_date: 20240115
|
|
duration: 600
|
|
views: 1234
|
|
likes: 56
|
|
tags: ["vaporwave", "nostalgia"]
|
|
thumbnail: http://img.test/abc.jpg
|
|
---
|
|
|
|
# Vaporwave and nostalgia
|
|
|
|
> [Ver en YouTube](https://www.youtube.com/watch?v=abc123)
|
|
|
|
## Transcripcion
|
|
|
|
### Intro (00:00)
|
|
|
|
**00:15** · Bienvenidos al video sobre vaporwave.
|
|
**00:32** · El vaporwave es una estetica muy interesante.
|
|
|
|
### Historia (01:15)
|
|
|
|
**01:15** · El vaporwave nacio en los 2010s.
|
|
**02:40** · dragon ball es un anime clasico.
|
|
"""
|
|
|
|
|
|
@pytest.fixture()
|
|
def store(tmp_path):
|
|
s = Store(tmp_path / "t.db")
|
|
s.upsert_channel("UC1", "@alpha", "Alpha", 1)
|
|
s.upsert_videos([VideoRef("abc123", "UC1", "Vaporwave and nostalgia", "https://y/watch?v=abc123", "20240115", 600)])
|
|
s.mark_done("abc123", "data/markdown/Alpha/x.md", "es", "auto", True)
|
|
return s
|
|
|
|
|
|
def test_parse_markdown_metadata():
|
|
parsed = segments.parse_markdown(SAMPLE_MD)
|
|
assert parsed.metadata["video_id"] == "abc123"
|
|
assert parsed.metadata["title"] == "Vaporwave and nostalgia"
|
|
assert parsed.metadata["channel"] == "Alpha"
|
|
assert len(parsed.segments) == 4
|
|
assert abs(parsed.segments[0].start - 15.0) < 0.01
|
|
assert "vaporwave" in parsed.segments[0].text.lower()
|
|
chapters = parsed.chapters
|
|
assert chapters[0]["title"] == "Intro"
|
|
assert chapters[1]["title"] == "Historia"
|
|
|
|
|
|
def test_backfill_from_markdown(store, tmp_path):
|
|
md = tmp_path / "md" / "Alpha" / "vapor.md"
|
|
md.parent.mkdir(parents=True)
|
|
md.write_text(SAMPLE_MD, encoding="utf-8")
|
|
n = segments.backfill_from_markdown(store, tmp_path / "md")
|
|
assert n == 1
|
|
v = store.get_video("abc123")
|
|
assert v.view_count == 1234
|
|
assert v.like_count == 56
|
|
assert json.loads(v.tags) == ["vaporwave", "nostalgia"]
|
|
assert v.thumbnail == "http://img.test/abc.jpg"
|
|
assert v.segments_json is not None
|
|
segs = store.get_segments("abc123")
|
|
assert len(segs) == 4
|
|
# search now works
|
|
hits = store.search_segments("vaporwave")
|
|
assert len(hits) >= 1
|
|
# idempotent
|
|
n2 = segments.backfill_from_markdown(store, tmp_path / "md")
|
|
assert n2 == 0
|
|
|
|
|
|
def test_ts_helpers():
|
|
assert segments.ts_to_seconds("01:30") == 90.0
|
|
assert segments.ts_to_seconds("1:02:03") == 3723.0
|
|
|
|
|
|
def test_export_json_csv(store, tmp_path):
|
|
store.store_segments("abc123", [Segment(0.0, 2.0, "hello vaporwave")])
|
|
store.update_video_metadata("abc123", view_count=1234, tags=["vaporwave"])
|
|
j = export.export_json(store, "UC1", tmp_path / "out.json")
|
|
data = json.loads(j.read_text(encoding="utf-8"))
|
|
assert len(data["videos"]) == 1
|
|
assert data["videos"][0]["video_id"] == "abc123"
|
|
|
|
c = export.export_csv(store, "UC1", tmp_path / "out.csv")
|
|
with open(c, encoding="utf-8") as f:
|
|
reader = csv.DictReader(f)
|
|
rows = list(reader)
|
|
assert rows[0]["video_id"] == "abc123"
|
|
assert rows[0]["view_count"] == "1234"
|
|
|
|
|
|
def test_export_srt(store, tmp_path):
|
|
segs = [Segment(1.5, 3.5, "first line"), Segment(5.0, 7.0, "second line")]
|
|
store.store_segments("abc123", segs)
|
|
out = export.export_srt_video(store, "abc123", tmp_path / "x.srt")
|
|
assert out is not None
|
|
text = out.read_text(encoding="utf-8")
|
|
assert "1\n" in text
|
|
assert "00:00:01,500 --> 00:00:03,500" in text
|
|
assert "first line" in text
|
|
assert "second line" in text
|
|
|
|
|
|
def test_export_html(store, tmp_path):
|
|
store.update_video_metadata("abc123", thumbnail="http://t.test/x.jpg", tags=["vaporwave"])
|
|
out = export.export_html(store, "UC1", tmp_path / "g.html")
|
|
html = out.read_text(encoding="utf-8")
|
|
assert "abc123" in html
|
|
assert "Gallery" in html
|
|
|
|
|
|
def test_word_frequency(store):
|
|
segs = [
|
|
Segment(0, 1, "vaporwave vaporwave nostalgia nostalgia nostalgia"),
|
|
Segment(2, 3, "dragon dragon anime"),
|
|
]
|
|
store.store_segments("abc123", segs)
|
|
freq = analysis.word_frequency(store, "UC1", top=5)
|
|
assert freq["nostalgia"] == 3
|
|
assert freq["vaporwave"] == 2
|
|
assert freq["dragon"] == 2
|
|
# stopwords filtered
|
|
assert "y" not in freq
|
|
assert "the" not in freq
|
|
|
|
|
|
def test_term_timeline(store):
|
|
segs = [Segment(0, 1, "vaporwave mentioned here")]
|
|
store.store_segments("abc123", segs)
|
|
tl = analysis.term_timeline(store, "vaporwave", "UC1")
|
|
assert len(tl) == 1
|
|
assert tl[0][0] == "2024-01"
|
|
|
|
|
|
def test_build_filename_stem():
|
|
# callers pass already-normalized dates; stem uses them verbatim
|
|
stem = build_filename_stem("2024-01-01", "Hola Mundo!", "{upload_date}_{slug}")
|
|
assert stem == "2024-01-01_hola-mundo"
|