Coverage for ckanext/udc/search/params.py: 11%

1from __future__ import annotations 

2 

3from collections import OrderedDict 

4from datetime import datetime 

5from typing import Any, Iterable, Optional 

6 

7import ckan.plugins as plugins 

8from ckan.common import request 

9 

10from ckanext.udc.solr.config import get_current_lang 

11 

12 

13UDC_PARAM_PREFIX = "ext_udc_" 

14CKAN_FTS_FIELDS = [ 

15 "title", 

16 "notes", 

17 "url", 

18 "version", 

19 "author", 

20 "author_email", 

21 "maintainer", 

22 "maintainer_email", 

23] 

24CORE_STRING_FACETS = ["organization", "groups", "license_id", "res_format"] 

25 

26 

27def _solr_field_for( 

28 param_kind: str, 

29 ui_field: str, 

30 lang: str, 

31 text_fields: set[str], 

32) -> Optional[str]: 

33 if ui_field in text_fields: 

34 if param_kind == "fts": 

35 return f"{ui_field}_{lang}_txt" 

36 if param_kind == "exact": 

37 return f"{ui_field}_{lang}_f" 

38 if param_kind in ("min", "max"): 

39 return f"extras_{ui_field}" 

40 

41 if ui_field == "tags": 

42 if param_kind == "fts": 

43 return f"tags_{lang}_txt" 

44 if param_kind == "exact": 

45 return f"tags_{lang}_f" 

46 

47 if ui_field in CKAN_FTS_FIELDS: 

48 if param_kind == "fts": 

49 return f"{ui_field}_{lang}_txt" 

50 return None 

51 

52 if ui_field in CORE_STRING_FACETS: 

53 if param_kind == "exact": 

54 return ui_field 

55 return None 

56 

57 if ui_field == "portal_type" and param_kind == "exact": 

58 return "extras_portal_type" 

59 

60 if param_kind in ("min", "max", "exact", "fts"): 

61 return f"extras_{ui_field}" 

62 

63 return None 

64 

65 

66def _decode_udc_param(param: str) -> tuple[Optional[str], Optional[str]]: 

67 if not param.startswith(UDC_PARAM_PREFIX): 

68 return None, None 

69 

70 suffix = param[len(UDC_PARAM_PREFIX):] 

71 for kind in ("filter_logic", "exact", "fts", "min", "max"): 

72 prefix = f"{kind}_" 

73 if suffix.startswith(prefix): 

74 return kind, suffix[len(prefix):] 

75 

76 return None, None 

77 

78 

79def _request_items() -> Iterable[tuple[str, str]]: 

80 try: 

81 return request.args.items(multi=True) 

82 except RuntimeError: 

83 return [] 

84 

85 

86def get_search_details( 

87 params: Optional[Iterable[tuple[str, str]]] = None, 

88) -> dict[str, Any]: 

89 fq = "" 

90 fields: list[tuple[str, str]] = [] 

91 fields_grouped: dict[str, Any] = {} 

92 filter_logics: dict[str, str] = {} 

93 include_undefined: set[str] = set() 

94 

95 udc = plugins.get_plugin("udc") 

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

97 date_fields = set(udc.date_fields or []) 

98 lang = get_current_lang() 

99 

100 for param, value in params if params is not None else _request_items(): 

101 if not value: 

102 continue 

103 

104 kind, ui_name = _decode_udc_param(param) 

105 if not kind or not ui_name: 

106 continue 

107 

108 if kind == "filter_logic": 

109 if value.lower() == "and": 

110 filter_logics[ui_name] = "AND" 

111 elif value in ("date", "number"): 

112 solr_key = _solr_field_for("min", ui_name, lang, text_fields) 

113 if solr_key: 

114 include_undefined.add(solr_key) 

115 continue 

116 

117 solr_kind = "exact" if kind == "exact" else kind 

118 solr_key = _solr_field_for(solr_kind, ui_name, lang, text_fields) 

119 if not solr_key: 

120 continue 

121 

122 if kind in ("fts", "exact"): 

123 fields.append((param, value)) 

124 group = fields_grouped.setdefault( 

125 solr_key, 

126 {"ui": ui_name, "fts": kind == "fts", "values": [], "params": []}, 

127 ) 

128 group["values"].append(value) 

129 group["params"].append(param) 

130 continue 

131 

132 group = fields_grouped.setdefault(solr_key, {"ui": ui_name}) 

133 group[kind] = value 

134 group[f"{kind}_param"] = param 

135 

136 for solr_key, opts in fields_grouped.items(): 

137 if "values" in opts: 

138 vals = opts["values"] 

139 ui_name = opts.get("ui", solr_key) 

140 logic_op = filter_logics.get(ui_name, "OR") 

141 if len(vals) > 1: 

142 joined = f" {logic_op} ".join([f'"{v}"' for v in vals]) 

143 fq += f" {solr_key}:({joined})" 

144 else: 

145 fq += f' {solr_key}:"{vals[0]}"' 

146 continue 

147 

148 min_value = opts.get("min") 

149 max_value = opts.get("max") 

150 ui_name = opts.get("ui") 

151 if ui_name and solr_key.startswith("extras_") and ui_name in date_fields: 

152 try: 

153 if min_value: 

154 min_value = datetime.strptime( 

155 min_value, "%Y-%m-%d" 

156 ).strftime("%Y-%m-%dT%H:%M:%SZ") 

157 if max_value: 

158 max_date = datetime.strptime(max_value, "%Y-%m-%d") 

159 max_value = max_date.replace( 

160 hour=23, minute=59, second=59 

161 ).strftime("%Y-%m-%dT%H:%M:%SZ") 

162 except ValueError: 

163 continue 

164 

165 if min_value and max_value: 

166 range_query = f" {solr_key}:[{min_value} TO {max_value}]" 

167 elif min_value: 

168 range_query = f" {solr_key}:[{min_value} TO *]" 

169 elif max_value: 

170 range_query = f" {solr_key}:[* TO {max_value}]" 

171 else: 

172 range_query = "" 

173 

174 if range_query: 

175 if solr_key in include_undefined: 

176 range_query = f"({range_query} OR (*:* -{solr_key}:[* TO *]))" 

177 fq += range_query 

178 

179 return { 

180 "fields": fields, 

181 "fields_grouped": fields_grouped, 

182 "filter_logics": filter_logics, 

183 "fq": fq, 

184 } 

185 

186 

187def facet_alias_map(facet_keys: Iterable[str], lang: Optional[str] = None): 

188 udc = plugins.get_plugin("udc") 

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

190 lang = lang or get_current_lang() 

191 

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

193 for key in facet_keys: 

194 if key == "tags": 

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

196 elif key == "portal_type": 

197 alias_to_solr[key] = "extras_portal_type" 

198 elif key == "version_dataset": 

199 alias_to_solr[key] = "version_dataset_url" 

200 elif key == "dataset_versions": 

201 alias_to_solr[key] = "dataset_versions_url" 

202 elif key.startswith("extras_"): 

203 alias_to_solr[key] = key 

204 elif key in text_fields: 

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

206 else: 

207 alias_to_solr[key] = key 

208 

209 solr_fields = list(dict.fromkeys(alias_to_solr.values())) 

210 return solr_fields, alias_to_solr