music-curator/music_curator.py

"""Build a local record of what actually gets listened to.

Ingests a Last.fm scrobble history into SQLite and keeps it current, so that
the later stages -- building playlists, and deciding what in the library has
gone cold -- can query listening habits locally instead of spending an API call
per question.

The store is derived state. It can be deleted and rebuilt from Last.fm at any
time. Nothing here writes to Last.fm, and nothing here touches the music
library or Lidarr.

Ingest is in two halves. A backfill walks the history backwards until it runs
out, and a catch-up fetches everything scrobbled since the newest scrobble
already held. Both take their bounds from the database rather than from a saved
cursor, so an interrupted run resumes from what it actually has.
"""

import argparse
import fcntl
import http.client
import io
import json
import logging
import os
import re
import signal
import sqlite3
import sys
import tempfile
import time
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

logger = logging.getLogger("music-curator")

API_ROOT = "https://ws.audioscrobbler.com/2.0/"

# The documented ceiling for user.getRecentTracks. getLovedTracks does not state
# one; asking for more than it allows is harmless, because paging is driven by
# the totalPages the response reports rather than by arithmetic on this number.
PAGE_SIZE = 200

# Error codes worth another attempt: backend failure, service offline, service
# temporarily unavailable, rate limit. Everything else is a fault in the request
# or the key and will fail again identically.
RETRYABLE_ERRORS = {8, 11, 16, 29}
RETRYABLE_STATUS = {429, 500, 502, 503, 504}

# Lidarr's own list, and 500 is deliberately absent. A 500 from Lidarr is an
# unhandled exception inside its serialisation, not a busy server; it will be
# raised again identically, and retrying only delays finding that out.
LIDARR_RETRYABLE_STATUS = {429, 502, 503, 504}

# Last.fm asks for no more than five requests a second averaged over five
# minutes. A full backfill is thousands of requests, so it is worth staying
# well inside that rather than discovering error 29 halfway through.
REQUEST_DELAY_SECONDS = 0.25
BACKOFF_SECONDS = 2.0
BACKOFF_CEILING_SECONDS = 60.0

# What music-mirror names its output, and where playlists are written inside the
# mirror. music-mirror's prune only deletes `*.mp3` and only removes directories
# it finds empty, so a directory of M3Us survives it untouched.
MIRROR_SUFFIX = ".mp3"
PLAYLIST_DIRECTORY = "_playlists"

# Tags are re-fetched this often. They move slowly, and the first pass over a
# library already costs one request per artist.
TAG_REFRESH_SECONDS = 90 * 86400

# How much of an artist's tag weight has to fall inside a vibe before their
# tracks are eligible for it. Weights are Last.fm's own 0-100 popularity, summed
# across whichever of the vibe's tags the artist carries.
VIBE_MIN_SCORE = 30

# A vibe writes a file named after itself, so the name has to be a filename.
SAFE_VIBE_NAME = re.compile(r"^[a-z0-9][a-z0-9-]*$")

# Group-readable, for the same reason music-mirror sets it: the mirror is read
# back by whatever serves it, and a playlist nobody can read is not a playlist.
GROUP_READ = 0o040

SCHEMA_VERSION = "2"

# Versions this build upgrades in place. Everything added since version 1 is a
# new table, and the schema script only ever creates what is missing, so running
# it is the whole migration -- an existing history is not re-downloaded.
MIGRATABLE_FROM = {"1"}

SCHEMA = """
-- One row per scrobble. The primary key collapses two plays of the same track
-- in the same second into one: Last.fm gives scrobbles no identifier, so they
-- are genuinely indistinguishable, and one lost play a decade is not worth a
-- surrogate key that would make re-ingest non-idempotent.
CREATE TABLE IF NOT EXISTS scrobble (
    uts         INTEGER NOT NULL,
    artist      TEXT    NOT NULL,
    track       TEXT    NOT NULL,
    album       TEXT    NOT NULL DEFAULT '',
    artist_mbid TEXT,
    album_mbid  TEXT,
    track_mbid  TEXT,
    PRIMARY KEY (uts, artist, track)
);
CREATE INDEX IF NOT EXISTS scrobble_uts ON scrobble (uts);
CREATE INDEX IF NOT EXISTS scrobble_artist_track ON scrobble (artist, track);
CREATE INDEX IF NOT EXISTS scrobble_track_mbid ON scrobble (track_mbid);

-- Loved tracks are current state, not history: a track can be unloved again.
-- The table is replaced on every pass rather than accumulated.
CREATE TABLE IF NOT EXISTS loved (
    artist     TEXT NOT NULL,
    track      TEXT NOT NULL,
    track_mbid TEXT,
    loved_at   INTEGER,
    PRIMARY KEY (artist, track)
);

CREATE TABLE IF NOT EXISTS state (
    key   TEXT PRIMARY KEY,
    value TEXT NOT NULL
);

-- The library as Lidarr sees it. Rebuilt from the API rather than by walking
-- the files, because Lidarr is the only thing holding the MusicBrainz ids that
-- make an exact join possible.
CREATE TABLE IF NOT EXISTS lidarr_artist (
    id        INTEGER PRIMARY KEY,
    mbid      TEXT,
    name      TEXT NOT NULL,
    norm_name TEXT NOT NULL,
    path      TEXT,
    monitored INTEGER NOT NULL DEFAULT 1
);
CREATE INDEX IF NOT EXISTS lidarr_artist_norm ON lidarr_artist (norm_name);

CREATE TABLE IF NOT EXISTS lidarr_album (
    id           INTEGER PRIMARY KEY,
    artist_id    INTEGER NOT NULL,
    mbid         TEXT,
    title        TEXT NOT NULL,
    monitored    INTEGER NOT NULL DEFAULT 1,
    release_date TEXT
);
CREATE INDEX IF NOT EXISTS lidarr_album_artist ON lidarr_album (artist_id);

CREATE TABLE IF NOT EXISTS lidarr_track (
    id             INTEGER PRIMARY KEY,
    artist_id      INTEGER NOT NULL,
    album_id       INTEGER NOT NULL,
    recording_mbid TEXT,
    title          TEXT NOT NULL,
    norm_artist    TEXT NOT NULL,
    norm_title     TEXT NOT NULL,
    has_file       INTEGER NOT NULL DEFAULT 0,
    path           TEXT,
    added          INTEGER,
    duration       INTEGER
);
CREATE INDEX IF NOT EXISTS lidarr_track_recording ON lidarr_track (recording_mbid);
CREATE INDEX IF NOT EXISTS lidarr_track_norm ON lidarr_track (norm_artist, norm_title);
-- Separate from the composite above, which a lookup by title alone cannot use:
-- its leading column is the artist. The report searches by title on its own, and
-- without this it scans every track for every unmatched key -- forty-seven
-- seconds on a library of eighty-four thousand.
CREATE INDEX IF NOT EXISTS lidarr_track_title ON lidarr_track (norm_title);
CREATE INDEX IF NOT EXISTS lidarr_track_album ON lidarr_track (album_id);

-- One row per distinct thing listened to, with the verdict on whether it could
-- be tied to the library. Matching belongs at this granularity rather than per
-- play: it is a property of the name pair, and there are three plays for every
-- pair.
CREATE TABLE IF NOT EXISTS scrobble_key (
    artist      TEXT NOT NULL,
    track       TEXT NOT NULL,
    norm_artist TEXT NOT NULL,
    norm_track  TEXT NOT NULL,
    track_mbid  TEXT,
    plays       INTEGER NOT NULL,
    first_uts   INTEGER NOT NULL,
    last_uts    INTEGER NOT NULL,
    track_id    INTEGER,
    method      TEXT NOT NULL DEFAULT 'none',
    PRIMARY KEY (artist, track)
);
CREATE INDEX IF NOT EXISTS scrobble_key_track ON scrobble_key (track_id);

-- Last.fm's crowd tags for each library artist. MusicBrainz genres come free
-- with the Lidarr index and are useless for this: they are sparse and formal,
-- and will not tell you a record is screamo or synthwave. People typing tags
-- will. Keyed on the normalised name rather than the Lidarr id so it survives
-- an artist being removed and re-added there.
CREATE TABLE IF NOT EXISTS artist_tag (
    norm_artist TEXT    NOT NULL,
    tag         TEXT    NOT NULL,
    weight      INTEGER NOT NULL,
    PRIMARY KEY (norm_artist, tag)
);
CREATE INDEX IF NOT EXISTS artist_tag_tag ON artist_tag (tag);

CREATE TABLE IF NOT EXISTS artist_tag_fetched (
    norm_artist TEXT PRIMARY KEY,
    fetched_at  INTEGER NOT NULL
);
"""


# Suffixes that a file tag carries and a scrobble does not, or the reverse.
# Stripped only from a trailing bracket or after a trailing dash, so a title
# like "(Don't Fear) The Reaper" keeps its opening parenthetical.
VERSION_WORDS = (
    "remaster",
    "remastered",
    "live",
    "mono",
    "stereo",
    "version",
    "edit",
    "mix",
    "remix",
    "deluxe",
    "bonus",
    "explicit",
    "acoustic",
    "demo",
    "radio",
    "single",
    "anniversary",
    "reissue",
    "instrumental",
    # Drum and bass and its neighbours mark versions their own way.
    "vip",
    "bootleg",
    "rework",
    "extended",
)
_VERSIONS = "|".join(VERSION_WORDS)
BRACKETED_VERSION = re.compile(
    rf"\s*[\(\[][^\)\]]*\b(?:{_VERSIONS})\b[^\)\]]*[\)\]]\s*$", re.IGNORECASE
)
# The suffix is matched lazily rather than as a run of non-hyphens, because the
# thing being stripped frequently contains hyphens of its own -- "Gold Dust -
# Shy FX Re-Edit", "Back To Your Roots - Friction & K-Tee Remix".
TRAILING_VERSION = re.compile(rf"\s+-\s+.*?\b(?:{_VERSIONS})\b.*$", re.IGNORECASE)

