feedBack/lib/sloppak.py

"""Sloppak — open song format loader.

A `.sloppak` is an open, hand-editable song package. It exists in two
interchangeable forms:

1. **Zip archive** — a `.sloppak` file containing a `manifest.yaml`,
   arrangement JSONs, stem OGGs, optional cover/lyrics. Distribution form.
2. **Directory** — a directory whose name ends in `.sloppak/` containing the
   same files. Authoring form.

See the format spec in the project's sloppak plan for the full layout.
"""

from __future__ import annotations

import logging
import math
import shutil
import threading
import zipfile
from dataclasses import dataclass, field
from pathlib import Path

log = logging.getLogger("feedBack.lib.sloppak")

# The feedpak format version this build targets / writes (manifest
# `feedpak_version`, a semver string per spec §4). Readers tolerate any version
# (additive/MINOR compatibility); writers stamp this.
FEEDPAK_VERSION = "1.2.0"

# Package suffixes. The format is byte-identical regardless of suffix; `.feedpak`
# is the current write extension, `.sloppak` the legacy one we still read.
FEEDPAK_EXT = ".feedpak"
SLOPPAK_EXT = ".sloppak"
SONG_EXTS = (FEEDPAK_EXT, SLOPPAK_EXT)  # accepted on read/discovery

import yaml

from jsonc import load_json
from safepath import safe_join
from song import (
    Song,
    Beat,
    Section,
    Arrangement,
    arrangement_from_wire,
    _finite_float,
    sanitize_tempos,
)
import drums as drums_mod
import notation as notation_mod


# ── Format detection ──────────────────────────────────────────────────────────

def is_sloppak(path: Path) -> bool:
    """True if path looks like a song package (zip file or directory).

    Accepts both the current `.feedpak` suffix and the legacy `.sloppak` one —
    same on-disk format, either form.
    """
    return path.name.lower().endswith(SONG_EXTS)


# ── Source resolution (zip unpack cache + directory passthrough) ──────────────

# Maps sloppak filename (relative to DLC_DIR) → (source_dir, mtime, size).
# For directory-form sloppaks, source_dir is the original path and we only
# track it so serving can locate it by filename.
# For zipped sloppaks, source_dir is a cache dir under the unpack root.
_source_cache: dict[str, tuple[Path, float, int]] = {}
_source_lock = threading.Lock()

# Full-archive unpacks (zip form) are expensive — they write every stem to
# disk. Cap how many run at once so a burst (e.g. many plays queued, or a stray
# caller looping the library) can't saturate disk/CPU, and serialize per-file so
# two callers never rmtree + re-extract the same dest simultaneously (which
# would corrupt the half-written dir the other is reading).
_UNPACK_MAX_CONCURRENCY = 2
_unpack_semaphore = threading.BoundedSemaphore(_UNPACK_MAX_CONCURRENCY)
_unpack_locks: dict[str, threading.Lock] = {}
_unpack_locks_guard = threading.Lock()


def _unpack_lock_for(filename: str) -> threading.Lock:
    """Return a stable per-file lock so concurrent unpacks of the same sloppak
    serialize instead of racing on the same destination dir."""
    with _unpack_locks_guard:
        lk = _unpack_locks.get(filename)
        if lk is None:
            lk = threading.Lock()
            _unpack_locks[filename] = lk
        return lk


def _unpack_zip(zip_path: Path, dest: Path) -> None:
    """Extract a sloppak zip archive into dest, replacing any previous contents.

    Members whose names escape ``dest`` via ``..`` segments, absolute paths, or
    Windows-style separators are skipped with a warning so a crafted sloppak
    can't write outside the unpack cache (zip-slip).
    """
    if dest.exists():
        shutil.rmtree(dest, ignore_errors=True)
    dest.mkdir(parents=True, exist_ok=True)
    dest_resolved = dest.resolve()
    with zipfile.ZipFile(str(zip_path), "r") as zf:
        for member in zf.infolist():
            target = safe_join(dest_resolved, member.filename)
            if target is None:
                log.warning("sloppak: rejected unsafe zip member %r", member.filename)
                continue
            # A contained-but-degenerate name (e.g. "." or "subdir/..") would
            # resolve back to the unpack root itself; opening that path for
            # write is meaningless and would mask a real bug, so skip it.
            if target == dest_resolved:
                log.warning("sloppak: rejected zip member resolving to unpack root %r", member.filename)
                continue
            try:
                if member.is_dir():
                    target.mkdir(parents=True, exist_ok=True)
                    continue
                target.parent.mkdir(parents=True, exist_ok=True)
                with zf.open(member) as src, open(target, "wb") as dst:
                    shutil.copyfileobj(src, dst)
            except (OSError, zipfile.BadZipFile, RuntimeError, NotImplementedError) as e:
                log.warning("sloppak: failed to extract zip member %r: %s", member.filename, e)
                continue


