Coverage for oc_meta / lib / rdf_patch.py: 90%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import hashlib 

8import multiprocessing 

9import os 

10from collections import defaultdict 

11from concurrent.futures import ProcessPoolExecutor 

12from dataclasses import dataclass 

13from typing import TypeVar, cast 

14 

15import orjson 

16import yaml 

17from oc_ocdm.graph import GraphSet 

18from oc_ocdm.graph.entities.bibliographic.agent_role import AgentRole 

19from oc_ocdm.graph.entities.bibliographic.responsible_agent import ResponsibleAgent 

20from oc_ocdm.graph.entities.identifier import Identifier 

21 

22from oc_meta.lib.file_manager import find_rdf_file 

23from oc_meta.run.meta.generate_csv import load_json_from_file 

24 

25_forkserver_context = multiprocessing.get_context("forkserver") 

26 

27HAS_IDENTIFIER = "http://purl.org/spar/datacite/hasIdentifier" 

28USES_IDENTIFIER_SCHEME = "http://purl.org/spar/datacite/usesIdentifierScheme" 

29HAS_LITERAL_VALUE = ( 

30 "http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue" 

31) 

32IS_DOCUMENT_CONTEXT_FOR = "http://purl.org/spar/pro/isDocumentContextFor" 

33IS_HELD_BY = "http://purl.org/spar/pro/isHeldBy" 

34WITH_ROLE = "http://purl.org/spar/pro/withRole" 

35HAS_NEXT = "https://w3id.org/oc/ontology/hasNext" 

36FOAF_NAME = "http://xmlns.com/foaf/0.1/name" 

37GIVEN_NAME = "http://xmlns.com/foaf/0.1/givenName" 

38FAMILY_NAME = "http://xmlns.com/foaf/0.1/familyName" 

39PROV_SPECIALIZATION_OF = "http://www.w3.org/ns/prov#specializationOf" 

40DATACITE_PREFIX = "http://purl.org/spar/datacite/" 

41ROLE_MAP = { 

42 "http://purl.org/spar/pro/author": "author", 

43 "http://purl.org/spar/pro/editor": "editor", 

44 "http://purl.org/spar/pro/publisher": "publisher", 

45} 

46 

47Item = TypeVar("Item") 

48 

49 

50@dataclass(frozen=True, slots=True) 

51class AuditConfig: 

52 rdf_dir: str 

53 dir_split: int 

54 items_per_file: int 

55 zip_output: bool 

56 

57 

58@dataclass(frozen=True, slots=True) 

59class EntityFileLocator: 

60 rdf_dir: str 

61 dir_split: int 

62 items_per_file: int 

63 zip_output: bool 

64 

65 def path(self, uri: str) -> str: 

66 return find_rdf_file( 

67 uri, 

68 self.rdf_dir, 

69 self.dir_split, 

70 self.items_per_file, 

71 self.zip_output, 

72 ) 

73 

74 

75def values(entity: dict[str, object], predicate: str, key: str) -> list[str]: 

76 raw = entity[predicate] if predicate in entity else [] 

77 if isinstance(raw, dict): 

78 raw = [raw] 

79 if not isinstance(raw, list): 

80 return [] 

81 return [ 

82 value[key] 

83 for value in raw 

84 if isinstance(value, dict) and isinstance(value.get(key), str) 

85 ] 

86 

87 

88def ids(entity: dict[str, object], predicate: str) -> list[str]: 

89 return values(entity, predicate, "@id") 

90 

91 

92def literals(entity: dict[str, object], predicate: str) -> list[str]: 

93 return values(entity, predicate, "@value") 

94 

95 

96def first(values: list[str]) -> str: 

97 return values[0] if values else "" 

98 

99 

100def batches(items: list[Item], size: int) -> list[list[Item]]: 

101 return [items[start : start + size] for start in range(0, len(items), size)] 

102 

103 

104def sha256(path: str) -> str: 

105 digest = hashlib.sha256() 

106 with open(path, "rb") as stream: 

107 while chunk := stream.read(1024 * 1024): 

108 digest.update(chunk) 

109 return digest.hexdigest() 

110 

111 

112def ensure_parent(path: str) -> None: 

113 os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) 

114 

115 

116def write_json(path: str, value: object) -> None: 

117 ensure_parent(path) 

118 temporary_path = f"{path}.tmp" 

119 with open(temporary_path, "wb") as stream: 

120 stream.write( 

121 orjson.dumps(value, option=orjson.OPT_INDENT_2 | orjson.OPT_SORT_KEYS) 

122 ) 

123 stream.write(b"\n") 

124 os.replace(temporary_path, path) 

125 

126 

127def read_json_object(path: str) -> dict[str, object]: 

128 with open(path, "rb") as stream: 

129 value = orjson.loads(stream.read()) 

130 if not isinstance(value, dict): 

131 raise ValueError(f"Expected a JSON object in {path}") 

132 return cast(dict[str, object], value) 

133 

134 

135def load_entities(path: str) -> dict[str, dict[str, object]]: 

136 if path.endswith(".zip"): 

137 graphs = load_json_from_file(path) 

138 else: 