# Last.fm routinely carries the guest credit where the file tag holds only the
# primary artist -- "Yellowcard feat. Tay Jardine" against a tag of
# "Yellowcard", or "Self vs Self (feat. In Flames)" against "Self vs Self". The
# opening bracket has to be allowed for: requiring whitespace immediately before
# the word misses every bracketed credit, which is most of them. `with` is
# deliberately absent -- it appears in far too many real titles to cut on sight.
GUEST_CREDIT = re.compile(r"[\s(\[]+(?:feat|ft|featuring)\b.*$", re.IGNORECASE)

LEADING_ARTICLE = re.compile(r"^the\s+")
# Deleted rather than spaced, so "Don't" and "Dont" agree. Every other mark
# becomes a space instead, so "AC/DC", "AC-DC" and "AC DC" all agree too; the
# two rules pull opposite ways and both are needed.
APOSTROPHES = re.compile(r"['‘’ʼ`´]")
NOT_ALPHANUMERIC = re.compile(r"[^0-9a-z ]+")
WHITESPACE = re.compile(r"\s+")


class LastfmError(Exception):
    """A Last.fm request that failed in a way retrying will not fix.

    `code` is the service's own error number where the failure came from the
    API rather than the transport. Callers need it to tell "this thing does not
    exist", which is final, from "something went wrong", which is not.
    """

    def __init__(self, message, code=None):
        super().__init__(message)
        self.code = code


class LidarrError(Exception):
    """A Lidarr request that failed.

    `transient` separates "the network or the server had a moment" from "this
    request will fail identically forever". The distinction matters twice: only
    the first is worth retrying, and only the second is worth investigating,
    since probing a library-wide outage artist by artist multiplies the load
    that caused it.
    """

    def __init__(self, message, transient=False):
        super().__init__(message)
        self.transient = transient


def normalise(text):
    """Return a comparison key for an artist or a track title.

    Scrobbles and file tags disagree in predictable ways, and every one of them
    has to be flattened or a track that was played reads as one that was not.
    That is the direction that costs music, so this leans towards collapsing too
    much rather than too little: a false match makes something look played, a
    missed match makes something look abandoned.
    """
    if not text:
        return ""
    text = unicodedata.normalize("NFKD", str(text)).casefold()
    text = "".join(character for character in text if not unicodedata.combining(character))
    text = GUEST_CREDIT.sub("", text)
    # A title can carry more than one, e.g. "Song (Live) (Remastered)".
    while True:
        stripped = BRACKETED_VERSION.sub("", text)
        if stripped == text:
            break
        text = stripped
    text = TRAILING_VERSION.sub("", text)
    text = text.replace("&", " and ")
    text = APOSTROPHES.sub("", text)
    text = NOT_ALPHANUMERIC.sub(" ", text)
    text = WHITESPACE.sub(" ", text).strip()
    return LEADING_ARTICLE.sub("", text)


@dataclass(frozen=True)
class Scrobble:
    """One play, as Last.fm recorded it."""

    uts: int
    artist: str
    track: str
    album: str = ""
    artist_mbid: str | None = None
    album_mbid: str | None = None
    track_mbid: str | None = None


def parse_interval(interval):
    """Return seconds for an interval such as ``30m``, ``6h`` or ``90``."""
    text = str(interval).strip().lower()
    match = re.fullmatch(r"([0-9]+)([smhd]?)", text)
    if not match:
        raise ValueError(f"unrecognised interval {interval!r}: expected e.g. 45m, 6h, 1d")
    value = int(match.group(1))
    return value * {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400}[match.group(2)]


def as_list(value):
    """Return a list for a field Last.fm renders as a bare object when singular."""
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]


def name_of(value):
    """Return the name out of a Last.fm sub-object.

    The same conceptual field is ``#text`` in some responses (recent tracks) and
    ``name`` in others (loved tracks, and anything fetched with extended=1).
    """
    if isinstance(value, dict):
        return (value.get("name") or value.get("#text") or "").strip()
    return (value or "").strip()


def mbid_of(value):
    """Return an MBID, or None where the field is present but empty.

    Last.fm sends an empty string rather than omitting the key, and an empty
    string would otherwise look like a usable join key further down the line.
    """
    if isinstance(value, dict):
        value = value.get("mbid")
    value = (value or "").strip()
    return value or None


def parse_scrobble(entry):
    """Return a Scrobble for one recent-tracks entry, or None to skip it.

    The currently-playing track comes back with no ``date`` at all. Storing it
    would mean a scrobble with no timestamp, and it would arrive again on every
    pass until the song ended, so it is dropped until it has a real one.
    """
    date = entry.get("date") or {}
    if "uts" not in date:
        return None
    artist = name_of(entry.get("artist"))
    track = (entry.get("name") or "").strip()
    if not artist or not track:
        return None
    return Scrobble(
        uts=int(date["uts"]),
        artist=artist,
        track=track,
        album=name_of(entry.get("album")),
        artist_mbid=mbid_of(entry.get("artist")),
        album_mbid=mbid_of(entry.get("album")),
        track_mbid=mbid_of(entry.get("mbid")),
    )


def http_get(url, timeout=30, headers=None):
    """Fetch a URL and return its body. Replaced in tests."""
    request = urllib.request.Request(url, headers=headers or {})
    with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310
        return response.read().decode("utf-8")


class Lastfm:
    """Minimal read-only Last.fm client.

    Only an API key is needed: none of the methods used here authenticate a
    user, so there is no session key, no signing and no secret to hold.
    """

    def __init__(
        self,
        api_key,
        delay=REQUEST_DELAY_SECONDS,
        attempts=5,
        backoff=BACKOFF_SECONDS,
        transport=None,
    ):
        self.api_key = api_key
        self.delay = delay
        self.attempts = attempts
        self.backoff = backoff
        # Resolved here rather than as a default argument so the tests can
        # replace the module-level fetcher.
        self.transport = transport or http_get
        self._previous_request = None

    def call(self, method, params):
        """Return the decoded response for one API method."""
        query = {key: value for key, value in params.items() if value is not None}
        query.update(method=method, api_key=self.api_key, format="json")
        url = f"{API_ROOT}?{urllib.parse.urlencode(query)}"

        for attempt in range(1, self.attempts + 1):
            self._throttle()
            try:
                payload = json.loads(self.transport(url))
            except urllib.error.HTTPError as error:
                if error.code not in RETRYABLE_STATUS:
                    raise LastfmError(f"{method}: HTTP {error.code}") from error
                self._retry_or_raise(method, attempt, f"HTTP {error.code}", error)
                continue
            except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as error:
                self._retry_or_raise(method, attempt, str(error), error)
                continue

            code = payload.get("error")
            if code is None:
                return payload
            detail = f"error {code}: {payload.get('message', '')}".strip()
            if code not in RETRYABLE_ERRORS:
                raise LastfmError(f"{method}: {detail}", code=code)
            self._retry_or_raise(method, attempt, detail, None)

        raise LastfmError(f"{method}: gave up after {self.attempts} attempts")

    def _retry_or_raise(self, method, attempt, detail, cause):
        """Sleep before the next attempt, or give up if this was the last one."""
        if attempt >= self.attempts:
            raise LastfmError(f"{method}: {detail}") from cause
        pause = min(self.backoff * 2 ** (attempt - 1), BACKOFF_CEILING_SECONDS)
        logger.warning("%s: %s; retrying in %.0fs", method, detail, pause)
        time.sleep(pause)

    def _throttle(self):
        """Keep consecutive requests at least `delay` apart."""
        if self.delay <= 0:
            return
        if self._previous_request is not None:
            waited = time.monotonic() - self._previous_request
            if waited < self.delay:
                time.sleep(self.delay - waited)
        self._previous_request = time.monotonic()


