merge dev (tag_rules) into split-config-docs

# Conflicts:
#	docs/guide/output-and-naming.md
#	docs/reference/configuration.md

tests/core/test_tag_rules.py

@@ -0,0 +1,223 @@
+"""Tests for the tag_rules engine behind conditional release-group tags."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from unshackle.core.utils.tag_rules import evaluate_tag_rules
+
+CONTEXT: dict[str, Any] = {
+    "tag": "NOGRP",
+    "title_type": "movie",
+    "quality": "2160p",
+    "resolution": "2160",
+    "hdr": "DV",
+    "lang_tag": "SUBBED",
+    "edition": "",
+}
+
+
+def test_first_match_wins() -> None:
+    rules = [
+        {"when": {"quality": "2160p"}, "tag": "FIRST"},
+        {"when": {"title_type": "movie"}, "tag": "SECOND"},
+    ]
+    assert evaluate_tag_rules(rules, CONTEXT) == "FIRST"
+
+
+def test_list_matches_any_entry() -> None:
+    rules = [{"when": {"hdr": ["HDR10P", "DV"]}, "tag": "UHDGROUP"}]
+    assert evaluate_tag_rules(rules, CONTEXT) == "UHDGROUP"
+
+
+def test_list_with_no_matching_entry() -> None:
+    rules = [{"when": {"hdr": ["HDR10P", "HLG"]}, "tag": "UHDGROUP"}]
+    assert evaluate_tag_rules(rules, CONTEXT) is None
+
+
+def test_case_insensitive() -> None:
+    rules = [{"when": {"lang_tag": "subbed", "quality": "2160P"}, "tag": "SUBMOVIES"}]
+    assert evaluate_tag_rules(rules, CONTEXT) == "SUBMOVIES"
+
+
+def test_conditions_are_anded() -> None:
+    rules = [{"when": {"title_type": "movie", "quality": "1080p"}, "tag": "NOPE"}]
+    assert evaluate_tag_rules(rules, CONTEXT) is None
+
+
+def test_unknown_key_skips_rule_but_later_rules_match(caplog: pytest.LogCaptureFixture) -> None:
+    rules = [
+        {"when": {"bogus": "value"}, "tag": "SKIPPED"},
+        {"when": {"title_type": "movie"}, "tag": "MATCHED"},
+    ]
+    assert evaluate_tag_rules(rules, CONTEXT) == "MATCHED"
+    assert "bogus" in caplog.text
+
+
+@pytest.mark.parametrize(
+    "rule",
+    [
+        {"when": {"title_type": "movie"}},
+        {"tag": "NOCONDITIONS"},
+        {"when": {}, "tag": "EMPTY"},
+    ],
+)
+def test_malformed_rule_skipped(rule: dict[str, Any]) -> None:
+    assert evaluate_tag_rules([rule], CONTEXT) is None
+
+
+def test_no_rules_and_no_match() -> None:
+    assert evaluate_tag_rules([], CONTEXT) is None
+    assert evaluate_tag_rules([{"when": {"quality": "720p"}, "tag": "SD"}], CONTEXT) is None
+
+
+def test_empty_context_value_matches_empty_condition() -> None:
+    rules = [{"when": {"edition": ""}, "tag": "PLAIN"}]
+    assert evaluate_tag_rules(rules, CONTEXT) == "PLAIN"
+
+
+@pytest.mark.parametrize(
+    ("condition", "expected"),
+    [
+        (">=2160", True),
+        (">=4320", False),
+        ("<=2160", True),
+        ("<1080", False),
+        (">1080", True),
+        ("==2160", True),
+        ("=2160", True),
+        ("!=1080", True),
+        ("!=2160", False),
+        (">= 2160", True),
+    ],
+)
+def test_numeric_operators_extract_first_number(condition: str, expected: bool) -> None:
+    """The actual value is "2160p", so the suffix must not defeat the comparison."""
+    rules = [{"when": {"quality": condition}, "tag": "MATCHED"}]
+    assert (evaluate_tag_rules(rules, CONTEXT) == "MATCHED") is expected
+
+
+def test_string_inequality() -> None:
+    assert evaluate_tag_rules([{"when": {"hdr": "!=DV"}, "tag": "NODV"}], CONTEXT) is None
+    assert evaluate_tag_rules([{"when": {"hdr": "!=HDR10P"}, "tag": "NOHDR10P"}], CONTEXT) == "NOHDR10P"
+
+
+def test_string_equality_operator_is_case_insensitive() -> None:
+    assert evaluate_tag_rules([{"when": {"lang_tag": "==subbed"}, "tag": "SUBGRP"}], CONTEXT) == "SUBGRP"
+
+
+def test_plain_equality_unchanged() -> None:
+    assert evaluate_tag_rules([{"when": {"quality": "2160p"}, "tag": "UHD"}], CONTEXT) == "UHD"
+    assert evaluate_tag_rules([{"when": {"quality": "2160"}, "tag": "UHD"}], CONTEXT) is None
+
+
+def test_numeric_operator_against_value_without_a_number() -> None:
+    rules = [{"when": {"hdr": ">=1080"}, "tag": "NOPE"}]
+    assert evaluate_tag_rules(rules, CONTEXT) is None
+
+
+def test_ordering_operator_with_non_numeric_operand(caplog: pytest.LogCaptureFixture) -> None:
+    rules = [{"when": {"hdr": ">DV"}, "tag": "NOPE"}]
+    assert evaluate_tag_rules(rules, CONTEXT) is None
+    assert "numeric operand" in caplog.text
+
+
+def test_operators_inside_lists() -> None:
+    rules = [{"when": {"quality": ["<720", ">=2160"]}, "tag": "MATCHED"}]
+    assert evaluate_tag_rules(rules, CONTEXT) == "MATCHED"
+    rules = [{"when": {"quality": ["<720", "1080p"]}, "tag": "NOPE"}]
+    assert evaluate_tag_rules(rules, CONTEXT) is None
+
+
+@pytest.mark.parametrize("rules", ["UHDGROUP", {"when": {"quality": "2160p"}, "tag": "UHD"}, None, 42])
+def test_rules_that_are_not_a_list(rules: Any) -> None:
+    """A mapping or string here iterates keys/characters, so it must be rejected outright."""
+    assert evaluate_tag_rules(rules, CONTEXT) is None
+
+
+@pytest.mark.parametrize("rule", [None, "UHDGROUP", 42, ["when", "tag"]])
+def test_rule_entry_that_is_not_a_mapping_skipped(rule: Any) -> None:
+    assert evaluate_tag_rules([rule], CONTEXT) is None
+    assert evaluate_tag_rules([rule, {"when": {"title_type": "movie"}, "tag": "MATCHED"}], CONTEXT) == "MATCHED"
+
+
+@pytest.mark.parametrize("when", [["title_type", "movie"], "title_type", 42])
+def test_when_that_is_not_a_mapping_skipped(when: Any) -> None:
+    assert evaluate_tag_rules([{"when": when, "tag": "NOPE"}], CONTEXT) is None
+
+
+@pytest.mark.parametrize(
+    "rule",
+    [
+        {"when": {"atmos": True}, "tag": "ATMOS"},
+        {"when": {"atmos": [True, "Atmos"]}, "tag": "ATMOS"},
+        {"when": {"title_type": "movie"}, "tag": True},
+    ],
+)
+def test_yaml_boolean_skipped(rule: dict[str, Any], caplog: pytest.LogCaptureFixture) -> None:
+    """`atmos: true` / `tag: yes` are YAML coercions the user meant as text."""
+    assert evaluate_tag_rules([rule], CONTEXT) is None
+    assert "boolean" in caplog.text
+
+
+def test_numbers_stay_valid() -> None:
+    assert evaluate_tag_rules([{"when": {"resolution": 2160}, "tag": "UHD"}], CONTEXT) == "UHD"
+    assert evaluate_tag_rules([{"when": {"resolution": 1080}, "tag": "NOPE"}], CONTEXT) is None
+
+
+def test_numeric_tag_becomes_text() -> None:
+    assert evaluate_tag_rules([{"when": {"title_type": "movie"}, "tag": 123}], CONTEXT) == "123"
+
+
+@pytest.mark.parametrize("tag", [["A", "B"], {"x": 1}, ("A",), {"A"}])
+def test_non_scalar_tag_skipped(tag: Any, caplog: pytest.LogCaptureFixture) -> None:
+    """A list or dict here used to reach the filename as its repr."""
+    assert evaluate_tag_rules([{"when": {"title_type": "movie"}, "tag": tag}], CONTEXT) is None
+    assert "must be text or a number" in caplog.text
+
+
+@pytest.mark.parametrize("tag", ["", 0])
+def test_empty_tag_skipped(tag: Any) -> None:
+    assert evaluate_tag_rules([{"when": {"title_type": "movie"}, "tag": tag}], CONTEXT) is None
+
+
+def test_skipped_rule_falls_through_to_a_later_one() -> None:
+    rules = [
+        {"when": {"atmos": True}, "tag": "SKIPPED"},
+        {"when": {"title_type": "movie"}, "tag": "MATCHED"},
+    ]
+    assert evaluate_tag_rules(rules, CONTEXT) == "MATCHED"
+
+
+def test_several_negations_in_a_list_always_match() -> None:
+    """Any-entry matching makes a list of negations useless, which the docs warn about."""
+    rules = [{"when": {"hdr": ["!=DV", "!=HLG"]}, "tag": "ALWAYS"}]
+    assert evaluate_tag_rules(rules, CONTEXT) == "ALWAYS"
+
+
+def test_leading_space_before_an_operator_is_literal() -> None:
+    """Current behaviour: the operator is only found at the very start of the value."""
+    rules = [{"when": {"quality": " >=2160"}, "tag": "NOPE"}]
+    assert evaluate_tag_rules(rules, CONTEXT) is None
+
+
+def test_inequality_holds_for_a_value_with_no_number() -> None:
+    """ "DV" really is not equal to 2160, so != holds while the others fail."""
+    assert evaluate_tag_rules([{"when": {"hdr": "!=2160"}, "tag": "MATCHED"}], CONTEXT) == "MATCHED"
+    assert evaluate_tag_rules([{"when": {"hdr": ">=2160"}, "tag": "NOPE"}], CONTEXT) is None
+    assert evaluate_tag_rules([{"when": {"hdr": "==2160"}, "tag": "NOPE"}], CONTEXT) is None
+
+
+@pytest.mark.parametrize("condition", ["!=", ">=", "<", "=", "=="])
+def test_bare_operator_fails(condition: str, caplog: pytest.LogCaptureFixture) -> None:
+    assert evaluate_tag_rules([{"when": {"hdr": condition}, "tag": "NOPE"}], CONTEXT) is None
+    assert "no value to compare against" in caplog.text
+
+
+def test_composite_hdr_does_not_match_plain_dv() -> None:
+    """DV titles carry a base-layer suffix, which the docs tell users to list explicitly."""
+    context = {**CONTEXT, "hdr": "DV.HDR10P"}
+    assert evaluate_tag_rules([{"when": {"hdr": "DV"}, "tag": "NOPE"}], context) is None
+    assert evaluate_tag_rules([{"when": {"hdr": ["DV", "DV.HDR10P"]}, "tag": "DOLBY"}], context) == "DOLBY"

tests/core/test_tag_rules_integration.py

@@ -0,0 +1,131 @@
+"""Tests for tag_rules reaching the filename context through Title."""
+
+from __future__ import annotations
+
+import inspect
+import types
+from typing import Any, Iterator
+
+import pytest
+from langcodes import Language
+
+from unshackle.core.config import config
+from unshackle.core.titles.movie import Movie
+
+
+class DummyService:
+    pass
+
+
+class TrackStub:
+    """Stand-in for Tracks: truthy and iterable, with the two lists lang_tag reads."""
+
+    def __init__(self, audio: list[Any], subtitles: list[Any]) -> None:
+        self.audio = audio
+        self.subtitles = subtitles
+
+    def __iter__(self) -> Iterator[Any]:
+        return iter([])
+
+    def __bool__(self) -> bool:
+        return True
+
+
+def make_audio(language: str) -> types.SimpleNamespace:
+    return types.SimpleNamespace(
+        language=language,
+        format="AAC",
+        channel_layout=None,
+        channellayout_original=None,
+        channel_s=2,
+        channels=2,
+        format_additionalfeatures=None,
+        joc=None,
+    )
+
+
+def make_media_info(languages: list[str]) -> types.SimpleNamespace:
+    return types.SimpleNamespace(video_tracks=[], audio_tracks=[make_audio(lang) for lang in languages])
+
+
+def make_movie() -> Movie:
+    return Movie(id_="movie-0001", service=DummyService, name="The Film", year=2024, language="en")
+
+
+@pytest.fixture(autouse=True)
+def clean_config(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(config, "tag", "NOGRP")
+    monkeypatch.setattr(config, "tag_rules", [])
+    monkeypatch.setattr(config, "language_tags", {})
+
+
+def context_for(rules: list[dict[str, Any]], monkeypatch: pytest.MonkeyPatch) -> dict:
+    monkeypatch.setattr(config, "tag_rules", rules)
+    return make_movie()._build_base_template_context(make_media_info(["en"]))
+
+
+def test_matching_rule_replaces_the_tag(monkeypatch: pytest.MonkeyPatch) -> None:
+    context = context_for([{"when": {"title_type": "movie"}, "tag": "MOVIEGRP"}], monkeypatch)
+    assert context["tag"] == "MOVIEGRP"
+
+
+def test_no_matching_rule_keeps_the_config_tag(monkeypatch: pytest.MonkeyPatch) -> None:
+    context = context_for([{"when": {"title_type": "series"}, "tag": "TVGRP"}], monkeypatch)
+    assert context["tag"] == "NOGRP"
+
+
+def test_no_rules_keeps_the_config_tag(monkeypatch: pytest.MonkeyPatch) -> None:
+    context = context_for([], monkeypatch)
+    assert context["tag"] == "NOGRP"
+
+
+def test_title_type_is_in_the_context(monkeypatch: pytest.MonkeyPatch) -> None:
+    assert context_for([], monkeypatch)["title_type"] == "movie"
+
+
+def test_rules_run_after_lang_tag(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A rule matching on lang_tag only works if lang_tag is resolved first."""
+    monkeypatch.setattr(config, "language_tags", {"rules": [{"audio": "ja", "tag": "SUBBED"}]})
+    monkeypatch.setattr(config, "tag_rules", [{"when": {"lang_tag": "SUBBED"}, "tag": "SUBGRP"}])
+
+    movie = make_movie()
+    movie.tracks = TrackStub(audio=[types.SimpleNamespace(language=Language.get("ja"))], subtitles=[])  # type: ignore[assignment]
+    context = movie._build_base_template_context(make_media_info(["ja"]))
+
+    assert context["lang_tag"] == "SUBBED"
+    assert context["tag"] == "SUBGRP"
+
+
+def test_a_quality_rule_matches_the_built_context(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Guards the context keys the docs tell users to match on."""
+    monkeypatch.setattr(config, "tag_rules", [{"when": {"quality": "2160p", "video": "H.265"}, "tag": "UHDGROUP"}])
+    video = types.SimpleNamespace(
+        width=3840,
+        height=2160,
+        format="HEVC",
+        scan_type="Progressive",
+        other_display_aspect_ratio=["16:9"],
+        hdr_format_commercial=None,
+        hdr_format=None,
+        transfer_characteristics=None,
+        transfer_characteristics_original=None,
+        frame_rate="23.976",
+    )
+    media_info = types.SimpleNamespace(video_tracks=[video], audio_tracks=[make_audio("en")])
+
+    context = make_movie()._build_base_template_context(media_info)
+
+    assert context["quality"] == "2160p"
+    assert context["video"] == "H.265"
+    assert context["tag"] == "UHDGROUP"
+
+
+def test_dl_clears_tag_rules_when_tag_is_given() -> None:
+    """Source-level pin: the block lives inside the click-driven dl.__init__."""
+    from unshackle.commands.dl import dl
+
+    source = inspect.getsource(dl.__init__)
+    block = source.split("if tag:")[1].split("\n\n")[0]
+
+    assert "config.tag = tag" in block
+    assert "config.tag_rules = []" in block

unshackle/commands/dl.py

@@ -574,7 +574,10 @@ class dl:
         help="Proxy URI to use. If a 2-letter country is provided, it will try to get a proxy from the config.",
     )
     @click.option(
-        "--tag", type=str, default=None, help="Set the Group Tag to be used, overriding the one in config if any."
+        "--tag",
+        type=str,
+        default=None,
+        help="Set the Group Tag to be used, overriding the config tag and any tag rules.",
     )
     @click.option("--repack", is_flag=True, default=False, help="Add REPACK tag to the output filename.")
     @click.option(
@@ -1252,6 +1255,7 @@ class dl:
 
         if tag:
             config.tag = tag
+            config.tag_rules = []
 
         # needs to be added this way instead of @cli.result_callback to be
         # able to keep `self` as the first positional

unshackle/core/config.py

@@ -130,6 +130,7 @@ class Config:
         self.redact_paths: bool = kwargs.get("redact_paths", True)
 
         self.language_tags: dict = kwargs.get("language_tags") or {}
+        self.tag_rules: list = kwargs.get("tag_rules") or []
         self.dual_multi_mode: str = (kwargs.get("dual_multi_mode") or "strict").lower()
         self.output_template: dict = kwargs.get("output_template") or {}
         folder_cfg = self.output_template.pop("folder", "")
@@ -203,6 +204,7 @@ class Config:
             "edition",
             "repack",
             "lang_tag",
+            "title_type",
         }
 
         unsafe_chars = r'[<>:"/\\|?*]'

unshackle/core/titles/title.py

@@ -95,6 +95,9 @@ class Title:
             "hfr": "",
             "edition": "",
             "lang_tag": "",
+            "title_type": {"Movie": "movie", "Episode": "series", "Song": "music"}.get(
+                type(self).__name__, type(self).__name__.lower()
+            ),
         }
 
         if self.tracks:
@@ -199,6 +202,13 @@ class Title:
             sub_langs = [s.language for s in self.tracks.subtitles]
             context["lang_tag"] = evaluate_language_tag(lang_tag_rules, audio_langs, sub_langs)
 
+        if config.tag_rules:
+            from unshackle.core.utils.tag_rules import evaluate_tag_rules
+
+            override = evaluate_tag_rules(config.tag_rules, context)
+            if override:
+                context["tag"] = override
+
         return context
 
     @abstractmethod

unshackle/core/utils/tag_rules.py

@@ -0,0 +1,116 @@
+"""Release-group tag rule engine for output filename templates."""
+
+from __future__ import annotations
+
+import logging
+import operator
+import re
+from collections.abc import Mapping
+from typing import Any, Callable, Optional
+
+log = logging.getLogger(__name__)
+
+_OPS: dict[str, Callable[[Any, Any], bool]] = {
+    ">=": operator.ge,
+    "<=": operator.le,
+    "!=": operator.ne,
+    "==": operator.eq,
+    ">": operator.gt,
+    "<": operator.lt,
+    "=": operator.eq,
+}
+_ORDERING = (">=", "<=", ">", "<")
+_NUMBER = re.compile(r"\d+(?:\.\d+)?")
+
+
+def evaluate_tag_rules(rules: Any, context: dict[str, Any]) -> Optional[str]:
+    """Evaluate tag rules against a filename template context.
+
+    Rules are evaluated in order; the first matching rule's tag is returned.
+    Conditions are AND-ed, compared case-insensitively, and a list matches if
+    any entry matches. A value may lead with a comparison operator, e.g.
+    ">=2160" or "!=DV".
+
+    Args:
+        rules: Rules straight from the user's YAML, so any shape is possible.
+            Each valid rule is a mapping with a ``when`` mapping and a ``tag``.
+        context: The already-built filename template context.
+
+    Returns:
+        The tag string from the first matching rule, or ``None`` if none match.
+    """
+    if not isinstance(rules, list):
+        log.warning("tag_rules must be a list of rules, ignoring: %r", rules)
+        return None
+
+    for rule in rules:
+        if not isinstance(rule, Mapping):
+            log.warning("Tag rule must be a mapping of 'when' and 'tag', skipping: %r", rule)
+            continue
+
+        when = rule.get("when")
+        tag = rule.get("tag")
+        if when is not None and not isinstance(when, Mapping):
+            log.warning("Tag rule 'when' must be a mapping of conditions, skipping: %s", rule)
+            continue
+        if not when or not tag:
+            log.warning("Tag rule needs both 'when' conditions and a 'tag', skipping: %s", rule)
+            continue
+        if isinstance(tag, bool) or any(_has_bool(value) for value in when.values()):
+            log.warning("Tag rule has a YAML boolean, quote the value to use it as text, skipping: %s", rule)
+            continue
+        if not isinstance(tag, (str, int, float)):
+            log.warning("Tag rule 'tag' must be text or a number, skipping: %s", rule)
+            continue
+
+        unknown = [key for key in when if key not in context]
+        if unknown:
+            log.warning("Tag rule has unknown 'when' key(s) %s, skipping: %s", ", ".join(sorted(unknown)), rule)
+            continue
+
+        if all(_matches(expected, context[key]) for key, expected in when.items()):
+            log.debug("Tag rule matched: %s -> %s", rule, tag)
+            return str(tag)
+
+    return None
+
+
+def _has_bool(expected: Any) -> bool:
+    """Check for a YAML boolean, which is an unquoted word the user meant as text."""
+    values = expected if isinstance(expected, list) else [expected]
+    return any(isinstance(x, bool) for x in values)
+
+
+def _matches(expected: Any, actual: Any) -> bool:
+    """Check one condition, where a list matches on any entry."""
+    values = expected if isinstance(expected, list) else [expected]
+    return any(_matches_one(x, actual) for x in values)
+
+
+def _matches_one(expected: Any, actual: Any) -> bool:
+    """Check a single value, which may carry a leading comparison operator."""
+    text = "" if expected is None else str(expected)
+    for op in sorted(_OPS, key=len, reverse=True):
+        if text.startswith(op):
+            return _compare(op, text[len(op) :].strip(), actual)
+    return str(actual or "").lower() == text.lower()
+
+
+def _compare(op: str, operand: str, actual: Any) -> bool:
+    """Compare against an operand, numerically when the operand is a number."""
+    if not operand:
+        log.warning("Tag rule condition '%s' has no value to compare against, condition failed", op)
+        return False
+
+    try:
+        wanted = float(operand)
+    except ValueError:
+        if op in _ORDERING:
+            log.warning("Tag rule condition '%s%s' needs a numeric operand, condition failed", op, operand)
+            return False
+        return _OPS[op](str(actual or "").lower(), operand.lower())
+
+    found = _NUMBER.search(str(actual or ""))
+    if not found:
+        return op == "!="
+    return _OPS[op](float(found.group()), wanted)

unshackle/unshackle-example.yaml

@@ -9,6 +9,12 @@
 # Group or username appended to download filenames after a dash.
 tag: "NOGRP"
 
+# Conditionally swap the tag above based on filename variables (first match wins).
+# See docs/reference/configuration.md#tag_rules for conditions and semantics.
+# tag_rules:
+#   - when: {lang_tag: SUBBED}
+#     tag: SUBGRP
+
 # --- Output filename & folder templates --- docs/guide/output-and-naming.md
 # The full variable list and all naming options are documented there.
 # movies/series have no built-in fallback (define them to download those kinds);