Lyra/worker/lyra_worker/library.py

import os
import re
import shutil

from lyra_worker.types import MBTarget

_ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]')


def safe_name(s: str) -> str:
    """A filesystem-safe path component (single segment, no separators)."""
    cleaned = _ILLEGAL.sub("_", s).strip().strip(".").strip()
    return cleaned or "Unknown"


def album_dir(dest_root: str, target: MBTarget) -> str:
    """`{dest_root}/{safe artist}/{safe album}[ (year)]`."""
    album = safe_name(target.album)
    if target.year:
        album = f"{album} ({target.year})"
    return f"{dest_root}/{safe_name(target.artist)}/{album}"


_AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"}
_COVER_STEMS = {"cover", "folder", "front"}
_COVER_EXT = {".jpg", ".jpeg", ".png"}


def list_audio_files(folder: str) -> list[str]:
    """The album's on-disk audio filenames (basenames, sorted) — captured into
    LibraryItem.trackNames so the UI can show the real tracks on disk. Top-level only
    (albums are one flat folder here); returns [] if the folder is missing/unreadable."""
    try:
        names = [
            n for n in os.listdir(folder)
            if os.path.isfile(os.path.join(folder, n)) and os.path.splitext(n)[1].lower() in _AUDIO_EXT
        ]
    except OSError:
        return []
    return sorted(names)


def staging_dir(staging_root: str, job_id: str) -> str:
    """A per-job download staging directory under `staging_root`. The staging root may be a
    volume separate from the library (e.g. local disk while the library is an SMB share):
    import_album assembles into a temp dir *on the library volume* and swaps atomically there,
    so cross-volume staging stays safe."""
    return f"{staging_root}/{job_id}"


def clear_staging_root(staging_root: str) -> None:
    """Remove every per-job staging dir under `staging_root` (worker-startup crash-orphan sweep).
    Clears the contents but not the root itself, so a mounted staging volume stays mounted."""
    if not os.path.isdir(staging_root):
        return
    for name in os.listdir(staging_root):
        path = os.path.join(staging_root, name)
        if os.path.isdir(path):
            shutil.rmtree(path, ignore_errors=True)
        else:
            try:
                os.remove(path)
            except OSError:
                pass


def _find_cover(staging: str) -> str | None:
    for root, _dirs, files in os.walk(staging):
        for name in sorted(files):
            stem, ext = os.path.splitext(name)
            if ext.lower() in _COVER_EXT and stem.lower() in _COVER_STEMS:
                return os.path.join(root, name)
    return None


def import_album(staging: str, final: str) -> str:
    """Atomically place the album's audio (+ one cover) at `final`, replacing any existing folder.
    Assembles into a sibling temp dir, then swaps via rename so a mid-assembly failure never
    destroys the existing copy. Non-audio artifacts/subfolders in `staging` are left behind.
    Tolerates a missing `staging`. Returns `final`."""
    # Recover from a swap interrupted by a crash: {final}.old holds the previous copy.
    old = final + ".old"
    if os.path.isdir(old) and not os.path.isdir(final):
        os.rename(old, final)                       # restore the last-good copy
    shutil.rmtree(old, ignore_errors=True)          # drop a leftover .old (swap had completed)
    shutil.rmtree(final + ".importing", ignore_errors=True)  # drop a stale half-assembly
    os.makedirs(os.path.dirname(final), exist_ok=True)
    tmp = final + ".importing"
    shutil.rmtree(tmp, ignore_errors=True)
    os.makedirs(tmp)
    imported = False
    if os.path.isdir(staging):
        for name in sorted(os.listdir(staging)):
            src = os.path.join(staging, name)
            if os.path.isfile(src) and os.path.splitext(name)[1].lower() in _AUDIO_EXT:
                shutil.move(src, os.path.join(tmp, name))
                imported = True
        cover = _find_cover(staging)
        if cover is not None:
            shutil.move(cover, os.path.join(tmp, "cover.jpg"))
            imported = True
    # swap: keep the old copy until the new one is assembled, then rename into place
    if os.path.isdir(final):
        if not imported:
            # Nothing new to import (e.g. a bare recovery pass with no/empty staging): leave the
            # existing (possibly just-recovered) `final` untouched rather than swapping in empty.
            shutil.rmtree(tmp, ignore_errors=True)
            return final
        old = final + ".old"
        shutil.rmtree(old, ignore_errors=True)
        os.rename(final, old)
        os.rename(tmp, final)
        shutil.rmtree(old, ignore_errors=True)
    else:
        os.rename(tmp, final)
    return final