Pierre-Olivier Mercier
75629c829a
All checks were successful
continuous-integration/drone/push Build is passing
A yt-dlp source pointing at a label/artist page flat-extracts to a mix of
/track/ and /album/ URLs. The provider used each verbatim as a locator, so an
/album/ URL was handed to the fetcher as if it were a track: yt-dlp then
(mis)downloaded the whole album into one file and tagged it from the
playlist-level info, which carries the album title and no artist — surfacing as
"Unknown artist" on the stream.
Drill picked container entries down to a single track before emitting a
locator, bounded by a small depth so nested containers (label -> album ->
track) resolve while a real track (which flat-extracts to just itself) is the
base case. Locators are now always downloadable tracks, so the existing
tag/fetch path and anti-repeat keying work as intended.
Also make guess_metadata trust the explicit artist tag over the "Artist -
Title" title split: some label uploads double the artist into the title
("Artist - Artist - Title"), which the blind last-" - " split mis-parsed. When
that artist prefixes the title we peel it off (repeatedly), falling back to the
split only when there is no artist tag.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
218 lines
8.4 KiB
Python
218 lines
8.4 KiB
Python
"""YtdlpProvider: picks tracks from a hand-maintained list of URLs.
|
|
|
|
The list is a plain text file (one URL per line, ``#`` comments allowed),
|
|
mounted into the container so it can be edited without a rebuild. Each line may
|
|
be either a direct track URL or a *container* URL (playlist, album, label,
|
|
artist page); container URLs are expanded with a flat yt-dlp extraction and one
|
|
entry is picked at random, honouring the anti-repeat window.
|
|
|
|
The provider only *resolves* references — it emits ``ytdlp`` tracks whose
|
|
locator is the chosen media URL. The actual download happens in the matching
|
|
fetcher (milestone 4). Flat extraction is cached per source URL to avoid
|
|
re-hitting the network on every pick.
|
|
"""
|
|
|
|
import logging
|
|
import random
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from .. import config, tagging
|
|
from ..db import Database
|
|
from ..fetchers.ytdlp import cached_file
|
|
from ..models import Track
|
|
|
|
log = logging.getLogger("radieo.provider.ytdlp")
|
|
|
|
|
|
class YtdlpProvider:
|
|
name = "ytdlp"
|
|
|
|
# How many container levels (label -> album -> track) to expand while
|
|
# drilling down to a single track. One flat extraction of a label page is
|
|
# not enough: its /album/ entries are themselves containers.
|
|
_MAX_EXPAND_DEPTH = 3
|
|
|
|
def __init__(self, urls_file: Path, db: Database, cache_dir: Path):
|
|
self._urls_file = urls_file
|
|
self._db = db
|
|
self._cache_dir = cache_dir
|
|
self._urls: list[str] = []
|
|
self._loaded_at = 0.0
|
|
# source URL -> (entries, loaded_at); entries are normalized dicts.
|
|
self._expanded: dict[str, tuple[list[dict], float]] = {}
|
|
|
|
# --- source list ------------------------------------------------------
|
|
|
|
def _load_urls(self) -> None:
|
|
now = time.time()
|
|
if self._urls and now - self._loaded_at < config.YTDLP_REFRESH:
|
|
return
|
|
try:
|
|
lines = self._urls_file.read_text(encoding="utf-8").splitlines()
|
|
except OSError:
|
|
self._urls = []
|
|
self._loaded_at = now
|
|
return
|
|
urls = []
|
|
for line in lines:
|
|
line = line.strip()
|
|
if line and not line.startswith("#"):
|
|
urls.append(line)
|
|
if urls != self._urls:
|
|
log.info("loaded %d yt-dlp source URL(s)", len(urls))
|
|
self._urls = urls
|
|
self._loaded_at = now
|
|
|
|
# --- resolution -------------------------------------------------------
|
|
|
|
def next(self) -> Track | None:
|
|
self._load_urls()
|
|
if not self._urls:
|
|
return None
|
|
recent = self._db.recent_locators(config.ANTIREPEAT_WINDOW)
|
|
# Try source lines in random order until one yields a usable track.
|
|
candidates = list(self._urls)
|
|
random.shuffle(candidates)
|
|
for src in candidates:
|
|
track = self._resolve(src, recent)
|
|
if track is not None:
|
|
return track
|
|
return None
|
|
|
|
def _resolve(self, src_url: str, recent: set[str]) -> Track | None:
|
|
try:
|
|
entry = self._pick_entry(src_url, recent, 0)
|
|
except Exception as exc: # yt-dlp raises many extractor-specific errors
|
|
log.warning("yt-dlp could not resolve %s: %s", src_url, exc)
|
|
return None
|
|
if entry is None:
|
|
return None
|
|
locator = entry["url"]
|
|
meta = self._resolved_tags(locator)
|
|
if meta is not None: # a previous download already wrote real tags
|
|
artist, title, mbid = meta.artist, meta.title, meta.mbid
|
|
else:
|
|
# Split the entry's "Artist - Title" the same way the fetcher will
|
|
# after downloading, so the scheduler's canonical anti-repeat keys
|
|
# this provisional track the same before and after the download.
|
|
guess = tagging.guess_metadata(
|
|
{"title": entry.get("title"), "artist": entry.get("artist")}
|
|
)
|
|
artist = guess.artist
|
|
title = guess.title if entry.get("title") else locator
|
|
mbid = None
|
|
return Track(
|
|
backend="ytdlp",
|
|
locator=locator,
|
|
artist=artist,
|
|
title=title,
|
|
origin=self.name,
|
|
mbid=mbid,
|
|
source_url=src_url if src_url != locator else None,
|
|
)
|
|
|
|
def _pick_entry(self, src_url: str, recent: set[str], depth: int) -> dict | None:
|
|
"""Pick one *track* entry under ``src_url``, drilling through containers.
|
|
|
|
A bandcamp label/artist page flat-extracts to a mix of ``/track/`` and
|
|
``/album/`` URLs, and an ``/album/`` URL is itself a container. If such a
|
|
URL were emitted as a locator, the fetcher would (mis)download the whole
|
|
album into a single file and tag it with album-level, artist-less
|
|
metadata — the "Unknown artist" bug. So when the picked entry is itself a
|
|
container we recurse into it, bounded by ``_MAX_EXPAND_DEPTH``, until the
|
|
locator is a single downloadable track.
|
|
|
|
A real track flat-extracts to just itself, which is the recursion's base
|
|
case. That leaf extraction is a full (non-flat) metadata fetch, so the
|
|
entry carries the track's real ``title``/``artist``; it is cached, so the
|
|
cost is paid once per refresh window.
|
|
"""
|
|
entries = self._entries_for(src_url)
|
|
if not entries:
|
|
return None
|
|
if len(entries) == 1 and entries[0]["url"] == src_url:
|
|
return entries[0] # leaf: a track resolves to itself
|
|
pool = [e for e in entries if e["url"] not in recent] or entries
|
|
entry = random.choice(pool)
|
|
url = entry["url"]
|
|
if url == src_url or depth >= self._MAX_EXPAND_DEPTH:
|
|
return entry
|
|
# The picked entry may itself be a container (an album); drill into it.
|
|
# Fall back to the entry as-is if that turns up nothing usable.
|
|
return self._pick_entry(url, recent, depth + 1) or entry
|
|
|
|
def _resolved_tags(self, locator: str):
|
|
"""Corrected identity a previous download wrote into the cached file.
|
|
|
|
The yt-dlp fetcher fixes bandcamp label accounts and writes the real
|
|
artist/title/MBID into the file's tags. Reading them back here — before
|
|
the scheduler keys and anti-repeat-checks the track — makes selection,
|
|
history and cache reuse agree on one identity. Returns None when the
|
|
file isn't cached (yet) or carries no usable tags.
|
|
"""
|
|
if not tagging.is_bandcamp({}, locator):
|
|
return None
|
|
cached = cached_file(self._cache_dir, locator)
|
|
return tagging.read_tags(cached) if cached else None
|
|
|
|
def _entries_for(self, src_url: str) -> list[dict]:
|
|
now = time.time()
|
|
cached = self._expanded.get(src_url)
|
|
if cached and now - cached[1] < config.YTDLP_REFRESH:
|
|
return cached[0]
|
|
entries = self._extract(src_url)
|
|
self._expanded[src_url] = (entries, now)
|
|
return entries
|
|
|
|
@staticmethod
|
|
def _extract(src_url: str) -> list[dict]:
|
|
from yt_dlp import YoutubeDL
|
|
|
|
opts = {
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
"extract_flat": "in_playlist",
|
|
"skip_download": True,
|
|
"socket_timeout": 30,
|
|
}
|
|
with YoutubeDL(opts) as ydl:
|
|
info = ydl.extract_info(src_url, download=False)
|
|
if info is None:
|
|
return []
|
|
|
|
raw_entries = info.get("entries")
|
|
if raw_entries is not None: # container URL -> list of tracks
|
|
out = []
|
|
for e in raw_entries:
|
|
if not e:
|
|
continue
|
|
url = e.get("url") or e.get("webpage_url") or e.get("id")
|
|
if not url:
|
|
continue
|
|
out.append(
|
|
{
|
|
"url": url,
|
|
"title": e.get("title"),
|
|
"artist": (
|
|
e.get("artist")
|
|
or e.get("uploader")
|
|
or e.get("channel")
|
|
or info.get("uploader")
|
|
),
|
|
}
|
|
)
|
|
return out
|
|
|
|
# direct track URL
|
|
return [
|
|
{
|
|
"url": info.get("webpage_url") or src_url,
|
|
"title": info.get("title"),
|
|
"artist": (
|
|
info.get("artist")
|
|
or info.get("uploader")
|
|
or info.get("channel")
|
|
),
|
|
}
|
|
]
|