diff --git a/src/second_brain/main.py b/src/second_brain/main.py index 6b91e47..29d870e 100644 --- a/src/second_brain/main.py +++ b/src/second_brain/main.py @@ -49,13 +49,53 @@ def cli() -> None: @click.option("--focus", "-f", default=None, help="Optional focus directive.") @click.option("--title", "-t", default=None, help="Override title.") def add(url: str, domain: str, focus: Optional[str], title: Optional[str]) -> None: - """Queue a source URL for processing.""" - from second_brain.sources_service import add_source + """Queue a source URL for processing. + + Pasting a YouTube playlist URL (the `/playlist?list=…` form) auto- + expands into one source row per video — no flag, just paste. + `watch?v=…&list=…` is treated as a single video (use the canonical + playlist URL to fan out). + """ + from second_brain.sources_service import ( + DEFAULT_PLAYLIST_MAX_ITEMS, + add_playlist, + add_source, + is_youtube_playlist_url, + ) config = load_config() config.ensure_dirs() db = get_database() + # --- playlist fan-out path --------------------------------------------- + if is_youtube_playlist_url(url): + try: + with db.session() as sess: + result = add_playlist( + sess, + url=url, + domain=domain, + focus=focus, + max_items=DEFAULT_PLAYLIST_MAX_ITEMS, + ) + except ValueError as exc: + click.echo(f"[add] Error: {exc}", err=True) + sys.exit(2) + + click.echo( + f"[add] Playlist expanded: {result.expanded} entr" + f"{'y' if result.expanded == 1 else 'ies'} " + f"(cap {DEFAULT_PLAYLIST_MAX_ITEMS})" + ) + click.echo( + f" added={result.added} duplicates={result.duplicates} " + f"failed={result.failed}" + ) + for u, why in result.failures[:5]: + click.echo(f" ! {u}: {why}", err=True) + return + + # --- single-source path ------------------------------------------------ try: with db.session() as sess: result = add_source( diff --git a/src/second_brain/sources_service.py b/src/second_brain/sources_service.py index 15c7dd9..ac59244 100644 --- a/src/second_brain/sources_service.py +++ b/src/second_brain/sources_service.py @@ -7,15 +7,18 @@ drift on validation, dedupe, or default semantics. from __future__ import annotations -from dataclasses import dataclass -from typing import Optional -from urllib.parse import urlparse +import logging +from dataclasses import dataclass, field +from typing import Callable, Optional +from urllib.parse import parse_qs, urlparse from sqlalchemy.orm import Session from second_brain.config import DOMAINS from second_brain.models import Source, SourceStatus, SourceType, utcnow +logger = logging.getLogger(__name__) + _VIDEO_HOSTS = { "youtube.com", @@ -109,4 +112,177 @@ def add_source( return AddResult(source=source, added=True) -__all__ = ["AddResult", "add_source", "is_article_url", "validate_url"] +# --------------------------------------------------------------------------- +# YouTube playlist fan-out +# --------------------------------------------------------------------------- + +_YT_HOSTS = {"youtube.com", "www.youtube.com", "m.youtube.com", "music.youtube.com"} + +# Hard ceiling shared between the CLI and the web route. yt-dlp will only +# enumerate up to this many entries via `playlistend`. Stops accidental +# 327-video queue floods. +DEFAULT_PLAYLIST_MAX_ITEMS = 50 + + +def is_youtube_playlist_url(url: str) -> bool: + """Detect a YouTube *playlist* URL — strict. + + Returns True only for the `/playlist?list=…` form on a known YouTube host. + A `/watch?v=…&list=…` URL is ambiguous (user typically pasted a single + video that happens to live inside a playlist) — we treat it as a single + video. To fan out from a watch URL, paste the canonical `/playlist?list=…` + form instead. + """ + if not url: + return False + try: + parsed = urlparse(url.strip()) + except ValueError: + return False + if parsed.scheme not in ("http", "https"): + return False + if parsed.netloc not in _YT_HOSTS: + return False + if parsed.path.rstrip("/") != "/playlist": + return False + q = parse_qs(parsed.query) + return bool(q.get("list")) + + +def expand_youtube_playlist( + url: str, *, max_items: int = DEFAULT_PLAYLIST_MAX_ITEMS +) -> list[dict]: + """Enumerate a playlist into per-video dicts without downloading. + + Uses yt-dlp's `extract_flat=True` — one network call, gets back the + playlist's entries with `id` + `title`. Builds a canonical + `https://www.youtube.com/watch?v={id}` URL for each. Empty / private / + placeholder entries (missing `id`) are skipped silently. + + `max_items` is plumbed straight to yt-dlp's `playlistend` so we never + enumerate more than asked even on a 1000-entry playlist. + """ + # Lazy import: yt-dlp is a runtime dep but importing at module level + # would pull it into every consumer of the service module, and the + # tower already pulls it for the downloader path. Lazy keeps the + # web/dev import graph thin and lets the test suite mock this whole + # function without yt-dlp installed. + import yt_dlp # noqa: PLC0415 + + ydl_opts = { + "quiet": True, + "no_warnings": True, + "skip_download": True, + "extract_flat": True, + "ignoreerrors": True, + "playlistend": max_items, + } + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(url, download=False) or {} + + entries = info.get("entries") or [] + out: list[dict] = [] + for entry in entries: + if not entry: + # yt-dlp returns None placeholders for private / removed videos + continue + vid = entry.get("id") + if not vid: + continue + out.append( + { + "url": f"https://www.youtube.com/watch?v={vid}", + "title": entry.get("title"), + } + ) + return out + + +@dataclass +class PlaylistAddResult: + """Per-entry tallies from a playlist fan-out. + + `expanded` is what came back from the expander (after yt-dlp's own + filtering); the three counters always sum to that number. + """ + + expanded: int = 0 + added: int = 0 + duplicates: int = 0 + failed: int = 0 + # (url, reason) per failure, capped so the flash doesn't blow up the page. + failures: list[tuple[str, str]] = field(default_factory=list) + + +def add_playlist( + sess: Session, + *, + url: str, + domain: str = "development", + focus: Optional[str] = None, + max_items: int = DEFAULT_PLAYLIST_MAX_ITEMS, + expander: Optional[Callable[..., list[dict]]] = None, +) -> PlaylistAddResult: + """Expand a playlist URL and queue one source row per video. + + Calls `add_source` for each entry so URL validation, source_type + detection, and the UNIQUE dedupe path are identical to the + single-add flow — there's only ever one way a source row lands in + the queue. + + Partial failures are tolerated: a bad entry doesn't abort the batch. + The per-entry `title` is taken from yt-dlp's playlist metadata + (overrides any caller-supplied title — single playlist title would + be wrong for N videos). + + `expander` is an injection seam for tests; default is the live + yt-dlp-backed `expand_youtube_playlist`. + """ + if domain not in DOMAINS: + raise ValueError( + f"unknown domain {domain!r}; valid: {', '.join(DOMAINS)}" + ) + + fn = expander or expand_youtube_playlist + try: + entries = fn(url, max_items=max_items) + except Exception as exc: # noqa: BLE001 — yt-dlp can raise lots of things + logger.warning("playlist expand failed for %s: %s", url, exc) + raise ValueError(f"could not expand playlist: {exc}") from exc + + result = PlaylistAddResult(expanded=len(entries)) + for entry in entries: + entry_url = entry.get("url", "") + try: + sub = add_source( + sess, + url=entry_url, + domain=domain, + focus=focus, + title=entry.get("title"), + ) + if sub.added: + result.added += 1 + else: + result.duplicates += 1 + except Exception as exc: # noqa: BLE001 — must not abort the batch + result.failed += 1 + if len(result.failures) < 10: + result.failures.append((entry_url, str(exc))) + logger.warning( + "playlist entry add failed: %s — %s", entry_url, exc + ) + return result + + +__all__ = [ + "AddResult", + "DEFAULT_PLAYLIST_MAX_ITEMS", + "PlaylistAddResult", + "add_playlist", + "add_source", + "expand_youtube_playlist", + "is_article_url", + "is_youtube_playlist_url", + "validate_url", +] diff --git a/src/second_brain/web/app.py b/src/second_brain/web/app.py index 880c0fb..350a2aa 100644 --- a/src/second_brain/web/app.py +++ b/src/second_brain/web/app.py @@ -29,7 +29,12 @@ from second_brain.settings_store import ( parse_time_or_none, update_settings, ) -from second_brain.sources_service import add_source +from second_brain.sources_service import ( + DEFAULT_PLAYLIST_MAX_ITEMS, + add_playlist, + add_source, + is_youtube_playlist_url, +) # --------------------------------------------------------------------------- # App setup @@ -291,6 +296,21 @@ def _extraction_to_dict(e: Optional[Extraction]) -> Optional[dict]: } +def _playlist_flash(pr) -> str: + """Human-readable summary for a playlist fan-out. Matches the CLI tone.""" + if pr.expanded == 0: + return "Playlist returned no entries" + parts = [f"Queued {pr.added} video{'' if pr.added == 1 else 's'}"] + extras = [] + if pr.duplicates: + extras.append(f"{pr.duplicates} duplicate{'' if pr.duplicates == 1 else 's'} skipped") + if pr.failed: + extras.append(f"{pr.failed} failed") + if extras: + parts.append(f"({', '.join(extras)})") + return " ".join(parts) + + def _source_to_response(s: Source) -> SourceResponse: return SourceResponse( id=s.id, @@ -375,20 +395,31 @@ async def sources_add( try: with db.session() as sess: - result = add_source( - sess, url=url, domain=domain, focus=focus, title=title - ) - # Snapshot the message inside the session — `result.source` - # detaches on commit. - if result.added: - add_flash = ( - f"Queued {result.source.source_type.value}: {result.source.url}" + if is_youtube_playlist_url(url): + pr = add_playlist( + sess, + url=url, + domain=domain, + focus=focus, + max_items=DEFAULT_PLAYLIST_MAX_ITEMS, ) + add_flash = _playlist_flash(pr) else: - add_flash = ( - f"Already queued (id={result.source.id}, " - f"status={result.source.status.value})" + result = add_source( + sess, url=url, domain=domain, focus=focus, title=title ) + # Snapshot the message inside the session — `result.source` + # detaches on commit. + if result.added: + add_flash = ( + f"Queued {result.source.source_type.value}: " + f"{result.source.url}" + ) + else: + add_flash = ( + f"Already queued (id={result.source.id}, " + f"status={result.source.status.value})" + ) except ValueError as exc: add_error = str(exc) diff --git a/tests/test_playlist.py b/tests/test_playlist.py new file mode 100644 index 0000000..d5d3e78 --- /dev/null +++ b/tests/test_playlist.py @@ -0,0 +1,252 @@ +"""Tests for YouTube playlist fan-out. + +URL-detection + count-aggregation are pure-Python and run anywhere. The +DB-backed `add_playlist` tests need a live petalbrain connection (same +auto-skip pattern as `tests/test_claim.py`). One opt-in live-network +test exercises `expand_youtube_playlist` against a real small public +playlist; it stays gated behind `SECOND_BRAIN_LIVE_NETWORK_TESTS=1` so +CI doesn't depend on YouTube reachability. +""" + +from __future__ import annotations + +import os +import uuid + +import psycopg +import pytest + + +# --------------------------------------------------------------------------- +# Pure-Python detection +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "url", + [ + "https://www.youtube.com/playlist?list=PLxxxxxxxxxx", + "https://youtube.com/playlist?list=PLxxxxxxxxxx", + "https://music.youtube.com/playlist?list=PLxxxxxxxxxx", + "https://m.youtube.com/playlist?list=PLxxxxxxxxxx", + "https://www.youtube.com/playlist/?list=PLxxxxxxxxxx", # trailing slash + ], +) +def test_is_youtube_playlist_url_positives(url): + from second_brain.sources_service import is_youtube_playlist_url + + assert is_youtube_playlist_url(url) is True + + +@pytest.mark.parametrize( + "url", + [ + # Watch URL with list= — ambiguous; we treat as single video. + "https://www.youtube.com/watch?v=abc123&list=PLxxxxxxxxxx", + "https://www.youtube.com/watch?v=abc123", + "https://youtu.be/abc123", + "https://www.youtube.com/@somechannel", + "https://www.youtube.com/playlist", # no list param + "https://example.com/playlist?list=PL", # not a YouTube host + "https://marktechpost.com/some/article", + "", + "not-a-url", + "ftp://www.youtube.com/playlist?list=PLxxxxxxxxxx", + ], +) +def test_is_youtube_playlist_url_negatives(url): + from second_brain.sources_service import is_youtube_playlist_url + + assert is_youtube_playlist_url(url) is False + + +# --------------------------------------------------------------------------- +# DB-backed: add_playlist with a mocked expander +# --------------------------------------------------------------------------- + + +def _db_reachable() -> bool: + url = os.environ.get("SECOND_BRAIN_DATABASE_URL") or os.environ.get( + "HERBYLAB_DATABASE_URL" + ) + if not url: + return False + raw = url.replace("postgresql+psycopg://", "postgresql://", 1) + try: + with psycopg.connect(raw, connect_timeout=2): + return True + except Exception: + return False + + +pytestmark_db = pytest.mark.skipif( + not _db_reachable(), + reason="needs SECOND_BRAIN_DATABASE_URL pointed at a real petalbrain", +) + + +@pytestmark_db +def test_add_playlist_aggregates_counts(): + """5 entries; 2 are pre-seeded duplicates → added=3, duplicates=2, failed=0.""" + from second_brain.database import Database, reset_database_singleton + from second_brain.models import Source, SourceStatus, SourceType, utcnow + from second_brain.sources_service import add_playlist + + reset_database_singleton() + db = Database() + + # Unique URL set per run so re-execution is idempotent. + tag = uuid.uuid4().hex[:8] + urls = [ + f"https://www.youtube.com/watch?v=playlist-{tag}-{i}" for i in range(5) + ] + + # Pre-seed entries 0 and 1 so they're already in the queue. + pre_seeded_ids: list[int] = [] + with db.session() as sess: + for u in urls[:2]: + src = Source( + url=u, + title=f"pre-{tag}", + domain="development", + source_type=SourceType.VIDEO, + status=SourceStatus.PENDING, + ingested_at=utcnow(), + ) + sess.add(src) + sess.flush() + pre_seeded_ids.append(src.id) + + def fake_expander(playlist_url, *, max_items): # noqa: ARG001 + return [{"url": u, "title": f"vid-{tag}-{i}"} for i, u in enumerate(urls)] + + try: + with db.session() as sess: + result = add_playlist( + sess, + url=f"https://www.youtube.com/playlist?list=FAKE-{tag}", + domain="development", + expander=fake_expander, + ) + + assert result.expanded == 5 + assert result.added == 3 + assert result.duplicates == 2 + assert result.failed == 0 + assert result.failures == [] + + with db.session() as sess: + count = sess.query(Source).filter(Source.url.in_(urls)).count() + assert count == 5 # all 5 URLs now in the queue (2 pre-existing + 3 new) + finally: + with db.session() as sess: + sess.query(Source).filter(Source.url.in_(urls)).delete( + synchronize_session=False + ) + reset_database_singleton() + + +@pytestmark_db +def test_add_playlist_tolerates_bad_entries(): + """A bad entry mid-batch must not abort the rest.""" + from second_brain.database import Database, reset_database_singleton + from second_brain.models import Source + from second_brain.sources_service import add_playlist + + reset_database_singleton() + db = Database() + tag = uuid.uuid4().hex[:8] + + entries = [ + {"url": f"https://www.youtube.com/watch?v=good-{tag}-1", "title": "ok 1"}, + {"url": "not-a-url", "title": "bad"}, # validate_url raises + {"url": f"https://www.youtube.com/watch?v=good-{tag}-2", "title": "ok 2"}, + ] + urls_added = [entries[0]["url"], entries[2]["url"]] + + def fake_expander(playlist_url, *, max_items): # noqa: ARG001 + return entries + + try: + with db.session() as sess: + result = add_playlist( + sess, + url=f"https://www.youtube.com/playlist?list=FAKE-{tag}", + expander=fake_expander, + ) + + assert result.expanded == 3 + assert result.added == 2 + assert result.duplicates == 0 + assert result.failed == 1 + assert len(result.failures) == 1 + assert result.failures[0][0] == "not-a-url" + + with db.session() as sess: + count = sess.query(Source).filter(Source.url.in_(urls_added)).count() + assert count == 2 + finally: + with db.session() as sess: + sess.query(Source).filter(Source.url.in_(urls_added)).delete( + synchronize_session=False + ) + reset_database_singleton() + + +@pytestmark_db +def test_add_playlist_empty_expansion(): + """Empty playlist → expanded=0, no rows touched.""" + from second_brain.database import Database, reset_database_singleton + from second_brain.sources_service import add_playlist + + reset_database_singleton() + db = Database() + + def fake_expander(playlist_url, *, max_items): # noqa: ARG001 + return [] + + try: + with db.session() as sess: + result = add_playlist( + sess, + url="https://www.youtube.com/playlist?list=EMPTY", + expander=fake_expander, + ) + assert result.expanded == 0 + assert result.added == 0 + assert result.duplicates == 0 + assert result.failed == 0 + finally: + reset_database_singleton() + + +# --------------------------------------------------------------------------- +# Opt-in live-network test: hit a real (small) public playlist. +# Gate with SECOND_BRAIN_LIVE_NETWORK_TESTS=1 so CI doesn't depend on YouTube. +# --------------------------------------------------------------------------- + + +@pytest.mark.skipif( + os.environ.get("SECOND_BRAIN_LIVE_NETWORK_TESTS") != "1", + reason="set SECOND_BRAIN_LIVE_NETWORK_TESTS=1 to hit yt-dlp against YouTube", +) +def test_expand_youtube_playlist_live(): + """Sanity-check the yt-dlp call shape against a real playlist. + + Uses Google's "YouTube Developers" playlist which has been stable for + years. We only assert the returned shape (list of dicts with `url` + and `title`), not the count — playlist contents change. + """ + from second_brain.sources_service import expand_youtube_playlist + + url = ( + "https://www.youtube.com/playlist?" + "list=PLOU2XLYxmsIKpaV8h0AGE05so0fAwwfTw" + ) + entries = expand_youtube_playlist(url, max_items=5) + assert isinstance(entries, list) + assert len(entries) > 0 + assert len(entries) <= 5 + for e in entries: + assert e["url"].startswith("https://www.youtube.com/watch?v=") + assert "title" in e