embycovers/services/music_covers.py

"""Music library maintenance, refactored from the original ``music-covers.py`` CLI.

The interactive terminal UI is gone; what remains is pure, importable logic the
web app drives. Two entry points matter:

* :func:`scan_library` — read-only analysis. Walks ``MUSIC_ROOT`` and reports, per
  album, what each maintenance mode *would* do. Safe to call any time.
* :func:`process_library` — performs the work. Honours ``dry_run`` (the web UI
  default) so nothing is renamed, deleted, or downloaded unless the caller opts in.

Both are synchronous (filesystem + blocking HTTP); call them from FastAPI via
``asyncio.to_thread``. Progress is reported through an optional ``log`` callback.
"""

from __future__ import annotations

import logging
import os
import queue
import re
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Iterator

import requests

try:  # Optional: only needed for the "find missing year/cover" online lookups.
    import musicbrainzngs

    musicbrainzngs.set_useragent("HomelabToolkit", "1.0", "homelab-toolkit@example.com")
    # musicbrainzngs logs "uncaught attribute"/"uncaught tag" at INFO whenever the
    # MusicBrainz XML carries fields it doesn't model (e.g. release-group type-id).
    # Harmless noise — keep only real warnings.
    logging.getLogger("musicbrainzngs").setLevel(logging.WARNING)
    _HAS_MUSICBRAINZ = True
except Exception:  # pragma: no cover - optional dependency
    _HAS_MUSICBRAINZ = False

from mutagen import File as MutagenFile, MutagenError

MUSIC_ROOT = Path(os.environ.get("MUSIC_ROOT", r"\\Matt-htpc\d\Music"))

COVER_NAME_PRIORITY = ("cover.jpg", "folder.jpg", "front.jpg")
COVER_NAMES = set(COVER_NAME_PRIORITY)
COVER_MISSING_MARKER = ".cover-not-found"
LYRICS_MISSING_MARKER = ".lyrics-not-found"
LYRICS_SIDECAR_EXTENSIONS = {".lrc", ".txt"}
AUDIO_EXTENSIONS = {".mp3", ".flac", ".m4a"}
YEAR_ALBUM_FOLDER_RE = re.compile(r"^\s*(\d{4})\s*-\s*(.+?)\s*$")

LogCallback = Callable[[dict], None]


DEFAULT_RECENT_WINDOW_SECONDS = 2 * 60 * 60


@dataclass
class ProcessOptions:
    folder_cleanup: bool = False
    rename: bool = False
    file_cleanup: bool = False
    lyrics: bool = False
    covers: bool = True
    dry_run: bool = True
    recent_only: bool = False
    recent_window_seconds: int = DEFAULT_RECENT_WINDOW_SECONDS

    @classmethod
    def from_dict(cls, data: dict) -> "ProcessOptions":
        try:
            window = int(data.get("recent_window_seconds") or DEFAULT_RECENT_WINDOW_SECONDS)
        except (TypeError, ValueError):
            window = DEFAULT_RECENT_WINDOW_SECONDS
        return cls(
            folder_cleanup=bool(data.get("folder_cleanup", False)),
            rename=bool(data.get("rename", False)),
            file_cleanup=bool(data.get("file_cleanup", False)),
            lyrics=bool(data.get("lyrics", False)),
            covers=bool(data.get("covers", True)),
            dry_run=bool(data.get("dry_run", True)),
            recent_only=bool(data.get("recent_only", False)),
            recent_window_seconds=window,
        )


@dataclass
class _Recorder:
    """Collects structured action records and forwards them to an optional sink."""

    sink: LogCallback | None = None
    actions: list[dict] = field(default_factory=list)
    counts: dict[str, int] = field(default_factory=dict)

    def emit(self, level: str, action: str, message: str, **extra) -> None:
        record = {"level": level, "action": action, "message": message, **extra}
        self.actions.append(record)
        self.counts[action] = self.counts.get(action, 0) + 1
        if self.sink:
            self.sink(record)


# ── pure string helpers ──────────────────────────────────────────────────────


def clean_name(text: str) -> str:
    text = re.sub(r"\[(.*?)\]|\((.*?)\)", "", text)
    text = text.replace("_", " ").replace("-", " ")
    return " ".join(text.split()).strip()


