Skip to content

Commit

Permalink
kms
Browse files Browse the repository at this point in the history
  • Loading branch information
@azuline
azuline committed Nov 1, 2023
1 parent bc8c0d4 commit af6e8ad
Show file tree
Hide file tree
Showing 4 changed files with 83 additions and 52 deletions.
9 changes: 6 additions & 3 deletions rose/rule_parser.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,8 @@ class InvalidRuleSpecError(RoseError):
"genre",
"label",
"releasetype",
"artist",
"trackartist",
"albumartist",
]

ALL_TAGS: list[Tag] = [
Expand All @@ -41,7 +42,8 @@ class InvalidRuleSpecError(RoseError):
"genre",
"label",
"releasetype",
"artist",
"trackartist",
"albumartist",
]


Expand All @@ -57,7 +59,8 @@ class InvalidRuleSpecError(RoseError):
MULTI_VALUE_TAGS: list[Tag] = [
"genre",
"label",
"artist",
"trackartist",
"albumartist",
]


Expand Down
8 changes: 4 additions & 4 deletions rose/rule_parser_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,15 +14,15 @@
def test_rule_to_str() -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["tracktitle"], pattern="Track"),
action=ReplaceAction(replacement="lalala", tags=["artist", "genre"]),
action=ReplaceAction(replacement="lalala", tags=["albumartist", "genre"]),
)
assert str(rule) == "matcher=tracktitle:Track action=artist,genre:replace:lalala"
assert str(rule) == "matcher=tracktitle:Track action=albumartist,genre:replace:lalala"

rule = MetadataRule(
matcher=MetadataMatcher(tags=["tracktitle", "artist", "genre"], pattern=":"),
matcher=MetadataMatcher(tags=["tracktitle", "albumartist", "genre"], pattern=":"),
action=SedAction(src=re.compile(r":"), dst="; "),
)
assert str(rule) == r'matcher=tracktitle,artist,genre:\: action="sed:\::; "'
assert str(rule) == r'matcher=tracktitle,albumartist,genre:\: action="sed:\::; "'


def test_rule_parser() -> None:
Expand Down
88 changes: 46 additions & 42 deletions rose/rules.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,13 @@
The rules module implements the Rules Engine for updating metadata. The rules engine accepts,
previews, and executes rules.
"""

import contextlib
import copy
import logging
from collections.abc import Callable
from pathlib import Path
from typing import Any

import click

Expand Down Expand Up @@ -96,13 +99,7 @@ def matches_rule(x: str) -> bool:
# that we directly use string interpolation here instead of prepared queries, because we are
# constructing a complex match string and everything is escaped and spaced-out with a random
# paragraph character, so there's no risk of SQL being interpreted.
columns: list[str] = []
for field in rule.matcher.tags:
if field == "artist":
columns.extend(["trackartist", "albumartist"])
else:
columns.append(field)
ftsquery = f"{{{' '.join(columns)}}} : {matchsql}"
ftsquery = f"{{{' '.join(rule.matcher.tags)}}} : {matchsql}"
query = f"""
SELECT DISTINCT t.source_path
FROM rules_engine_fts
Expand All @@ -127,41 +124,9 @@ def matches_rule(x: str) -> bool:
# audiotags into the `audiotag` list. Print planned changes for user confirmation.
# - 2nd pass: Flush the changes.

# Factor out the logic for executing an action on a single-value tag and a multi-value tag.
def execute_single_action(value: str | None) -> str | None:
if not matches_rule(value or ""):
return value
if isinstance(rule.action, ReplaceAction):
return rule.action.replacement
elif isinstance(rule.action, SedAction):
if not value:
return value
return rule.action.src.sub(rule.action.dst, str(value or ""))
elif isinstance(rule.action, DeleteAction):
return None
raise InvalidRuleActionError(f"Invalid action {type(rule.action)} for single-value tag")

def execute_multi_value_action(values: list[str]) -> list[str]:
if isinstance(rule.action, ReplaceAllAction):
return rule.action.replacement

rval: list[str] = []
for v in values:
if not matches_rule(v):
rval.append(v)
continue
with contextlib.suppress(InvalidRuleActionError):
if newv := execute_single_action(v):
rval.append(newv)
continue
if isinstance(rule.action, SplitAction):
for newv in v.split(rule.action.delimiter):
if newv:
rval.append(newv.strip())
continue
raise InvalidRuleActionError(f"Invalid action {type(rule.action)} for multi-value tag")
return rval

# In this first pass, we have two steps:
#
# 1. Make sure that this audio track should indeed
audiotags: list[AudioTags] = []
for tpath in track_paths:
tags = AudioTags.from_file(tpath)
Expand Down Expand Up @@ -323,3 +288,42 @@ def execute_multi_value_action(values: list[str]) -> list[str]:
logger.info(f"Flushing rule-applied tags for {tags.path}.")
tags.flush()
logger.info(f"Successfully flushed all {len(audiotags)} rule-applied tags")


