Coverage for heritrace/utils/filters.py: 97%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import threading 

8from typing import TYPE_CHECKING 

9from urllib.parse import quote, urlparse 

10 

11from dateutil import parser as dateutil_parser 

12from flask import url_for 

13from flask_babel import format_datetime, gettext, lazy_gettext 

14from SPARQLWrapper import JSON 

15 

16from heritrace.apis.orcid import format_orcid_attribution, is_orcid_url 

17from heritrace.apis.zenodo import format_zenodo_source, is_zenodo_url 

18from heritrace.sparql import ( 

19 SPARQLWrapperWithRetry, 

20 get_sparql_bindings, 

21 select_results, 

22) 

23from heritrace.utils.uri_utils import is_valid_url 

24 

25if TYPE_CHECKING: 

26 from rdflib import Dataset, Graph, URIRef 

27 

28 

29class Filter: 

30 def __init__( 

31 self, context: dict, display_rules: list[dict] | None, sparql_endpoint: str 

32 ) -> None: 

33 self.context = context 

34 self.display_rules = display_rules 

35 self.sparql_endpoint = sparql_endpoint 

36 self._thread_local = threading.local() 

37 self._query_lock = threading.Lock() 

38 

39 def _get_sparql(self) -> SPARQLWrapperWithRetry: 

40 if not hasattr(self._thread_local, "sparql"): 

41 sparql = SPARQLWrapperWithRetry(self.sparql_endpoint, timeout=30.0) 

42 sparql.setReturnFormat(JSON) 

43 self._thread_local.sparql = sparql 

44 return self._thread_local.sparql 

45 

46 @staticmethod 

47 def _find_display_name_from_rule( 

48 rule: dict, 

49 predicate_uri: str, 

50 object_shape_uri: str | None, 

51 ) -> str | None: 

52 if "displayProperties" not in rule: 

53 return None 

54 for display_property in rule["displayProperties"]: 

55 prop_uri = display_property.get("property") or display_property.get( 

56 "virtual_property" 

57 ) 

58 if prop_uri == str(predicate_uri): 

59 if "displayRules" in display_property: 

60 if object_shape_uri: 

61 for display_rule in display_property["displayRules"]: 

62 if display_rule.get("shape") == object_shape_uri: 

63 return display_rule["displayName"] 

64 return display_property["displayRules"][0]["displayName"] 

65 if "displayName" in display_property: 

66 return display_property["displayName"] 

67 return None 

68 

69 def human_readable_predicate( 

70 self, 

71 predicate_uri: str, 

72 entity_key: tuple[str | None, str | None], 

73 *, 

74 is_link: bool = False, 

75 object_shape_uri: str | None = None, 

76 ) -> str: 

77 from heritrace.utils.display_rules_utils import ( # noqa: PLC0415 

78 find_matching_rule, 

79 ) 

80 

81 class_uri, shape_uri = entity_key 

82 rule = find_matching_rule(class_uri, shape_uri, self.display_rules) 

83 

84 if rule: 

85 display_name = self._find_display_name_from_rule( 

86 rule, predicate_uri, object_shape_uri 

87 ) 

88 if display_name is not None: 

89 return display_name 

90 

91 first_part, _ = split_namespace(predicate_uri) 

92 if first_part in self.context: 

93 return format_uri_as_readable(predicate_uri) 

94 if is_valid_url(predicate_uri) and is_link: 

95 href = url_for("entity.about", subject=quote(predicate_uri)) 

96 alt = gettext( 

97 "Link to the entity %(entity)s", 

98 entity=predicate_uri, 

99 ) 

100 return f"<a href='{href}' alt='{alt}'>{predicate_uri}</a>" 

101 return str(predicate_uri) 

102 

103 def human_readable_class( 

104 self, entity_key: tuple[str | None, str | None] | None 

105 ) -> str: 

106 """ 

107 Converts a class URI to human-readable format. 

108 

109 Args: 

110 entity_key (tuple): A tuple containing (class_uri, shape_uri) 

111 

112 Returns: 

113 str: Human-readable representation of the class 

114 """ 

115 from heritrace.utils.display_rules_utils import ( # noqa: PLC0415 

116 find_matching_rule, 

117 ) 

118 from heritrace.utils.shacl_utils import ( # noqa: PLC0415 

119 determine_shape_for_classes, 

120 ) 

121 

122 if entity_key is None: 

123 return "Unknown" 

124 

125 class_uri, shape_uri = entity_key 

126 

127 if class_uri is None and shape_uri is None: 

128 return "Unknown" 

129 

130 if shape_uri is None and class_uri is not None: 

131 shape_uri = determine_shape_for_classes([class_uri]) 

132 rule = find_matching_rule(class_uri, shape_uri, self.display_rules) 

133 

134 if rule and "displayName" in rule: 

135 return rule["displayName"] 

136 

137 if class_uri is None: 

138 return "Unknown" 

139 return format_uri_as_readable(class_uri) 

140 

141 def human_readable_entity( 

142 self, 

143 uri: str | URIRef, 

144 entity_key: tuple[str | None, str | None], 

145 graph: Graph | Dataset | None = None, 

146 ) -> str: 

147 from heritrace.utils.display_rules_utils import ( # noqa: PLC0415 

148 find_matching_rule, 

149 ) 

150 