def clean_album_folder_name(text: str) -> str:
    match = YEAR_ALBUM_FOLDER_RE.match(text)
    if match:
        text = match.group(2)
    return clean_name(text)


def get_year_from_album_folder_name(text: str) -> str | None:
    match = YEAR_ALBUM_FOLDER_RE.match(text)
    return match.group(1) if match else None


def safe_filename(text: str) -> str:
    text = re.sub(r'[<>:"/\\|?*]', "", text)
    text = text.strip().rstrip(".")
    return " ".join(text.split())


def clean_track_number(value) -> str | None:
    if not value:
        return None
    text = str(value[0] if isinstance(value, list) else value).strip()
    text = text.split("/")[0].strip()
    return text.zfill(2) if text.isdigit() else None


def extract_year(value: str | None) -> str | None:
    if not value:
        return None
    match = re.search(r"\b(19\d{2}|20\d{2})\b", str(value))
    return match.group(1) if match else None


# ── metadata reading ─────────────────────────────────────────────────────────


def _load_audio(path: Path):
    try:
        return MutagenFile(path, easy=True)
    except (MutagenError, OSError):
        return None


def _first_tag(audio, names):
    for name in names:
        value = audio.get(name)
        if value:
            return str(value[0]).strip()
    return None


def get_album_metadata_from_files(album_folder: Path):
    for file in album_folder.iterdir():
        if not file.is_file() or file.suffix.lower() not in AUDIO_EXTENSIONS:
            continue
        audio = _load_audio(file)
        if audio is None:
            continue
        artist = _first_tag(audio, ["albumartist", "artist"])
        album = _first_tag(audio, ["album"])
        year = extract_year(_first_tag(audio, ["date", "originaldate", "year"]))
        if artist or album or year:
            return artist, album, year
    return None, None, None


def get_track_metadata(path: Path):
    audio = _load_audio(path)
    if audio is None:
        return None, None, None, None
    artist = _first_tag(audio, ["artist", "albumartist"])
    album = _first_tag(audio, ["album"])
    title = _first_tag(audio, ["title"])
    duration = None
    info = getattr(audio, "info", None)
    if info and getattr(info, "length", None):
        duration = round(info.length)
    return artist, album, title, duration


# ── album inspection ─────────────────────────────────────────────────────────


def _cover_to_keep(album_folder: Path) -> str | None:
    existing = {p.name.lower() for p in album_folder.iterdir() if p.is_file()}
    for name in COVER_NAME_PRIORITY:
        if name in existing:
            return name
    return None


def _has_cover(album_folder: Path) -> bool:
    existing = {p.name.lower() for p in album_folder.iterdir() if p.is_file()}
    return any(name in existing for name in COVER_NAMES)


def _should_keep_file(path: Path, cover_to_keep: str | None) -> bool:
    name = path.name.lower()
    suffix = path.suffix.lower()
    return (
        suffix in AUDIO_EXTENSIONS
        or suffix in LYRICS_SIDECAR_EXTENSIONS
        or name == cover_to_keep
    )


def _has_lyrics(audio_file: Path) -> bool:
    return any(
        audio_file.with_suffix(extension).exists()
        for extension in LYRICS_SIDECAR_EXTENSIONS
    )


def analyze_album(album_folder: Path) -> dict:
    """Read-only summary of an album folder and the pending maintenance work."""
    tag_artist, tag_album, tag_year = get_album_metadata_from_files(album_folder)
    artist = tag_artist or clean_name(album_folder.parent.name)
    album = tag_album or clean_album_folder_name(album_folder.name)
    year = tag_year or get_year_from_album_folder_name(album_folder.name)

    audio_files = [
        f for f in album_folder.iterdir()
        if f.is_file() and f.suffix.lower() in AUDIO_EXTENSIONS
    ]
    cover_to_keep = _cover_to_keep(album_folder)
    extra_files = [
        f.name for f in album_folder.iterdir()
        if f.is_file() and not _should_keep_file(f, cover_to_keep)
    ]

    suggested_folder = None
    if album and year:
        candidate = f"{year} - {safe_filename(album)}"
        if candidate != album_folder.name:
            suggested_folder = candidate

    missing_lyrics = sum(1 for f in audio_files if not _has_lyrics(f))

    return {
        "path": str(album_folder),
        "folder_name": album_folder.name,
        "artist": artist or "",
        "album": album or "",
        "year": year,
        "track_count": len(audio_files),
        "has_cover": _has_cover(album_folder),
        "suggested_folder": suggested_folder,
        "needs_folder_rename": suggested_folder is not None,
        "extra_files": extra_files,
        "extra_file_count": len(extra_files),
        "missing_lyrics_count": missing_lyrics,
    }


