musicmouse/python-backend/musicmouse/library/scanner.py

"""Turning folders on disk into :class:`~musicmouse.library.models.Album` objects.

One level under each section folder, every directory holding audio files is one album.
Tags come from mutagen; where a section's tags are known to be useless the folder name
wins instead (see :mod:`musicmouse.library.sections`). Nothing here raises on bad
input - an unreadable file loses its metadata, not the boot.
"""

from __future__ import annotations

import logging
import time
from collections import Counter
from collections.abc import Iterator, Mapping
from pathlib import Path

from musicmouse.library.cache import Fingerprint, LibraryCache
from musicmouse.library.colors import colors_from_cover, colors_from_id
from musicmouse.library.models import Album, LibraryTrack, album_id
from musicmouse.library.sections import SECTIONS, AlbumKind, Section

_log = logging.getLogger(__name__)

__all__ = ["scan_library"]

#: Checked in order for a cover sitting next to the audio.
_COVER_NAMES = ("cover.jpg", "cover.jpeg", "cover.png", "folder.jpg")

#: Extensions tried for an episode's own same-stem sidecar cover image (see
#: :func:`_cover_for_episode`), matching what `podcast_feeds.py`'s
#: ``episode_cover_filename`` can produce.
_SIDECAR_COVER_EXTENSIONS = (".jpg", ".jpeg", ".png", ".webp")

#: How often a long scan reports where it is, so a library of thousands of albums
#: doesn't sit silent for minutes with nothing on the console to say it is still going.
_PROGRESS_INTERVAL_SECONDS = 5.0


def _album_folders(folder: Path, extensions: frozenset[str]) -> Iterator[Path]:
    """Every leaf album folder under ``folder``, however deep it is nested.

    A folder that holds audio files directly *is* an album - an artist who groups
    their books under an extra "ab 3" / "ab 5" age-range folder, or a series folder
    that groups its episodes one directory further down than usual, still bottoms out
    here without needing to be special-cased. Only descended into when a folder holds
    no audio of its own, so an ordinary album folder is never mistaken for a grouping
    folder just because it also happens to contain subfolders.
    """
    if _audio_files(folder, extensions):
        yield folder
        return
    for sub in sorted(folder.iterdir(), key=lambda path: path.name):
        if sub.is_dir() and not sub.name.startswith("."):
            yield from _album_folders(sub, extensions)


def _audio_files(folder: Path, extensions: frozenset[str]) -> list[Path]:
    """The playable files in ``folder``, alphabetically.

    Dotfiles are skipped outright and everything else must match ``audio_extensions``,
    which is what keeps a podcast folder's ``archive.json`` and its half-finished
    ``.podcast-dl-*.download.tmp`` out of the playlist.
    """
    return sorted(
        (
            path
            for path in folder.iterdir()
            if path.is_file()
            and not path.name.startswith(".")
            and path.suffix.lower() in extensions
        ),
        key=lambda path: path.name,
    )


def _tags(path: Path) -> tuple[dict[str, str], float]:
    """``(tags, duration)`` for one file. Empty and ``0.0`` when it cannot be read."""
    try:
        import mutagen

        audio = mutagen.File(path, easy=True)
        if audio is None:
            return {}, 0.0
        tags = {key: values[0] for key, values in dict(audio).items() if values}
        return tags, float(getattr(audio.info, "length", 0.0) or 0.0)
    except Exception:  # pragma: no cover - mutagen raises freely on damaged files
        _log.debug("No readable tags in %s", path)
        return {}, 0.0


def _embedded_art(path: Path) -> bytes | None:
    try:
        import mutagen

        raw = mutagen.File(path)
        if raw is None or raw.tags is None:
            return None
        for key in raw.tags:
            if key.startswith("APIC"):
                data: bytes = raw.tags[key].data
                return data
    except Exception:  # pragma: no cover - defensive
        _log.debug("No readable embedded art in %s", path)
    return None


def _split_folder_name(name: str) -> tuple[str, str]:
    """``"Conni - Conni in den Bergen"`` -> ``("Conni", "Conni in den Bergen")``."""
    artist, separator, title = name.partition(" - ")
    return (artist, title) if separator else ("", name)