class Store:
    """The local scrobble database."""

    def __init__(self, path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.connection = sqlite3.connect(self.path)
        self.connection.row_factory = sqlite3.Row
        # WAL so a long backfill does not lock a reader out, and a relaxed sync
        # because the whole store can be rebuilt from Last.fm if a crash eats
        # the tail of it.
        self.connection.execute("PRAGMA journal_mode = WAL")
        self.connection.execute("PRAGMA synchronous = NORMAL")
        # Registered so matching runs in SQL, rather than pulling twenty
        # thousand name pairs into Python to compare them one at a time.
        self.connection.create_function("normalise", 1, normalise, deterministic=True)
        self.connection.executescript(SCHEMA)
        self._check_version()

    def _check_version(self):
        held = self.get_state("schema_version")
        if held is None or held in MIGRATABLE_FROM:
            if held is not None:
                logger.info("store migrated from schema version %s to %s", held, SCHEMA_VERSION)
            self.set_state("schema_version", SCHEMA_VERSION)
        elif held != SCHEMA_VERSION:
            raise RuntimeError(
                f"{self.path} is schema version {held}, this build expects {SCHEMA_VERSION}; "
                "delete it and let it rebuild"
            )

    def close(self):
        self.connection.close()

    def get_state(self, key):
        row = self.connection.execute("SELECT value FROM state WHERE key = ?", (key,)).fetchone()
        return row["value"] if row else None

    def set_state(self, key, value):
        with self.connection:
            self.connection.execute(
                "INSERT INTO state (key, value) VALUES (?, ?)"
                " ON CONFLICT (key) DO UPDATE SET value = excluded.value",
                (key, value),
            )

    def add_scrobbles(self, scrobbles):
        """Insert scrobbles, ignoring any already held. Returns how many were new."""
        if not scrobbles:
            return 0
        with self.connection:
            before = self.connection.total_changes
            self.connection.executemany(
                "INSERT OR IGNORE INTO scrobble"
                " (uts, artist, track, album, artist_mbid, album_mbid, track_mbid)"
                " VALUES (?, ?, ?, ?, ?, ?, ?)",
                [
                    (
                        item.uts,
                        item.artist,
                        item.track,
                        item.album,
                        item.artist_mbid,
                        item.album_mbid,
                        item.track_mbid,
                    )
                    for item in scrobbles
                ],
            )
            return self.connection.total_changes - before

    def replace_loved(self, rows):
        """Replace the loved-track list wholesale, in one transaction."""
        with self.connection:
            self.connection.execute("DELETE FROM loved")
            self.connection.executemany(
                "INSERT OR IGNORE INTO loved (artist, track, track_mbid, loved_at)"
                " VALUES (?, ?, ?, ?)",
                rows,
            )

    def replace_library(self, artists, albums, tracks):
        """Swap in a freshly indexed library, in one transaction."""
        with self.connection:
            for table in ("lidarr_track", "lidarr_album", "lidarr_artist"):
                self.connection.execute(f"DELETE FROM {table}")
            self.connection.executemany(
                "INSERT INTO lidarr_artist (id, mbid, name, norm_name, path, monitored)"
                " VALUES (?, ?, ?, ?, ?, ?)",
                artists,
            )
            self.connection.executemany(
                "INSERT INTO lidarr_album"
                " (id, artist_id, mbid, title, monitored, release_date)"
                " VALUES (?, ?, ?, ?, ?, ?)",
                albums,
            )
            self.connection.executemany(
                "INSERT INTO lidarr_track"
                " (id, artist_id, album_id, recording_mbid, title, norm_artist, norm_title,"
                "  has_file, path, added, duration)"
                " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                tracks,
            )

    def replace_tags(self, norm_artist, pairs, now):
        """Replace one artist's tags, and record that they were fetched.

        The fetch is recorded even when nothing came back, so an artist Last.fm
        has never heard of is not asked about again on every pass.
        """
        with self.connection:
            self.connection.execute("DELETE FROM artist_tag WHERE norm_artist = ?", (norm_artist,))
            self.connection.executemany(
                "INSERT OR IGNORE INTO artist_tag (norm_artist, tag, weight) VALUES (?, ?, ?)",
                [(norm_artist, tag, weight) for tag, weight in pairs],
            )
            self.connection.execute(
                "INSERT INTO artist_tag_fetched (norm_artist, fetched_at) VALUES (?, ?)"
                " ON CONFLICT (norm_artist) DO UPDATE SET fetched_at = excluded.fetched_at",
                (norm_artist, now),
            )

    def rebuild_keys(self):
        """Collapse the scrobble history into one row per distinct track.

        Rows whose plays have not changed keep their existing verdict until the
        matcher overwrites it; rows for tracks no longer in the history are
        dropped, so an edited history does not leave orphans behind.
        """
        with self.connection:
            self.connection.execute(
                "DELETE FROM scrobble_key WHERE (artist, track) NOT IN"
                " (SELECT artist, track FROM scrobble)"
            )
            self.connection.execute(
                "INSERT INTO scrobble_key"
                " (artist, track, norm_artist, norm_track, track_mbid, plays,"
                "  first_uts, last_uts)"
                " SELECT artist, track, normalise(artist), normalise(track),"
                "        MAX(track_mbid), COUNT(*), MIN(uts), MAX(uts)"
                " FROM scrobble GROUP BY artist, track"
                " ON CONFLICT (artist, track) DO UPDATE SET"
                "   norm_artist = excluded.norm_artist,"
                "   norm_track  = excluded.norm_track,"
                "   track_mbid  = excluded.track_mbid,"
                "   plays       = excluded.plays,"
                "   first_uts   = excluded.first_uts,"
                "   last_uts    = excluded.last_uts"
            )

    def scalar(self, sql):
        return self.connection.execute(sql).fetchone()[0]

    def newest_uts(self):
        return self.scalar("SELECT MAX(uts) FROM scrobble")

    def oldest_uts(self):
        return self.scalar("SELECT MIN(uts) FROM scrobble")

    def count(self):
        return self.scalar("SELECT COUNT(*) FROM scrobble")


def fetch_page(client, user, to=None, since=None, page=1):
    """Return the scrobbles on one page of the recent-tracks history."""
    payload = client.call(
        "user.getRecentTracks",
        {"user": user, "limit": PAGE_SIZE, "page": page, "from": since, "to": to},
    )
    block = payload.get("recenttracks") or {}
    entries = as_list(block.get("track"))
    parsed = [scrobble for scrobble in map(parse_scrobble, entries) if scrobble is not None]
    attributes = block.get("@attr") or {}
    return parsed, int(attributes.get("totalPages") or 1)


def backfill(client, store, user, now, limit=0):
    """Walk the history backwards until it runs out. Returns scrobbles added.

    Each request asks for the newest page of everything at or before a cursor,
    and the cursor is then moved to the oldest scrobble that came back. There is
    no page number to keep across runs, so an interrupted backfill resumes from
    whatever the database holds.
    """
    if store.get_state("backfill_complete") == "yes":
        return 0

    cursor = store.oldest_uts()
    if cursor is None:
        cursor = now
    page = 1
    added = 0
    requests = 0

    while limit <= 0 or requests < limit:
        scrobbles, _ = fetch_page(client, user, to=cursor, page=page)
        requests += 1
        if not scrobbles:
            store.set_state("backfill_complete", "yes")
            logger.info("backfill complete: %d scrobbles held", store.count())
            return added

        added += store.add_scrobbles(scrobbles)
        page_oldest = min(scrobble.uts for scrobble in scrobbles)
        if page_oldest < cursor:
            cursor, page = page_oldest, 1
        else:
            # Every scrobble on this page shares the cursor's second. Moving the
            # window would step over the rest of them, so take the next page of
            # the same window instead.
            page += 1
        logger.info("backfill: %d added, reached %s", added, format_time(page_oldest))

    logger.info("backfill paused after %d requests; resumes next pass", requests)
    return added


def catch_up(client, store, user, now):
    """Fetch everything scrobbled since the newest scrobble held.

    Bounded at both ends, so the window cannot shift under the paging while new
    scrobbles arrive mid-run; anything that lands during the pass is picked up
    by the next one.
    """
    newest = store.newest_uts()
    if newest is None:
        return 0

    added = 0
    page = 1
    while True:
        scrobbles, total_pages = fetch_page(client, user, to=now, since=newest, page=page)
        added += store.add_scrobbles(scrobbles)
        if page >= total_pages:
            return added
        page += 1


def sync_loved(client, store, user):
    """Rebuild the loved-track table. Returns how many are loved."""
    rows = []
    page = 1
    while True:
        payload = client.call(
            "user.getLovedTracks", {"user": user, "limit": PAGE_SIZE, "page": page}
        )
        block = payload.get("lovedtracks") or {}
        for entry in as_list(block.get("track")):
            artist = name_of(entry.get("artist"))
            track = (entry.get("name") or "").strip()
            if not artist or not track:
                continue
            loved_at = (entry.get("date") or {}).get("uts")
            rows.append((artist, track, mbid_of(entry.get("mbid")), int(loved_at or 0) or None))
        total_pages = int((block.get("@attr") or {}).get("totalPages") or 1)
        if page >= total_pages:
            break
        page += 1

    # Written only once the whole list has been fetched: a failure halfway
    # through must not leave a truncated set of protected tracks behind.
    store.replace_loved(rows)
    return len(rows)


class KeepAlive:
    """A transport that holds one connection open per host.

    urllib opens a fresh TCP connection -- and performs a fresh DNS lookup --
    for every request it makes. Indexing a library is two requests per artist,
    which on a large collection is thousands of lookups inside a few minutes.
    That is enough to exhaust a container's resolver, and the failure it
    produces is `[Errno -3] Try again` on everything at once. Resolving once and
    reusing the socket removes the cause rather than papering over it, and is
    considerably faster besides.
    """

    def __init__(self, timeout=60):
        self.timeout = timeout
        self._connections = {}

    def __call__(self, url, timeout=None, headers=None):
        parsed = urllib.parse.urlparse(url)
        key = (parsed.scheme, parsed.hostname, parsed.port)
        target = parsed.path + (f"?{parsed.query}" if parsed.query else "")
        request_headers = {**(headers or {}), "Accept": "application/json"}

        # Two attempts, because a kept-alive connection the server has since
        # closed fails on use rather than announcing itself. The second attempt
        # is on a fresh socket.
        for attempt in (1, 2):
            connection = self._connections.get(key)
            if connection is None:
                connection = self._connect(parsed, timeout or self.timeout)
                self._connections[key] = connection
            try:
                connection.request("GET", target, headers=request_headers)
                response = connection.getresponse()
                body = response.read()
            except (http.client.HTTPException, OSError) as error:
                self.close(key)
                if attempt == 2:
                    raise urllib.error.URLError(error) from error
                continue

            if response.status >= 300:
                # Includes redirects: this client does not follow them, and one
                # here means the URL is pointing somewhere unintended.
                raise urllib.error.HTTPError(
                    url, response.status, response.reason, response.headers, io.BytesIO(body)
                )
            return body.decode("utf-8")
        raise urllib.error.URLError("unreachable")

    @staticmethod
    def _connect(parsed, timeout):
        if parsed.scheme == "https":
            return http.client.HTTPSConnection(parsed.hostname, parsed.port, timeout=timeout)
        return http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=timeout)

    def close(self, key=None):
        for handle in [self._connections.pop(key, None)] if key else self._connections.values():
            if handle is not None:
                handle.close()
        if key is None:
            self._connections.clear()


