From a475893403d21fa763ad3b191b88d19f838674ef Mon Sep 17 00:00:00 2001 From: Travis Herbranson Date: Mon, 25 May 2026 13:59:35 -0400 Subject: [PATCH] playlists: queue-time YouTube fan-out via yt-dlp extract_flat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pasting a YouTube playlist URL into either entry point now expands into one source row per video. Single-video URLs and non-YouTube URLs keep their existing behaviour untouched. Service module additions: - is_youtube_playlist_url(url): strict detector. Only `/playlist?list=…` on a known YouTube host (youtube.com / m / music / no-www) counts. A `watch?v=…&list=…` URL is ambiguous (user usually pasted a single video that happens to sit inside a playlist) and intentionally falls through to single-add. To fan out, paste the canonical playlist URL. - expand_youtube_playlist(url, *, max_items=50): yt-dlp with extract_flat=True, playlistend=max_items, skip_download. Builds a canonical https://www.youtube.com/watch?v={id} URL per entry and silently drops placeholders for private/removed videos. - add_playlist(sess, *, url, domain, focus, max_items, expander=None): loops expansion entries through add_source so URL validation, source_type detection, and the UNIQUE dedupe path stay identical to the single-add flow. Per-entry titles win over any caller-supplied title (a single playlist title would be wrong for N videos). Partial failures don't abort the batch — failed entries are tallied with up-to-10 (url, reason) tuples for the flash. `expander` is an injection seam for tests so the suite never hits YouTube unless explicitly opted in. - DEFAULT_PLAYLIST_MAX_ITEMS = 50 — shared ceiling, no throttle change. CLI: `second-brain add ` auto-detects and reports `expanded / added / duplicates / failed`. No new flag needed. Web: POST /sources/add same detection. _playlist_flash() builds the HTMX flash — "Queued N videos (M duplicates skipped, F failed)" with sensible plural forms and graceful omission of zero counters. Tests: - 15 pure-Python detection cases (positives + negatives, including the ambiguous watch?v=…&list=… rule). - 3 DB-backed add_playlist tests (with a mocked expander, so no network): count aggregation across new + pre-seeded duplicates, bad-entry tolerance, and the empty-playlist case. - 1 opt-in live-network test gated on SECOND_BRAIN_LIVE_NETWORK_TESTS=1 exercising expand_youtube_playlist against a real public playlist. Live-verified end to end: - web POST of a real 13-entry public playlist queued 13 video rows with titles, flash showed "Queued 13 videos". - re-POST returned "Queued 0 videos (13 duplicates skipped)". - watch?v=…&list=… correctly stayed a single-add. - CLI parity confirmed against the same playlist. --- src/second_brain/main.py | 44 ++++- src/second_brain/sources_service.py | 184 +++++++++++++++++++- src/second_brain/web/app.py | 55 ++++-- tests/test_playlist.py | 252 ++++++++++++++++++++++++++++ 4 files changed, 517 insertions(+), 18 deletions(-) create mode 100644 tests/test_playlist.py 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