def _safe_id(filename: str) -> str:
    """Turn a filename into a filesystem-safe cache key (no path separators)."""
    return filename.replace("/", "__").replace("\\", "__").replace(" ", "_")


def resolve_source_dir(
    filename: str,
    dlc_root: Path,
    unpack_cache_root: Path,
) -> Path:
    """Return the on-disk directory containing a sloppak's files.

    - Directory-form: returns the sloppak dir itself (no copy).
    - Zip-form:       unpacks to ``unpack_cache_root/{id}/`` on first use,
                      re-unpacks if mtime/size changed, then returns that dir.

    Caches the resolution so subsequent calls are ~free.
    """
    path = dlc_root / filename
    stat = path.stat()
    mtime, size = stat.st_mtime, stat.st_size

    with _source_lock:
        cached = _source_cache.get(filename)
        if cached:
            cached_dir, cached_mtime, cached_size = cached
            if (
                cached_mtime == mtime
                and cached_size == size
                and cached_dir.exists()
            ):
                return cached_dir

    if path.is_dir():
        resolved = path
    else:
        # Zip form — unpack to the cache. Serialize per-file (so concurrent
        # callers don't rmtree + re-extract the same dest at once) and cap
        # global unpack concurrency (so a burst can't saturate disk/CPU).
        dest = unpack_cache_root / _safe_id(filename)
        with _unpack_lock_for(filename):
            # Re-check the cache inside the per-file lock — a prior holder may
            # have just finished unpacking this exact (mtime, size).
            with _source_lock:
                cached = _source_cache.get(filename)
            if (
                cached
                and cached[1] == mtime
                and cached[2] == size
                and cached[0].exists()
            ):
                resolved = cached[0]
            else:
                with _unpack_semaphore:
                    _unpack_zip(path, dest)
                resolved = dest

    with _source_lock:
        _source_cache[filename] = (resolved, mtime, size)
    return resolved


def get_cached_source_dir(filename: str) -> Path | None:
    """Return the cached source dir for a sloppak if one is known."""
    with _source_lock:
        cached = _source_cache.get(filename)
        return cached[0] if cached else None


# ── Manifest + song loading ───────────────────────────────────────────────────

def _read_manifest(source_dir: Path) -> dict:
    mf = source_dir / "manifest.yaml"
    if not mf.exists():
        mf = source_dir / "manifest.yml"
    if not mf.exists():
        raise FileNotFoundError(f"manifest.yaml not found in {source_dir}")
    with mf.open("r", encoding="utf-8") as fh:
        data = yaml.safe_load(fh)
    if not isinstance(data, dict):
        raise ValueError("manifest.yaml must contain a mapping at the top level")
    return data


def _read_manifest_from_zip(zip_path: Path) -> dict:
    """Read just manifest.yaml from a zipped sloppak without unpacking stems."""
    with zipfile.ZipFile(str(zip_path), "r") as zf:
        for name in ("manifest.yaml", "manifest.yml"):
            try:
                with zf.open(name) as fh:
                    data = yaml.safe_load(fh.read().decode("utf-8"))
                    if isinstance(data, dict):
                        return data
            except KeyError:
                continue
    raise FileNotFoundError(f"manifest.yaml not found in zip {zip_path}")


def load_manifest(path: Path) -> dict:
    """Return the parsed manifest dict for a sloppak (dir or zip)."""
    if path.is_dir():
        return _read_manifest(path)
    return _read_manifest_from_zip(path)


_COVER_MEDIA_TYPES = {
    ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
    ".png": "image/png", ".webp": "image/webp",
}


def _cover_media_type(name: str) -> str:
    return _COVER_MEDIA_TYPES.get(Path(name).suffix.lower(), "image/jpeg")