def _was_created_recently(folder: Path, window_seconds: int) -> bool:
    """True if the folder was created/modified within the window (recent-only mode)."""
    try:
        stat = folder.stat()
    except OSError:
        return False
    age = time.time() - max(stat.st_ctime, stat.st_mtime)
    return 0 <= age <= window_seconds


def _iter_album_folders(root: Path):
    for first_level in root.iterdir():
        if not first_level.is_dir():
            continue
        try:
            has_audio = any(
                p.is_file() and p.suffix.lower() in AUDIO_EXTENSIONS
                for p in first_level.iterdir()
            )
        except OSError:
            continue
        if has_audio:
            # Top-level folder holding audio directly is not an Artist/Album tree.
            continue
        for album_folder in first_level.iterdir():
            if album_folder.is_dir():
                yield album_folder


def scan_library(root: Path | None = None) -> dict:
    root = root or MUSIC_ROOT
    if not root.exists():
        return {"root": str(root), "exists": False, "albums": []}

    albums = []
    for album_folder in _iter_album_folders(root):
        try:
            albums.append(analyze_album(album_folder))
        except OSError:
            continue
    albums.sort(key=lambda a: (a["artist"].lower(), a["year"] or "", a["album"].lower()))

    return {
        "root": str(root),
        "exists": True,
        "album_count": len(albums),
        "missing_cover_count": sum(1 for a in albums if not a["has_cover"]),
        "needs_rename_count": sum(1 for a in albums if a["needs_folder_rename"]),
        "extra_file_count": sum(a["extra_file_count"] for a in albums),
        "albums": albums,
    }


# ── online lookups (year / cover / lyrics) ───────────────────────────────────


def find_album_year(artist: str, album: str) -> str | None:
    if not _HAS_MUSICBRAINZ:
        return None
    try:
        result = musicbrainzngs.search_releases(artist=artist, release=album, limit=5)
        for release in result.get("release-list", []):
            year = extract_year(release.get("date"))
            if year:
                return year
    except Exception:
        return None
    return None


def find_album_cover(artist: str, album: str) -> bytes | None:
    if not _HAS_MUSICBRAINZ:
        return None
    try:
        result = musicbrainzngs.search_releases(artist=artist, release=album, limit=3)
        releases = result.get("release-list", [])
        if not releases:
            return None
        mbid = releases[0]["id"]
        url = f"https://coverartarchive.org/release/{mbid}/front-500"
        response = requests.get(url, timeout=20, allow_redirects=True)
        if response.status_code == 200 and response.headers.get("content-type", "").startswith("image"):
            return response.content
    except Exception:
        return None
    return None


def find_track_lyrics(artist: str, album: str, title: str, duration: int | None):
    if not duration:
        return None
    try:
        response = requests.get(
            "https://lrclib.net/api/get",
            params={
                "artist_name": artist,
                "track_name": title,
                "album_name": album,
                "duration": duration,
            },
            headers={"User-Agent": "HomelabToolkit/1.0 (local music library tool)"},
            timeout=20,
        )
        if response.status_code != 200:
            return None
        data = response.json()
        if data.get("syncedLyrics"):
            return ".lrc", data["syncedLyrics"].strip() + "\n"
        if data.get("plainLyrics"):
            return ".txt", data["plainLyrics"].strip() + "\n"
    except Exception:
        return None
    return None


# ── mutating operations (respect dry_run) ────────────────────────────────────


def _is_top_level(folder: Path, root: Path) -> bool:
    try:
        return folder.resolve().parent == root.resolve()
    except OSError:
        return folder.parent == root


