Coverage for ramose/filters.py: 100%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7from collections.abc import Mapping 

8from pathlib import Path 

9from re import Match, findall, sub 

10 

11import yaml 

12 

13from ramose.escaping import escape_iri, escape_literal, escape_local_name 

14 

15FiltersConfig = Mapping[str, Mapping[str, "str | dict[str, str]"]] 

16 

17_VALUE_PATTERN = r"\{\{\s*value\s*\}\}" 

18_QLEVER_HAS_WORD = ("ql:has-word", "<http://qlever.cs.uni-freiburg.de/builtin-functions/has-word>") 

19_ALWAYS_EMPTY_FILTER = "FILTER(false)" 

20 

21 

22def render(template: str, value: str) -> str: 

23 def bind(placeholder: Match[str]) -> str: 

24 before = template[placeholder.start() - 1 : placeholder.start()] 

25 after = template[placeholder.end() : placeholder.end() + 1] 

26 if before in ('"', "'") and after == before: 

27 return escape_literal(value) 

28 if before == "<" and after == ">": 

29 return escape_iri(value) 

30 return escape_local_name(value) 

31 

32 return sub(_VALUE_PATTERN, bind, template) 

33 

34 

35def _select_template(key: str, spec: str | dict[str, str], value: str) -> str: 

36 if isinstance(spec, str): 

37 return spec 

38 if value not in spec: 

39 msg = f"The value '{value}' is not valid for filter '{key}', valid values are {', '.join(sorted(spec))}" 

40 raise ValueError(msg) 

41 return spec[value] 

42 

43 

44def _render_fragments(template: str, value: str) -> list[str]: 

45 if not any(predicate in template for predicate in _QLEVER_HAS_WORD): 

46 return [render(template, value)] 

47 words = findall(r"[^\W_]+", value.lower()) 

48 if not words: 

49 return [_ALWAYS_EMPTY_FILTER] 

50 return [render(template, word) for word in words] 

51 

52 

53def _is_always_empty(fragment: str) -> bool: 

54 return fragment.strip().upper() == _ALWAYS_EMPTY_FILTER.upper() 

55 

56 

57def apply_filters(config: FiltersConfig, values: list[str]) -> dict[str, str]: 

58 slots: list[str] = [] 

59 for slot_map in config.values(): 

60 for slot in slot_map: 

61 if slot not in slots: 

62 slots.append(slot) 

63 result = dict.fromkeys(slots, "") 

64 fragments: list[tuple[str, str]] = [] 

65 

66 raw_pairs = (pair.strip() for value in values for pair in value.split(",")) 

67 for pair in (pair for pair in raw_pairs if pair): 

68 key, value = pair.split(":", 1) 

69 if key not in config: 

70 msg = f"The filter '{key}' is not configured, configured filters are {', '.join(sorted(config))}" 

71 raise ValueError(msg) 

72 for slot, spec in config[key].items(): 

73 template = _select_template(key, spec, value) 

74 fragments.extend((slot, fragment) for fragment in _render_fragments(template, value)) 

75 

76 always_empty_slots = {slot for slot, fragment in fragments if _is_always_empty(fragment)} 

77 if always_empty_slots: 

78 for slot in always_empty_slots: 

79 result[slot] = _ALWAYS_EMPTY_FILTER 

80 return result 

81 

82 for slot, fragment in fragments: 

83 result[slot] = f"{result[slot]}\n{fragment}" if result[slot] else fragment 

84 return result 

85 

86 

87def load_yaml_config(path: str) -> FiltersConfig: 

88 return yaml.safe_load(Path(path).read_text(encoding="utf-8"))