Coverage for oc_meta / run / migration / rdf_from_export.py: 0%
202 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1#!/usr/bin/python
3# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
4#
5# SPDX-License-Identifier: ISC
7import argparse
8import gzip
9import logging
10import multiprocessing
11import os
12import re
13import time
14import uuid
15from zipfile import ZIP_DEFLATED, ZipFile
17import orjson
18from rdflib import Dataset
19from rdflib.exceptions import ParserError
20from oc_ocdm.support.support import find_paths
21from rich_argparse import RichHelpFormatter
22from tqdm import tqdm
25def store(triples, graph_identifier, stored_g: Dataset) -> Dataset:
26 for triple in triples:
27 stored_g.add((triple[0], triple[1], triple[2], graph_identifier))
28 return stored_g
31def store_in_file(cur_g: Dataset, cur_file_path: str, zip_output: bool) -> None:
32 dir_path = os.path.dirname(cur_file_path)
33 if not os.path.exists(dir_path):
34 os.makedirs(dir_path, exist_ok=True)
36 cur_json_ld = orjson.loads(cur_g.serialize(format="json-ld"))
38 if zip_output:
39 with ZipFile(
40 cur_file_path, mode="w", compression=ZIP_DEFLATED, allowZip64=True
41 ) as zip_file:
42 json_str = orjson.dumps(cur_json_ld).decode("utf-8")
43 zip_file.writestr(
44 os.path.basename(cur_file_path.replace(".zip", ".json")), json_str
45 )
46 else:
47 with open(cur_file_path, "wb") as f:
48 f.write(orjson.dumps(cur_json_ld))
51def load_graph(file_path: str, cur_format: str = "json-ld"):
52 loaded_graph = Dataset(default_union=True)
53 if file_path.endswith(".zip"):
54 with ZipFile(
55 file=file_path, mode="r", compression=ZIP_DEFLATED, allowZip64=True
56 ) as archive:
57 for zf_name in archive.namelist():
58 with archive.open(zf_name) as f:
59 if cur_format == "json-ld":
60 json_ld_file = orjson.loads(f.read())
61 if isinstance(json_ld_file, dict):
62 json_ld_file = [json_ld_file]
63 for json_ld_resource in json_ld_file:
64 loaded_graph.parse(
65 data=orjson.dumps(json_ld_resource).decode("utf-8"),
66 format=cur_format,
67 )
68 else:
69 loaded_graph.parse(file=f, format=cur_format) # type: ignore[arg-type]
70 else:
71 with open(file_path, "rb") as f:
72 if cur_format == "json-ld":
73 json_ld_file = orjson.loads(f.read())
74 if isinstance(json_ld_file, dict):
75 json_ld_file = [json_ld_file]
76 for json_ld_resource in json_ld_file:
77 loaded_graph.parse(
78 data=orjson.dumps(json_ld_resource).decode("utf-8"),
79 format=cur_format,
80 )
81 else:
82 loaded_graph.parse(file=f, format=cur_format) # type: ignore[arg-type]
84 return loaded_graph
87def process_graph(
88 context, graph_identifier, output_root, base_iri, file_limit, item_limit, zip_output
89):
90 modifications_by_file = {}
91 triples = 0
92 unique_id = generate_unique_id()
94 for triple in context:
95 triples += len(triple)
96 entity_uri = triple[0]
97 _, cur_file_path = find_paths(
98 entity_uri, output_root, base_iri, "_", file_limit, item_limit, True
99 )
100 if cur_file_path is None:
101 logging.warning(f"Skipping triple due to invalid URI: {entity_uri}")
102 continue
104 # Estrai il nome base del file (numero) e aggiungi l'ID unico
105 base_name = os.path.splitext(os.path.basename(cur_file_path))[0]
106 new_file_name = f"{base_name}_{unique_id}"
108 cur_file_path = os.path.join(
109 os.path.dirname(cur_file_path),
110 new_file_name + (".zip" if zip_output else ".json"),
111 )
113 if cur_file_path not in modifications_by_file:
114 modifications_by_file[cur_file_path] = {
115 "graph_identifier": graph_identifier,
116 "triples": [],
117 }
118 modifications_by_file[cur_file_path]["triples"].append(triple)
120 for file_path, data in modifications_by_file.items():
121 stored_g = load_graph(file_path) if os.path.exists(file_path) else Dataset()
122 stored_g = store(data["triples"], data["graph_identifier"], stored_g)
123 store_in_file(stored_g, file_path, zip_output)
124 return triples
127def merge_files(output_root, base_file_name, file_extension, zip_output):
128 """Funzione per fondere i file generati dai diversi processi"""
129 files_to_merge = [
130 f
131 for f in os.listdir(output_root)
132 if f.startswith(base_file_name) and f.endswith(file_extension)
133 ]
135 merged_graph = Dataset()
137 for file_path in files_to_merge:
138 cur_full_path = os.path.join(output_root, file_path)
139 loaded_graph = load_graph(cur_full_path)
140 merged_graph += loaded_graph
142 final_file_path = os.path.join(output_root, base_file_name + file_extension)
143 store_in_file(merged_graph, final_file_path, zip_output) # type: ignore[arg-type]
146def merge_files_in_directory(directory, zip_output, stop_file):
147 """Function to merge files in a specific directory"""
148 if check_stop_file(stop_file):
149 logging.info("Stop file detected. Stopping merge process.")
150 return
152 files = [
153 f
154 for f in os.listdir(directory)
155 if f.endswith(".zip" if zip_output else ".json")
156 ]
158 # Group files by their base name (number without the unique ID)
159 file_groups = {}
160 for file in files:
161 match = re.match(r"^((?:\d+)|(?:se))(?:_[^.]+)?\.", file)
162 if match:
163 base_name = match.group(1)
164 if base_name not in file_groups:
165 file_groups[base_name] = []
166 file_groups[base_name].append(file)
168 for base_file_name, files_to_merge in file_groups.items():
169 if check_stop_file(stop_file):
170 logging.info("Stop file detected. Stopping merge process.")
171 return
173 # Only proceed with merging if there's at least one file with an underscore
174 if not any("_" in file for file in files_to_merge):
175 continue
177 merged_graph = Dataset()
179 for file_path in files_to_merge:
180 cur_full_path = os.path.join(directory, file_path)
181 loaded_graph = load_graph(cur_full_path)
182 for context in loaded_graph.graphs():
183 graph_identifier = context.identifier
184 for triple in context:
185 merged_graph.add(triple + (graph_identifier,)) # type: ignore[arg-type]
187 final_file_path = os.path.join(
188 directory, f"{base_file_name}" + (".zip" if zip_output else ".json")
189 )
190 store_in_file(merged_graph, final_file_path, zip_output)
192 # Remove the original files after merging
193 for file_path in files_to_merge:
194 if file_path != os.path.basename(final_file_path):
195 os.remove(os.path.join(directory, file_path))
198def generate_unique_id():
199 return f"{int(time.time())}-{uuid.uuid4()}"
202def merge_files_wrapper(args):
203 directory, zip_output, stop_file = args
204 merge_files_in_directory(directory, zip_output, stop_file)
207def merge_all_files_parallel(output_root, zip_output, stop_file):
208 """Function to merge files in parallel"""
209 if check_stop_file(stop_file):
210 logging.info("Stop file detected. Stopping merge process.")
211 return
213 directories_to_process = []
214 for root, dirs, files in os.walk(output_root):
215 if any(f.endswith(".zip" if zip_output else ".json") for f in files):
216 directories_to_process.append(root)
218 with multiprocessing.Pool() as pool:
219 list(
220 tqdm(
221 pool.imap(
222 merge_files_wrapper,
223 [(dir, zip_output, stop_file) for dir in directories_to_process],
224 ),
225 total=len(directories_to_process),
226 desc="Merging files in directories",
227 )
228 )
231def process_file_content(
232 file_path, output_root, base_iri, file_limit, item_limit, zip_output, rdf_format
233):
234 with gzip.open(file_path, "rb") as f:
235 content = f.read()
236 assert isinstance(content, bytes)
237 data = content.decode("utf-8")
238 graph = Dataset()
239 try:
240 graph.parse(data=data, format=rdf_format)
241 except ParserError as e:
242 logging.error(f"Failed to parse {file_path}: {e}")
243 return
245 for context in graph.graphs():
246 graph_identifier = context.identifier
247 process_graph(
248 context,
249 graph_identifier,
250 output_root,
251 base_iri,
252 file_limit,
253 item_limit,
254 zip_output,
255 )
258def process_file_wrapper(args):
259 (
260 file_path,
261 output_root,
262 base_iri,
263 file_limit,
264 item_limit,
265 zip_output,
266 rdf_format,
267 cache_file,
268 stop_file,
269 ) = args
270 if check_stop_file(stop_file):
271 return
272 if not is_file_processed(file_path, cache_file):
273 process_file_content(
274 file_path,
275 output_root,
276 base_iri,
277 file_limit,
278 item_limit,
279 zip_output,
280 rdf_format,
281 )
282 mark_file_as_processed(file_path, cache_file)
285def process_chunk(
286 chunk,
287 output_root,
288 base_iri,
289 file_limit,
290 item_limit,
291 zip_output,
292 rdf_format,
293 cache_file,
294 stop_file,
295):
296 with multiprocessing.Pool() as pool:
297 list(
298 tqdm(
299 pool.imap(
300 process_file_wrapper,
301 [
302 (
303 file_path,
304 output_root,
305 base_iri,
306 file_limit,
307 item_limit,
308 zip_output,
309 rdf_format,
310 cache_file,
311 stop_file,
312 )
313 for file_path in chunk
314 ],
315 ),
316 total=len(chunk),
317 desc="Processing files",
318 )
319 )
322def create_cache_file(cache_file):
323 if cache_file:
324 if not os.path.exists(cache_file):
325 with open(cache_file, "w", encoding="utf8"):
326 pass # Create an empty file
327 else:
328 logging.info("No cache file specified. Skipping cache creation.")
331def is_file_processed(file_path, cache_file):
332 if not cache_file:
333 return False
334 with open(cache_file, "r", encoding="utf8") as f:
335 processed_files = f.read().splitlines()
336 return file_path in processed_files
339def mark_file_as_processed(file_path, cache_file):
340 if cache_file:
341 with open(cache_file, "a", encoding="utf8") as f:
342 f.write(f"{file_path}\n")
343 else:
344 logging.debug(
345 f"No cache file specified. Skipping marking {file_path} as processed."
346 )
349def check_stop_file(stop_file):
350 return os.path.exists(stop_file)
353def main():
354 parser = argparse.ArgumentParser(
355 description="Process gzipped input files into OC Meta RDF",
356 formatter_class=RichHelpFormatter,
357 )
358 parser.add_argument(
359 "input_folder", type=str, help="Input folder containing gzipped input files"
360 )
361 parser.add_argument(
362 "output_root", type=str, help="Root folder for output OC Meta RDF files"
363 )
364 parser.add_argument(
365 "--base_iri",
366 type=str,
367 default="https://w3id.org/oc/meta/",
368 help="The base URI of entities on Meta",
369 )
370 parser.add_argument(
371 "--file_limit", type=int, default=10000, help="Number of files per folder"
372 )
373 parser.add_argument(
374 "--item_limit", type=int, default=1000, help="Number of items per file"
375 )
376 parser.add_argument(
377 "-v",
378 "--zip_output",
379 default=True,
380 dest="zip_output",
381 action="store_true",
382 help="Zip output json files",
383 )
384 parser.add_argument(
385 "--input_format",
386 type=str,
387 default="jsonld",
388 choices=["jsonld", "nquads"],
389 help="Format of the input files",
390 )
391 parser.add_argument(
392 "--chunk_size",
393 type=int,
394 default=1000,
395 help="Number of files to process before merging",
396 )
397 parser.add_argument(
398 "--cache_file",
399 type=str,
400 default=None,
401 help="File to store processed file names (optional)",
402 )
403 parser.add_argument(
404 "--stop_file",
405 type=str,
406 default="./.stop",
407 help="File to signal process termination",
408 )
409 args = parser.parse_args()
411 create_cache_file(args.cache_file)
413 file_extension = ".nq.gz" if args.input_format == "nquads" else ".jsonld.gz"
414 rdf_format = "nquads" if args.input_format == "nquads" else "json-ld"
416 files_to_process = [
417 os.path.join(args.input_folder, file)
418 for file in os.listdir(args.input_folder)
419 if file.endswith(file_extension)
420 ]
421 chunks = [
422 files_to_process[i : i + args.chunk_size]
423 for i in range(0, len(files_to_process), args.chunk_size)
424 ]
425 for i, chunk in enumerate(tqdm(chunks, desc="Processing chunks")):
426 if check_stop_file(args.stop_file):
427 logging.info("Stop file detected. Gracefully terminating the process.")
428 break
429 logging.info(f"Processing chunk {i + 1}/{len(chunks)}")
430 process_chunk(
431 chunk,
432 args.output_root,
433 args.base_iri,
434 args.file_limit,
435 args.item_limit,
436 args.zip_output,
437 rdf_format,
438 args.cache_file,
439 args.stop_file,
440 )
441 logging.info(f"Merging files for chunk {i + 1}")
442 merge_all_files_parallel(args.output_root, args.zip_output, args.stop_file)
444 logging.info("Processing complete")
447if __name__ == "__main__":
448 main()