# Factor out the logic for executing an action on a single-value tag and a multi-value tag.
def execute_single_action(rule: MetadataRule, value: str | None) -> str | None:
if isinstance(rule.action, ReplaceAction):
return rule.action.replacement
elif isinstance(rule.action, SedAction):
if not value:
return value
return rule.action.src.sub(rule.action.dst, str(value or ""))
elif isinstance(rule.action, DeleteAction):
return None
raise InvalidRuleActionError(f"Invalid action {type(rule.action)} for single-value tag")


def execute_multi_value_action(
rule: MetadataRule,
values: list[str],
matches_rule: Callable[[Any], bool],
) -> list[str]:
if isinstance(rule.action, ReplaceAllAction):
return rule.action.replacement

rval: list[str] = []
for v in values:
if not matches_rule(v):
rval.append(v)
continue
with contextlib.suppress(InvalidRuleActionError):
if newv := execute_single_action(rule, v):
rval.append(newv)
continue
if isinstance(rule.action, SplitAction):
for newv in v.split(rule.action.delimiter):
if newv:
rval.append(newv.strip())
continue
raise InvalidRuleActionError(f"Invalid action {type(rule.action)} for multi-value tag")
return rval
30 changes: 27 additions & 3 deletions rose/rules_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -100,8 +100,7 @@ def test_rules_execution_match_superstrict(config: Config, source_dir: Path) ->
assert af.title == "lalala"


def test_all_fields_match(config: Config, source_dir: Path) -> None:
"""Test that all fields can match. This checks that we're querying and shit correctly."""
def test_rules_fields_match_tracktitle(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["tracktitle"], pattern="Track"),
action=ReplaceAction(replacement="8"),
Expand All @@ -110,6 +109,8 @@ def test_all_fields_match(config: Config, source_dir: Path) -> None:
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.title == "8"


def test_rules_fields_match_year(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["year"], pattern="1990"),
action=ReplaceAction(replacement="8"),
Expand All @@ -118,6 +119,8 @@ def test_all_fields_match(config: Config, source_dir: Path) -> None:
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.year == 8


def test_rules_fields_match_releasetype(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["releasetype"], pattern="album"),
action=ReplaceAction(replacement="live"),
Expand All @@ -126,6 +129,8 @@ def test_all_fields_match(config: Config, source_dir: Path) -> None:
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.release_type == "live"


def test_rules_fields_match_tracknumber(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["tracknumber"], pattern="1"),
action=ReplaceAction(replacement="8"),
Expand All @@ -134,6 +139,8 @@ def test_all_fields_match(config: Config, source_dir: Path) -> None:
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.track_number == "8"


def test_rules_fields_match_discnumber(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["discnumber"], pattern="1"),
action=ReplaceAction(replacement="8"),
Expand All @@ -142,6 +149,8 @@ def test_all_fields_match(config: Config, source_dir: Path) -> None:
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.disc_number == "8"


def test_rules_fields_match_albumtitle(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["albumtitle"], pattern="Love Blackpink"),
action=ReplaceAction(replacement="8"),
Expand All @@ -150,6 +159,8 @@ def test_all_fields_match(config: Config, source_dir: Path) -> None:
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.album == "8"


def test_rules_fields_match_genre(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["genre"], pattern="K-Pop"),
action=ReplaceAction(replacement="8"),
Expand All @@ -158,6 +169,8 @@ def test_all_fields_match(config: Config, source_dir: Path) -> None:
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.genre == ["8", "Pop"]


def test_rules_fields_match_label(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["label"], pattern="Cool"),
action=ReplaceAction(replacement="8"),
Expand All @@ -166,13 +179,24 @@ def test_all_fields_match(config: Config, source_dir: Path) -> None:
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.label == ["8"]


def test_rules_fields_match_albumartist(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["artist"], pattern="BLACKPINK"),
matcher=MetadataMatcher(tags=["albumartist"], pattern="BLACKPINK"),
action=ReplaceAction(replacement="8"),
)
execute_metadata_rule(config, rule, False)
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.album_artists.main == ["8"]


def test_rules_fields_match_trackartist(config: Config, source_dir: Path) -> None:
rule = MetadataRule(
matcher=MetadataMatcher(tags=["trackartist"], pattern="BLACKPINK"),
action=ReplaceAction(replacement="8"),
)
execute_metadata_rule(config, rule, False)
af = AudioTags.from_file(source_dir / "Test Release 1" / "01.m4a")
assert af.artists.main == ["8"]


Expand Down

0 comments on commit af6e8ad

Please sign in to comment.