Coverage for ckanext/udc/helpers.py: 65%

1from __future__ import annotations 

2 

3import json 

4import traceback 

5import re 

6from collections import OrderedDict 

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

8 

9from ckan.types import Schema, Context 

10from ckan.common import _ 

11import ckan 

12import ckan.plugins as plugins 

13import ckan.logic as logic 

14import ckan.model as model 

15import ckan.plugins.toolkit as tk 

16from ckan.plugins.toolkit import (chained_action, side_effect_free, chained_helper) 

17import ckan.lib.helpers as h 

18from ckan.common import current_user, _ 

19 

20from .graph.logic import onUpdateCatalogue, onDeleteCatalogue, get_catalogue_graph 

21from .search.params import get_search_details 

22from ckanext.udc.file_format.logic import before_package_update as before_package_update_for_file_format 

23 

24import logging 

25import json 

26import chalk 

27 

28log = logging.getLogger(__name__) 

29 

30 

31import time 

32 

33# Register a chained action after `config_option_update(...)` is triggered, i.e. config is saved from the settings page. 

34# We need to reload the UDC plugin to make sure the maturity model is up to date. 

35@side_effect_free 

36@chained_action 

37def config_option_update(original_action, context, data_dict): 

38 try: 

39 # Call our plugin to update the config 

40 log.info("config_option_update: Update UDC Config") 

41 plugins.get_plugin('udc').reload_config( 

42 json.loads(data_dict["ckanext.udc.config"])) 

43 except: 

44 log.error 

45 

46 res = original_action(context, data_dict) 

47 return res 

48 

49@side_effect_free 

50@chained_action 

51def package_update(original_action, context, data_dict): 

52 # Pre-process custom file format 

53 before_package_update_for_file_format(context, data_dict) 

54 

55 result = original_action(context, data_dict) 

56 try: 

57 if not plugins.get_plugin('udc').disable_graphdb: 

58 onUpdateCatalogue(context, result) 

59 except Exception as e: 

60 log.error(e) 

61 print(e) 

62 raise logic.ValidationError([_("Error occurred in updating the knowledge graph, please contact administrator:\n") + str(e)]) 

63 return result 

64 

65@side_effect_free 

66@chained_action 

67def package_delete(original_action, context, data_dict): 

68 print(f"Package Delete: ", data_dict) 

69 result = original_action(context, data_dict) 

70 try: 

71 if not plugins.get_plugin('udc').disable_graphdb: 

72 onDeleteCatalogue(context, data_dict) 

73 except Exception as e: 

74 log.error(e) 

75 print(e) 

76 raise logic.ValidationError([_("Error occurred in updating the knowledge graph, please contact administrator:\n") + str(e)]) 

77 return result 

78 

79 

80# Register a chained helpers for humanize_entity_type() to change labels. 

81@chained_helper 

82def humanize_entity_type(next_helper: Callable[..., Any], 

83 entity_type: str, object_type: str, purpose: str): 

84 

85 if (entity_type, object_type) == ("package", "catalogue"): 

86 if purpose == "main nav": 

87 return _("Catalogue") 

88 elif purpose == "search placeholder": 

89 return _("Search Catalogue Entries") 

90 elif purpose == "search_placeholder": 

91 # Don't know where is this used. 

92 return _("Catalogue Entry") 

93 elif purpose == "create title": 

94 return _("Create Catalogue Entry") 

95 elif purpose == "create label": 

96 return _("Create Catalogue Entry") 

97 elif purpose == "add link": 

98 return _("Add Catalogue Entry") 

99 elif purpose == "no description": 

100 return _("There is no description for this catalogue entry") 

101 elif purpose == "view label": 

102 return _("View Catalogue Entry") 

103 

104 

105 original_text = next_helper(entity_type, object_type, purpose) 

106 # print(entity_type, object_type, purpose, original_text) 

107 

108 return original_text 

109 

110 

111def render_markdown(data: str, **kwargs): 

112 allow_html = tk.asbool(tk.config.get("ckanext.udc.render_markdown_allow_html", False)) 

113 kwargs.setdefault("allow_html", allow_html) 

114 return h.render_markdown(data, **kwargs) 

115 

116 

117def get_default_facet_titles(): 

118 facets: dict[str, str] = OrderedDict() 

119 

120 # Copied from ckan.views.dataset.search 

121 org_label = h.humanize_entity_type( 

122 u'organization', 

123 h.default_group_type(u'organization'), 

124 u'facet label') or _(u'Organizations') 

125 

126 group_label = h.humanize_entity_type( 

127 u'group', 

128 h.default_group_type(u'group'), 

129 u'facet label') or _(u'Groups') 

130 

131 default_facet_titles = { 

132 u'organization': org_label, 

133 u'groups': group_label, 

134 u'tags': _(u'Tags'), 

135 u'res_format': _(u'Formats'), 

136 u'license_id': _(u'Licenses'), 

137 } 

