"""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"), "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)