def _most_common(values: list[str]) -> str:
    """The tag value most files in an album agree on, ignoring blanks."""
    counted = Counter(value for value in values if value)
    return counted.most_common(1)[0][0] if counted else ""


def _cover_for(
    folder: Path, paths: list[Path], identifier: str, cache: LibraryCache
) -> tuple[Path | None, bytes | None]:
    for name in _COVER_NAMES:
        candidate = folder / name
        if candidate.is_file():
            return candidate, candidate.read_bytes()
    for path in paths[:3]:
        # Podcast feeds sometimes art only some episodes; a couple of tries is enough.
        art = _embedded_art(path)
        if art is not None:
            return cache.store_cover(identifier, art), art
    return None, None


def _cover_for_episode(
    folder: Path, path: Path, identifier: str, cache: LibraryCache
) -> tuple[Path | None, bytes | None]:
    """The reverse priority from :func:`_cover_for`: with one album per episode, art
    specific to this one episode wins over the folder's shared cover. A same-stem
    sidecar image (what ``podcast_feeds.py`` saves when a feed has real per-episode
    art) is tried first - it's a plain file stat, cheaper than reading tags - then the
    episode's own embedded art, then the folder's shared cover as the last resort for
    an episode with neither."""
    for extension in _SIDECAR_COVER_EXTENSIONS:
        candidate = path.with_suffix(extension)
        if candidate.is_file():
            return candidate, candidate.read_bytes()
    art = _embedded_art(path)
    if art is not None:
        return cache.store_cover(identifier, art), art
    for name in _COVER_NAMES:
        candidate = folder / name
        if candidate.is_file():
            return candidate, candidate.read_bytes()
    return None, None


def scan_album(
    folder: Path,
    *,
    root: Path,
    section_name: str,
    section: Section,
    extensions: frozenset[str],
    cache: LibraryCache,
    figure_kinds: Mapping[str, AlbumKind] | None = None,
) -> tuple[Album, Fingerprint] | None:
    paths = _audio_files(folder, extensions)
    if not paths:
        _log.debug("No audio files in %s", folder)
        return None
    if section.order == "newest_first":
        paths.reverse()

    fingerprint = Fingerprint.of(paths)
    identifier = album_id(root, folder)

    tracks: list[LibraryTrack] = []
    albums: list[str] = []
    artists: list[str] = []
    for path in paths:
        tags, duration = _tags(path)
        tracks.append(
            LibraryTrack(path=path, title=tags.get("title") or path.stem, duration=duration)
        )
        albums.append(tags.get("album", ""))
        artists.append(tags.get("albumartist") or tags.get("artist", ""))

    figure = folder.name if section.figures else None
    # Every other shelf is named after what is on it. A figure folder is named after the
    # figurine, so its media type has to be declared in the config.
    kind = (figure_kinds or {}).get(figure, "music") if figure else section.kind

    folder_artist, folder_title = _split_folder_name(folder.name)

    if section.title_from == "folder":
        # A figure folder is named in lowercase ("fuchs"); it sits next to real album
        # titles in the browse grid, so give it a capital.
        title = folder.name[:1].upper() + folder.name[1:] if section.figures else folder.name
    else:
        title = _most_common(albums) or folder_title or folder.name

    if section.artist_from == "folder":
        artist = folder.name
    else:
        artist = _most_common(artists) or folder_artist

    cover, art = _cover_for(folder, paths, identifier, cache)
    colors = colors_from_cover(art, identifier) if art else colors_from_id(identifier)

    album = Album(
        id=identifier,
        section=section_name,
        kind=kind,
        title=title,
        artist=artist,
        # Books group by who or what they are about; music groups by artist. The browse
        # view's category row is built straight off this, so it takes only the part
        # before the first comma - an ``album_artist`` of "Bobo Siebenschlaefer, Markus
        # Osterwalder, ..." is a credit list whose first name is the character.
        series=artist.split(",")[0].strip() if kind == "book" else None,
        figure=figure,
        colors=colors,
        folder=folder,
        cover=cover,
        tracks=tuple(tracks),
    )
    return album, fingerprint


