Coverage for oc_meta / run / merge / entities.py: 89%
180 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2024-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import argparse
6import csv
7import logging
8import os
9import tempfile
10from typing import Dict, List, Sequence, TypedDict
12from oc_graphenricher.deduplication import GraphDeduplicator
13from oc_graphenricher.storage import DirectoryStorage, directory_storage
14from oc_ocdm.counter_handler.filesystem_counter_handler import FilesystemCounterHandler
15from oc_ocdm.graph import GraphSet
16from rich_argparse import RichHelpFormatter
18from oc_meta.core.editor import MetaEditor
19from oc_meta.run.merge.closure import (
20 compute_identifier_merge_closure,
21 compute_related_closure,
22)
23from oc_meta.run.merge.csv_utils import parse_merged_entities
25logging.basicConfig(
26 level=logging.INFO,
27 format="%(asctime)s - %(levelname)s - %(message)s",
28 datefmt="%Y-%m-%d %H:%M:%S",
29)
30logger = logging.getLogger(__name__)
32REINDEX_SENTINEL_FILENAME = "reindex_required.out"
35class MergeRow(TypedDict):
36 surviving_entity: str
37 merged_entities: list[str]
40class EntityMerger:
41 def __init__(
42 self,
43 meta_config: str,
44 resp_agent: str,
45 entity_types: Sequence[str] = ("ra", "br", "id"),
46 stop_file_path: str = "stop.out",
47 ):
48 self.meta_config = meta_config
49 self.resp_agent = resp_agent
50 self.entity_types = entity_types
51 self.stop_file_path = stop_file_path
52 self.batch_size = 10
53 self.identifier_batch_size = 1000
55 @staticmethod
56 def get_entity_type(entity_url: str) -> str | None:
57 parts = entity_url.split("/")
58 if "oc" in parts and "meta" in parts:
59 try:
60 return parts[parts.index("meta") + 1]
61 except IndexError:
62 return None
63 return None
65 @staticmethod
66 def read_csv(csv_file: str) -> List[Dict]:
67 data = []
68 with open(csv_file, mode="r", newline="", encoding="utf-8") as file:
69 csv_reader = csv.DictReader(file)
70 for row in csv_reader:
71 if "Done" not in row:
72 row["Done"] = "False"
73 data.append(row)
74 return data
76 @staticmethod
77 def write_csv(csv_file: str, data: List[Dict]):
78 fieldnames = data[0].keys()
79 directory = os.path.dirname(os.path.abspath(csv_file))
80 with tempfile.NamedTemporaryFile(
81 mode="w", newline="", encoding="utf-8", dir=directory, delete=False
82 ) as file:
83 writer = csv.DictWriter(file, fieldnames=fieldnames)
84 writer.writeheader()
85 for row in data:
86 writer.writerow(row)
87 tmp_path = file.name
88 os.replace(tmp_path, csv_file)
90 @staticmethod
91 def count_csv_rows(csv_file: str) -> int:
92 with open(csv_file, "r", encoding="utf-8") as f:
93 return sum(1 for _ in f) - 1
95 @staticmethod
96 def build_merge_clusters(
97 rows_to_process: List[tuple[str, List[str]]],
98 ) -> Dict[str, List[str]]:
99 clusters: Dict[str, List[str]] = {}
100 for surviving_entity, merged_entities in rows_to_process:
101 if surviving_entity not in clusters:
102 clusters[surviving_entity] = []
103 for merged_entity in merged_entities:
104 if merged_entity not in clusters[surviving_entity]:
105 clusters[surviving_entity].append(merged_entity)
106 return clusters
108 @staticmethod
109 def create_storage(meta_editor: MetaEditor, g_set: GraphSet) -> DirectoryStorage:
110 return directory_storage(
111 meta_editor.base_dir,
112 items_per_directory=meta_editor.dir_split,
113 items_per_file=meta_editor.n_file_item,
114 supplier_prefix="",
115 zip_output=meta_editor.zip_output_rdf,
116 modified_entities=set(g_set.res_to_entity.keys()),
117 wanted_label=False,
118 counter_handler=meta_editor.counter_handler,
119 )
121 @staticmethod
122 def merge_clusters_and_save(
123 g_set: GraphSet,
124 storage: DirectoryStorage,
125 clusters: Dict[str, List[str]],
126 ) -> None:
127 deduplicator = GraphDeduplicator(g_set, storage=storage)
128 deduplicator.merge_clusters_and_save(clusters)
130 def should_stop_processing(self) -> bool:
131 return os.path.exists(self.stop_file_path)
133 def process_rows(self, rows: List[MergeRow]) -> None:
134 """Merge the given rows in one batch against a single triplestore snapshot"""
135 meta_editor = MetaEditor(self.meta_config, self.resp_agent, save_queries=True)
136 g_set = GraphSet(
137 meta_editor.base_iri, custom_counter_handler=meta_editor.counter_handler
138 )
140 surviving_entities = [row["surviving_entity"] for row in rows]
141 merged_entities = [merged for row in rows for merged in row["merged_entities"]]
142 logger.info(
143 f"Computing merge closure for {len(merged_entities)} merged entities and {len(surviving_entities)} surviving entities"
144 )
146 identifiers_only = (
147 bool(surviving_entities or merged_entities)
148 and all(
149 self.get_entity_type(entity) == "id" for entity in surviving_entities
150 )
151 and all(self.get_entity_type(entity) == "id" for entity in merged_entities)
152 )
153 if identifiers_only:
154 closure = compute_identifier_merge_closure(
155 meta_editor.endpoint,
156 surviving_entities,
157 merged_entities,
158 self.identifier_batch_size,
159 )
160 import_batch_size = self.identifier_batch_size
161 else:
162 closure = compute_related_closure(
163 meta_editor.endpoint,
164 set(surviving_entities) | set(merged_entities),
165 self.batch_size,
166 )
167 import_batch_size = self.batch_size
168 logger.info(f"Merge closure contains {len(closure)} entities")
170 entities_to_import = {
171 e for e in closure if not meta_editor.entity_cache.is_cached(e)
172 }
174 if entities_to_import:
175 logger.info(f"Importing {len(entities_to_import)} new entities")
176 meta_editor.reader.import_entities_from_triplestore(
177 g_set=g_set,
178 ts_url=meta_editor.endpoint,
179 entities=list(entities_to_import),
180 resp_agent=meta_editor.resp_agent,
181 enable_validation=False,
182 batch_size=import_batch_size,
183 )
184 for entity in entities_to_import:
185 meta_editor.entity_cache.add(entity)
186 logger.info("Entity import completed successfully")
188 clusters = self.build_merge_clusters(
189 [(row["surviving_entity"], row["merged_entities"]) for row in rows]
190 )
191 logger.info(
192 f"Merging {len(merged_entities)} entities in {len(clusters)} survivor clusters"
193 )
194 self.merge_clusters_and_save(
195 g_set,
196 self.create_storage(meta_editor, g_set),
197 clusters,
198 )
199 if isinstance(meta_editor.counter_handler, FilesystemCounterHandler):
200 meta_editor.counter_handler.flush()
202 logger.info(f"Successfully processed {len(merged_entities)} merges")
204 def process_file(self, csv_file: str) -> bool:
205 """Process a single CSV file of merge instructions.
207 Return True when merges were applied, False when there was nothing to
208 process or a stop file halted the run."""
209 logger.info(f"Starting to process file: {csv_file}")
210 data = self.read_csv(csv_file)
211 logger.info(f"Read {len(data)} rows from {csv_file}")
213 if self.should_stop_processing():
214 logger.info("Stop file detected, halting processing")
215 return False
217 rows: List[MergeRow] = [
218 {
219 "surviving_entity": row["surviving_entity"],
220 "merged_entities": parse_merged_entities(row["merged_entities"]),
221 }
222 for row in data
223 if row["Done"] != "True"
224 and self.get_entity_type(row["surviving_entity"]) in self.entity_types
225 ]
227 if not rows:
228 logger.info(f"No rows to process in {csv_file}")
229 return False
231 logger.info(f"Found {len(rows)} rows to process in {csv_file}")
232 self.process_rows(rows)
234 marked_done = 0
235 for row in data:
236 if (
237 row["Done"] != "True"
238 and self.get_entity_type(row["surviving_entity"]) in self.entity_types
239 ):
240 row["Done"] = "True"
241 marked_done += 1
243 logger.info(f"Marked {marked_done} rows as done")
244 self.write_csv(csv_file, data)
245 logger.info(f"Saved changes to {csv_file}")
247 return True
249 @staticmethod
250 def reindex_sentinel_path(csv_folder: str) -> str:
251 return os.path.join(csv_folder, REINDEX_SENTINEL_FILENAME)
253 @staticmethod
254 def reindex_sentinel_path_for_csv(csv_file: str) -> str:
255 return os.path.join(
256 os.path.dirname(os.path.abspath(csv_file)), REINDEX_SENTINEL_FILENAME
257 )
259 def process_path(self, csv_path: str) -> str | None:
260 if os.path.isfile(csv_path):
261 return self.process_single_csv(csv_path)
262 return self.process_folder(csv_path)
264 def process_single_csv(self, csv_file: str) -> str | None:
265 if os.path.exists(self.stop_file_path):
266 os.remove(self.stop_file_path)
268 sentinel_path = self.reindex_sentinel_path_for_csv(csv_file)
269 if os.path.exists(sentinel_path):
270 raise RuntimeError(
271 f"{sentinel_path} exists: a previous run already merged a CSV file "
272 "on top of the current triplestore snapshot. Re-index the "
273 "triplestore from the RDF files, then delete the sentinel to "
274 "process another CSV file."
275 )
277 if self.process_file(csv_file):
278 with open(sentinel_path, "w", encoding="utf-8") as sentinel:
279 sentinel.write(
280 f"{csv_file} was merged on top of the current triplestore "
281 "snapshot.\nRe-index the triplestore from the RDF files, "
282 "then delete this file to let another merge run start.\n"
283 )
284 logger.info(
285 f"Processed {csv_file}. Re-index the triplestore from the RDF "
286 f"files, then delete {sentinel_path} to process another CSV file."
287 )
288 return csv_file
290 return None
292 def process_folder(self, csv_folder: str) -> str | None:
293 """Process the first CSV file with pending rows, then require a re-index.
295 Every merge batch is computed against one triplestore snapshot, so a
296 second file must not be processed until the triplestore has been
297 re-indexed from the RDF files this run wrote. After a file is merged a
298 sentinel is created next to the CSVs and the next invocation refuses to
299 start until the sentinel is removed. Return the processed file, or None
300 when no file had pending rows."""
301 if os.path.exists(self.stop_file_path):
302 os.remove(self.stop_file_path)
304 sentinel_path = self.reindex_sentinel_path(csv_folder)
305 if os.path.exists(sentinel_path):
306 raise RuntimeError(
307 f"{sentinel_path} exists: a previous run already merged a CSV file "
308 "on top of the current triplestore snapshot. Re-index the "
309 "triplestore from the RDF files, then delete the sentinel to "
310 "process the next CSV file."
311 )
313 csv_files = sorted(
314 os.path.join(csv_folder, file)
315 for file in os.listdir(csv_folder)
316 if file.endswith(".csv")
317 )
319 for csv_file in csv_files:
320 if self.should_stop_processing():
321 break
322 if self.process_file(csv_file):
323 with open(sentinel_path, "w", encoding="utf-8") as sentinel:
324 sentinel.write(
325 f"{csv_file} was merged on top of the current triplestore "
326 "snapshot.\nRe-index the triplestore from the RDF files, "
327 "then delete this file to let the next merge run start.\n"
328 )
329 logger.info(
330 f"Processed {csv_file}. Re-index the triplestore from the RDF "
331 f"files, then delete {sentinel_path} to process the next file."
332 )
333 return csv_file
335 logger.info("No CSV files with pending rows found")
336 return None
339def main():
340 parser = argparse.ArgumentParser(
341 description=(
342 "Merge entities from a CSV file or from the first pending CSV file "
343 "in a folder. One file is processed per run: re-index the "
344 f"triplestore from the RDF files and delete {REINDEX_SENTINEL_FILENAME} "
345 "before the next run."
346 ),
347 formatter_class=RichHelpFormatter,
348 )
349 parser.add_argument(
350 "csv_path", type=str, help="Path to a merge CSV file or folder of CSV files"
351 )
352 parser.add_argument("meta_config", type=str, help="Meta configuration string")
353 parser.add_argument("resp_agent", type=str, help="Responsible agent string")
354 parser.add_argument(
355 "--entity_types",
356 nargs="+",
357 default=["ra", "br", "id"],
358 help="Types of entities to merge (ra, br, id)",
359 )
360 parser.add_argument(
361 "--stop_file", type=str, default="stop.out", help="Path to the stop file"
362 )
364 args = parser.parse_args()
366 merger = EntityMerger(
367 meta_config=args.meta_config,
368 resp_agent=args.resp_agent,
369 entity_types=args.entity_types,
370 stop_file_path=args.stop_file,
371 )
373 merger.process_path(args.csv_path)
376if __name__ == "__main__":
377 main()