class Lidarr:
    """Minimal read-only Lidarr client.

    Retries only what is worth retrying. A dropped connection or a resolver
    hiccup is transient; an HTTP 500 out of Lidarr is an exception in its own
    serialisation and will be thrown again identically, so spending three
    attempts on it only slows down finding out.
    """

    def __init__(self, url, api_key, timeout=60, attempts=3, backoff=1.0, transport=None):
        self.root = url.rstrip("/")
        self.api_key = api_key
        self.timeout = timeout
        self.attempts = attempts
        self.backoff = backoff
        self.transport = transport or KeepAlive(timeout)

    def get(self, path, params=None):
        """Return the decoded response for one API path."""
        query = urllib.parse.urlencode(params or {})
        url = f"{self.root}/api/v1/{path}" + (f"?{query}" if query else "")

        for attempt in range(1, self.attempts + 1):
            try:
                body = self.transport(
                    url, timeout=self.timeout, headers={"X-Api-Key": self.api_key}
                )
            except urllib.error.HTTPError as error:
                detail = error_detail(error)
                message = f"GET {url}: HTTP {error.code}{': ' + detail if detail else ''}"
                if error.code not in LIDARR_RETRYABLE_STATUS:
                    raise LidarrError(message)
                if attempt >= self.attempts:
                    raise LidarrError(message, transient=True)
            except (urllib.error.URLError, TimeoutError) as error:
                message = f"GET {url}: {error}"
                if attempt >= self.attempts:
                    raise LidarrError(message, transient=True) from error
            else:
                try:
                    return json.loads(body)
                except json.JSONDecodeError as error:
                    raise LidarrError(f"GET {url}: malformed response") from error

            pause = min(self.backoff * 2 ** (attempt - 1), BACKOFF_CEILING_SECONDS)
            logger.warning("%s; retrying in %.0fs", message, pause)
            time.sleep(pause)

        raise LidarrError(f"GET {url}: gave up after {self.attempts} attempts", transient=True)


def error_detail(error, limit=300):
    """Return whatever the server said about a failure, for the log.

    A bare status code sends you looking in the wrong place; Lidarr puts the
    actual exception in the body.
    """
    try:
        body = error.read().decode("utf-8", "replace").strip()
    except Exception:  # noqa: BLE001 - diagnostics must never raise
        return ""
    if not body:
        return ""
    try:
        payload = json.loads(body)
    except json.JSONDecodeError:
        return body[:limit]
    for key in ("message", "description", "error"):
        if payload.get(key):
            return str(payload[key])[:limit]
    return body[:limit]


def parse_added(value):
    """Return a unix timestamp for a Lidarr ISO-8601 date, or None."""
    if not value:
        return None
    try:
        return int(datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp())
    except ValueError:
        return None


def find_bad_albums(client, artist_id, artist_name):
    """Name the specific albums Lidarr cannot serialise for one artist.

    Runs only once that artist's album fetch has already failed, so the extra
    requests are spent on a problem that already exists. Tracks come from an
    endpoint that still works, and their album ids give a list to probe one at a
    time; the ones that throw are the culprits. A track title from each is
    enough to recognise the album in the UI, which the id alone is not.
    """
    try:
        tracks = client.get("track", {"artistId": artist_id})
    except LidarrError as error:
        logger.warning("could not probe %s for the offending album: %s", artist_name, error)
        return []

    sample = {}
    for track in tracks:
        sample.setdefault(track.get("albumId"), track.get("title") or "")

    bad = []
    for album_id, title in sorted(sample.items(), key=lambda item: item[0] or 0):
        if not album_id:
            continue
        try:
            client.get("album", {"albumIds": album_id})
        except LidarrError:
            bad.append((album_id, title))
    return bad


def fetch_albums(client, artists):
    """Return albums grouped by artist id, plus the artists whose albums failed.

    Every album endpoint maps through a resource that picks the release with
    `SingleOrDefault(x => x.Monitored)`, which throws for an album with two
    monitored releases -- "Sequence contains more than one element" -- and takes
    the entire response down with it.

    The unfiltered endpoint is one request and is the only one that also skips
    albums whose artist metadata is missing, so it is tried first. When it dies,
    asking per artist confines the loss to whichever artist owns the offending
    album, and names them, which is the only way to find it.
    """
    try:
        grouped = {}
        for album in client.get("album"):
            grouped.setdefault(album.get("artistId"), []).append(album)
        return grouped, []
    except LidarrError as error:
        logger.warning("fetching all albums failed (%s)", error)
        logger.warning("falling back to one request per artist to isolate the bad album")

    grouped, failed = {}, []
    for artist in artists:
        artist_id = artist["id"]
        try:
            grouped[artist_id] = client.get("album", {"artistId": artist_id})
        except LidarrError as error:
            name = artist.get("artistName") or str(artist_id)
            logger.warning("could not fetch albums for %s: %s", name, error)
            failed.append(name)
            # Only worth probing a deterministic failure. When the network or
            # the resolver is the problem, every artist fails, and probing each
            # of them album by album multiplies the load that caused it.
            if not error.transient:
                for album_id, sample in find_bad_albums(client, artist_id, name):
                    logger.warning(
                        "  album id %d is the one Lidarr cannot serialise (it holds the"
                        " track %r). Open it in Lidarr and leave exactly one release"
                        " monitored.",
                        album_id,
                        sample,
                    )
    return grouped, failed


def index_library(client, store):
    """Rebuild the library index from Lidarr. Returns (artists, tracks).

    Rebuilt wholesale rather than reconciled: Lidarr is the authority, the index
    is small, and a delete in Lidarr has to disappear here or it lingers as a
    library entry with no file behind it.
    """
    artists = client.get("artist")
    artist_rows, album_rows, track_rows = [], [], []
    skipped = []

    albums_by_artist, album_failures = fetch_albums(client, artists)

    for artist in artists:
        artist_id = artist["id"]
        name = artist.get("artistName") or ""
        artist_rows.append(
            (
                artist_id,
                mbid_of(artist.get("foreignArtistId")),
                name,
                normalise(name),
                artist.get("path"),
                1 if artist.get("monitored") else 0,
            )
        )

        for album in albums_by_artist.get(artist_id, []):
            album_rows.append(
                (
                    album["id"],
                    artist_id,
                    mbid_of(album.get("foreignAlbumId")),
                    album.get("title") or "",
                    1 if album.get("monitored") else 0,
                    (album.get("releaseDate") or "")[:10] or None,
                )
            )

        # Tracks and files have no unfiltered endpoint -- Lidarr rejects a call
        # with no filter at all -- so these stay per artist. One artist it
        # cannot serve must not cost the whole index, but it cannot pass
        # silently either: their tracks end up absent, and every scrobble of
        # theirs then reads as unmatched.
        try:
            # Paths and the date a track landed live on the file, not the track.
            files = {
                handle["id"]: handle
                for handle in client.get("trackfile", {"artistId": artist_id})
            }
            tracks = client.get("track", {"artistId": artist_id})
        except LidarrError as error:
            logger.warning("could not index %s: %s", name or artist_id, error)
            skipped.append(name or str(artist_id))
            continue

        for track in tracks:
            handle = files.get(track.get("trackFileId") or 0) or {}
            title = track.get("title") or ""
            track_rows.append(
                (
                    track["id"],
                    artist_id,
                    track.get("albumId") or 0,
                    mbid_of(track.get("foreignRecordingId")),
                    title,
                    normalise(name),
                    normalise(title),
                    1 if track.get("hasFile") else 0,
                    handle.get("path"),
                    parse_added(handle.get("dateAdded")),
                    track.get("duration"),
                )
            )

    store.replace_library(artist_rows, album_rows, track_rows)
    store.set_state("index_skipped", str(len(skipped)))
    store.set_state("index_albums_skipped", str(len(album_failures)))
    logger.info(
        "library indexed: %d artists, %d albums, %d tracks (%d with files)",
        len(artist_rows),
        len(album_rows),
        len(track_rows),
        sum(row[7] for row in track_rows),
    )
    if skipped:
        logger.warning(
            "%d artists could not be indexed, so their tracks are missing: %s",
            len(skipped),
            ", ".join(sorted(skipped)[:10]),
        )
    if album_failures:
        logger.warning(
            "no albums indexed for %d artists: %s."
            " In Lidarr, open each one and check the Releases tab of their albums:"
            " exactly one release per album may be monitored.",
            len(album_failures),
            ", ".join(sorted(album_failures)[:10]),
        )
    return len(artist_rows), len(track_rows)


