Initial commit

Library.py

@@ -0,0 +1,136 @@
+"""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())