Coverage for heritrace / routes / entity / _about.py: 88%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5from flask import abort, current_app, render_template 

6from flask_login import current_user, login_required 

7from rdflib import RDF, Graph, Literal, URIRef 

8from time_agnostic_library.agnostic_entity import AgnosticEntity 

9 

10from heritrace.extensions import ( 

11 get_change_tracking_config, 

12 get_dataset_is_quadstore, 

13 get_display_rules, 

14 get_form_fields, 

15 get_shacl_graph, 

16) 

17from heritrace.forms import UpdateTripleForm 

18from heritrace.routes.entity._blueprint import entity_bp 

19from heritrace.utils.datatypes import get_datatype_options 

20from heritrace.utils.display_rules_utils import ( 

21 get_grouped_triples, 

22 get_highest_priority_class, 

23) 

24from heritrace.utils.primary_source_utils import get_default_primary_source 

25from heritrace.utils.shacl_utils import determine_shape_for_entity_triples 

26from heritrace.utils.shacl_validation import get_valid_predicates 

27from heritrace.utils.sparql_utils import ( 

28 convert_to_rdflib_graphs, 

29 fetch_data_graph_for_subject, 

30 get_triples_from_graph, 

31) 

32from heritrace.utils.virtual_properties import get_virtual_properties_for_entity 

33 

34 

35def get_deleted_entity_context_info( 

36 *, 

37 is_deleted: bool, 

38 sorted_timestamps: list[str], 

39 history: dict, 

40 subject: URIRef, 

41) -> tuple[Graph | None, str | None, str | None]: 

42 if is_deleted and len(sorted_timestamps) > 1: 

43 context_snapshot = history[str(subject)][sorted_timestamps[-2]] 

44 

45 subject_classes = [ 

46 str(o) 

47 for _, _, o in get_triples_from_graph( 

48 context_snapshot, (subject, RDF.type, None) 

49 ) 

50 ] 

51 

52 highest_priority_class = get_highest_priority_class(subject_classes) 

53 entity_shape = determine_shape_for_entity_triples( 

54 list(get_triples_from_graph(context_snapshot, (subject, None, None))) 

55 ) 

56 

57 return context_snapshot, highest_priority_class, entity_shape 

58 return None, None, None 

59 

60 

61def _build_live_entity_context( 

62 subject_uri: URIRef, 

63 history: dict, 

64 subject: str, 

65) -> tuple[dict, list, list, dict, dict, dict, str | None, str | None]: 

66 data_graph = fetch_data_graph_for_subject(subject_uri) 

67 

68 if not history.get(subject) and (not data_graph or len(data_graph) == 0): 

69 abort(404) 

70 

71 if not data_graph: 

72 return {}, [], [], {}, {}, {}, None, None 

73 

74 triples: list[tuple[URIRef, URIRef, URIRef | Literal]] = [ 

75 ( 

76 URIRef(str(s)), 

77 URIRef(str(p)), 

78 URIRef(str(o)) if isinstance(o, URIRef) else o, 

79 ) # type: ignore[misc] 

80 for s, p, o in get_triples_from_graph(data_graph, (None, None, None)) 

81 ] 

82 subject_classes = [ 

83 str(o) 

84 for _, _, o in get_triples_from_graph(data_graph, (subject_uri, RDF.type, None)) 

85 ] 

86 subject_triples = list( 

87 get_triples_from_graph(data_graph, (subject_uri, None, None)) 

88 ) 

89 

90 highest_priority_class = get_highest_priority_class(subject_classes) 

91 entity_shape = determine_shape_for_entity_triples(subject_triples) 

92 

93 if not highest_priority_class: 

94 return {}, [], [], {}, {}, {}, highest_priority_class, entity_shape 

95 

96 ( 

97 can_be_added, 

98 can_be_deleted, 

99 datatypes, 

100 mandatory_values, 

101 optional_values, 

102 valid_predicates_set, 

103 ) = get_valid_predicates( 

104 triples, highest_priority_class=URIRef(highest_priority_class) 

105 ) 

106 valid_predicates = list(valid_predicates_set) 

107 

108 grouped_triples, relevant_properties = get_grouped_triples( 

109 subject_uri, 

110 triples, 

111 valid_predicates, 

112 entity_key=(highest_priority_class, entity_shape), 

113 ) 

114 

115 if entity_shape: 

116 virtual_properties = get_virtual_properties_for_entity( 

117 highest_priority_class, entity_shape 

118 ) 

