Coverage for oc_meta / run / find / duplicates.py: 70%
287 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import argparse
6import csv
7import logging
8import multiprocessing as mp
9import os
10import shutil
11import tempfile
12import zipfile
13from collections import defaultdict
14from collections.abc import Iterable, Mapping, Sequence
16import orjson
17from rdflib import Dataset, URIRef
18from rich_argparse import RichHelpFormatter
19from tqdm import tqdm
21from oc_meta.lib.file_manager import collect_files
23ERROR_LOG_FILENAME = "error_log_find_duplicated_resources.txt"
24LOGGER = logging.getLogger(__name__)
26PathLikeString = str | os.PathLike[str]
27IdentifierKey = tuple[str, str]
30class UnionFind:
31 def __init__(self) -> None:
32 self.parent: dict[str, str] = {}
33 self.rank: dict[str, int] = {}
35 def find(self, item: str) -> str:
36 if item not in self.parent:
37 self.parent[item] = item
38 self.rank[item] = 0
39 return item
41 if self.parent[item] != item:
42 self.parent[item] = self.find(self.parent[item])
43 return self.parent[item]
45 def union(self, x: str, y: str) -> None:
46 xroot = self.find(x)
47 yroot = self.find(y)
49 if xroot == yroot:
50 return
52 if self.rank[xroot] < self.rank[yroot]:
53 self.parent[xroot] = yroot
54 elif self.rank[xroot] > self.rank[yroot]:
55 self.parent[yroot] = xroot
56 else:
57 self.parent[yroot] = xroot
58 self.rank[xroot] += 1
61def get_zip_files(folder_path: PathLikeString) -> list[str]:
62 return sorted(
63 collect_files(
64 os.fspath(folder_path),
65 pattern="*.zip",
66 path_filter=lambda p: os.path.basename(p) != "se.zip",
67 )
68 )
71def save_merge_rows_to_csv(
72 duplicates: Iterable[tuple[str, Sequence[str]]], csv_path: PathLikeString
73) -> None:
74 with open(csv_path, mode="w", newline="", encoding="utf-8") as csv_file:
75 csv_writer = csv.writer(csv_file)
76 csv_writer.writerow(["surviving_entity", "merged_entities"])
78 for surviving_entity, merged_group in duplicates:
79 csv_writer.writerow([surviving_entity, "; ".join(merged_group)])
82def process_identifier_zip_file(
83 zip_path: PathLikeString,
84) -> dict[IdentifierKey, set[str]]:
85 entity_info: defaultdict[IdentifierKey, set[str]] = defaultdict(set)
86 datacite_uses_identifier_scheme = URIRef(
87 "http://purl.org/spar/datacite/usesIdentifierScheme"
88 )
89 literal_reification_has_literal_value = URIRef(
90 "http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue"
91 )
93 try:
94 with zipfile.ZipFile(zip_path, "r") as zip_ref:
95 for zip_file in zip_ref.namelist():
96 try:
97 with zip_ref.open(zip_file) as rdf_file:
98 graph = Dataset(default_union=True)
99 graph.parse(data=rdf_file.read(), format="json-ld")
101 for subject, _, identifier_scheme_ref in graph.triples(
102 (None, datacite_uses_identifier_scheme, None)
103 ):
104 entity_id = str(subject)
105 identifier_scheme = str(identifier_scheme_ref)
106 literal_value = graph.value(
107 subject, literal_reification_has_literal_value
108 )
109 if identifier_scheme and literal_value:
110 key = (identifier_scheme, str(literal_value))
111 entity_info[key].add(entity_id)
112 except Exception as e:
113 print(f"Error processing file {zip_file} in {zip_path}: {str(e)}")
114 except zipfile.BadZipFile:
115 print(f"Corrupted or invalid ZIP file: {zip_path}")
116 except Exception as e:
117 print(f"Error opening ZIP file {zip_path}: {str(e)}")
119 return dict(entity_info)
122def save_identifier_chunk_to_temp_csv(
123 entity_info: Mapping[IdentifierKey, set[str]], temp_file_path: PathLikeString
124) -> None:
125 with open(temp_file_path, mode="w", newline="", encoding="utf-8") as csv_file:
126 csv_writer = csv.writer(csv_file)
127 csv_writer.writerow(["identifier_scheme", "literal_value", "entity_ids"])
128 for (scheme, value), ids in entity_info.items():
129 csv_writer.writerow([scheme, value, ";".join(sorted(ids))])
132def load_and_merge_identifier_temp_csv(
133 temp_file_path: PathLikeString, entity_info: defaultdict[IdentifierKey, set[str]]
134) -> None:
135 with open(temp_file_path, mode="r", encoding="utf-8") as csv_file:
136 csv_reader = csv.DictReader(csv_file)
137 for row in csv_reader:
138 key = (row["identifier_scheme"], row["literal_value"])
139 ids = set(row["entity_ids"].split(";"))
140 entity_info[key].update(ids)
143def process_identifier_chunk(
144 zip_files_chunk: Sequence[str], temp_dir: PathLikeString, chunk_index: int
145) -> str:
146 entity_info: defaultdict[IdentifierKey, set[str]] = defaultdict(set)
148 # Use forkserver to avoid deadlocks when forking in a multi-threaded environment.
149 ctx = mp.get_context("forkserver")
150 with ctx.Pool(processes=mp.cpu_count()) as pool:
151 results = pool.map(process_identifier_zip_file, zip_files_chunk)
153 for result in results:
154 for key, value in result.items():
155 entity_info[key].update(value)
157 temp_file_path = get_chunk_temp_file_path(temp_dir, chunk_index)
158 save_identifier_chunk_to_temp_csv(entity_info, temp_file_path)
160 return temp_file_path
163def get_chunk_temp_file_path(temp_dir: PathLikeString, chunk_index: int) -> str:
164 return os.path.join(temp_dir, f"chunk_{chunk_index}.csv")
167def find_duplicate_ids(
168 folder_path: PathLikeString, csv_path: PathLikeString, chunk_size: int = 5000
169) -> None:
170 id_folder_path = os.path.join(folder_path, "id")
172 if not os.path.exists(id_folder_path):
173 print(f"Error: The 'id' subfolder does not exist in path: {folder_path}")
174 return
176 zip_files = get_zip_files(id_folder_path)
177 output_dir = os.path.dirname(os.path.abspath(csv_path))
178 temp_dir = tempfile.mkdtemp(prefix="oc_meta_duplicates_", dir=output_dir)
180 try:
181 chunks = [
182 zip_files[i : i + chunk_size] for i in range(0, len(zip_files), chunk_size)
183 ]
184 temp_files = [
185 get_chunk_temp_file_path(temp_dir, chunk_index)
186 for chunk_index in range(len(chunks))
187 ]
189 print(
190 f"Processing {len(zip_files)} ZIP files in {len(chunks)} chunks of max {chunk_size} files each"
191 )
192 print(f"Temporary files will be stored in: {temp_dir}")
194 for chunk_index, chunk in enumerate(tqdm(chunks, desc="Processing chunks")):
195 process_identifier_chunk(chunk, temp_dir, chunk_index)
197 print("Merging chunk results...")
198 entity_info: defaultdict[IdentifierKey, set[str]] = defaultdict(set)
199 for temp_file in tqdm(temp_files, desc="Merging chunks"):
200 load_and_merge_identifier_temp_csv(temp_file, entity_info)
202 save_identifier_duplicates_to_csv(entity_info, csv_path)
204 finally:
205 shutil.rmtree(temp_dir, ignore_errors=True)
208def get_identifier_duplicate_rows(
209 entity_info: Mapping[IdentifierKey, set[str]],
210) -> list[tuple[str, list[str]]]:
211 duplicates = []
212 for ids in entity_info.values():
213 if len(ids) > 1:
214 ids_list = sorted(ids)
215 duplicates.append((ids_list[0], ids_list[1:]))
216 return duplicates
219def save_identifier_duplicates_to_csv(
220 entity_info: Mapping[IdentifierKey, set[str]], csv_path: PathLikeString
221) -> None:
222 try:
223 save_merge_rows_to_csv(get_identifier_duplicate_rows(entity_info), csv_path)
224 except Exception as e:
225 print(f"Error saving CSV file {csv_path}: {str(e)}")
228def find_duplicate_brs(folder_path: PathLikeString, csv_path: PathLikeString) -> None:
229 find_duplicate_resources_by_type(folder_path, csv_path, "br")
232def find_duplicate_ras(folder_path: PathLikeString, csv_path: PathLikeString) -> None:
233 find_duplicate_resources_by_type(folder_path, csv_path, "ra")
236def find_duplicate_resources_by_type(
237 folder_path: PathLikeString, csv_path: PathLikeString, resource_dir: str
238) -> None:
239 resources: dict[str, set[str]] = {}
240 qualities: dict[str, tuple[int, ...]] = {}
241 error_log_handler, error_log_path = configure_error_log(csv_path)
243 try:
244 entity_folder_path = os.path.join(folder_path, resource_dir)
245 process_entity_folder(entity_folder_path, resources, qualities, resource_dir)
247 save_entity_duplicates_to_csv(resources, csv_path, qualities)
248 finally:
249 close_error_log(error_log_handler, error_log_path)
252def configure_error_log(csv_path: PathLikeString) -> tuple[logging.FileHandler, str]:
253 error_log_path = os.path.join(
254 os.path.dirname(os.path.abspath(csv_path)), ERROR_LOG_FILENAME
255 )
256 handler = logging.FileHandler(error_log_path, mode="w", encoding="utf-8")
257 handler.setLevel(logging.ERROR)
258 handler.setFormatter(logging.Formatter("%(levelname)s - %(message)s"))
259 LOGGER.addHandler(handler)
260 LOGGER.setLevel(logging.ERROR)
261 return handler, error_log_path
264def close_error_log(
265 handler: logging.FileHandler, error_log_path: PathLikeString
266) -> None:
267 LOGGER.removeHandler(handler)
268 handler.close()
269 if os.path.exists(error_log_path) and os.path.getsize(error_log_path) == 0:
270 os.remove(error_log_path)
273def process_entity_folder(
274 folder_path: PathLikeString,
275 resources: dict[str, set[str]],
276 qualities: dict[str, tuple[int, ...]],
277 expected_type: str,
278) -> None:
279 if not os.path.exists(folder_path):
280 LOGGER.error(
281 f"La sottocartella '{expected_type}' non esiste nel percorso: {folder_path}"
282 )
283 return
285 zip_files = get_zip_files(folder_path)
287 for zip_path in tqdm(zip_files, desc=f"Analizzando i file ZIP in {expected_type}"):
288 try:
289 with zipfile.ZipFile(zip_path, "r") as zip_ref:
290 for zip_file in zip_ref.namelist():
291 try:
292 with zip_ref.open(zip_file) as json_file:
293 data = orjson.loads(json_file.read())
294 analyze_entity_json(
295 data,
296 resources,
297 qualities,
298 zip_path,
299 zip_file,
300 expected_type,
301 )
302 except orjson.JSONDecodeError:
303 LOGGER.error(
304 f"Errore nel parsing JSON del file {zip_file} in {zip_path}"
305 )
306 except Exception as e:
307 LOGGER.error(
308 f"Errore nell'elaborazione del file {zip_file} in {zip_path}: {str(e)}"
309 )
310 except zipfile.BadZipFile:
311 LOGGER.error(f"File ZIP corrotto o non valido: {zip_path}")
312 except Exception as e:
313 LOGGER.error(f"Errore nell'apertura del file ZIP {zip_path}: {str(e)}")
316def analyze_entity_json(
317 data,
318 resources: dict[str, set[str]],
319 qualities: dict[str, tuple[int, ...]],
320 zip_path: PathLikeString,
321 zip_file: str,
322 expected_type: str,
323) -> None:
324 for graph in data:
325 for entity in graph["@graph"]:
326 try:
327 entity_id = entity["@id"]
328 entity_type = get_entity_type(entity)
330 if entity_type is None:
331 print(
332 f"Tipo non specificato per l'entità {entity_id} nel file {zip_file} all'interno di {zip_path}. Assumendo tipo {expected_type}."
333 )
334 entity_type = expected_type
336 if entity_type == expected_type:
337 identifiers = get_identifiers(entity)
339 if entity_id not in resources:
340 resources[entity_id] = set()
341 resources[entity_id].update(identifiers)
342 qualities[entity_id] = get_entity_quality(entity, entity_type)
343 except KeyError as e:
344 entity_id = entity["@id"] if "@id" in entity else "ID sconosciuto"
345 LOGGER.error(
346 f"Chiave mancante nell'entità {entity_id} "
347 f"nel file {zip_file} all'interno di {zip_path}: {str(e)}"
348 )
349 except Exception as e:
350 entity_id = entity["@id"] if "@id" in entity else "ID sconosciuto"
351 LOGGER.error(
352 f"Errore nell'analisi dell'entità {entity_id} "
353 f"nel file {zip_file} all'interno di {zip_path}: {str(e)}"
354 )
357def get_entity_type(entity) -> str | None:
358 entity_types = entity["@type"] if "@type" in entity else []
359 if "http://purl.org/spar/fabio/Expression" in entity_types:
360 return "br"
361 if "http://xmlns.com/foaf/0.1/Agent" in entity_types:
362 return "ra"
363 return None
366def get_identifiers(entity) -> list[str]:
367 identifiers = []
368 predicate = "http://purl.org/spar/datacite/hasIdentifier"
369 if predicate not in entity:
370 return identifiers
372 for identifier in entity[predicate]:
373 if isinstance(identifier, dict) and "@id" in identifier:
374 identifiers.append(identifier["@id"])
375 return identifiers
378def get_entity_quality(entity, entity_type: str) -> tuple[int, ...]:
379 if entity_type == "br":
380 pub_date = get_literal(
381 entity, "http://prismstandard.org/namespaces/basic/2.0/publicationDate"
382 )
383 return (
384 int(bool(get_literal(entity, "http://purl.org/dc/terms/title"))),
385 len(pub_date),
386 int(bool(get_literal(entity, "http://purl.org/spar/fabio/hasSubtitle"))),
387 int(bool(get_uri(entity, "http://purl.org/vocab/frbr/core#partOf"))),
388 int(
389 bool(
390 get_literal(
391 entity, "http://purl.org/spar/fabio/hasSequenceIdentifier"
392 )
393 )
394 ),
395 int(bool(get_literal(entity, "http://purl.org/spar/fabio/hasEdition"))),
396 len(entity["@type"] if "@type" in entity else []),
397 )
398 return (
399 sum(
400 bool(get_literal(entity, predicate))
401 for predicate in [
402 "http://xmlns.com/foaf/0.1/name",
403 "http://xmlns.com/foaf/0.1/givenName",
404 "http://xmlns.com/foaf/0.1/familyName",
405 ]
406 ),
407 sum(
408 len(get_literal(entity, predicate))
409 for predicate in [
410 "http://xmlns.com/foaf/0.1/name",
411 "http://xmlns.com/foaf/0.1/givenName",
412 "http://xmlns.com/foaf/0.1/familyName",
413 ]
414 ),
415 )
418def get_literal(entity, predicate: str) -> str:
419 if predicate not in entity:
420 return ""
421 values = entity[predicate]
422 if isinstance(values, dict):
423 values = [values]
424 for value in values:
425 if isinstance(value, dict) and "@value" in value:
426 return value["@value"]
427 return ""
430def get_uri(entity, predicate: str) -> str:
431 if predicate not in entity:
432 return ""
433 values = entity[predicate]
434 if isinstance(values, dict):
435 values = [values]
436 for value in values:
437 if isinstance(value, dict) and "@id" in value:
438 return value["@id"]
439 return ""
442def save_entity_duplicates_to_csv(
443 resources: Mapping[str, set[str]],
444 csv_path: PathLikeString,
445 qualities: Mapping[str, tuple[int, ...]] | None = None,
446) -> None:
447 try:
448 save_merge_rows_to_csv(find_entity_duplicates(resources, qualities), csv_path)
449 except Exception as e:
450 LOGGER.error(f"Errore nel salvataggio del file CSV {csv_path}: {str(e)}")
453def find_entity_duplicates(
454 resources: Mapping[str, set[str]],
455 qualities: Mapping[str, tuple[int, ...]] | None = None,
456) -> list[tuple[str, list[str]]]:
457 union_find = UnionFind()
459 for entity, identifiers in resources.items():
460 for identifier in identifiers:
461 union_find.union(entity, identifier)
463 groups: dict[str, list[str]] = {}
464 for entity in resources:
465 representative = union_find.find(entity)
466 if representative not in groups:
467 groups[representative] = []
468 groups[representative].append(entity)
470 duplicate_groups = []
471 for group in groups.values():
472 if len(group) > 1:
473 surviving_entity = select_surviving_entity(group, qualities)
474 merged_entities = sorted(
475 entity for entity in group if entity != surviving_entity
476 )
477 duplicate_groups.append((surviving_entity, merged_entities))
478 return duplicate_groups
481def select_surviving_entity(
482 group: Sequence[str], qualities: Mapping[str, tuple[int, ...]] | None = None
483) -> str:
484 if qualities is None:
485 qualities = {}
487 def sort_key(entity: str):
488 quality = qualities[entity] if entity in qualities else (0,)
489 return tuple(-value for value in quality) + (entity,)
491 return min(group, key=sort_key)
494def add_folder_and_csv_arguments(
495 parser: argparse.ArgumentParser, folder_help: str
496) -> None:
497 parser.add_argument("folder_path", type=str, help=folder_help)
498 parser.add_argument(
499 "csv_path", type=str, help="Path to the CSV file to save duplicates"
500 )
503def build_argument_parser() -> argparse.ArgumentParser:
504 parser = argparse.ArgumentParser(
505 description="Find duplicate IDs, RAs, and BRs in RDF ZIP archives.",
506 formatter_class=RichHelpFormatter,
507 )
508 subparsers = parser.add_subparsers(dest="command", required=True)
510 ids_parser = subparsers.add_parser(
511 "ids",
512 help="Find duplicate IDs in the 'id' subfolder",
513 formatter_class=RichHelpFormatter,
514 )
515 add_folder_and_csv_arguments(
516 ids_parser, "Path to the folder containing the 'id' subfolder"
517 )
518 ids_parser.add_argument(
519 "--chunk-size",
520 type=int,
521 default=5000,
522 help="Number of ZIP files to process per chunk (default: 5000)",
523 )
525 ras_parser = subparsers.add_parser(
526 "ras",
527 help="Find duplicate responsible agents in the 'ra' subfolder",
528 formatter_class=RichHelpFormatter,
529 )
530 add_folder_and_csv_arguments(
531 ras_parser, "Path to the folder containing the 'ra' subfolder"
532 )
534 brs_parser = subparsers.add_parser(
535 "brs",
536 help="Find duplicate bibliographic resources in the 'br' subfolder",
537 formatter_class=RichHelpFormatter,
538 )
539 add_folder_and_csv_arguments(
540 brs_parser, "Path to the folder containing the 'br' subfolder"
541 )
543 return parser
546def main() -> None: # pragma: no cover
547 parser = build_argument_parser()
548 args = parser.parse_args()
550 if args.command == "ids":
551 find_duplicate_ids(args.folder_path, args.csv_path, args.chunk_size)
552 elif args.command == "ras":
553 find_duplicate_ras(args.folder_path, args.csv_path)
554 elif args.command == "brs":
555 find_duplicate_brs(args.folder_path, args.csv_path)
558if __name__ == "__main__": # pragma: no cover
559 main()