def read_cover_bytes(
    path: Path, manifest: dict | None = None
) -> tuple[bytes, str] | None:
    """Return ``(image_bytes, media_type)`` for a sloppak's cover, or ``None``.

    Reads ONLY the cover image. For a zipped sloppak this opens the single
    cover member rather than unpacking the whole archive (stems included), so
    serving album art on the library grid never triggers a full extraction —
    the dominant cost behind slow cover loading on scroll.
    """
    try:
        if manifest is None:
            manifest = load_manifest(path)
    except Exception:
        manifest = {}
    cover_rel = str((manifest or {}).get("cover") or "cover.jpg")

    if path.is_dir():
        # Directory form — read the file, guarding against escape.
        cover_path = (path / cover_rel).resolve()
        try:
            cover_path.relative_to(path.resolve())
        except ValueError:
            return None
        if cover_path.is_file():
            try:
                return cover_path.read_bytes(), _cover_media_type(cover_path.name)
            except OSError as e:
                log.warning("sloppak: failed to read cover %r: %s", cover_path, e)
        return None

    # Zip form — read just the cover member, no unpack. Normalize the manifest
    # name the way the filesystem would (collapse './' and 'a/../b', backslash →
    # slash) so a non-canonical-but-valid cover like './cover.jpg' still resolves
    # to the archive member 'cover.jpg' — matching the old unpack-then-resolve
    # behavior — and reject zip-slip escape before opening.
    _zip_root = Path("/_root").resolve()
    safe = safe_join(_zip_root, cover_rel)
    # `safe is None` → escape; `safe == _zip_root` → a degenerate name like "."
    # or "subdir/.." that collapses to the root (member would be "."). Reject
    # both, mirroring _unpack_zip's degenerate-root guard.
    if safe is None or safe == _zip_root:
        log.warning("sloppak: rejected unsafe cover name %r in %r", cover_rel, path)
        return None
    member = safe.relative_to(_zip_root).as_posix()
    try:
        with zipfile.ZipFile(str(path), "r") as zf:
            try:
                data = zf.read(member)
            except KeyError:
                return None
        return data, _cover_media_type(member)
    except (OSError, zipfile.BadZipFile, RuntimeError) as e:
        log.warning("sloppak: failed to read cover from zip %r: %s", path, e)
        return None


def _sanitize_time_signatures(events) -> list[dict]:
    """Clean a time-signature event list (``[{time, ts:[num, den]}]``): keep
    entries with a finite non-bool ``time`` and a ``ts`` of two integers >= 1,
    sorted by time. Non-list / all-invalid input -> ``[]``."""
    out: list[dict] = []
    if isinstance(events, list):
        for ev in events:
            if not isinstance(ev, dict):
                continue
            t = ev.get("time")
            ts = ev.get("ts")
            if (not isinstance(t, (int, float)) or isinstance(t, bool)
                    or not math.isfinite(t)):
                continue
            if not isinstance(ts, list) or len(ts) != 2:
                continue
            if not all(isinstance(x, int) and not isinstance(x, bool) and x >= 1
                       for x in ts):
                continue
            out.append({"time": float(t), "ts": [int(ts[0]), int(ts[1])]})
        out.sort(key=lambda e: e["time"])
    return out


@dataclass
class LoadedSloppak:
    """Result of loading a sloppak: the Song object plus stem descriptors."""
    song: Song
    stems: list[dict]           # [{"id": str, "file": str, "default": bool}]
    source_dir: Path
    manifest: dict
    # The pack's declared format version (manifest `feedpak_version`, a semver
    # string per spec §4). None when absent (legacy / pre-versioning packs).
    feedpak_version: str | None = None
    # Parsed `drum_tab.json` payload when the manifest carries a `drum_tab:`
    # key pointing at a readable, schema-valid file. None otherwise (older
    # sloppaks, sloppaks without drums, sloppaks whose drum tab failed to
    # parse). The drums plugin reads this through the highway WS rather than
    # the file directly — see server.py highway_ws for the wire shape.
    drum_tab: dict | None = None
    # Parsed `song_timeline.json` payload when the manifest carries a
    # `song_timeline:` key pointing at a readable, schema-valid file.
    # When present, its beats/sections take priority over any beats/sections
    # embedded in the arrangement JSONs.
    song_timeline: dict | None = None
    # Parsed `keys.json` payload (manifest `keys:` key) — a song-level,
    # instrument-independent key/scale-change track (spec §7.7). None when
    # absent / unreadable / malformed. Streamed over the highway WS as a
    # `keys` message; consumers (renderers, plugins) read it from there.
    keys: dict | None = None
    # Sanitized song-level tempo + time-signature maps from `song_timeline.json`
    # (feedpak 1.2.0). `tempos`: [{time, bpm}]; `time_signatures`: [{time, ts}].
    # None when absent/empty. Streamed over the highway WS (`tempos` /
    # `time_signatures` messages); a per-chart arrangement `tempos` overrides
    # `tempos` for that chart (spec §6.10).
    tempos: list | None = None
    time_signatures: list | None = None
    # Maps arrangement id → validated notation payload.  None when no
    # arrangement passed schema validation; a non-empty dict only when at least
    # one arrangement carried a `notation:` sub-key whose file loaded and passed
    # schema validation.  The dict is never an empty mapping at runtime.
    notation_by_id: dict[str, dict] | None = None
    # Manifest arrangement id for each entry in song.arrangements, in the same
    # order.  None where the manifest entry had no id field.  Parallel to
    # song.arrangements (not to manifest["arrangements"]) — skipped entries are
    # absent so indexing by song.arrangements index is safe.
    arrangement_ids: list[str | None] = field(default_factory=list)
    # Manifest-relative path to the single full-mix audio file, taken from the
    # manifest `original_audio:` key (e.g. "original/full.ogg"). This is the
    # pre-separation mixdown that exists alongside the per-instrument `stems`.
    # None when the key is absent, points outside source_dir, or the file is
    # missing on disk. Served to the front-end via the highway WS as
    # `original_audio_url`; the stems plugin uses it to play the untouched mix
    # when every stem slider is at unity (and the separate stems otherwise).
    original_audio: str | None = None