def _rename_album_folder(album_folder, artist, album, year, root, opts, rec) -> Path:
    if not album or not year or _is_top_level(album_folder, root):
        return album_folder
    new_name = f"{year} - {safe_filename(album)}"
    if album_folder.name == new_name:
        return album_folder
    new_folder = album_folder.with_name(new_name)
    if new_folder.exists():
        rec.emit("skip", "folder", f"Rename target already exists: {new_name}")
        return album_folder
    rec.emit(
        "dry" if opts.dry_run else "ok",
        "folder",
        f"{album_folder.name}  →  {new_name}",
        path=str(album_folder),
    )
    if opts.dry_run:
        return album_folder
    album_folder.rename(new_folder)
    return new_folder


def _rename_track(path: Path, opts, rec) -> Path:
    audio = _load_audio(path)
    if audio is None:
        return path
    artist = _first_tag(audio, ["artist", "albumartist"])
    album = _first_tag(audio, ["album"])
    title = _first_tag(audio, ["title"])
    track = clean_track_number(audio.get("tracknumber"))
    if not (artist and album and title and track):
        return path
    new_name = f"{track} - {safe_filename(title)}{path.suffix.lower()}"
    if path.name == new_name:
        return path
    new_path = path.with_name(new_name)
    if new_path.exists():
        rec.emit("skip", "rename", f"Target exists: {new_name}")
        return path
    rec.emit("dry" if opts.dry_run else "ok", "rename", f"{path.name}  →  {new_name}")
    if opts.dry_run:
        return path
    path.rename(new_path)
    return new_path


def _clean_files(album_folder: Path, opts, rec) -> None:
    cover_to_keep = _cover_to_keep(album_folder)
    for file in album_folder.iterdir():
        if not file.is_file() or _should_keep_file(file, cover_to_keep):
            continue
        rec.emit("dry" if opts.dry_run else "ok", "remove", f"Remove {file.name}", path=str(file))
        if opts.dry_run:
            continue
        try:
            file.unlink()
        except OSError as exc:
            rec.emit("warn", "remove", f"Could not remove {file.name}: {exc}")


def _fetch_cover(album_folder, artist, album, opts, rec) -> None:
    if _has_cover(album_folder):
        return
    if not (artist and album):
        return
    rec.emit("info", "cover", f"Looking up cover: {artist} - {album}")
    image = find_album_cover(artist, album)
    if not image:
        rec.emit("skip", "cover", f"No cover found: {artist} - {album}")
        return
    output = album_folder / "cover.jpg"
    rec.emit("dry" if opts.dry_run else "ok", "cover", f"Save cover.jpg for {album}")
    if not opts.dry_run:
        output.write_bytes(image)
        time.sleep(1)


def _fetch_lyrics(audio_file, album_artist, album_name, opts, rec) -> None:
    if _has_lyrics(audio_file):
        return
    t_artist, t_album, title, duration = get_track_metadata(audio_file)
    artist = t_artist or album_artist
    album = t_album or album_name
    if not (artist and album and title):
        return
    lyrics = find_track_lyrics(artist, album, title, duration)
    if not lyrics:
        rec.emit("skip", "lyrics", f"No lyrics: {artist} - {title}")
        return
    extension, text = lyrics
    output = audio_file.with_suffix(extension)
    rec.emit("dry" if opts.dry_run else "ok", "lyrics", f"Save lyrics for {title}")
    if not opts.dry_run:
        output.write_text(text, encoding="utf-8")
        time.sleep(1)


def _process_album(album_folder: Path, root: Path, opts: ProcessOptions, rec: _Recorder) -> None:
    tag_artist, tag_album, tag_year = get_album_metadata_from_files(album_folder)
    artist = tag_artist or clean_name(album_folder.parent.name)
    album = tag_album or clean_album_folder_name(album_folder.name)
    year = tag_year or get_year_from_album_folder_name(album_folder.name)

    if opts.folder_cleanup and not year and artist and album:
        year = find_album_year(artist, album)
    if opts.folder_cleanup:
        album_folder = _rename_album_folder(album_folder, artist, album, year, root, opts, rec)

    audio_files = [
        f for f in album_folder.iterdir()
        if f.is_file() and f.suffix.lower() in AUDIO_EXTENSIONS
    ]
    if opts.rename:
        audio_files = [_rename_track(f, opts, rec) for f in audio_files]

    if opts.lyrics:
        for file in audio_files:
            _fetch_lyrics(file, artist, album, opts, rec)

    if opts.covers:
        _fetch_cover(album_folder, artist, album, opts, rec)

    if opts.file_cleanup:
        _clean_files(album_folder, opts, rec)


