MediaShelf/mediashelf/rules.py

"""Saved-view rule grammar → parameterized SQL (§7.1).

Everything here is whitelist-driven. Field names map to columns through a dict;
operators come from a fixed set; values are always bound parameters. There is no
string interpolation of user input anywhere in this module, and anything outside
the whitelist is a RuleError (surfaced as HTTP 400), never a best-effort guess.
"""

from __future__ import annotations

from typing import Any

DAY = 86400


class RuleError(ValueError):
    """Malformed or non-whitelisted rule. Always a 400, never a 500."""


# field name -> (sql column, type)
FIELDS: dict[str, tuple[str, str]] = {
    "library_id":             ("i.library_id", "int"),
    "kind":                   ("i.kind", "str"),
    "title":                  ("i.title", "str"),
    "year":                   ("i.year", "int"),
    "size_bytes":             ("i.size_bytes", "int"),
    "added_at":               ("i.added_at", "ts"),
    "provider_added_at":      ("i.provider_added_at", "ts"),
    "added_at_source":        ("i.added_at_source", "str"),
    "first_seen_at":          ("i.first_seen_at", "ts"),
    "updated_at":             ("i.updated_at", "ts"),
    "episode_count":          ("i.episode_count", "int"),
    "part_count":             ("i.part_count", "int"),
    "duration_ms":            ("i.duration_ms", "int"),
    "primary_path":           ("i.primary_path", "str"),
    "resolution":             ("i.resolution", "str"),
    "video_codec":            ("i.video_codec", "str"),
    "watch_count":            ("i.watch_count", "int"),
    "partial_count":          ("i.partial_count", "int"),
    "abandoned_count":        ("i.abandoned_count", "int"),
    "avg_percent_complete":   ("i.avg_percent_complete", "float"),
    "last_watched_at":        ("i.last_watched_at", "ts"),
    "last_touched_at":        ("i.last_touched_at", "ts"),
    "first_watched_at":       ("i.first_watched_at", "ts"),
    "distinct_watcher_count": ("i.distinct_watcher_count", "int"),
    "pre_history":            ("i.pre_history", "bool"),
    "kept":                   ("i.kept", "bool"),
    "kept_via":               ("i.kept_via", "str"),
    "status":                 ("i.status", "str"),
    "guid":                   ("i.guid", "str"),
    # derived, provided by the query builder as a correlated expression
    "watch_ratio":            ("(CASE WHEN i.episode_count > 0 "
                               "THEN CAST(i.watch_count AS REAL) / i.episode_count "
                               "ELSE CAST(i.watch_count AS REAL) END)", "float"),
    "duplicate_count":        ("COALESCE(d.dupe_count, 1)", "int"),
}

SIMPLE_OPS = {
    "eq": "=", "ne": "!=", "lt": "<", "lte": "<=", "gt": ">", "gte": ">=",
}
LIST_OPS = {"in": "IN", "not_in": "NOT IN"}
LIKE_OPS = {"contains": "%{}%", "starts_with": "{}%", "ends_with": "%{}"}
NULL_OPS = {"is_null": "IS NULL", "is_not_null": "IS NOT NULL"}
REL_OPS = {"older_than_days", "newer_than_days"}
SPECIAL_OPS = {"never"}

ALL_OPS = (set(SIMPLE_OPS) | set(LIST_OPS) | set(LIKE_OPS)
           | set(NULL_OPS) | REL_OPS | SPECIAL_OPS)