def load_song(
    filename: str,
    dlc_root: Path,
    unpack_cache_root: Path,
) -> LoadedSloppak:
    """Fully load a sloppak: resolve its source dir, parse manifest + all
    arrangements + optional lyrics, and return a ready-to-stream Song."""
    source_dir = resolve_source_dir(filename, dlc_root, unpack_cache_root)
    manifest = _read_manifest(source_dir)

    song = Song(
        title=str(manifest.get("title", "")),
        artist=str(manifest.get("artist", "")),
        album=str(manifest.get("album", "")),
        year=int(manifest.get("year", 0) or 0),
        song_length=float(manifest.get("duration", 0.0) or 0.0),
    )

    # Load each arrangement from its JSON file.
    notation_acc: dict[str, dict] = {}
    any_notation = False
    arrangement_ids_acc: list[str | None] = []  # parallel to song.arrangements
    for entry in manifest.get("arrangements", []) or []:
        if not isinstance(entry, dict):
            log.warning("sloppak: non-dict arrangement entry skipped (%r)", type(entry).__name__)
            continue
        rel_raw = entry.get("file")
        rel = rel_raw.strip() if isinstance(rel_raw, str) else ""
        notation_raw = entry.get("notation")
        has_notation_key = isinstance(notation_raw, str) and bool(notation_raw.strip())
        if not rel and not has_notation_key:
            continue
        data = None
        if rel:
            try:
                arr_path = (source_dir / rel).resolve()
                arr_path.relative_to(source_dir.resolve())
            except ValueError:
                log.warning("sloppak: arrangement path %r escapes source_dir — skipped", rel)
                continue
            except OSError as e:
                log.warning("sloppak: arrangement path resolution failed (%s) — skipped", e)
                continue
            if not arr_path.exists():
                continue
            try:
                data = load_json(arr_path)
            except Exception as e:
                log.debug("sloppak: failed to parse arrangement %r: %s", rel, e)
                continue
            arr = arrangement_from_wire(data)
        else:
            arr = arrangement_from_wire({
                "notes": [], "chords": [], "anchors": [],
                "handshapes": [], "templates": [],
            })
        # Manifest-level overrides take precedence over anything embedded in
        # the arrangement JSON (name, tuning, capo, centOffset).
        if entry.get("name"):
            arr.name = str(entry["name"])
        if "tuning" in entry:
            arr.tuning = list(entry["tuning"])
        if "capo" in entry:
            arr.capo = int(entry["capo"])
        if "centOffset" in entry:
            # _finite_float keeps a malformed manifest NaN/Infinity from
            # poisoning the song_info JSON (same guard as the wire path).
            arr.cent_offset = _finite_float(entry["centOffset"])

        # Beats/sections can live on the arrangement itself in the wire format.
        # If the manifest-level arrangement JSON carries them, pull them onto
        # the song object the first time we see them.
        if data is not None:
            if not song.beats:
                for b in data.get("beats", []) or []:
                    song.beats.append(
                        Beat(time=float(b.get("time", 0)), measure=int(b.get("measure", -1)))
                    )
            if not song.sections:
                for s in data.get("sections", []) or []:
                    song.sections.append(
                        Section(
                            name=str(s.get("name", "")),
                            number=int(s.get("number", 0)),
                            start_time=float(s.get("time", s.get("start_time", 0))),
                        )
                    )
        song.arrangements.append(arr)
        arr_id = str(entry.get("id", "")).strip()
        arrangement_ids_acc.append(arr_id or None)

        if not arr_id:
            log.warning("sloppak: arrangement entry has no id — notation skipped")
            continue
        notation_rel = entry.get("notation")
        if not isinstance(notation_rel, str):
            continue
        notation_rel = notation_rel.strip()
        if not notation_rel:
            continue
        try:
            nt_path = (source_dir / notation_rel).resolve()
            nt_path.relative_to(source_dir.resolve())
        except ValueError:
            log.warning("sloppak: notation path %r escapes source_dir — skipped", notation_rel)
            nt_path = None
        except OSError as e:
            log.warning("sloppak: notation path resolution failed (%s) — skipped", e)
            nt_path = None
        raw_nt = None
        if nt_path is not None and nt_path.exists():
            try:
                raw_nt = load_json(nt_path)
            except Exception as e:
                log.warning("sloppak: failed to parse notation %r: %s", notation_rel, e)
        if raw_nt is not None:
            ok, reason = notation_mod.validate_notation(raw_nt)
            if ok:
                if arr_id in notation_acc:
                    log.warning(
                        "sloppak: duplicate arrangement id %r — notation overwritten", arr_id
                    )
                notation_acc[arr_id] = raw_nt
                any_notation = True
            else:
                log.warning("sloppak: notation %r failed validation: %s", notation_rel, reason)
    notation_by_id_data = notation_acc if any_notation else None

    # Optional drum_tab.json — top-level manifest key per sloppak-spec §5.3.
    # The file lives off to the side (its own JSON), and the manifest opts in
    # via `drum_tab: drum_tab.json`. The loader stays permissive: a missing file
    # silently disables drum playback; a malformed or invalid tab disables it
    # with a warning.
    drum_tab_data: dict | None = None
    drum_tab_rel = manifest.get("drum_tab")
    if isinstance(drum_tab_rel, str) and drum_tab_rel:
        # Constrain to source_dir to prevent a crafted manifest from reading
        # files outside the sloppak directory via path traversal (e.g. ../../etc).
        # Wrap both resolve() calls in a broad handler: symlink loops and
        # permission errors on .resolve() should disable drums, not abort load.
        try:
            dt_path = (source_dir / drum_tab_rel).resolve()
            dt_path.relative_to(source_dir.resolve())
        except ValueError:
            log.warning("sloppak: drum_tab path %r escapes source_dir — skipped", drum_tab_rel)
            dt_path = None
        except OSError as e:
            log.warning("sloppak: drum_tab path resolution failed (%s) — skipped", e)
            dt_path = None
        if dt_path is not None and dt_path.exists():
            try:
                raw = load_json(dt_path)
            except Exception as e:
                log.warning("sloppak: failed to parse drum_tab %r: %s", drum_tab_rel, e)
                raw = None
            if raw is not None:
                ok, reason = drums_mod.validate_drum_tab(raw)
                if ok:
                    drum_tab_data = raw
                else:
                    log.warning("sloppak: drum_tab %r failed validation: %s",
                                drum_tab_rel, reason)

    # Drum-only sloppak: every GP track was percussion, so it ships a
    # drum_tab but no pitched arrangements. The highway WS rejects an empty
    # arrangements list with "No arrangements found" *before* it serves the
    # drum_tab, leaving the drums unplayable even in the drum highway.
    # Synthesize a minimal placeholder arrangement so the stream proceeds and
    # the drum_tab reaches the drum highway. It carries no notes (the guitar
    # highway just shows an empty board) and, when the manifest omits a
    # duration, derives a song length from the last drum hit so the timeline
    # isn't zero-length.
    if not song.arrangements and drum_tab_data is not None:
        if song.song_length <= 0:
            # validate_drum_tab() intentionally does NOT type-check individual
            # hits (they're sanitized at WS-stream time), so a hit may carry a
            # non-numeric "t". Skip anything that won't convert rather than let
            # one malformed hit abort the whole load.
            _max_t = 0.0
            for _h in drum_tab_data.get("hits") or []:
                if not isinstance(_h, dict):
                    continue
                try:
                    _max_t = max(_max_t, float(_h.get("t", 0) or 0))
                except (TypeError, ValueError):
                    continue
            if _max_t > 0:
                song.song_length = _max_t + 2.0
        song.arrangements.append(Arrangement(name="Drums"))
        arrangement_ids_acc.append(None)

    # Optional song_timeline.json — top-level manifest key per sloppak-spec §5.3.
    # When present, its beats and sections override whatever the arrangement JSONs
    # already loaded onto the song object — song_timeline is the authoritative
    # source for timeline data in sloppaks that carry it.
    song_timeline_data: dict | None = None
    tempos_data: list | None = None
    time_sigs_data: list | None = None
    song_timeline_rel = manifest.get("song_timeline")
    if isinstance(song_timeline_rel, str) and song_timeline_rel:
        try:
            st_path = (source_dir / song_timeline_rel).resolve()
            st_path.relative_to(source_dir.resolve())
        except ValueError:
            log.warning("sloppak: song_timeline path %r escapes source_dir — skipped", song_timeline_rel)
            st_path = None
        except OSError as e:
            log.warning("sloppak: song_timeline path resolution failed (%s) — skipped", e)
            st_path = None
        if st_path is not None and st_path.exists():
            try:
                raw = load_json(st_path)
            except Exception as e:
                log.warning("sloppak: failed to parse song_timeline %r: %s", song_timeline_rel, e)
                raw = None
            if raw is not None:
                if not isinstance(raw, dict):
                    log.warning("sloppak: song_timeline %r ignored — expected dict, got %s",
                                song_timeline_rel, type(raw).__name__)
                elif not isinstance(raw.get("beats"), list):
                    log.warning("sloppak: song_timeline %r ignored — 'beats' must be a list",
                                song_timeline_rel)
                elif not isinstance(raw.get("sections"), list):
                    log.warning("sloppak: song_timeline %r ignored — 'sections' must be a list",
                                song_timeline_rel)
                else:
                    song.beats = []
                    song.sections = []
                    for b in raw["beats"]:
                        if not isinstance(b, dict):
                            log.warning(
                                "sloppak: song_timeline %r — non-dict beat entry skipped (%r)",
                                song_timeline_rel, type(b).__name__,
                            )
                            continue
                        try:
                            song.beats.append(
                                Beat(
                                    # _finite_float prevents NaN/Infinity from
                                    # slipping through json.loads and poisoning
                                    # the highway WS JSON with invalid tokens.
                                    time=_finite_float(b.get("time", 0)),
                                    measure=int(b.get("measure", -1)),
                                )
                            )
                        except (TypeError, ValueError):
                            log.warning(
                                "sloppak: song_timeline %r — invalid beat entry skipped (%r)",
                                song_timeline_rel, b,
                            )
                            continue
                    for s in raw["sections"]:
                        if not isinstance(s, dict):
                            log.warning(
                                "sloppak: song_timeline %r — non-dict section entry skipped (%r)",
                                song_timeline_rel, type(s).__name__,
                            )
                            continue
                        try:
                            song.sections.append(
                                Section(
                                    name=str(s.get("name", "")),
                                    number=int(s.get("number", 0)),
                                    # Same key fallback as the arrangement-JSON
                                    # section parser: `time` with `start_time`
                                    # as the legacy alias.
                                    # _finite_float: same NaN/Infinity guard as
                                    # beat timestamps above.
                                    start_time=_finite_float(
                                        s.get("time", s.get("start_time", 0))
                                    ),
                                )
                            )
                        except (TypeError, ValueError):
                            log.warning(
                                "sloppak: song_timeline %r — invalid section entry skipped (%r)",
                                song_timeline_rel, s,
                            )
                            continue
                    song_timeline_data = raw
            # tempos / time_signatures (feedpak 1.2.0) are independent of the
            # beats/sections validation above — all are optional — so load them
            # whenever the payload parsed to a dict.
            if isinstance(raw, dict):
                tempos_data = sanitize_tempos(raw.get("tempos")) or None
                time_sigs_data = _sanitize_time_signatures(
                    raw.get("time_signatures")) or None

    # Optional shared lyrics file. Same safety posture as the drum_tab
    # loader above: constrain the manifest-declared path to source_dir
    # (a crafted sloppak with `lyrics: ../../etc/passwd.json` would
    # otherwise read arbitrary files), and ignore the payload unless
    # it's the documented shape — a flat list of syllable dicts.
    # Anything else (a dict at the root, a string, malformed entries)
    # leaves `song.lyrics` empty rather than streaming surprise data
    # downstream through the WS path.
    lyrics_rel = manifest.get("lyrics")
    if isinstance(lyrics_rel, str) and lyrics_rel:
        try:
            lyr_path = (source_dir / lyrics_rel).resolve()
            lyr_path.relative_to(source_dir.resolve())
        except ValueError:
            log.warning("sloppak: lyrics path %r escapes source_dir — skipped", lyrics_rel)
            lyr_path = None
        except OSError as e:
            log.warning("sloppak: lyrics path resolution failed (%s) — skipped", e)
            lyr_path = None
        if lyr_path is not None and lyr_path.exists():
            try:
                raw = load_json(lyr_path)
            except Exception as e:
                log.debug("sloppak: failed to parse lyrics %r: %s", lyrics_rel, e)
                raw = None
            if isinstance(raw, list):
                # Filter to entries that at least look like syllables —
                # presence of all three required keys with the right
                # primitive types. Drops anything weird without poisoning
                # the whole list.
                song.lyrics = [
                    e for e in raw
                    if isinstance(e, dict)
                    and isinstance(e.get("w"), str)
                    and isinstance(e.get("t"), (int, float))
                    and isinstance(e.get("d"), (int, float))
                ]
                if song.lyrics:
                    # Provenance. The feedpak spec (§7.1) vocabulary is
                    # {authored, transcribed, user}; older manifests + the
                    # in-tree readers also use the source-format names
                    # (xml/notechart) and the WhisperX engine name
                    # (whisperx). Accept the union so both spec-compliant
                    # writers (e.g. the stem_splitter plugin emitting
                    # `transcribed`) and legacy packs validate. Validate
                    # against the closed enum so a hand-edited (or otherwise
                    # malformed) manifest can't propagate a YAML dict / list /
                    # arbitrary string into the highway WS `lyrics.source`
                    # field and out to plugin badges. Anything outside the
                    # enum (or the wrong type) falls back to "xml" — the
                    # back-compat default — instead of being stringified and
                    # trusted.
                    # Post-alias values only: `whisperx` is normalised to
                    # `transcribed` before the membership check below, so (like
                    # `sng`) it is intentionally absent from this set.
                    _ALLOWED_LYRICS_SOURCES = {
                        "xml", "notechart", "user",
                        "authored", "transcribed",
                    }
                    # Legacy aliases: older manifests labelled note-chart-derived
                    # lyrics with the source format's name, and the WhisperX
                    # fallback with the engine name — normalise both to the
                    # spec vocabulary the badges now expect.
                    _LYRICS_SOURCE_ALIASES = {"sng": "notechart", "whisperx": "transcribed"}
                    raw_source = manifest.get("lyrics_source")
                    if isinstance(raw_source, str):
                        raw_source = _LYRICS_SOURCE_ALIASES.get(raw_source, raw_source)
                    if isinstance(raw_source, str) and raw_source in _ALLOWED_LYRICS_SOURCES:
                        song.lyrics_source = raw_source
                    else:
                        if raw_source is not None and (
                            not isinstance(raw_source, str)
                            or raw_source not in _ALLOWED_LYRICS_SOURCES
                        ):
                            log.warning(
                                "sloppak: ignoring invalid lyrics_source %r — "
                                "must be one of %s; falling back to 'xml'",
                                raw_source, sorted(_ALLOWED_LYRICS_SOURCES),
                            )
                        song.lyrics_source = "xml"
            elif raw is not None:
                log.warning("sloppak: lyrics %r ignored — expected list, got %s",
                            lyrics_rel, type(raw).__name__)

    # Stem descriptors — normalized for callers. File paths are resolved but
    # returned as ``file`` relative strings so URL construction stays caller-side.
    stems: list[dict] = []
    for s in manifest.get("stems", []) or []:
        if not isinstance(s, dict):
            continue
        sid = str(s.get("id", ""))
        sfile = str(s.get("file", ""))
        if not sid or not sfile:
            continue
        default_val = s.get("default", True)
        if isinstance(default_val, str):
            default_on = default_val.lower() not in ("off", "false", "0", "no")
        else:
            default_on = bool(default_val)
        stems.append({"id": sid, "file": sfile, "default": default_on})

    # Optional keys.json — song-level, instrument-independent key/scale track
    # (manifest `keys:` key, spec §7.7). Permissive like the other side-files:
    # missing / unreadable / malformed -> None, never fatal. Stored as a
    # sanitized {version, events:[{t, key, scale?}]} (finite t, non-empty string
    # key, sorted) so the highway WS can stream it without re-validating.
    keys_data: dict | None = None
    keys_rel = manifest.get("keys")
    if isinstance(keys_rel, str) and keys_rel:
        try:
            k_path = (source_dir / keys_rel).resolve()
            k_path.relative_to(source_dir.resolve())
        except ValueError:
            log.warning("sloppak: keys path %r escapes source_dir — skipped", keys_rel)
            k_path = None
        except OSError as e:
            log.warning("sloppak: keys path resolution failed (%s) — skipped", e)
            k_path = None
        if k_path is not None and k_path.exists():
            try:
                raw = load_json(k_path)
            except Exception as e:
                log.warning("sloppak: failed to parse keys %r: %s", keys_rel, e)
                raw = None
            if raw is not None and not isinstance(raw, dict):
                log.warning("sloppak: keys %r ignored — expected dict, got %s",
                            keys_rel, type(raw).__name__)
            elif isinstance(raw, dict):
                if not isinstance(raw.get("events"), list):
                    log.warning("sloppak: keys %r ignored — 'events' must be a list", keys_rel)
                else:
                    clean_events: list[dict] = []
                    for ev in raw["events"]:
                        if not isinstance(ev, dict):
                            continue
                        # Drop events with a missing / non-numeric / non-finite
                        # time rather than silently rewriting them to 0.0 — a
                        # bad `t` makes the whole event meaningless.
                        t = ev.get("t")
                        if (not isinstance(t, (int, float)) or isinstance(t, bool)
                                or not math.isfinite(t)):
                            continue
                        t = float(t)
                        key = ev.get("key")
                        if not isinstance(key, str) or not key:
                            continue
                        entry = {"t": t, "key": key}
                        scale = ev.get("scale")
                        if isinstance(scale, str) and scale:
                            entry["scale"] = scale
                        clean_events.append(entry)
                    clean_events.sort(key=lambda e: e["t"])
                    # int only — a float version (incl. NaN/Inf, which json.loads
                    # accepts) would raise on int(); default rather than abort the
                    # load of an optional side-file.
                    _ver = raw.get("version")
                    keys_data = {
                        "version": _ver if isinstance(_ver, int)
                                   and not isinstance(_ver, bool) else 1,
                        "events": clean_events,
                    }

    _fpv = manifest.get("feedpak_version")
    # Optional full-mix audio — manifest `original_audio:` key. The single
    # pre-separation mixdown that ships alongside the per-instrument stems.
    # Same permissive, path-traversal-guarded posture as drum_tab above: a
    # missing/escaping/absent file simply leaves the full mix unavailable (the
    # player falls back to the separate stems) rather than aborting the load.
    # We store the manifest-relative string so server.py can build its URL the
    # same way it builds stem URLs (via the /api/sloppak/.../file/ endpoint).
    original_audio_data: str | None = None
    original_audio_rel = manifest.get("original_audio")
    if isinstance(original_audio_rel, str) and original_audio_rel.strip():
        rel = original_audio_rel.strip()
        try:
            oa_path = (source_dir / rel).resolve()
            oa_path.relative_to(source_dir.resolve())
        except ValueError:
            log.warning("sloppak: original_audio path %r escapes source_dir — skipped", rel)
            oa_path = None
        except OSError as e:
            log.warning("sloppak: original_audio path resolution failed (%s) — skipped", e)
            oa_path = None
        if oa_path is not None and oa_path.is_file():
            original_audio_data = rel

    return LoadedSloppak(
        song=song,
        stems=stems,
        source_dir=source_dir,
        manifest=manifest,
        feedpak_version=_fpv if isinstance(_fpv, str) and _fpv else None,
        drum_tab=drum_tab_data,
        song_timeline=song_timeline_data,
        tempos=tempos_data,
        time_signatures=time_sigs_data,
        keys=keys_data,
        notation_by_id=notation_by_id_data,
        arrangement_ids=arrangement_ids_acc,
        original_audio=original_audio_data,
    )


