SaraTune/Library.py

"""Music mode's library: every audio file below the configured folders, with
their tags. The tags are cached in ~/.cache/SaraSuite/SaraTune/Library.toml
so a start only reads files that are new or changed since the last scan."""

import json
import os
import tomllib
from collections import defaultdict
from pathlib import Path
from typing import Callable, Optional

from Core import IsAudioFile, LibraryCachePath
from Metadata import Track, ReadTrack

CacheVersion = 1


def _TomlValue(Value) -> str:
    # JSON's string escapes are valid TOML basic-string escapes.
    if isinstance(Value, str):
        return json.dumps(Value, ensure_ascii=False)
    return repr(Value)


class Library:
    def __init__(self):
        self.Tracks: dict[str, Track] = {}

    # --- Cache --------------------------------------------------------------

    def LoadCache(self) -> None:
        try:
            Data = tomllib.loads(LibraryCachePath.read_text(encoding="utf-8"))
        except (OSError, tomllib.TOMLDecodeError):
            return
        if Data.get("Version") != CacheVersion:
            return
        self.Tracks = {}
        for Entry in Data.get("Tracks", []):
            if "Path" in Entry:
                self.Tracks[Entry["Path"]] = Track.FromDict(Entry)

    def SaveCache(self) -> None:
        # Written by hand: tomlkit takes seconds for a few thousand tables.
        Lines = [f"Version = {CacheVersion}", ""]
        for Item in sorted(self.Tracks.values(), key=lambda T: T.Path):
            Lines.append("[[Tracks]]")
            Lines += [f"{Key} = {_TomlValue(Value)}" for Key, Value in Item.ToDict().items()]
            Lines.append("")
        LibraryCachePath.parent.mkdir(parents=True, exist_ok=True)
        Temporary = LibraryCachePath.with_suffix(".new")
        Temporary.write_text("\n".join(Lines), encoding="utf-8")
        Temporary.replace(LibraryCachePath)

    def ClearCache(self) -> None:
        self.Tracks = {}
        LibraryCachePath.unlink(missing_ok=True)

    # --- Scanning -----------------------------------------------------------

    def Scan(self, Folders: list[str], OnProgress: Optional[Callable[[int], None]] = None) -> tuple[int, int]:
        """Reads new and changed files, forgets ones that are gone. Blocking
        (run it in a thread). Returns (added or changed, removed)."""
        Found: dict[str, float] = {}
        for Folder in Folders:
            Root = Path(Folder).expanduser()
            if not Root.is_dir():
                continue
            for DirPath, DirNames, FileNames in os.walk(Root, followlinks=True):
                DirNames[:] = [Name for Name in DirNames if not Name.startswith(".")]
                for Name in FileNames:
                    FilePath = Path(DirPath) / Name
                    if IsAudioFile(FilePath):
                        try:
                            Found[str(FilePath)] = FilePath.stat().st_mtime
                        except OSError:
                            pass
        Changed = 0
        for Index, (FilePath, Modified) in enumerate(Found.items()):
            Known = self.Tracks.get(FilePath)
            if Known is None or Known.Modified != Modified:
                self.Tracks[FilePath] = ReadTrack(FilePath)
                Changed += 1
                if OnProgress and Changed % 25 == 0:
                    OnProgress(Index + 1)
        Gone = [FilePath for FilePath in self.Tracks if FilePath not in Found]
        for FilePath in Gone:
            del self.Tracks[FilePath]
        if Changed or Gone:
            self.SaveCache()
        return Changed, len(Gone)

    # --- Views --------------------------------------------------------------

    def Get(self, FilePath: str) -> Track:
        """The library's Track, or one read on the spot for files outside it
        (playlists may point anywhere)."""
        return self.Tracks.get(FilePath) or ReadTrack(FilePath)

    def Songs(self) -> list[Track]:
        return sorted(self.Tracks.values(), key=SongKey)

    def Albums(self) -> list[list[Track]]:
        """Each album's tracks in disc/track order; albums by artist, then name."""
        Groups: dict[tuple[str, str], list[Track]] = defaultdict(list)
        for Item in self.Tracks.values():
            if Item.Album:
                Groups[Item.AlbumKey].append(Item)
        return [sorted(Tracks, key=AlbumOrderKey) for _, Tracks in sorted(Groups.items())]

    def Artists(self) -> list[tuple[str, list[Track]]]:
        Groups: dict[str, list[Track]] = defaultdict(list)
        Names: dict[str, str] = {}
        for Item in self.Tracks.values():
            Name = Item.AlbumArtist or Item.Artist or "Unknown artist"
            Groups[Name.lower()].append(Item)
            Names.setdefault(Name.lower(), Name)
        return [(Names[Key], sorted(Groups[Key], key=ArtistOrderKey)) for Key in sorted(Groups)]


def SongKey(Item: Track) -> tuple:
    return Item.DisplayArtist.lower(), Item.DisplayAlbum.lower(), Item.DiscNumber, Item.TrackNumber, Item.DisplayTitle.lower()


def AlbumOrderKey(Item: Track) -> tuple:
    return Item.DiscNumber, Item.TrackNumber, Item.DisplayTitle.lower()


def ArtistOrderKey(Item: Track) -> tuple:
    return Item.Year, Item.DisplayAlbum.lower(), Item.DiscNumber, Item.TrackNumber, Item.DisplayTitle.lower()


def Matches(Item: Track, Query: str) -> bool:
    """Every word of the search has to appear in the title, artist, album or genre."""
    Haystack = " ".join((Item.DisplayTitle, Item.Artist, Item.AlbumArtist, Item.Album, Item.Genre)).lower()
    return all(Word in Haystack for Word in Query.lower().split())