class _Builder:
    def __init__(self, now: int):
        self.now = now
        self.params: dict[str, Any] = {}
        self._n = 0

    def bind(self, value) -> str:
        self._n += 1
        key = "r%d" % self._n
        self.params[key] = value
        return ":" + key

    def coerce(self, value, ftype: str):
        if ftype == "int":
            try:
                return int(value)
            except (TypeError, ValueError):
                raise RuleError("expected an integer, got %r" % (value,))
        if ftype == "float":
            try:
                return float(value)
            except (TypeError, ValueError):
                raise RuleError("expected a number, got %r" % (value,))
        if ftype == "bool":
            if isinstance(value, bool):
                return 1 if value else 0
            if str(value).lower() in ("1", "true", "yes"):
                return 1
            if str(value).lower() in ("0", "false", "no"):
                return 0
            raise RuleError("expected a boolean, got %r" % (value,))
        if ftype == "ts":
            try:
                return int(value)
            except (TypeError, ValueError):
                raise RuleError("expected a timestamp, got %r" % (value,))
        return str(value)

    def group(self, node: dict, depth: int = 0) -> str:
        if depth > 8:
            raise RuleError("rule nesting too deep")
        if not isinstance(node, dict):
            raise RuleError("rule node must be an object")

        if "rules" in node:
            op = str(node.get("op", "and")).lower()
            if op not in ("and", "or"):
                raise RuleError("group operator must be 'and' or 'or'")
            rules = node.get("rules") or []
            if not isinstance(rules, list):
                raise RuleError("'rules' must be a list")
            parts = [self.group(r, depth + 1) for r in rules]
            parts = [p for p in parts if p]
            if not parts:
                return ""
            joiner = " AND " if op == "and" else " OR "
            return "(" + joiner.join(parts) + ")"

        return self.condition(node)

    def condition(self, node: dict) -> str:
        field = node.get("field")
        op = str(node.get("op", "")).lower()
        if field not in FIELDS:
            raise RuleError("unknown field %r" % (field,))
        if op not in ALL_OPS:
            raise RuleError("unknown operator %r" % (op,))

        col, ftype = FIELDS[field]
        value = node.get("value")

        if op in NULL_OPS:
            return "%s %s" % (col, NULL_OPS[op])

        if op == "never":
            # "never watched" is null-or-zero, which is not the same as IS NULL
            return "(%s IS NULL OR %s = 0)" % (col, col)

        if op in REL_OPS:
            days = self.coerce(value, "int")
            cutoff = self.bind(self.now - days * DAY)
            if op == "older_than_days":
                return "(%s IS NOT NULL AND %s > 0 AND %s < %s)" % (col, col, col, cutoff)
            return "(%s IS NOT NULL AND %s >= %s)" % (col, col, cutoff)

        if op in LIST_OPS:
            if not isinstance(value, (list, tuple)) or not value:
                raise RuleError("operator %r needs a non-empty list" % op)
            if len(value) > 500:
                raise RuleError("list too long")
            placeholders = ", ".join(self.bind(self.coerce(v, ftype)) for v in value)
            return "%s %s (%s)" % (col, LIST_OPS[op], placeholders)

        if op in LIKE_OPS:
            if ftype != "str":
                raise RuleError("operator %r only applies to text fields" % op)
            pattern = LIKE_OPS[op].format(_escape_like(str(value)))
            return "%s LIKE %s ESCAPE '\\'" % (col, self.bind(pattern))

        return "%s %s %s" % (col, SIMPLE_OPS[op], self.bind(self.coerce(value, ftype)))


def _escape_like(s: str) -> str:
    return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def compile_rules(rules: dict | None, now: int) -> tuple[str, dict]:
    """Compile a rule group to (sql_fragment, params). Empty rules -> ('', {})."""
    if not rules:
        return "", {}
    b = _Builder(now)
    sql = b.group(rules)
    return sql, b.params


SORTABLE = {
    "title": "i.sort_title, i.title",
    "year": "i.year",
    "size_bytes": "i.size_bytes",
    "added_at": "i.added_at",
    "updated_at": "i.updated_at",
    "last_watched_at": "i.last_watched_at",
    "last_touched_at": "i.last_touched_at",
    "watch_count": "i.watch_count",
    "partial_count": "i.partial_count",
    "abandoned_count": "i.abandoned_count",
    "distinct_watcher_count": "i.distinct_watcher_count",
    "episode_count": "i.episode_count",
    "avg_percent_complete": "i.avg_percent_complete",
    "reclaim_score": "reclaim_score",
    "library": "lib.title",
    "kind": "i.kind",
}


def compile_sort(spec: str | None) -> str:
    """'reclaim_score:desc,size_bytes:desc' -> ORDER BY clause. Whitelisted."""
    if not spec:
        return "reclaim_score DESC, i.size_bytes DESC"
    out = []
    for part in str(spec).split(","):
        part = part.strip()
        if not part:
            continue
        if ":" in part:
            name, direction = part.split(":", 1)
        else:
            name, direction = part, "asc"
        name = name.strip()
        direction = "DESC" if direction.strip().lower() == "desc" else "ASC"
        if name not in SORTABLE:
            raise RuleError("cannot sort by %r" % name)
        out.append("%s %s" % (SORTABLE[name], direction))
    if not out:
        return "reclaim_score DESC, i.size_bytes DESC"
    out.append("i.id ASC")           # stable tiebreak so paging never repeats a row
    return ", ".join(out)