def process_library(
    options: ProcessOptions,
    *,
    root: Path | None = None,
    log: LogCallback | None = None,
    album_paths: list[str] | None = None,
) -> dict:
    """Run the selected maintenance modes. Honours ``options.dry_run``.

    ``album_paths`` optionally limits the run to specific album folders.
    """
    root = root or MUSIC_ROOT
    rec = _Recorder(sink=log)

    if not root.exists():
        rec.emit("warn", "root", f"Music root does not exist: {root}")
        return {"root": str(root), "dry_run": options.dry_run, "actions": rec.actions, "counts": rec.counts}

    if album_paths:
        wanted = {str(Path(p)) for p in album_paths}
        folders = [Path(p) for p in album_paths] if all(Path(p).exists() for p in album_paths) else [
            f for f in _iter_album_folders(root) if str(f) in wanted
        ]
    else:
        folders = list(_iter_album_folders(root))

    if options.recent_only and not album_paths:
        before = len(folders)
        folders = [f for f in folders if _was_created_recently(f, options.recent_window_seconds)]
        rec.emit(
            "info",
            "recent",
            f"Recent-only: {len(folders)} of {before} albums modified in the last "
            f"{options.recent_window_seconds // 3600}h",
        )

    rec.emit(
        "info",
        "start",
        f"{'Dry run' if options.dry_run else 'Applying'} across {len(folders)} album(s)",
    )
    for album_folder in folders:
        try:
            _process_album(album_folder, root, options, rec)
        except OSError as exc:
            rec.emit("warn", "album", f"Error processing {album_folder.name}: {exc}")

    rec.emit("info", "done", "Finished")
    return {
        "root": str(root),
        "dry_run": options.dry_run,
        "actions": rec.actions,
        "counts": rec.counts,
    }


# ── streaming variants (disk-efficient; yield as work happens) ────────────────


def scan_library_stream(root: Path | None = None) -> Iterator[dict]:
    """Yield one album analysis at a time so the UI can render incrementally.

    Memory stays flat (no full list is accumulated) and the slow NAS walk
    streams results to the caller as each album folder is inspected.
    """
    root = root or MUSIC_ROOT
    if not root.exists():
        yield {"type": "error", "message": f"Music root not found: {root}"}
        return

    yield {"type": "start", "root": str(root)}
    album_count = missing_cover = needs_rename = extra_files = 0
    for album_folder in _iter_album_folders(root):
        try:
            album = analyze_album(album_folder)
        except OSError:
            continue
        album_count += 1
        if not album["has_cover"]:
            missing_cover += 1
        if album["needs_folder_rename"]:
            needs_rename += 1
        extra_files += album["extra_file_count"]
        yield {"type": "album", "album": album, "scanned": album_count}

    yield {
        "type": "summary",
        "root": str(root),
        "album_count": album_count,
        "missing_cover_count": missing_cover,
        "needs_rename_count": needs_rename,
        "extra_file_count": extra_files,
    }


def process_library_stream(
    options: ProcessOptions,
    *,
    root: Path | None = None,
    album_paths: list[str] | None = None,
) -> Iterator[dict]:
    """Run maintenance and yield each action the moment it happens.

    ``process_library`` already reports through a ``log`` callback; we bridge that
    to a queue drained by this generator so the HTTP response streams live.
    """
    events: queue.Queue = queue.Queue()
    sentinel = object()

    def worker():
        try:
            process_library(options, root=root, log=events.put, album_paths=album_paths)
        except Exception as exc:  # pragma: no cover - defensive
            events.put({"level": "warn", "action": "error", "message": str(exc)})
        finally:
            events.put(sentinel)

    threading.Thread(target=worker, name="music_process_stream", daemon=True).start()
    while True:
        item = events.get()
        if item is sentinel:
            break
        yield item