5131f3ccf6
Adds per-record edit mode and retry to the source detail page, plus a
CLI `second-brain retry <id>` for parity.
Edit mode exposes domain (select), title, focus on /sources/{id}.
URL stays read-only — it's the UNIQUE dedupe key and the embeddings
identity tuple. Save goes through a new service helper
`update_source_metadata` so the web route stays a thin wrapper.
Retry exposes two modes, both available on any record:
- full: status -> PENDING (re-pull + re-process)
- extract: status -> TRANSCRIBED (re-extract only, requires an
existing transcript; offered when source_type=video and a
transcript is present, to avoid the expensive re-download path)
Both modes clear claimed_by/claimed_at and error_message so the queue
dance picks it up cleanly.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
401 lines
12 KiB
Python
401 lines
12 KiB
Python
"""Service-layer helpers for the `sources` table.
|
|
|
|
Single source of truth for "queue a URL" — both the `second-brain add`
|
|
CLI and the web POST /sources/add route call into here, so they can't
|
|
drift on validation, dedupe, or default semantics.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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",
|
|
"www.youtube.com",
|
|
"youtu.be",
|
|
"vimeo.com",
|
|
"www.vimeo.com",
|
|
}
|
|
|
|
|
|
def is_article_url(url: str) -> bool:
|
|
"""Match the CLI's `_is_article` heuristic: anything not on a known
|
|
video host is treated as an article."""
|
|
parsed = urlparse(url)
|
|
return parsed.netloc not in _VIDEO_HOSTS
|
|
|
|
|
|
def validate_url(url: str) -> str:
|
|
"""Normalize and validate the input URL. Raises ValueError on bad shape."""
|
|
if url is None:
|
|
raise ValueError("url is required")
|
|
url = url.strip()
|
|
if not url:
|
|
raise ValueError("url is required")
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ("http", "https"):
|
|
raise ValueError("url must use http or https")
|
|
if not parsed.netloc:
|
|
raise ValueError("url is missing a host")
|
|
return url
|
|
|
|
|
|
@dataclass
|
|
class AddResult:
|
|
"""Outcome of `add_source`. `added=False` means the URL was already queued."""
|
|
|
|
source: Source
|
|
added: bool
|
|
|
|
|
|
def add_source(
|
|
sess: Session,
|
|
*,
|
|
url: str,
|
|
domain: str = "development",
|
|
focus: Optional[str] = None,
|
|
title: Optional[str] = None,
|
|
) -> AddResult:
|
|
"""Queue a single source URL.
|
|
|
|
Mirrors the CLI add path exactly:
|
|
- normalize + validate URL
|
|
- reject unknown domains
|
|
- if the URL is already queued, return the existing row with added=False
|
|
- otherwise insert a PENDING row and return it with added=True
|
|
|
|
Does NOT open or commit a session — caller owns the transaction
|
|
boundary (matches the rest of the codebase). For the CLI, that's
|
|
`db.session()`; for the web route, the request handler's
|
|
`with db.session()` block.
|
|
"""
|
|
url = validate_url(url)
|
|
|
|
if domain not in DOMAINS:
|
|
raise ValueError(
|
|
f"unknown domain {domain!r}; valid: {', '.join(DOMAINS)}"
|
|
)
|
|
|
|
title = (title or "").strip() or None
|
|
focus = (focus or "").strip() or None
|
|
|
|
existing = sess.query(Source).filter_by(url=url).first()
|
|
if existing is not None:
|
|
return AddResult(source=existing, added=False)
|
|
|
|
source_type = (
|
|
SourceType.ARTICLE if is_article_url(url) else SourceType.VIDEO
|
|
)
|
|
|
|
source = Source(
|
|
url=url,
|
|
title=title,
|
|
domain=domain,
|
|
focus=focus,
|
|
source_type=source_type,
|
|
status=SourceStatus.PENDING,
|
|
ingested_at=utcnow(),
|
|
)
|
|
sess.add(source)
|
|
sess.flush()
|
|
return AddResult(source=source, added=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edit + retry helpers (shared by web routes and CLI)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
# Fields the edit-mode form is allowed to mutate. URL is intentionally
|
|
# excluded — it's the UNIQUE dedupe key and editing it risks collisions
|
|
# plus breaks the embeddings 4-tuple identity (source_id is stable but
|
|
# semantic identity drifts).
|
|
EDITABLE_METADATA_FIELDS = ("domain", "title", "focus")
|
|
|
|
|
|
def update_source_metadata(
|
|
sess: Session,
|
|
*,
|
|
source_id: int,
|
|
domain: Optional[str] = None,
|
|
title: Optional[str] = None,
|
|
focus: Optional[str] = None,
|
|
) -> Source:
|
|
"""Update editable metadata on a source row.
|
|
|
|
Empty strings collapse to NULL for title/focus (matches `add_source`
|
|
semantics). Domain is required to stay in the configured set —
|
|
bumping the constraint is a separate decision.
|
|
|
|
Raises ValueError on validation failures and LookupError if the row
|
|
is missing. Caller owns the transaction boundary.
|
|
"""
|
|
source = sess.query(Source).filter(Source.id == source_id).first()
|
|
if source is None:
|
|
raise LookupError(f"source id={source_id} not found")
|
|
|
|
if domain is not None:
|
|
if domain not in DOMAINS:
|
|
raise ValueError(
|
|
f"unknown domain {domain!r}; valid: {', '.join(DOMAINS)}"
|
|
)
|
|
source.domain = domain
|
|
|
|
if title is not None:
|
|
source.title = title.strip() or None
|
|
|
|
if focus is not None:
|
|
source.focus = focus.strip() or None
|
|
|
|
source.updated_at = utcnow()
|
|
sess.flush()
|
|
return source
|
|
|
|
|
|
# Retry modes — keep the surface small. Anything else is a separate ask.
|
|
RETRY_FULL = "full"
|
|
RETRY_EXTRACT = "extract"
|
|
RETRY_MODES = (RETRY_FULL, RETRY_EXTRACT)
|
|
|
|
|
|
def reset_source_for_retry(
|
|
sess: Session,
|
|
*,
|
|
source_id: int,
|
|
mode: str = RETRY_FULL,
|
|
) -> Source:
|
|
"""Reset a source so the pipeline re-processes it.
|
|
|
|
Modes:
|
|
- "full": status → PENDING. Article pull or video download +
|
|
transcribe will happen again on the appropriate worker.
|
|
- "extract": status → TRANSCRIBED. Requires the row to already
|
|
carry a transcript (transcript_text). The dev scheduler will
|
|
re-run extraction without re-pulling the source. Cheap.
|
|
|
|
Always clears `claimed_by` / `claimed_at` / `error_message` so the
|
|
queue dance can pick it up cleanly and stale errors don't bleed
|
|
into the UI.
|
|
|
|
Raises ValueError on bad mode / missing transcript for extract mode,
|
|
and LookupError if the row is missing.
|
|
"""
|
|
if mode not in RETRY_MODES:
|
|
raise ValueError(
|
|
f"unknown retry mode {mode!r}; valid: {', '.join(RETRY_MODES)}"
|
|
)
|
|
|
|
source = sess.query(Source).filter(Source.id == source_id).first()
|
|
if source is None:
|
|
raise LookupError(f"source id={source_id} not found")
|
|
|
|
if mode == RETRY_EXTRACT:
|
|
if not (source.transcript_text and source.transcript_text.strip()):
|
|
raise ValueError(
|
|
"extract-only retry requires an existing transcript; "
|
|
"use full retry instead"
|
|
)
|
|
source.status = SourceStatus.TRANSCRIBED
|
|
else:
|
|
source.status = SourceStatus.PENDING
|
|
|
|
source.claimed_by = None
|
|
source.claimed_at = None
|
|
source.error_message = None
|
|
source.updated_at = utcnow()
|
|
sess.flush()
|
|
return source
|
|
|
|
|
|
__all__ = [
|
|
"AddResult",
|
|
"DEFAULT_PLAYLIST_MAX_ITEMS",
|
|
"EDITABLE_METADATA_FIELDS",
|
|
"PlaylistAddResult",
|
|
"RETRY_EXTRACT",
|
|
"RETRY_FULL",
|
|
"RETRY_MODES",
|
|
"add_playlist",
|
|
"add_source",
|
|
"expand_youtube_playlist",
|
|
"is_article_url",
|
|
"is_youtube_playlist_url",
|
|
"reset_source_for_retry",
|
|
"update_source_metadata",
|
|
"validate_url",
|
|
]
|