From a58fe4aef965fe9f5adcff8930e4cc097e7b0bd0 Mon Sep 17 00:00:00 2001 From: blissful Date: Mon, 6 May 2024 00:50:12 -0400 Subject: [PATCH] templating --- rose/releases.py | 14 +++++++- rose/rule_parser.py | 27 ++++++++++++++-- rose/tracks.py | 14 +++++++- rose_vfs/virtualfs.py | 74 ++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 124 insertions(+), 5 deletions(-) diff --git a/rose/releases.py b/rose/releases.py index 4630b97..6f1275f 100644 --- a/rose/releases.py +++ b/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) diff --git a/rose/rule_parser.py b/rose/rule_parser.py index 88ccb89..4345855 100644 --- a/rose/rule_parser.py +++ b/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) diff --git a/rose/tracks.py b/rose/tracks.py index d7ab047..c2de4ab 100644 --- a/rose/tracks.py +++ b/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) diff --git a/rose_vfs/virtualfs.py b/rose_vfs/virtualfs.py index 60f12c4..9d40094 100644 --- a/rose_vfs/virtualfs.py +++ b/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":