119 else: 

120 virtual_properties = [] 

121 

122 can_be_added = [uri for uri in can_be_added if uri in relevant_properties] + [ 

123 vp[0] for vp in virtual_properties 

124 ] 

125 can_be_deleted = [uri for uri in can_be_deleted if uri in relevant_properties] + [ 

126 vp[0] for vp in virtual_properties 

127 ] 

128 

129 return ( 

130 grouped_triples, 

131 can_be_added, 

132 can_be_deleted, 

133 datatypes, 

134 mandatory_values, 

135 optional_values, 

136 highest_priority_class, 

137 entity_shape, 

138 ) 

139 

140 

141@entity_bp.route("/about/<path:subject>") 

142@login_required 

143def about(subject: str) -> str: 

144 subject_uri = URIRef(subject) 

145 change_tracking_config = get_change_tracking_config() 

146 

147 default_primary_source = get_default_primary_source(current_user.orcid) 

148 

149 agnostic_entity = AgnosticEntity( 

150 res=subject, 

151 config=change_tracking_config, 

152 include_related_objects=False, 

153 include_merged_entities=False, 

154 include_reverse_relations=False, 

155 ) 

156 history, provenance = agnostic_entity.get_history(include_prov_metadata=True) 

157 history = convert_to_rdflib_graphs(history, is_quadstore=get_dataset_is_quadstore()) 

158 

159 is_deleted = False 

160 context_snapshot = None 

161 highest_priority_class = None 

162 entity_shape = None 

163 

164 if history.get(subject): 

165 sorted_timestamps = sorted(history[subject].keys()) 

166 latest_metadata = next( 

167 ( 

168 meta 

169 for _, meta in provenance[subject].items() 

170 if meta["generatedAtTime"] == sorted_timestamps[-1] 

171 ), 

172 None, 

173 ) 

174 

175 is_deleted = bool( 

176 latest_metadata 

177 and "invalidatedAtTime" in latest_metadata 

178 and latest_metadata["invalidatedAtTime"] 

179 ) 

180 

181 context_snapshot, highest_priority_class, entity_shape = ( 

182 get_deleted_entity_context_info( 

183 is_deleted=is_deleted, 

184 sorted_timestamps=sorted_timestamps, 

185 history=history, 

186 subject=subject_uri, 

187 ) 

188 ) 

189 

190 if is_deleted: 

191 grouped_triples: dict = {} 

192 can_be_added: list = [] 

193 can_be_deleted: list = [] 

194 datatypes: dict = {} 

195 mandatory_values: dict = {} 

196 optional_values: dict = {} 

197 else: 

198 ( 

199 grouped_triples, 

200 can_be_added, 

201 can_be_deleted, 

202 datatypes, 

203 mandatory_values, 

204 optional_values, 

205 highest_priority_class, 

206 entity_shape, 

207 ) = _build_live_entity_context(subject_uri, history, subject) 

208 

209 update_form = UpdateTripleForm() 

210 form_fields = get_form_fields() 

211 datatype_options = get_datatype_options() 

212 

213 predicate_details_map = {} 

214 for entity_type_key, predicates in form_fields.items(): 

215 for predicate_uri, details_list in predicates.items(): 

216 for details in details_list: 

217 shape = details.get("nodeShape") 

218 key = (predicate_uri, entity_type_key, shape) 

219 predicate_details_map[key] = details 

220 

221 return render_template( 

222 "entity/about.jinja", 

223 subject=subject, 

224 history=history, 

225 can_be_added=can_be_added, 

226 can_be_deleted=can_be_deleted, 

227 datatypes=datatypes, 

228 update_form=update_form, 

229 mandatory_values=mandatory_values, 

230 optional_values=optional_values, 

231 shacl=bool(len(get_shacl_graph())), 

232 grouped_triples=grouped_triples, 

233 display_rules=get_display_rules(), 

234 form_fields=form_fields, 

235 entity_type=highest_priority_class, 

236 entity_shape=entity_shape, 

237 predicate_details_map=predicate_details_map, 

238 dataset_db_triplestore=current_app.config["DATASET_DB_TRIPLESTORE"], 

239 dataset_db_text_index_enabled=current_app.config[ 

240 "DATASET_DB_TEXT_INDEX_ENABLED" 

241 ], 

242 is_deleted=is_deleted, 

243 context=context_snapshot, 

244 default_primary_source=default_primary_source, 

245 datatype_options=datatype_options, 

246 )