def match_library(store):
    """Tie every distinct thing listened to back to a library track.

    Two tiers. A MusicBrainz recording id is an exact join and is tried first;
    everything else falls to the normalised name pair. There is no third tier on
    purpose -- a guess that is nearly right is worse here than an admitted miss,
    because the whole point of the number this produces is to say how far the
    matching can be trusted.
    """
    store.rebuild_keys()
    with store.connection:
        store.connection.execute("UPDATE scrobble_key SET track_id = NULL, method = 'none'")
        store.connection.execute(
            "UPDATE scrobble_key SET"
            "   track_id = (SELECT t.id FROM lidarr_track t"
            "               WHERE t.recording_mbid = scrobble_key.track_mbid"
            "               ORDER BY t.has_file DESC, t.id LIMIT 1),"
            "   method = 'mbid'"
            " WHERE track_mbid IS NOT NULL"
            "   AND EXISTS (SELECT 1 FROM lidarr_track t"
            "               WHERE t.recording_mbid = scrobble_key.track_mbid)"
        )
        store.connection.execute(
            "UPDATE scrobble_key SET"
            "   track_id = (SELECT t.id FROM lidarr_track t"
            "               WHERE t.norm_artist = scrobble_key.norm_artist"
            "                 AND t.norm_title = scrobble_key.norm_track"
            "               ORDER BY t.has_file DESC, t.id LIMIT 1),"
            "   method = 'name'"
            " WHERE track_id IS NULL"
            "   AND norm_artist <> '' AND norm_track <> ''"
            "   AND EXISTS (SELECT 1 FROM lidarr_track t"
            "               WHERE t.norm_artist = scrobble_key.norm_artist"
            "                 AND t.norm_title = scrobble_key.norm_track)"
        )


def format_time(uts):
    """Return a UTC timestamp as a readable date."""
    if uts is None:
        return "never"
    return datetime.fromtimestamp(uts, tz=timezone.utc).strftime("%Y-%m-%d %H:%M")


# Playlists written into the mirror. Each is a rule over the listening history
# and the library index; none of them look at the audio.
#
# The rotating ones are shuffled by week rather than by pass. A pass runs every
# few hours, and a playlist that reorders itself every time is one that has to
# be re-imported every time -- the Music app does not track a file, it imports a
# snapshot of one.
ROTATION_PERIOD_SECONDS = 7 * 86400
SHUFFLE_MULTIPLIER = 2654435761
SHUFFLE_MODULUS = 104729

# Common table expressions the rules share. `played` is every library track that
# has ever been matched to a scrobble, with its count and the last time it was
# heard; `recent` is the same restricted to a window.
PLAYED_CTE = """
WITH played AS (
    SELECT sk.track_id AS track_id, COUNT(*) AS plays, MAX(s.uts) AS last_uts
      FROM scrobble s
      JOIN scrobble_key sk ON sk.artist = s.artist AND sk.track = s.track
     WHERE sk.track_id IS NOT NULL
     GROUP BY sk.track_id
)
"""
RECENT_CTE = """
WITH recent AS (
    SELECT sk.track_id AS track_id, COUNT(*) AS plays
      FROM scrobble s
      JOIN scrobble_key sk ON sk.artist = s.artist AND sk.track = s.track
     WHERE sk.track_id IS NOT NULL AND s.uts >= :year_ago
     GROUP BY sk.track_id
)
"""
SELECT_TRACK = """
SELECT t.id AS id, a.name AS artist, t.title AS title,
       t.duration AS duration, t.path AS path
  FROM lidarr_track t
  JOIN lidarr_artist a ON a.id = t.artist_id
"""
PLAYABLE = " t.has_file = 1 AND t.path IS NOT NULL AND t.path <> '' "
SHUFFLE = f" ((t.id * {SHUFFLE_MULTIPLIER}) + :week) % {SHUFFLE_MODULUS} "

PLAYLISTS = (
    (
        "heavy-rotation",
        "most played over the last twelve months",
        RECENT_CTE + SELECT_TRACK + f"""
          JOIN recent r ON r.track_id = t.id
         WHERE {PLAYABLE}
         ORDER BY r.plays DESC, a.name, t.title
         LIMIT :limit
        """,
    ),
    (
        "all-time",
        "most played ever",
        PLAYED_CTE + SELECT_TRACK + f"""
          JOIN played p ON p.track_id = t.id
         WHERE {PLAYABLE}
         ORDER BY p.plays DESC, a.name, t.title
         LIMIT :limit
        """,
    ),
    (
        "neglected",
        "played heavily once, silent for twelve months",
        PLAYED_CTE + SELECT_TRACK + f"""
          JOIN played p ON p.track_id = t.id
         WHERE {PLAYABLE} AND p.last_uts < :year_ago
         ORDER BY p.plays DESC, a.name, t.title
         LIMIT :limit
        """,
    ),
    (
        "deep-cuts",
        "never played, from albums whose other tracks you play constantly",
        PLAYED_CTE + """,
        album_plays AS (
            SELECT t.album_id AS album_id, SUM(p.plays) AS plays
              FROM played p JOIN lidarr_track t ON t.id = p.track_id
             GROUP BY t.album_id
        )
        """ + SELECT_TRACK + f"""
          JOIN album_plays ap ON ap.album_id = t.album_id
          LEFT JOIN played p ON p.track_id = t.id
         WHERE {PLAYABLE} AND p.track_id IS NULL
         ORDER BY ap.plays DESC, t.id
         LIMIT :limit
        """,
    ),
    (
        "unheard-favourites",
        "never played, by the artists you play most; rotates weekly",
        PLAYED_CTE + """,
        artist_plays AS (
            SELECT t.artist_id AS artist_id, SUM(p.plays) AS plays
              FROM played p JOIN lidarr_track t ON t.id = p.track_id
             GROUP BY t.artist_id
        )
        """ + SELECT_TRACK + f"""
          LEFT JOIN played p ON p.track_id = t.id
         WHERE {PLAYABLE} AND p.track_id IS NULL
           AND t.artist_id IN (SELECT artist_id FROM artist_plays
                                ORDER BY plays DESC LIMIT 50)
         ORDER BY {SHUFFLE}
         LIMIT :limit
        """,
    ),
    (
        "unheard",
        "never played, anywhere in the library; rotates weekly",
        PLAYED_CTE + SELECT_TRACK + f"""
          LEFT JOIN played p ON p.track_id = t.id
         WHERE {PLAYABLE} AND p.track_id IS NULL
         ORDER BY {SHUFFLE}
         LIMIT :limit
        """,
    ),
)


# Moods, as sets of Last.fm tags. Chosen for this library -- drum and bass,
# punk and its descendants, big-room dance, classic rock -- rather than as a
# general taxonomy. `--vibes` replaces the lot with a JSON file of the same
# shape, so a new one does not need a new release.
#
# `years` filters on the album's release date, which is what separates eighties
# synth records from everything a synthpop tag would otherwise drag in.
DEFAULT_VIBES = (
    # Built from the tag distribution of this library rather than from a general
    # taxonomy, which is why some obvious-looking tags are absent and some
    # unobvious ones are here. `synthwave`, `edm`, `big room` and `hardstyle`
    # carry nothing at all; `techstep`, `easycore` and `hospital records` carry
    # real weight.
    #
    # Three kinds of tag are deliberately never used. Nationality -- american,
    # british, swedish -- describes a passport, not a sound, and `american`
    # alone spans 241 artists. `rock` and `electronic` span 340 and 275, which
    # is most of the library and therefore no mood at all. And Last.fm's most
    # popular tag for an artist is frequently their own name, so `green day`,
    # `paramore` and `queen` are single-artist playlists waiting to happen.
    {
        "name": "drum-and-bass",
        "tags": [
            "drum and bass", "dnb", "drum n bass", "drum'n'bass", "drum & bass",
            "drum 'n' bass", "liquid funk", "neurofunk", "jungle", "techstep",
            "darkstep", "drumstep", "breakbeat", "hospital records",
        ],
    },
    {
        "name": "bass",
        "tags": ["dubstep", "brostep", "grime", "trip-hop", "big beat"],
    },
    {
        "name": "dance",
        "tags": [
            "house", "electro house", "progressive house", "tech house", "trance",
            "techno", "electro", "rave", "dance", "minimal",
        ],
    },
    {
        "name": "pop-punk",
        "tags": [
            "pop punk", "pop-punk", "punk rock", "punk", "skate punk", "emo",
            "emocore", "easycore", "powerpop", "power pop", "post-grunge",
        ],
    },
    {
        "name": "screamo",
        "tags": [
            "screamo", "post-hardcore", "metalcore", "melodic metalcore",
            "melodic hardcore", "hardcore", "trancecore", "deathcore",
        ],
    },
    {
        "name": "heavy-metal",
        "tags": [
            "heavy metal", "metal", "thrash metal", "thrash", "speed metal",
            "power metal", "death metal", "progressive metal", "nwobhm",
            "classic metal",
        ],
    },
    {
        "name": "hair-metal",
        "tags": [
            "hair metal", "glam metal", "glam rock", "arena rock", "aor",
            "rock and roll", "rock n roll",
        ],
        "years": [1975, 1994],
    },
    {
        "name": "nu-metal",
        "tags": [
            "nu metal", "nu-metal", "alternative metal", "rapcore",
            "industrial metal", "industrial rock", "industrial",
        ],
    },
    {
        "name": "classic-rock",
        "tags": [
            "classic rock", "progressive rock", "psychedelic rock", "psychedelic",
            "blues rock", "blues", "southern rock", "art rock", "space rock",
            "british invasion", "folk rock", "70s", "60s",
        ],
    },
    {
        "name": "80s-synths",
        # Without the bare `80s` tag, which in this library sits on Def Leppard
        # and Bon Jovi rather than on anything with a synthesiser in it. The
        # year window is kept as a second filter, not as the only one.
        "tags": [
            "synthpop", "electropop", "new wave", "post-punk",
            "post-punk revival", "new romantic",
        ],
        "years": [1975, 1992],
    },
    {
        "name": "indie",
        "tags": ["indie", "indie rock", "indie pop", "britpop", "singer-songwriter"],
    },
)


