petal-power/second-brain
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

playlists: queue-time YouTube fan-out via yt-dlp extract_flat

Browse Source 
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 <playlist-url>` 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.
This commit is contained in:
Travis Herbranson 2026-05-25 13:59:35 -04:00
parent d055d1798d
commit a475893403
4 changed files with 517 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
src/second_brain/main.py
View File

@ -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(

184
src/second_brain/sources_service.py
View File

@ -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",
]

55
src/second_brain/web/app.py
View File

@ -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)

252
tests/test_playlist.py Normal file
View File

@ -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