138 

139 for facet in h.facets(): 

140 if facet in default_facet_titles: 

141 facets[facet] = default_facet_titles[facet] 

142 else: 

143 facets[facet] = facet 

144 

145 # Facet titles 

146 for plugin in plugins.PluginImplementations(plugins.IFacets): 

147 facets = plugin.dataset_facets(facets, "catalogue") 

148 return facets 

149 

150def process_facets_fields(facets_fields: dict): 

151 """For search page displaying search filters""" 

152 results = {} 

153 for field in facets_fields: 

154 if field.startswith("filter-logic"): 

155 continue 

156 

157 field_value = facets_fields[field] 

158 field_name = field_value.get("ui") if isinstance(field_value, dict) else None 

159 if field_name: 

160 pass 

161 elif field.startswith("extras_"): 

162 field_name = field[7:] 

163 elif field.endswith("_ngram"): 

164 field_name = field[:-6] 

165 else: 

166 field_name = field 

167 

168 if field_name not in results: 

169 results[field_name] = {"logic": "or", "values": []} 

170 

171 if isinstance(field_value, list): 

172 for item in field_value: 

173 results[field_name]["values"].append({ 

174 "ori_field": field, 

175 "ori_value": item, 

176 "value": item, 

177 }) 

178 continue 

179 

180 if isinstance(field_value, dict) and 'values' in field_value: 

181 values = field_value['values'] 

182 is_fts = field_value.get('fts', False) 

183 params = field_value.get("params", []) 

184 for index, item in enumerate(values): 

185 results[field_name]["values"].append({ 

186 "ori_field": params[index] if index < len(params) else field, 

187 "ori_value": item, 

188 "value": f'Search for "{item}"' if is_fts else item, 

189 }) 

190 elif isinstance(field_value, dict): 

191 # Date or number ranges 

192 min = field_value.get('min') 

193 max = field_value.get('max') 

194 

195 if min: 

196 results[field_name]["values"].append({ 

197 "ori_field": field_value.get("min_param", "min_" + field_name), 

198 "ori_value": min, 

199 "value": f"From: {min}", 

200 }) 

201 if max: 

202 results[field_name]["values"].append({ 

203 "ori_field": field_value.get("max_param", "max_" + field_name), 

204 "ori_value": max, 

205 "value": f"To: {max}", 

206 }) 

207 

208 

209 if "filter-logic-" + field in facets_fields and facets_fields["filter-logic-" + field][0] == "and": 

210 results[field_name]["logic"] = "and" 

211 

212 return results 

213 

214def udc_search_details(default_fields: Optional[dict] = None): 

215 details = get_search_details() 

216 fields_grouped = {} 

217 if isinstance(default_fields, dict): 

218 fields_grouped.update(default_fields) 

219 fields_grouped.update(details["fields_grouped"]) 

220 details["fields_grouped"] = fields_grouped 

221 return details 

222 

223def get_maturity_percentages(config, pkg_dict): 

224 percentages = [] 

225 for idx, level in enumerate(config): 

226 num_not_empty = 0 

227 total_size = 0 

228 for field in level["fields"]: 

229 if field.get("ckanField"): 

230 # Skip custom_fields 

231 if field.get("ckanField") in ['custom_fields']: 

232 continue 

233 # organization_and_visibility is always filled 

234 if field.get("ckanField") == 'organization_and_visibility': 

235 num_not_empty += 2 

236 total_size += 1 

237 # `description` is stored as `notes` 

238 elif field.get("ckanField") == 'description' and pkg_dict.get("notes"): 

239 num_not_empty += 1 

240 # `source` is stored as `url` 

241 elif field.get("ckanField") == 'source' and pkg_dict.get("url"): 

242 num_not_empty += 1 

243 elif pkg_dict.get(field["ckanField"]): 

244 num_not_empty += 1 

245 else: 

246 if field.get("name") and field.get("label") and pkg_dict.get(field["name"]): 

247 num_not_empty += 1 

248 

249 total_size += 1 

250 percentages.append(str(round(num_not_empty / total_size * 100)) + "%") 

251 

252 return percentages 

253 

254 

255def get_system_info(name: str): 

256 return model.system_info.get_system_info(name) 

257 

258 

259def udc_json_attr(value): 

260 """Return a JSON string safe to embed in an HTML attribute. 

261 

262 - If value is already a string, assume it is JSON (or plain text) and 

263 return it as-is; Jinja's autoescape will handle HTML encoding. 

264 - If value is a dict/list/etc, json.dumps it to a compact string. 

265 """ 

266 

267 if value is None: 

268 return "" 

269 if isinstance(value, str): 

270 return value 

271 try: 

272 return json.dumps(value, ensure_ascii=False) 

273 except Exception: 

274 return ""