def load_vibes(path):
    """Return the vibe definitions, from a file when one is given.

    Validated up front rather than at write time: a bad name would otherwise
    surface as a file created somewhere unintended, which is a poor way to find
    out about a typo.
    """
    if not path:
        return DEFAULT_VIBES
    try:
        loaded = json.loads(Path(path).read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise ValueError(f"cannot read vibes from {path}: {error}") from error

    if not isinstance(loaded, list):
        raise ValueError(f"{path}: expected a list of vibes")
    for vibe in loaded:
        name = vibe.get("name") if isinstance(vibe, dict) else None
        if not name or not SAFE_VIBE_NAME.fullmatch(str(name)):
            raise ValueError(f"{path}: {name!r} is not a usable vibe name (a-z, 0-9, -)")
        if not vibe.get("tags"):
            raise ValueError(f"{path}: vibe {name!r} lists no tags")
    return tuple(loaded)


def parse_tags(payload):
    """Return (tag, weight) pairs from an artist.getTopTags response.

    The documented sample carries only a name and a URL per tag; a live response
    also carries a 0-100 `count`. Rather than depend on which, the count is used
    when present and the documented ordering -- by popularity -- stands in for it
    when it is not.
    """
    block = payload.get("toptags") or {}
    pairs = []
    for position, entry in enumerate(as_list(block.get("tag"))):
        tag = (entry.get("name") or "").strip().casefold()
        if not tag:
            continue
        weight = int(entry.get("count") or 0) or max(1, 100 - position * 5)
        pairs.append((tag, weight))
    return pairs


def fetch_tags(client, name, mbid):
    """Return an artist's tags, asking by name first.

    By name, not by MusicBrainz id, despite Lidarr having an id for everything.
    Last.fm's mbid index is stale and partial -- it answers "the artist you
    supplied could not be found" for Devo, Escape the Fate and a few hundred
    others whose pages plainly exist -- while the name index is the one its own
    site runs on. The id is kept only as a fallback for a name Lidarr spells
    differently.

    Returns an empty list when the artist is genuinely unknown, which the caller
    records so it is not asked again.
    """
    attempts = [{"artist": name}] if name else []
    if mbid:
        attempts.append({"mbid": mbid})

    for query in attempts:
        try:
            return parse_tags(client.call("artist.getTopTags", {**query, "autocorrect": 1}))
        except LastfmError as error:
            # Error 6 here means "no such artist", which the next key may still
            # answer. Anything else is a real failure and belongs to the caller.
            if error.code != 6:
                raise
    return []


def sync_tags(client, store, now, limit=0):
    """Fetch crowd tags for library artists that have none, or stale ones.

    One request per artist, once, and then only for artists newly added. An
    artist Last.fm has never heard of is recorded as fetched with no tags, so it
    is not asked about again every pass.
    """
    stale = store.connection.execute(
        "SELECT a.norm_name AS norm_name, a.name AS name, a.mbid AS mbid"
        "  FROM lidarr_artist a"
        "  LEFT JOIN artist_tag_fetched f ON f.norm_artist = a.norm_name"
        " WHERE a.norm_name <> ''"
        "   AND (f.fetched_at IS NULL OR f.fetched_at < :cutoff)"
        " ORDER BY a.name",
        {"cutoff": now - TAG_REFRESH_SECONDS},
    ).fetchall()
    if not stale:
        return 0

    if limit > 0 and len(stale) > limit:
        logger.info("tagging %d of %d artists this pass; the rest follow next", limit, len(stale))
        stale = stale[:limit]
    else:
        logger.info("fetching tags for %d artists", len(stale))

    resolved = 0
    unknown = 0
    for artist in stale:
        try:
            pairs = fetch_tags(client, artist["name"], artist["mbid"])
        except LastfmError as error:
            # Something went wrong rather than the artist not existing. Left
            # unrecorded on purpose, so the next pass tries again.
            logger.warning("no tags for %s: %s", artist["name"], error)
            continue
        store.replace_tags(artist["norm_name"], pairs, now)
        resolved += 1
        if not pairs:
            unknown += 1

    logger.info("tagged %d artists (%d with nothing to say about them)", resolved, unknown)
    return resolved


def build_vibe_playlists(store, vibes, mirror_root, library_root, limit, now):
    """Write one playlist per mood. Returns how many tracks were listed."""
    if not store.scalar("SELECT COUNT(*) FROM artist_tag"):
        return 0

    directory = Path(mirror_root) / PLAYLIST_DIRECTORY
    week = now // ROTATION_PERIOD_SECONDS
    total = 0

    for vibe in vibes:
        tags = [str(tag).strip().casefold() for tag in vibe["tags"]]
        placeholders = ",".join("?" * len(tags))
        years = vibe.get("years")
        parameters = [*tags, vibe.get("min_score", VIBE_MIN_SCORE)]
        year_clause = ""
        if years:
            year_clause = (
                " AND CAST(substr(al.release_date, 1, 4) AS INTEGER) BETWEEN ? AND ?"
            )
            parameters += [int(years[0]), int(years[1])]
        parameters += [week, limit]

        sql = f"""
            WITH vibe AS (
                SELECT norm_artist, SUM(weight) AS score
                  FROM artist_tag
                 WHERE tag IN ({placeholders})
                 GROUP BY norm_artist
                HAVING SUM(weight) >= ?
            )
            SELECT t.id AS id, a.name AS artist, t.title AS title,
                   t.duration AS duration, t.path AS path
              FROM lidarr_track t
              JOIN lidarr_artist a ON a.id = t.artist_id
              JOIN vibe v ON v.norm_artist = a.norm_name
              LEFT JOIN lidarr_album al ON al.id = t.album_id
             WHERE {PLAYABLE}{year_clause}
             ORDER BY ((t.id * {SHUFFLE_MULTIPLIER}) + ?) % {SHUFFLE_MODULUS}
             LIMIT ?
        """

        entries = []
        for row in store.connection.execute(sql, parameters):
            mirror = mirror_path_for(row["path"], library_root, mirror_root)
            if mirror is None or not mirror.is_file():
                continue
            entries.append({**dict(row), "mirror": mirror})

        write_playlist(directory / f"{vibe['name']}.m3u", entries)
        total += len(entries)
        logger.info("playlist %-20s %4d tracks -- by tag", vibe["name"], len(entries))

    return total


def library_root_of(store):
    """Return the directory Lidarr's artist folders sit under.

    Derived rather than configured, because it has to agree with what Lidarr
    reports and no one wants to keep a second copy of that in step by hand.
    """
    paths = [
        row["path"]
        for row in store.connection.execute(
            "SELECT path FROM lidarr_artist WHERE path IS NOT NULL AND path <> ''"
        )
    ]
    if not paths:
        return None
    if len(paths) == 1:
        return str(Path(paths[0]).parent)
    try:
        return os.path.commonpath(paths)
    except ValueError:
        return None


def mirror_path_for(source, library_root, mirror_root):
    """Return where music-mirror would have put the MP3 for a source file."""
    try:
        relative = Path(source).relative_to(library_root)
    except ValueError:
        return None
    return (Path(mirror_root) / relative).with_suffix(MIRROR_SUFFIX)


def write_playlist(path, entries):
    """Write one extended M3U, atomically.

    Paths are relative to the playlist file, so the same playlist works from the
    NAS, from a Mac over SMB and from Linux without rewriting.
    """
    lines = ["#EXTM3U"]
    for entry in entries:
        seconds = round((entry["duration"] or 0) / 1000)
        lines.append(f"#EXTINF:{seconds},{entry['artist']} - {entry['title']}")
        lines.append(os.path.relpath(entry["mirror"], path.parent))

    path.parent.mkdir(parents=True, exist_ok=True)
    handle, temporary = tempfile.mkstemp(dir=path.parent, suffix=".m3u.part")
    os.close(handle)
    temporary = Path(temporary)
    try:
        temporary.write_text("\n".join(lines) + "\n", encoding="utf-8")
        # The mirror is read back by something else; see music-mirror, which had
        # to learn this the hard way.
        mode = temporary.stat().st_mode
        if not mode & GROUP_READ:
            temporary.chmod(mode | GROUP_READ)
        os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)


