Coverage for ckanext/udc/search/logic/actions.py: 25%

1from __future__ import annotations 

2from typing import Any, Callable, Collection, KeysView, Optional, Union, cast 

3from ckanext.udc.search.logic.utils import profile_func, cache_for 

4from ckan import model, authz, logic 

5from ckan.common import _, config, current_user 

6import ckan.plugins as plugins 

7import ckan.lib.helpers as h 

8from ckan.plugins.toolkit import side_effect_free 

9 

10import logging 

11 

12log = logging.getLogger(__name__) 

13 

14 

15# @profile_func 

16@side_effect_free 

17def filter_facets_get(context, data_dict): 

18 return _filter_facets_get(data_dict) 

19 

20from collections import OrderedDict 

21from typing import Any 

22import ckan.plugins as plugins 

23import ckan.plugins.toolkit as tk 

24import ckan.logic as logic 

25import ckan.lib.helpers as h 

26 

27from ckanext.udc.solr.config import get_current_lang 

28 

29 

30def _facet_cache_key(data_dict=None, *_, **kwargs): 

31 if isinstance(data_dict, dict): 

32 lang = data_dict.get("lang") 

33 else: 

34 lang = None 

35 

36 if not lang and "lang" in kwargs: 

37 lang = kwargs["lang"] 

38 

39 if not lang and isinstance(kwargs.get("data_dict"), dict): 

40 lang = kwargs["data_dict"].get("lang") 

41 

42 if not lang: 

43 lang = get_current_lang() 

44 

45 return lang or "__default__" 

46 

47 

48@cache_for(60, key_func=_facet_cache_key) 

49def _filter_facets_get(data_dict) -> dict[str, Any]: 

50 """ 

51 data_dict only needs to contain "lang" (optional). 

52  

53 Multilingual facets with stable outward keys: 

54 

55 'tags' -> Solr 'tags_<lang>_f' 

56 '<text_field>' -> Solr '<text_field>_<lang>_f' 

57 'extras_<name>' -> Solr 'extras_<name>' (non-text stays as-is) 

58 

59 The Solr response is renamed back to the stable keys so the UI 

60 and your existing code keep working unchanged. 

61 """ 

62 lang = data_dict.get("lang") or get_current_lang() 

63 

64 # 1) Gather facet keys (prefer plugin-provided; then CKAN defaults) 

65 ordered_plugin_keys: list[str] = [] 

66 for p in plugins.PluginImplementations(plugins.IFacets): 

67 try: 

68 provided = p.dataset_facets(OrderedDict(), "catalogue") or {} 

69 for k in provided.keys(): 

70 if k not in ordered_plugin_keys: 

71 ordered_plugin_keys.append(k) 

72 except Exception: 

73 continue 

74 

75 facet_keys: list[str] = [] 

76 for k in ordered_plugin_keys + list(h.facets()): 

77 if k not in facet_keys: 

78 facet_keys.append(k) 

79 

80 # 2) Build alias -> Solr mapping 

81 try: 

82 udc = plugins.get_plugin("udc") 

83 text_fields = set(udc.text_fields or []) 

84 dropdown_options = udc.dropdown_options or {} 

85 except Exception: 

86 text_fields = set() 

87 dropdown_options = {} 

88 

89 alias_to_solr: OrderedDict[str, str] = OrderedDict() 

90 for key in facet_keys: 

91 if key == "tags": 

92 alias_to_solr[key] = f"tags_{lang}_f" 

93 # version relationship helper facets: use *_title_url for display 

94 # but map outward stable keys to URL-only Solr fields 

95 elif key == "version_dataset": 

96 alias_to_solr[key] = "version_dataset_url" 

97 elif key == "dataset_versions": 

98 alias_to_solr[key] = "dataset_versions_url" 

99 elif key.startswith("extras_"): 

100 # Non-text maturity fields (date/number/select) facet via extras_* 

101 alias_to_solr[key] = key 

102 elif key in text_fields: 

103 # Text maturity fields use language-specific facet fields 

104 alias_to_solr[key] = f"{key}_{lang}_f" 

105 else: 

106 # Any other core facet stays as-is 

107 alias_to_solr[key] = key 

108 

109 facet_fields_solr = list(dict.fromkeys(alias_to_solr.values())) # de-dupe preserve order 

110 

111 # 3) Query Solr 

112 try: 

113 default_limit = int(tk.config.get("search.facets.default", 10)) 

114 except Exception: 

115 default_limit = 10 

116 

117 data_dict: dict[str, Any] = { 

118 "q": "*:*", 

119 "facet.limit": -1, 

120 "facet.field": facet_fields_solr, 

121 "rows": default_limit, 

122 "start": 0, 

123 "fq": 'capacity:"public"', 

124 } 

125 query = logic.get_action("package_search")({}, data_dict) 

126 raw_facets: dict[str, Any] = query.get("search_facets", {}) 

127 

128 # 4) Rename Solr facet keys back to stable outward keys 

129 facets: dict[str, Any] = {} 

130 for alias, solr_name in alias_to_solr.items(): 

131 if solr_name in raw_facets: 

132 facets[alias] = raw_facets[solr_name] 

133 

134 # 5) Localize dropdown option labels for extras_* facets 

135 for stable_key, payload in facets.items(): 

136 # For extras_*, strip prefix to lookup configured option labels 

137 base = stable_key[7:] if stable_key.startswith("extras_") else stable_key 

138 options_map = dropdown_options.get(base) 

139 if not options_map: 

140 continue 

141 for item in payload.get("items", []): 

142 label = options_map.get(item.get("name")) 

143 if label: 

144 item["display_name"] = label 

145 

146 return facets