139 with open(path, "rb") as stream: 

140 graphs = orjson.loads(stream.read()) 

141 entities = {} 

142 for graph in graphs: 

143 for raw_entity in graph["@graph"]: 

144 entity = cast(dict[str, object], raw_entity) 

145 uri = entity["@id"] 

146 if isinstance(uri, str): 

147 entities[uri] = entity 

148 return entities 

149 

150 

151def data_files(directory: str, zip_output: bool) -> list[str]: 

152 extension = ".zip" if zip_output else ".json" 

153 paths = [] 

154 for root, _, filenames in os.walk(directory): 

155 relative_parts = os.path.relpath(root, directory).split(os.sep) 

156 if "prov" in relative_parts: 

157 continue 

158 for filename in filenames: 

159 if not filename.endswith(extension): 

160 continue 

161 paths.append(os.path.join(root, filename)) 

162 return sorted(paths) 

163 

164 

165def load_audit_config(path: str) -> AuditConfig: 

166 with open(path, encoding="utf-8") as stream: 

167 loaded = yaml.safe_load(stream) 

168 if not isinstance(loaded, dict): 

169 raise ValueError("Meta configuration must be a YAML mapping") 

170 config = cast(dict[str, object], loaded) 

171 output_key = "output_rdf_dir" if "output_rdf_dir" in config else "base_output_dir" 

172 output_dir = os.path.abspath(cast(str, config[output_key])) 

173 return AuditConfig( 

174 rdf_dir=os.path.join(output_dir, "rdf"), 

175 dir_split=cast(int, config["dir_split_number"]), 

176 items_per_file=cast(int, config["items_per_file"]), 

177 zip_output=cast(bool, config["zip_output_rdf"]), 

178 ) 

179 

180 

181def _load_target_batch( 

182 tasks: list[tuple[str, frozenset[str]]], 

183) -> dict[str, dict[str, object]]: 

184 result = {} 

185 for path, targets in tasks: 

186 if not os.path.exists(path): 

187 continue 

188 for uri, entity in load_entities(path).items(): 

189 if uri in targets: 

190 result[uri] = entity 

191 return result 

192 

193 

194def load_available_entities( 

195 uris: set[str], locator: EntityFileLocator, workers: int 

196) -> dict[str, dict[str, object]]: 

197 targets_by_path: dict[str, set[str]] = defaultdict(set) 

198 for uri in uris: 

199 targets_by_path[locator.path(uri)].add(uri) 

200 tasks = [(path, frozenset(targets)) for path, targets in targets_by_path.items()] 

201 result = {} 

202 with ProcessPoolExecutor( 

203 max_workers=workers, 

204 mp_context=_forkserver_context, 

205 ) as executor: 

206 for partial in executor.map(_load_target_batch, batches(tasks, 24)): 

207 result.update(partial) 

208 return result 

209 

210 

211def provenance_path(data_path: str, zip_output: bool) -> str: 

212 stem = os.path.splitext(data_path)[0] 

213 extension = "zip" if zip_output else "json" 

214 return os.path.join(stem, "prov", f"se.{extension}") 

215 

216 

217def snapshot_number(uri: str) -> int: 

218 value = uri.rsplit("/", 1)[-1] 

219 return int(value) if value.isdigit() else 0 

220 

221 

222def load_progress(path: str, plan_sha256: str, review_sha256: str) -> set[str]: 

223 if not os.path.exists(path): 

224 return set() 

225 progress = read_json_object(path) 

226 if progress["plan_sha256"] != plan_sha256: 

227 raise ValueError("Progress file belongs to a different correction plan") 

228 if progress["review_sha256"] != review_sha256: 

229 raise ValueError("Review decisions changed after execution started") 

230 completed = progress["completed_groups"] 

231 if not isinstance(completed, list) or not all( 

232 isinstance(group_id, str) for group_id in completed 

233 ): 

234 raise ValueError("Invalid completed_groups in progress file") 

235 return set(cast(list[str], completed)) 

236 

237 

238def save_progress( 

239 path: str, plan_sha256: str, review_sha256: str, completed: set[str] 

240) -> None: 

241 write_json( 

242 path, 

243 { 

244 "plan_sha256": plan_sha256, 

245 "review_sha256": review_sha256, 

246 "completed_groups": sorted(completed), 

247 }, 

248 ) 

249 

250 

251def responsible_agent(g_set: GraphSet, uri: str) -> ResponsibleAgent: 

252 entity = g_set.get_entity(uri) 

253 if not isinstance(entity, ResponsibleAgent): 

254 raise ValueError(f"Responsible agent not imported: {uri}") 

255 return entity 

256 

257 

258def identifier(g_set: GraphSet, uri: str) -> Identifier: 

259 entity = g_set.get_entity(uri) 

260 if not isinstance(entity, Identifier): 

261 raise ValueError(f"Identifier not imported: {uri}") 

262 return entity 

263 

264 

265def agent_role(g_set: GraphSet, uri: str) -> AgentRole: 

266 entity = g_set.get_entity(uri) 

267 if not isinstance(entity, AgentRole): 

268 raise ValueError(f"Agent role not imported: {uri}") 

269 return entity