151 uri_string = str(uri) 

152 rule = find_matching_rule(entity_key[0], entity_key[1], self.display_rules) 

153 if not rule: 

154 return uri_string 

155 

156 if "fetchUriDisplay" in rule: 

157 return self.get_fetch_uri_display(uri_string, rule, graph) 

158 

159 if "displayName" in rule: 

160 return rule["displayName"] 

161 

162 return uri_string 

163 

164 def get_fetch_uri_display( 

165 self, 

166 uri: str | URIRef, 

167 rule: dict, 

168 graph: Graph | Dataset | None = None, 

169 ) -> str: 

170 uri_string = str(uri) 

171 query = rule["fetchUriDisplay"].replace("[[uri]]", f"<{uri_string}>") 

172 

173 if graph is not None: 

174 with self._query_lock: 

175 results = graph.query(query) 

176 for row in select_results(results): 

177 display = row["display"] 

178 if display is not None: 

179 return str(display) 

180 else: 

181 sparql = self._get_sparql() 

182 sparql.setQuery(query) 

183 bindings = get_sparql_bindings(sparql.query().convert()) 

184 if bindings: 

185 return bindings[0]["display"]["value"] 

186 

187 msg = f"fetchUriDisplay returned no result for {uri_string}" 

188 raise ValueError(msg) 

189 

190 def human_readable_datetime(self, dt_str: str) -> str: 

191 dt = dateutil_parser.parse(dt_str) 

192 return format_datetime(dt, format="long") 

193 

194 def human_readable_primary_source(self, primary_source: str | None) -> str: 

195 if primary_source is None: 

196 return str(lazy_gettext("Unknown")) 

197 if "/prov/se" in primary_source: 

198 version_url = f"/entity-version/{primary_source.replace('/prov/se', '')}" 

199 return ( 

200 f"<a href='{version_url}'" 

201 f" alt='{lazy_gettext('Link to the primary source description')}'>" 

202 + lazy_gettext("Version") 

203 + " " 

204 + primary_source.split("/prov/se/")[-1] 

205 + "</a>" 

206 ) 

207 if is_valid_url(primary_source): 

208 alt = lazy_gettext("Link to the primary source description") 

209 return ( 

210 f"<a href='{primary_source}'" 

211 f" alt='{alt}" 

212 f" target='_blank'>" 

213 f"{primary_source}</a>" 

214 ) 

215 return primary_source 

216 

217 def format_source_reference(self, url: str) -> str: 

218 """ 

219 Format a source reference for display, handling various URL types including 

220 Zenodo DOIs and generic URLs. 

221 

222 Args: 

223 url (str): The source URL or identifier to format 

224 human_readable_primary_source (callable): Function to handle generic/unknown 

225 source types 

226 

227 Returns: 

228 str: Formatted HTML string representing the source 

229 """ 

230 if not url: 

231 return "Unknown" 

232 

233 # First check if it's a Zenodo DOI since this is more specific than a generic 

234 # URL 

235 if is_zenodo_url(url): 

236 return format_zenodo_source(url) 

237 

238 # If not Zenodo, use the provided generic handler 

239 return self.human_readable_primary_source(url) 

240 

241 def format_agent_reference(self, url: str) -> str: 

242 """ 

243 Format an agent reference for display, handling various URL types including 

244 ORCID and others. 

245 

246 Args: 

247 url (str): The agent URL or identifier to format 

248 

249 Returns: 

250 str: Formatted HTML string representing the agent 

251 """ 

252 if not url: 

253 return "Unknown" 

254 

255 if is_orcid_url(url): 

256 return format_orcid_attribution(url) 

257 

258 # For now, just return a simple linked version for other URLs 

259 if is_valid_url(url): 

260 return f'<a href="{url}" target="_blank">{url}</a>' 

261 

262 # If it's not a URL at all, just return the raw value 

263 return url 

264 

265 

266def split_namespace(uri: str) -> tuple[str, str]: 

267 """ 

268 Split a URI into namespace and local part. 

269 

270 Args: 

271 uri: The URI to split 

272 

273 Returns: 

274 Tuple of (namespace, local_part) 

275 """ 

276 parsed = urlparse(uri) 

277 if parsed.fragment: 

278 first_part = parsed.scheme + "://" + parsed.netloc + parsed.path + "#" 

279 last_part = parsed.fragment 

280 else: 

281 first_part = ( 

282 parsed.scheme 

283 + "://" 

284 + parsed.netloc 

285 + "/".join(parsed.path.split("/")[:-1]) 

286 + "/" 

287 ) 

288 last_part = parsed.path.split("/")[-1] 

289 return first_part, last_part 

290 

291 

292def format_uri_as_readable(uri: str) -> str: 

293 """ 

294 Format a URI as human-readable text by extracting and formatting the local part. 

295 

296 Args: 

297 uri: The URI to format 

298 

299 Returns: 

300 Human-readable string 

301 """ 

302 _, last_part = split_namespace(uri) 

303 

304 if last_part.islower(): 

305 return last_part 

306 # Convert CamelCase to space-separated words 

307 words = [] 

308 word = "" 

309 for char in last_part: 

310 if char.isupper() and word: 

311 words.append(word) 

312 word = char 

313 else: 

314 word += char 

315 words.append(word) 

316 return " ".join(words).lower()