Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/jackmawer/plex-lb-sync. Synchronise all ListenBrainz playlists (not just the auto-generated ones) to Plex
Something went wrong. Try again.
plex-lb-sync sync.py
10 kB · 284 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285#!/usr/bin/env python3"""Sync ListenBrainz playlists into a Plex music library.
Reads every playlist belonging to a ListenBrainz user (including private ones,if the token permits), resolves each track against a Plex music section, andcreates or replaces the matching Plex playlist. Unresolved tracks are writtento a CSV so you can see what your library is missing."""
from __future__ import annotations
import csvimport loggingimport osimport reimport sysimport timefrom dataclasses import dataclass, fieldfrom pathlib import Pathfrom typing import Any, Iterator
import requestsimport yamlfrom plexapi.server import PlexServer
LB_API = "https://api.listenbrainz.org/1"MB_TRACK_NS = "https://musicbrainz.org/doc/jspf#track"PAGE_SIZE = 25
log = logging.getLogger("lb-plex-sync")
# --------------------------------------------------------------------------# config# --------------------------------------------------------------------------
@dataclassclass Config: lb_user: str plex_url: str plex_token: str lb_token: str | None = None library: str = "Music" prefix: str = "" include: list[str] = field(default_factory=list) # regex, empty = all exclude: list[str] = field(default_factory=list) # regex created_for: bool = True # also pull Weekly Jams / Exploration etc. min_match: float = 0.0 # 0.0-1.0; skip playlist if match rate is worse missing_csv: str = "/config/missing.csv" dry_run: bool = False
@classmethod def load(cls, path: str) -> "Config": raw: dict[str, Any] = {} if Path(path).exists(): raw = yaml.safe_load(Path(path).read_text()) or {} # env vars win, so the container stays declarative either way for f in cls.__dataclass_fields__: env = os.environ.get(f.upper()) if env is None: continue if f in ("include", "exclude"): raw[f] = [x for x in env.split(",") if x] elif f in ("created_for", "dry_run"): raw[f] = env.strip().lower() in ("1", "true", "yes") elif f == "min_match": raw[f] = float(env) else: raw[f] = env missing = [k for k in ("lb_user", "plex_url", "plex_token") if not raw.get(k)] if missing: sys.exit(f"missing required config: {', '.join(missing)}") known = {k: v for k, v in raw.items() if k in cls.__dataclass_fields__} return cls(**known)
def wanted(self, title: str) -> bool: if any(re.search(p, title, re.I) for p in self.exclude): return False if not self.include: return True return any(re.search(p, title, re.I) for p in self.include)
# --------------------------------------------------------------------------# listenbrainz# --------------------------------------------------------------------------
class ListenBrainz: def __init__(self, token: str | None): self.s = requests.Session() if token: self.s.headers["Authorization"] = f"Token {token}"
def _get(self, path: str, **params) -> dict: for attempt in range(5): r = self.s.get(f"{LB_API}{path}", params=params, timeout=30) if r.status_code == 429: wait = int(r.headers.get("X-RateLimit-Reset-In", 5)) log.warning("rate limited, sleeping %ss", wait) time.sleep(wait) continue r.raise_for_status() return r.json() raise RuntimeError(f"gave up on {path}")
def playlists(self, user: str, created_for: bool) -> Iterator[dict]: endpoints = [f"/user/{user}/playlists"] if created_for: endpoints.append(f"/user/{user}/playlists/createdfor") for ep in endpoints: offset = 0 while True: data = self._get(ep, count=PAGE_SIZE, offset=offset) batch = data.get("playlists", []) for item in batch: yield item["playlist"] offset += len(batch) if not batch or offset >= data.get("playlist_count", 0): break
def tracks(self, mbid: str) -> list[dict]: data = self._get(f"/playlist/{mbid}") return data["playlist"].get("track", [])
def playlist_mbid(playlist: dict) -> str: # identifier is a URL ending in the playlist MBID ident = playlist["identifier"] if isinstance(ident, list): ident = ident[0] return ident.rstrip("/").rsplit("/", 1)[-1]
def track_fields(track: dict) -> tuple[str | None, str, str]: """Return (recording_mbid, artist, title).""" mbid = None ident = track.get("identifier") if isinstance(ident, str): ident = [ident] for i in ident or []: if "/recording/" in i: mbid = i.rstrip("/").rsplit("/", 1)[-1] break return mbid, track.get("creator", ""), track.get("title", "")
# --------------------------------------------------------------------------# plex resolution# --------------------------------------------------------------------------
NOISE = re.compile( r"\s*[\(\[][^)\]]*(remaster(ed)?|radio edit|single (edit|version)|" r"original mix|alternate mix|mono|stereo|deluxe|live)[^)\]]*[\)\]]", re.I,)
def normalise(s: str) -> str: s = NOISE.sub("", s) s = re.sub(r"[^\w\s]", " ", s.lower()) return re.sub(r"\s+", " ", s).strip()
class PlexResolver: def __init__(self, server: PlexServer, library: str): self.section = server.library.section(library) self._mbid_index: dict[str, Any] | None = None self._cache: dict[tuple[str, str], Any] = {}
def _build_mbid_index(self) -> dict[str, Any]: """Map recording MBIDs to Plex tracks, if the library is MB-tagged.
Plex exposes the MusicBrainz id in guid for libraries scanned with an agent that keeps it. Libraries tagged with Picard but scanned by the default agent will simply produce an empty index, and we fall back to text matching. """ index: dict[str, Any] = {} for track in self.section.searchTracks(): for guid in [track.guid] + [g.id for g in (track.guids or [])]: if guid and "mbid://" in str(guid): index[str(guid).rsplit("/", 1)[-1]] = track log.info("indexed %d tracks by MBID", len(index)) return index
def resolve(self, mbid: str | None, artist: str, title: str): if mbid: if self._mbid_index is None: self._mbid_index = self._build_mbid_index() hit = self._mbid_index.get(mbid) if hit: return hit
key = (normalise(artist), normalise(title)) if key in self._cache: return self._cache[key]
found = None for candidate in self.section.searchTracks(title=title, maxresults=25): if normalise(candidate.title) != key[1]: continue cand_artist = normalise( getattr(candidate, "originalTitle", None) or candidate.grandparentTitle ) if cand_artist == key[0] or key[0] in cand_artist or cand_artist in key[0]: found = candidate break self._cache[key] = found return found
# --------------------------------------------------------------------------# main# --------------------------------------------------------------------------
def sync(cfg: Config) -> int: lb = ListenBrainz(cfg.lb_token) plex = PlexServer(cfg.plex_url, cfg.plex_token) resolver = PlexResolver(plex, cfg.library) existing = {p.title: p for p in plex.playlists()} missing_rows: list[tuple[str, str, str]] = [] failures = 0
for meta in lb.playlists(cfg.lb_user, cfg.created_for): title = meta.get("title", "").strip() if not cfg.wanted(title): log.debug("skipping %s (filtered)", title) continue
target = f"{cfg.prefix}{title}" try: tracks = lb.tracks(playlist_mbid(meta)) except Exception as exc: log.error("could not fetch %s: %s", title, exc) failures += 1 continue
items, total = [], 0 for t in tracks: total += 1 mbid, artist, name = track_fields(t) hit = resolver.resolve(mbid, artist, name) if hit: items.append(hit) else: missing_rows.append((title, artist, name))
rate = len(items) / total if total else 0 log.info("%s: %d/%d matched (%.0f%%)", title, len(items), total, rate * 100)
if not items: log.warning("%s: nothing in your library, skipping", title) continue if rate < cfg.min_match: log.warning("%s: below min_match %.2f, skipping", title, cfg.min_match) continue if cfg.dry_run: continue
# replace rather than merge, so removed tracks actually disappear old = existing.get(target) if old: old.delete() plex.createPlaylist(target, items=items) log.info("wrote playlist %s", target)
if missing_rows and not cfg.dry_run: path = Path(cfg.missing_csv) path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="") as fh: w = csv.writer(fh) w.writerow(["playlist", "artist", "title"]) w.writerows(missing_rows) log.info("wrote %d unmatched tracks to %s", len(missing_rows), path)
return 1 if failures else 0
if __name__ == "__main__": logging.basicConfig( level=os.environ.get("LOG_LEVEL", "INFO").upper(), format="%(asctime)s %(levelname)s %(message)s", ) sys.exit(sync(Config.load(os.environ.get("CONFIG", "/config/config.yaml"))))