def build_playlists(store, mirror_root, library_root, limit, now):
    """Write every playlist into the mirror. Returns how many tracks were listed.

    A track is only listed once its mirror file has been confirmed to exist. The
    index knows what Lidarr holds, which is the lossless source; whether the MP3
    beside it has been encoded yet is music-mirror's business and is checked
    rather than assumed.
    """
    directory = Path(mirror_root) / PLAYLIST_DIRECTORY
    parameters = {
        "limit": limit,
        "year_ago": now - 365 * 86400,
        "week": now // ROTATION_PERIOD_SECONDS,
    }

    total = 0
    missing = 0
    for name, description, sql in PLAYLISTS:
        entries = []
        for row in store.connection.execute(sql, parameters):
            mirror = mirror_path_for(row["path"], library_root, mirror_root)
            if mirror is None or not mirror.is_file():
                missing += 1
                continue
            entries.append({**dict(row), "mirror": mirror})
        write_playlist(directory / f"{name}.m3u", entries)
        total += len(entries)
        logger.info("playlist %-20s %4d tracks -- %s", name, len(entries), description)

    if missing:
        logger.warning(
            "%d selected tracks had no file in the mirror and were left out."
            " A handful means music-mirror has not encoded them yet; a large"
            " number means --library-root or --mirror is pointing at the wrong"
            " place, since the paths are then being mapped to nothing.",
            missing,
        )
    return total


def coverage_report(store):
    """Log how much of the listening history could be tied to the library.

    The split that matters is not matched against unmatched. Most unmatched
    listening is music that was never in the library at all, which says nothing
    about the matcher. The number to watch is unmatched listening by an artist
    the library *does* hold: that is a track that was played, sitting next to a
    file it should have matched, and every one of those is a candidate for being
    wrongly called cold later on.
    """
    tracks = store.scalar("SELECT COUNT(*) FROM lidarr_track")
    if not tracks:
        return

    pairs = store.scalar("SELECT COUNT(*) FROM scrobble_key")
    if not pairs:
        return

    logger.info("--- match coverage ---")
    skipped = int(store.get_state("index_skipped") or 0)
    if skipped:
        logger.warning(
            "the index is incomplete: %d artists are missing their tracks, so every"
            " figure below is a floor. Do not cull against it.",
            skipped,
        )
    albums_skipped = int(store.get_state("index_albums_skipped") or 0)
    if albums_skipped:
        logger.warning(
            "%d artists have no albums indexed. Matching is unaffected -- it runs off"
            " tracks -- but a cull cannot be run until this is fixed.",
            albums_skipped,
        )
    for method in ("mbid", "name", "none"):
        row = store.connection.execute(
            "SELECT COUNT(*) AS pairs, COALESCE(SUM(plays), 0) AS plays"
            " FROM scrobble_key WHERE method = ?",
            (method,),
        ).fetchone()
        logger.info(
            "%-5s: %6d pairs (%4.1f%%), %7d plays",
            method,
            row["pairs"],
            100 * row["pairs"] / pairs,
            row["plays"],
        )

    # Why the mbid tier performs the way it does. A recording id on both sides
    # that still fails to join means the two disagree about which recording the
    # song is -- MusicBrainz holds a separate recording per release, and Last.fm
    # and Lidarr need not have picked the same one. That is not a fault to fix
    # in the matcher; it is the reason the name tier has to carry the load.
    library_with_mbid = store.scalar(
        "SELECT COUNT(*) FROM lidarr_track WHERE recording_mbid IS NOT NULL"
    )
    logger.info(
        "library tracks carrying a recording MBID: %d of %d (%.1f%%)",
        library_with_mbid,
        tracks,
        100 * library_with_mbid / tracks,
    )
    disagreed = store.connection.execute(
        "SELECT COUNT(*) AS pairs, COALESCE(SUM(plays), 0) AS plays FROM scrobble_key"
        " WHERE track_mbid IS NOT NULL AND method = 'name'"
    ).fetchone()
    logger.info(
        "carried a recording MBID, joined on name instead: %d pairs, %d plays"
        " -- both sides know the song, they disagree on which recording it is",
        disagreed["pairs"],
        disagreed["plays"],
    )

    suspect = store.connection.execute(
        "SELECT COUNT(*) AS pairs, COALESCE(SUM(plays), 0) AS plays FROM scrobble_key k"
        " WHERE k.track_id IS NULL"
        "   AND EXISTS (SELECT 1 FROM lidarr_artist a WHERE a.norm_name = k.norm_artist)"
    ).fetchone()
    logger.info(
        "unmatched by an artist the library holds: %d pairs, %d plays",
        suspect["pairs"],
        suspect["plays"],
    )

    # Owning an artist is a weak proxy for owning a track, so that figure alone
    # overstates the matcher's failings. Split it -- but not on the title alone.
    # Across fifty thousand tracks, titles collide constantly: "Everyday" is
    # Rusko and also Def Leppard, "Kaleidoscope" is Delta Heavy and also
    # Chappell Roan. Matching those would be worse than missing them.
    #
    # The signal for a genuine attribution miss is that the library's own title
    # credits the artist the scrobble is filed under -- "Voodoo People (Pendulum
    # Remix)" against a play credited to Pendulum. The normalised title has that
    # suffix stripped, which is exactly what let them meet, so the raw one has to
    # be searched for the name.
    attribution = store.connection.execute(
        "SELECT COUNT(*) AS pairs, COALESCE(SUM(plays), 0) AS plays FROM scrobble_key k"
        " WHERE k.track_id IS NULL"
        "   AND EXISTS (SELECT 1 FROM lidarr_artist a WHERE a.norm_name = k.norm_artist)"
        "   AND EXISTS (SELECT 1 FROM lidarr_track t"
        "                WHERE t.norm_title = k.norm_track"
        "                  AND instr(lower(t.title), lower(k.artist)) > 0)"
    ).fetchone()
    collision = store.connection.execute(
        "SELECT COUNT(*) AS pairs, COALESCE(SUM(plays), 0) AS plays FROM scrobble_key k"
        " WHERE k.track_id IS NULL"
        "   AND EXISTS (SELECT 1 FROM lidarr_artist a WHERE a.norm_name = k.norm_artist)"
        "   AND EXISTS (SELECT 1 FROM lidarr_track t WHERE t.norm_title = k.norm_track)"
    ).fetchone()
    logger.info(
        "  the library's title credits the scrobbled artist: %d pairs, %d plays"
        " -- remixes and guest spots, and the genuine misses",
        attribution["pairs"],
        attribution["plays"],
    )
    logger.info(
        "  same title under an unrelated artist: %d pairs, %d plays"
        " -- title collisions, not misses; matching these would be a mistake",
        collision["pairs"] - attribution["pairs"],
        collision["plays"] - attribution["plays"],
    )
    logger.info(
        "  the rest, %d pairs, %d plays: you own the artist but not the track",
        suspect["pairs"] - collision["pairs"],
        suspect["plays"] - collision["plays"],
    )

    mismatched = store.connection.execute(
        "SELECT k.artist, k.track, k.plays,"
        "       (SELECT t.title FROM lidarr_track t"
        "         WHERE t.norm_title = k.norm_track"
        "           AND instr(lower(t.title), lower(k.artist)) > 0 LIMIT 1) AS library_title"
        "  FROM scrobble_key k"
        " WHERE k.track_id IS NULL"
        "   AND EXISTS (SELECT 1 FROM lidarr_artist a WHERE a.norm_name = k.norm_artist)"
        "   AND EXISTS (SELECT 1 FROM lidarr_track t"
        "                WHERE t.norm_title = k.norm_track"
        "                  AND instr(lower(t.title), lower(k.artist)) > 0)"
        " ORDER BY k.plays DESC, k.artist LIMIT 10"
    ).fetchall()
    for position, row in enumerate(mismatched, start=1):
        logger.info(
            "  attribution %2d: %-45s %4d plays, library has %r",
            position,
            f"{row['artist']} - {row['track']}"[:45],
            row["plays"],
            row["library_title"],
        )

    with_files = store.scalar("SELECT COUNT(*) FROM lidarr_track WHERE has_file = 1")
    played = store.scalar(
        "SELECT COUNT(DISTINCT k.track_id) FROM scrobble_key k"
        " JOIN lidarr_track t ON t.id = k.track_id WHERE t.has_file = 1"
    )
    logger.info(
        "library tracks with a file: %d, of which played at least once: %d (%.1f%%)",
        with_files,
        played,
        100 * played / with_files if with_files else 0,
    )

    worst = store.connection.execute(
        "SELECT k.artist, k.track, k.plays FROM scrobble_key k"
        " WHERE k.track_id IS NULL"
        "   AND EXISTS (SELECT 1 FROM lidarr_artist a WHERE a.norm_name = k.norm_artist)"
        " ORDER BY k.plays DESC, k.artist LIMIT 15"
    ).fetchall()
    for position, row in enumerate(worst, start=1):
        label = f"{row['artist']} - {row['track']}"
        logger.info("  unmatched %2d: %-60s %d plays", position, label[:60], row["plays"])