# ── Fast metadata extractor (scanner path) ────────────────────────────────────

def _tuning_for_meta(arrangements_manifest: list[dict]) -> list[int]:
    """Best-effort guitar-first tuning for the library index."""
    for entry in arrangements_manifest:
        name = str(entry.get("name", "")).lower()
        tun = entry.get("tuning")
        if tun and isinstance(tun, list) and name in ("lead", "rhythm", "combo"):
            return list(tun)
    # Fallback: first arrangement with a tuning
    for entry in arrangements_manifest:
        tun = entry.get("tuning")
        if tun and isinstance(tun, list):
            return list(tun)
    return [0] * 6


def extract_meta(path: Path) -> dict:
    """Fast metadata for the library scanner. Reads only the manifest."""
    manifest = load_manifest(path)
    arr_list = manifest.get("arrangements", []) or []

    arrangements = []
    for i, entry in enumerate(arr_list):
        arrangements.append(
            {
                "index": i,
                "name": str(entry.get("name", entry.get("id", f"Arr{i}"))),
                "notes": 0,  # unknown without loading; fine for the index
            }
        )
    # Sort like archive path: Lead > Combo > Rhythm > Bass
    priority = {"Lead": 0, "Combo": 1, "Rhythm": 2, "Bass": 3}
    arrangements.sort(key=lambda a: priority.get(a["name"], 99))
    for i, a in enumerate(arrangements):
        a["index"] = i

    has_lyrics = bool(manifest.get("lyrics"))
    tuning_offsets = _tuning_for_meta(arr_list)

    stems_list = manifest.get("stems", []) or []
    stem_ids: list[str] = []
    for s in stems_list:
        if not isinstance(s, dict):
            continue
        sid = s.get("id")
        sfile = s.get("file")
        # Match `load_song()`'s validation: a stem entry needs BOTH a
        # non-empty id AND a non-empty file to be playable. Indexing a
        # half-formed entry would advertise a stem that load_song will
        # later refuse to surface, so the library filter would lie.
        if (
            isinstance(sid, str) and sid
            and isinstance(sfile, str) and sfile
        ):
            stem_ids.append(sid)
    stem_count = len(stem_ids)

    return {
        "title": str(manifest.get("title", "")),
        "artist": str(manifest.get("artist", "")),
        "album": str(manifest.get("album", "")),
        "year": str(manifest.get("year", "") or ""),
        # Primary genre from the feedpak `genres` list (spec 1.12.0); [0] = primary.
        "genre": (lambda g: str(g[0]) if isinstance(g, list) and g else "")(manifest.get("genres")),
        # Album track order from the feedpak `track`/`disc` fields (spec 1.12.0);
        # None when unauthored (the album view then falls back to title order).
        "track_number": (lambda v: int(v) if str(v if v is not None else "").strip().isdigit() else None)(manifest.get("track")),
        "disc": (lambda v: int(v) if str(v if v is not None else "").strip().isdigit() else None)(manifest.get("disc")),
        "duration": float(manifest.get("duration", 0) or 0),
        "tuning_offsets": tuning_offsets,  # caller maps to a name via tunings.tuning_name
        "arrangements": arrangements,
        "has_lyrics": has_lyrics,
        "stem_count": stem_count,
        # feedBack#129: per-stem filter needs the id list, not just count.
        "stem_ids": stem_ids,
    }