templating

rose/releases.py

@@ -35,7 +35,7 @@
 )
 from rose.common import Artist, ArtistMapping, RoseError, RoseExpectedError
 from rose.config import Config
-from rose.rule_parser import MetadataAction, MetadataMatcher
+from rose.rule_parser import ALL_TAGS, MetadataAction, MetadataMatcher
 from rose.rules import (
     execute_metadata_actions,
     fast_search_for_matching_releases,
@@ -450,6 +450,18 @@ def edit_release(
 
 
 def find_releases_matching_rule(c: Config, matcher: MetadataMatcher) -> list[Release]:
+    # Implement optimizations for common lookups. Only applies to strict lookups.
+    # TODO: Morning
+    if matcher.pattern.pattern.startswith("^") and matcher.pattern.pattern.endswith("$"):
+        if matcher.tags == ALL_TAGS["artist"]:
+            pass
+        if matcher.tags == ["genre"]:
+            pass
+        if matcher.tags == ["label"]:
+            pass
+        if matcher.tags == ["descriptor"]:
+            pass
+
     release_ids = [x.id for x in fast_search_for_matching_releases(c, matcher)]
     releases = list_releases(c, release_ids)
     return filter_release_false_positives_using_read_cache(matcher, releases)

rose/rule_parser.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 import io
 import logging
 import re
@@ -76,9 +77,11 @@ def __str__(self) -> str:
     "label",
 ]
 
+ExpandableTag = Tag | Literal["artist", "trackartist", "releaseartist"]
+
 # Map of a tag to its "resolved" tags. Most tags simply resolve to themselves; however, we let
 # certain tags be aliases for multiple other tags, purely for convenience.
-ALL_TAGS: dict[str, list[Tag]] = {
+ALL_TAGS: dict[ExpandableTag, list[Tag]] = {
     "tracktitle": ["tracktitle"],
     "trackartist": [
         "trackartist[main]",
@@ -275,7 +278,7 @@ def __str__(self) -> str:
         return r
 
 
-@dataclass
+@dataclass()
 class MetadataMatcher:
     # Tags to test against the pattern. If any tags match the pattern, the action will be ran
     # against the track.
@@ -289,6 +292,13 @@ def __str__(self) -> str:
         r += str(self.pattern)
         return r
 
+    def __init__(self, tags: Sequence[ExpandableTag], pattern: MatcherPattern) -> None:
+        _tags: set[Tag] = set()
+        for t in tags:
+            _tags.update(ALL_TAGS[t])
+        self.tags = list(_tags)
+        self.pattern = pattern
+
     @classmethod
     def parse(cls, raw: str, *, rule_name: str = "matcher") -> MetadataMatcher:
         idx = 0
@@ -375,6 +385,19 @@ class MetadataAction:
     # upon.
     pattern: MatcherPattern | None = None
 
+    def __init__(
+        self,
+        behavior: ReplaceAction | SedAction | SplitAction | AddAction | DeleteAction,
+        tags: Sequence[ExpandableTag],
+        pattern: MatcherPattern | None = None,
+    ) -> None:
+        self.behavior = behavior
+        _tags: set[Tag] = set()
+        for t in tags:
+            _tags.update(ALL_TAGS[t])
+        self.tags = list(_tags)
+        self.pattern = pattern
+
     def __str__(self) -> str:
         r = ""
         r += stringify_tags(self.tags)

rose/tracks.py

@@ -14,7 +14,7 @@
 )
 from rose.common import RoseExpectedError
 from rose.config import Config
-from rose.rule_parser import MetadataAction, MetadataMatcher
+from rose.rule_parser import ALL_TAGS, MetadataAction, MetadataMatcher
 from rose.rules import (
     execute_metadata_actions,
     fast_search_for_matching_tracks,
@@ -29,6 +29,18 @@ class TrackDoesNotExistError(RoseExpectedError):
 
 
 def find_tracks_matching_rule(c: Config, matcher: MetadataMatcher) -> list[Track]:
+    # Implement optimizations for common lookups. Only applies to strict lookups.
+    # TODO: Morning
+    if matcher.pattern.pattern.startswith("^") and matcher.pattern.pattern.endswith("$"):
+        if matcher.tags == ALL_TAGS["artist"]:
+            pass
+        if matcher.tags == ["genre"]:
+            pass
+        if matcher.tags == ["label"]:
+            pass
+        if matcher.tags == ["descriptor"]:
+            pass
+
     track_ids = [t.id for t in fast_search_for_matching_tracks(c, matcher)]
     tracks = list_tracks(c, track_ids)
     return filter_track_false_positives_using_read_cache(matcher, tracks)

rose_vfs/virtualfs.py

@@ -107,7 +107,10 @@
 )
 from rose.cache import (
     list_releases_delete_this,
+    release_within_collage,
 )
+from rose.rule_parser import MatcherPattern, MetadataMatcher
+from rose.tracks import find_tracks_matching_rule
 
 logger = logging.getLogger(__name__)
 
@@ -160,6 +163,9 @@ def get(self, key: K, default: T) -> V | T:
             return default
 
 
+ALL_TRACKS = "!All Tracks"
+
+
 @dataclass(frozen=True, slots=True)
 class VirtualPath:
     view: (
@@ -184,6 +190,9 @@ class VirtualPath:
     label: str | None = None
     collage: str | None = None
     playlist: str | None = None
+    # Release may be set to `ALL_TRACKS`, in which case it is never attempted to be resolved to a
+    # release. Instead, it is treated as a special directory. There may be name conflicts; I don't
+    # care.
     release: str | None = None
     file: str | None = None
 
@@ -421,7 +430,7 @@ def list_release_paths(
         directory names.
         """
         # For collision number generation.
-        seen: set[str] = set()
+        seen: set[str] = {ALL_TRACKS}
         prefix_pad_size = len(str(len(releases)))
         for idx, release in enumerate(releases):
             # Determine the proper template.
@@ -954,6 +963,14 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
         if p.collage:
             if not get_collage(self.config, p.collage):
                 raise llfuse.FUSEError(errno.ENOENT)
+            if p.release == ALL_TRACKS:
+                if not p.file:
+                    return self.stat("dir")
+                if (track := get_track(self.config, self._get_track_id(p))) and (
+                    release_within_collage(self.config, track.release.id, p.collage)
+                ):
+                    return self.stat("file", track.source_path)
+                raise llfuse.FUSEError(errno.ENOENT)
             if p.release:
                 return self._getattr_release(p)
             return self.stat("dir")
@@ -963,6 +980,14 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
             la = self.sanitizer.unsanitize(p.label, p.label_parent)
             if not label_exists(self.config, la) or not self.can_show.label(la):
                 raise llfuse.FUSEError(errno.ENOENT)
+            if p.release == ALL_TRACKS:
+                if not p.file:
+                    return self.stat("dir")
+                if (track := get_track(self.config, self._get_track_id(p))) and (
+                    p.label in track.release.labels
+                ):
+                    return self.stat("file", track.source_path)
+                raise llfuse.FUSEError(errno.ENOENT)
             if p.release:
                 return self._getattr_release(p)
             return self.stat("dir")
@@ -972,6 +997,14 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
             d = self.sanitizer.unsanitize(p.descriptor, p.descriptor_parent)
             if not descriptor_exists(self.config, d) or not self.can_show.descriptor(d):
                 raise llfuse.FUSEError(errno.ENOENT)
+            if p.release == ALL_TRACKS:
+                if not p.file:
+                    return self.stat("dir")
+                if (track := get_track(self.config, self._get_track_id(p))) and (
+                    p.descriptor in track.release.descriptors
+                ):
+                    return self.stat("file", track.source_path)
+                raise llfuse.FUSEError(errno.ENOENT)
             if p.release:
                 return self._getattr_release(p)
             return self.stat("dir")
@@ -981,6 +1014,17 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
             g = self.sanitizer.unsanitize(p.genre, p.genre_parent)
             if not genre_exists(self.config, g) or not self.can_show.genre(g):
                 raise llfuse.FUSEError(errno.ENOENT)
+            if p.release == ALL_TRACKS:
+                if not p.file:
+                    return self.stat("dir")
+                if (track := get_track(self.config, self._get_track_id(p))) and (
+                    p.genre in track.release.genres
+                    or p.genre in track.release.parent_genres
+                    or p.genre in track.release.secondary_genres
+                    or p.genre in track.release.parent_secondary_genres
+                ):
+                    return self.stat("file", track.source_path)
+                raise llfuse.FUSEError(errno.ENOENT)
             if p.release:
                 return self._getattr_release(p)
             return self.stat("dir")
@@ -990,6 +1034,16 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
             a = self.sanitizer.unsanitize(p.artist, p.artist_parent)
             if not artist_exists(self.config, a) or not self.can_show.artist(a):
                 raise llfuse.FUSEError(errno.ENOENT)
+            if p.release == ALL_TRACKS:
+                if not p.file:
+                    return self.stat("dir")
+                if (track := get_track(self.config, self._get_track_id(p))) and any(
+                    p.artist == a.name
+                    for _, artists in track.release.releaseartists.items()
+                    for a in artists
+                ):
+                    return self.stat("file", track.source_path)
+                raise llfuse.FUSEError(errno.ENOENT)
             if p.release:
                 return self._getattr_release(p)
             return self.stat("dir")
@@ -1034,6 +1088,22 @@ def readdir(self, p: VirtualPath) -> Iterator[tuple[str, dict[str, Any]]]:
             ]
             return
 
+        if p.release == ALL_TRACKS:
+            if p.artist:
+                matcher = MetadataMatcher(tags=["artist"], pattern=MatcherPattern(f"^{p.artist}$"))
+            if p.genre:
+                matcher = MetadataMatcher(tags=["genre"], pattern=MatcherPattern(f"^{p.genre}$"))
+            if p.descriptor:
+                matcher = MetadataMatcher(
+                    tags=["descriptor"], pattern=MatcherPattern(f"^{p.descriptor}$")
+                )
+            if p.label:
+                matcher = MetadataMatcher(tags=["label"], pattern=MatcherPattern(f"^{p.label}$"))
+            tracks = find_tracks_matching_rule(self.config, matcher)
+            for trk, vname in self.vnames.list_track_paths(p, tracks):
+                yield vname, self.stat("file", trk.source_path)
+            return
+
         if p.release:
             if (release_id := self.vnames.lookup_release(p)) and (
                 release := get_release(self.config, release_id)
@@ -1066,6 +1136,7 @@ def readdir(self, p: VirtualPath) -> Iterator[tuple[str, dict[str, Any]]]:
             # fmt: on
             for rls, vname in self.vnames.list_release_paths(p, releases):
                 yield vname, self.stat("dir", rls.source_path)
+            yield ALL_TRACKS, self.stat("dir")
             return
 
         if p.view == "Artists":
@@ -1106,6 +1177,7 @@ def readdir(self, p: VirtualPath) -> Iterator[tuple[str, dict[str, Any]]]:
             releases = get_collage_releases(self.config, p.collage)
             for rls, vname in self.vnames.list_release_paths(p, releases):
                 yield vname, self.stat("dir", rls.source_path)
+            yield ALL_TRACKS, self.stat("dir")
             return
 
         if p.view == "Collages":