def report(store, now):
    """Log what the store holds.

    The MBID coverage line is the one to watch: scrobbles carrying a MusicBrainz
    recording id can be joined to the library exactly, and everything else has
    to go through name matching. That percentage predicts how well the next
    stage will work, and whether it can be trusted to decide what is unplayed.
    """
    total = store.count()
    logger.info("scrobbles: %d", total)
    if not total:
        return

    with_mbid = store.scalar("SELECT COUNT(*) FROM scrobble WHERE track_mbid IS NOT NULL")
    logger.info(
        "range: %s to %s", format_time(store.oldest_uts()), format_time(store.newest_uts())
    )
    logger.info("distinct artists: %d", store.scalar("SELECT COUNT(DISTINCT artist) FROM scrobble"))
    logger.info(
        "distinct tracks: %d",
        store.scalar("SELECT COUNT(DISTINCT artist || ' - ' || track) FROM scrobble"),
    )
    logger.info("recording MBID present: %d (%.1f%%)", with_mbid, 100 * with_mbid / total)
    logger.info("loved tracks: %d", store.scalar("SELECT COUNT(*) FROM loved"))
    logger.info("backfill complete: %s", store.get_state("backfill_complete") == "yes")

    top = store.connection.execute(
        "SELECT artist, COUNT(*) AS plays FROM scrobble"
        " GROUP BY artist ORDER BY plays DESC, artist LIMIT 10"
    ).fetchall()
    for position, row in enumerate(top, start=1):
        logger.info("  top artist %2d: %-40s %d", position, row["artist"][:40], row["plays"])

    coverage_report(store)

    recent = store.connection.execute(
        "SELECT artist, track, COUNT(*) AS plays FROM scrobble WHERE uts >= ?"
        " GROUP BY artist, track ORDER BY plays DESC, artist LIMIT 10",
        (now - 90 * 86400,),
    ).fetchall()
    for position, row in enumerate(recent, start=1):
        label = f"{row['artist']} - {row['track']}"
        logger.info("  top track 90d %2d: %-50s %d", position, label[:50], row["plays"])


def run_once(
    client, store, user, now, backfill_limit, lidarr=None, mirror=None,
    library_root=None, playlist_limit=100, vibes=DEFAULT_VIBES, tag_limit=0,
):
    """Run a single pass. Returns the number of scrobbles added."""
    started = time.monotonic()
    added = catch_up(client, store, user, now)
    added += backfill(client, store, user, now, limit=backfill_limit)
    loved = sync_loved(client, store, user)

    if lidarr is not None:
        index_library(lidarr, store)
    # Matching is worth redoing even without a fresh index: new scrobbles have
    # arrived, and they need a verdict too.
    if store.scalar("SELECT COUNT(*) FROM lidarr_track"):
        match_library(store)
        if mirror is not None:
            root = library_root or library_root_of(store)
            if root is None:
                logger.warning(
                    "cannot work out where Lidarr's music lives, so no playlists were"
                    " written; set --library-root"
                )
            else:
                build_playlists(store, mirror, root, playlist_limit, now)
                sync_tags(client, store, now, limit=tag_limit)
                build_vibe_playlists(store, vibes, mirror, root, playlist_limit, now)

    logger.info(
        "pass complete in %.1fs: %d scrobbles added, %d loved",
        time.monotonic() - started,
        added,
        loved,
    )
    return added


def acquire_lock(database):
    """Take an exclusive lock so two passes cannot ingest into one store."""
    database.parent.mkdir(parents=True, exist_ok=True)
    handle = open(database.with_suffix(".lock"), "w")  # noqa: SIM115 - held for the process
    try:
        fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        handle.close()
        return None
    return handle


def build_parser():
    """Return the argument parser. Every option also reads an environment
    variable, so the container can be configured without a command line."""
    parser = argparse.ArgumentParser(
        prog="music-curator",
        description="Ingest a Last.fm scrobble history into a local store.",
    )
    parser.add_argument(
        "--user",
        default=os.getenv("MUSIC_CURATOR_LASTFM_USER"),
        help="Last.fm username to read (env MUSIC_CURATOR_LASTFM_USER)",
    )
    parser.add_argument(
        "--api-key",
        default=os.getenv("MUSIC_CURATOR_LASTFM_API_KEY"),
        help="Last.fm API key (env MUSIC_CURATOR_LASTFM_API_KEY)",
    )
    parser.add_argument(
        "--db",
        default=os.getenv("MUSIC_CURATOR_DB", "/data/curator.db"),
        help="path to the SQLite store (env MUSIC_CURATOR_DB)",
    )
    parser.add_argument(
        "--interval",
        default=os.getenv("MUSIC_CURATOR_INTERVAL"),
        help="repeat forever, waiting this long between passes, e.g. 6h (env MUSIC_CURATOR_INTERVAL)",
    )
    parser.add_argument(
        "--request-delay",
        type=float,
        default=float(os.getenv("MUSIC_CURATOR_REQUEST_DELAY", REQUEST_DELAY_SECONDS)),
        help="seconds between API requests (env MUSIC_CURATOR_REQUEST_DELAY)",
    )
    parser.add_argument(
        "--backfill-limit",
        type=int,
        default=int(os.getenv("MUSIC_CURATOR_BACKFILL_LIMIT", "0")),
        help="cap the backfill at this many requests per pass; 0 for no cap"
        " (env MUSIC_CURATOR_BACKFILL_LIMIT)",
    )
    parser.add_argument(
        "--lidarr-url",
        default=os.getenv("MUSIC_CURATOR_LIDARR_URL"),
        help="base URL of Lidarr, e.g. http://lidarr:8686 (env MUSIC_CURATOR_LIDARR_URL)",
    )
    parser.add_argument(
        "--lidarr-api-key",
        default=os.getenv("MUSIC_CURATOR_LIDARR_API_KEY"),
        help="Lidarr API key (env MUSIC_CURATOR_LIDARR_API_KEY)",
    )
    parser.add_argument(
        "--mirror",
        default=os.getenv("MUSIC_CURATOR_MIRROR"),
        help="root of the MP3 mirror; playlists are written into it"
        " (env MUSIC_CURATOR_MIRROR)",
    )
    parser.add_argument(
        "--library-root",
        default=os.getenv("MUSIC_CURATOR_LIBRARY_ROOT"),
        help="prefix to strip from Lidarr's track paths when mapping them into the"
        " mirror; derived from the indexed artist folders when unset"
        " (env MUSIC_CURATOR_LIBRARY_ROOT)",
    )
    parser.add_argument(
        "--playlist-limit",
        type=int,
        default=int(os.getenv("MUSIC_CURATOR_PLAYLIST_LIMIT", "100")),
        help="most tracks to put in any one playlist (env MUSIC_CURATOR_PLAYLIST_LIMIT)",
    )
    parser.add_argument(
        "--vibes",
        default=os.getenv("MUSIC_CURATOR_VIBES"),
        help="JSON file of mood definitions, replacing the built-in set"
        " (env MUSIC_CURATOR_VIBES)",
    )
    parser.add_argument(
        "--tag-limit",
        type=int,
        default=int(os.getenv("MUSIC_CURATOR_TAG_LIMIT", "0")),
        help="cap artist tag lookups per pass; 0 for no cap (env MUSIC_CURATOR_TAG_LIMIT)",
    )
    parser.add_argument(
        "--skip-index",
        action="store_true",
        help="do not re-index the library this pass; match against the index already held",
    )
    parser.add_argument(
        "--report-only",
        action="store_true",
        help="report on the existing store without fetching anything",
    )
    return parser


def main(argv=None, clock=time.time):
    """Entry point. Returns a process exit code."""
    logging.basicConfig(format="%(asctime)s %(levelname)s %(message)s", level=logging.INFO)
    args = build_parser().parse_args(argv)

    if not args.report_only and (not args.user or not args.api_key):
        logger.error("both a Last.fm user and an API key are required")
        return 2

    try:
        interval = parse_interval(args.interval) if args.interval else None
        vibes = load_vibes(args.vibes)
    except ValueError as error:
        logger.error("%s", error)
        return 2

    database = Path(args.db).resolve()
    lock = acquire_lock(database)
    if lock is None:
        logger.error("another pass is already using %s", database)
        return 3

    store = Store(database)

    if args.report_only:
        report(store, int(clock()))
        store.close()
        lock.close()
        return 0

    client = Lastfm(args.api_key, delay=args.request_delay)

    lidarr = None
    if args.lidarr_url and args.lidarr_api_key and not args.skip_index:
        lidarr = Lidarr(args.lidarr_url, args.lidarr_api_key)
    elif not args.lidarr_url:
        logger.info("no Lidarr URL configured; ingesting scrobbles only")

    stopping = False

    def stop(signum, _frame):
        nonlocal stopping
        stopping = True
        logger.info("signal %d received; finishing the current pass", signum)

    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)

    try:
        while True:
            now = int(clock())
            try:
                run_once(
                    client,
                    store,
                    args.user,
                    now,
                    args.backfill_limit,
                    lidarr,
                    args.mirror,
                    args.library_root,
                    args.playlist_limit,
                    vibes,
                    args.tag_limit,
                )
            except (LastfmError, LidarrError) as error:
                logger.error("%s", error)
                if interval is None:
                    return 1
            report(store, now)

            if interval is None or stopping:
                return 0
            logger.info("sleeping %ds", interval)
            for _ in range(interval):
                if stopping:
                    return 0
                time.sleep(1)
    finally:
        store.close()
        lock.close()


def run():
    """Console-script entry point."""
    sys.exit(main())


if __name__ == "__main__":
    run()