def scan_episodes(
    folder: Path,
    *,
    root: Path,
    section_name: str,
    section: Section,
    extensions: frozenset[str],
    cache: LibraryCache,
    known: dict[str, tuple[Album, Fingerprint]],
) -> dict[str, tuple[Album, Fingerprint]]:
    """One album per audio file, for a section whose folder is a show rather than a
    single release - a podcast feed's hundreds of episodes, most obviously.

    Fingerprinted per file rather than per folder, so a new episode landing in an
    already-scanned show only costs scanning that one file, not the whole show.
    """
    paths = _audio_files(folder, extensions)
    if section.order == "newest_first":
        paths.reverse()

    out: dict[str, tuple[Album, Fingerprint]] = {}
    for path in paths:
        identifier = album_id(root, path)
        fingerprint = Fingerprint.of([path])
        cached = known.get(identifier)
        if cached is not None and cached[1] == fingerprint:
            out[identifier] = cached
            continue

        tags, duration = _tags(path)
        title = tags.get("title") or path.stem
        cover, art = _cover_for_episode(folder, path, identifier, cache)
        colors = colors_from_cover(art, identifier) if art else colors_from_id(identifier)

        album = Album(
            id=identifier,
            section=section_name,
            kind=section.kind,
            title=title,
            artist=folder.name,
            # The folder is the show; every episode in it groups under the same series,
            # exactly like an audiobook's chapters group under its book.
            series=folder.name,
            figure=None,
            colors=colors,
            folder=folder,
            cover=cover,
            tracks=(LibraryTrack(path=path, title=title, duration=duration),),
        )
        out[identifier] = (album, fingerprint)
    return out


def scan_library(
    root: Path,
    extensions: frozenset[str],
    cache: LibraryCache,
    *,
    known: dict[str, tuple[Album, Fingerprint]] | None = None,
    figure_kinds: Mapping[str, AlbumKind] | None = None,
) -> dict[str, tuple[Album, Fingerprint]]:
    """Scan every section under ``root``.

    ``known`` is the previously cached index: a folder whose files, sizes and mtimes are
    unchanged is taken from it without a single tag being read. ``figure_kinds`` maps a
    figure name to what it holds, which is the one thing the folders cannot say.
    """
    cache.prepare()
    known = known or {}
    out: dict[str, tuple[Album, Fingerprint]] = {}
    last_report = time.monotonic()

    for section_name, section in SECTIONS.items():
        section_root = root / section_name
        if not section_root.is_dir():
            _log.warning("Library section %r has no folder at %s", section_name, section_root)
            continue

        for top in sorted(section_root.iterdir(), key=lambda path: path.name):
            if not top.is_dir() or top.name.startswith("."):
                continue

            if section.album_unit == "episode":
                # Fingerprinted per episode inside scan_episodes; the whole-folder
                # shortcut below does not apply since one folder yields many albums,
                # and an episode show is never nested deeper than this either.
                out.update(
                    scan_episodes(
                        top,
                        root=root,
                        section_name=section_name,
                        section=section,
                        extensions=extensions,
                        cache=cache,
                        known=known,
                    )
                )
                continue

            for folder in _album_folders(top, extensions):
                now = time.monotonic()
                if now - last_report >= _PROGRESS_INTERVAL_SECONDS:
                    _log.info(
                        "Scanning library: %d albums found so far, now in %s",
                        len(out),
                        folder.relative_to(root),
                    )
                    last_report = now

                identifier = album_id(root, folder)
                cached = known.get(identifier)
                if cached is not None:
                    paths = _audio_files(folder, extensions)
                    if paths and Fingerprint.of(paths) == cached[1]:
                        out[identifier] = cached
                        continue
                scanned = scan_album(
                    folder,
                    root=root,
                    section_name=section_name,
                    section=section,
                    extensions=extensions,
                    cache=cache,
                    figure_kinds=figure_kinds,
                )
                if scanned is not None:
                    out[identifier] = scanned

    _log.info("Library: %d albums under %s", len(out), root)
    return out