Coverage for oc_meta / run / meta_process.py: 72%
307 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2019 Silvio Peroni <silvio.peroni@unibo.it>
2# SPDX-FileCopyrightText: 2019-2020 Fabio Mariani <fabio.mariani555@gmail.com>
3# SPDX-FileCopyrightText: 2021 Simone Persiani <iosonopersia@gmail.com>
4# SPDX-FileCopyrightText: 2021-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
5#
6# SPDX-License-Identifier: ISC
8from __future__ import annotations
10import bisect
11import csv
12import multiprocessing
13import os
14import sys
15import traceback
16from argparse import ArgumentParser
17from datetime import datetime
18from sys import executable, platform
19from typing import Any, Dict, List, Optional, Tuple
21import orjson
22import yaml
23from oc_ocdm import Storer
24from oc_ocdm.counter_handler.filesystem_counter_handler import FilesystemCounterHandler
25from oc_ocdm.prov import ProvSet
26from oc_ocdm.support.reporter import Reporter
27from piccione.upload.on_triplestore import upload_sparql_updates
28from rich_argparse import RichHelpFormatter
29from time_agnostic_library.support import generate_config_file
31from oc_meta.core.creator import Creator
32from oc_meta.core.curator import Curator
33from oc_meta.lib.console import console, create_progress
34from oc_meta.lib.file_manager import (
35 get_csv_data,
36 init_cache,
37 normalize_path,
38 pathoo,
39 sort_files,
40)
41from oc_meta.lib.timer import ProcessTimer
42from oc_meta.run.benchmark.plotting import plot_incremental_progress
45def _upload_to_triplestore(
46 endpoint: str,
47 folder: str,
48 redis_host: str | None,
49 redis_port: int | None,
50 redis_db: int | None,
51 failed_file: str,
52 stop_file: str,
53 description: str = "Processing files",
54 show_progress: bool = False,
55) -> None:
56 try:
57 if redis_host is None:
58 upload_sparql_updates(
59 endpoint=endpoint,
60 folder=folder,
61 failed_file=failed_file,
62 stop_file=stop_file,
63 description=description,
64 show_progress=show_progress,
65 )
66 return
68 if redis_port is None or redis_db is None:
69 raise ValueError(
70 "redis_port and redis_cache_db are required when redis_host is set"
71 )
73 upload_sparql_updates(
74 endpoint=endpoint,
75 folder=folder,
76 failed_file=failed_file,
77 stop_file=stop_file,
78 redis_host=redis_host,
79 redis_port=redis_port,
80 redis_db=redis_db,
81 description=description,
82 show_progress=show_progress,
83 )
84 except Exception as e:
85 console.print(f"[red]Upload to {endpoint} failed: {e}[/red]")
86 sys.exit(1)
89def _generate_queries_worker(
90 storer: Storer, triplestore_url: str, base_dir: str
91) -> None:
92 storer.upload_all(
93 triplestore_url=triplestore_url,
94 base_dir=base_dir,
95 batch_size=10,
96 )
99def _store_rdf_worker(storer: Storer, base_dir, base_iri):
100 storer.store_all(
101 base_dir=base_dir,
102 base_iri=base_iri,
103 )
106class MetaProcess:
107 def __init__(
108 self,
109 settings: dict,
110 meta_config_path: str,
111 timer: Optional[ProcessTimer] = None,
112 ):
113 self.settings = settings
114 # Mandatory settings
115 self.triplestore_url = settings["triplestore_url"] # Main triplestore for data
116 self.provenance_triplestore_url = settings[
117 "provenance_triplestore_url"
118 ] # Separate triplestore for provenance
119 self.input_csv_dir = normalize_path(settings["input_csv_dir"])
120 self.base_output_dir = normalize_path(settings["base_output_dir"])
121 self.resp_agent = settings["resp_agent"]
122 self.output_csv_dir = os.path.join(self.base_output_dir, "csv")
123 self.output_rdf_dir = (
124 normalize_path(settings["output_rdf_dir"]) + os.sep + "rdf" + os.sep
125 )
126 self.cache_path = os.path.join(self.base_output_dir, "cache.txt")
127 self.errors_path = os.path.join(self.base_output_dir, "errors.txt")
128 self.timer = timer or ProcessTimer(enabled=False)
129 # Optional settings
130 self.base_iri = settings["base_iri"]
131 self.dir_split_number = settings["dir_split_number"]
132 self.items_per_file = settings["items_per_file"]
133 self.default_dir = settings["default_dir"]
134 self.zip_output_rdf = settings["zip_output_rdf"]
135 self.source = settings["source"]
136 supplier_prefix: str = settings["supplier_prefix"]
137 self.supplier_prefix = (
138 supplier_prefix if supplier_prefix.endswith("0") else f"{supplier_prefix}0"
139 )
140 self.silencer = settings["silencer"]
141 self.rdf_files_only = settings.get("rdf_files_only", False)
142 # Time-Agnostic_library integration
143 self.time_agnostic_library_config = os.path.join(
144 os.path.dirname(meta_config_path), "time_agnostic_library_config.json"
145 )
146 if not os.path.exists(self.time_agnostic_library_config):
147 generate_config_file(
148 config_path=self.time_agnostic_library_config,
149 dataset_urls=[self.triplestore_url],
150 dataset_dirs=list(),
151 provenance_urls=[self.provenance_triplestore_url]
152 if self.provenance_triplestore_url
153 not in settings["provenance_endpoints"]
154 else settings["provenance_endpoints"],
155 provenance_dirs=list(),
156 blazegraph_full_text_search=settings["blazegraph_full_text_search"],
157 fuseki_full_text_search=settings["fuseki_full_text_search"],
158 virtuoso_full_text_search=settings["virtuoso_full_text_search"],
159 graphdb_connector_name=settings["graphdb_connector_name"],
160 )
162 info_dir = (
163 os.path.join(self.base_output_dir, "info_dir", self.supplier_prefix)
164 + os.sep
165 )
166 self.counter_handler = FilesystemCounterHandler(
167 info_dir=info_dir, supplier_prefix=self.supplier_prefix
168 )
170 self.redis_host = settings.get("redis_host")
171 self.redis_port = settings.get("redis_port")
172 self.redis_cache_db = settings.get("redis_cache_db")
174 # Triplestore upload settings
175 self.ts_failed_queries = settings.get("ts_failed_queries", "failed_queries.txt")
176 self.ts_stop_file = settings.get("ts_stop_file", ".stop_upload")
178 self.data_update_dir = os.path.join(self.base_output_dir, "to_be_uploaded_data")
179 self.prov_update_dir = os.path.join(self.base_output_dir, "to_be_uploaded_prov")
181 def prepare_folders(self) -> List[str]:
182 completed = init_cache(self.cache_path)
183 files_in_input_csv_dir = {
184 filename
185 for filename in os.listdir(self.input_csv_dir)
186 if filename.endswith(".csv")
187 }
188 files_to_be_processed = sort_files(
189 list(files_in_input_csv_dir.difference(completed))
190 )
191 pathoo(self.output_csv_dir)
192 csv.field_size_limit(128)
193 return files_to_be_processed
195 def curate_and_create(
196 self,
197 filename: str,
198 cache_path: str,
199 errors_path: str,
200 settings: dict | None = None,
201 meta_config_path: str | None = None,
202 progress=None,
203 ) -> Tuple[dict, str, str, str]:
204 try:
205 with self.timer.timer("total_processing"):
206 filepath = os.path.join(self.input_csv_dir, filename)
207 console.print(filepath)
208 data = get_csv_data(filepath)
209 self.timer.record_metric("input_records", len(data))
211 min_rows_parallel = (
212 settings.get("min_rows_parallel", 1000) if settings else 1000
213 )
214 curator_obj = Curator(
215 data=data,
216 ts=self.triplestore_url,
217 prov_config=self.time_agnostic_library_config,
218 counter_handler=self.counter_handler,
219 base_iri=self.base_iri,
220 prefix=self.supplier_prefix,
221 settings=settings,
222 silencer=self.silencer,
223 meta_config_path=meta_config_path,
224 timer=self.timer,
225 progress=progress,
226 min_rows_parallel=min_rows_parallel,
227 )
228 name = f"{filename.replace('.csv', '')}_{datetime.now().strftime('%Y-%m-%dT%H-%M-%S')}"
229 curator_obj.curator(filename=name, path_csv=self.output_csv_dir)
230 self.timer.record_metric("curated_records", len(curator_obj.data))
232 local_g_size = len(curator_obj.finder.graph)
233 self.timer.record_metric("local_g_triples", local_g_size)
234 preexisting_count = len(curator_obj.preexisting_entities)
235 self.timer.record_metric(
236 "preexisting_entities_count", preexisting_count
237 )
239 RDF_BATCH_SIZE = 100_000
240 data = curator_obj.data
241 n_batches = (len(data) + RDF_BATCH_SIZE - 1) // RDF_BATCH_SIZE
242 total_entities = 0
243 total_modified = 0
245 batch_task_id = None
246 if progress is not None and n_batches > 1:
247 batch_task_id = progress.add_task(
248 f" [cyan]RDF batches[/cyan] ({filename})",
249 total=n_batches,
250 )
252 for batch_idx in range(n_batches):
253 batch_start = batch_idx * RDF_BATCH_SIZE
254 batch_end = min(batch_start + RDF_BATCH_SIZE, len(data))
255 batch_data = data[batch_start:batch_end]
257 with self.timer.timer("rdf_creation"):
258 creator_obj = Creator(
259 data=batch_data,
260 finder=curator_obj.finder,
261 base_iri=self.base_iri,
262 counter_handler=self.counter_handler,
263 supplier_prefix=self.supplier_prefix,
264 resp_agent=self.resp_agent,
265 ra_index=curator_obj.index_id_ra,
266 br_index=curator_obj.index_id_br,
267 re_index_csv=curator_obj.re_index,
268 ar_index_csv=curator_obj.ar_index,
269 vi_index=curator_obj.VolIss,
270 silencer=self.silencer,
271 progress=progress,
272 )
273 creator = creator_obj.creator(source=self.source)
274 total_entities += sum(
275 1
276 for e in creator.res_to_entity.values()
277 if not e._preexisting_triples
278 )
280 prov = ProvSet(
281 creator,
282 self.base_iri,
283 wanted_label=False,
284 supplier_prefix=self.supplier_prefix,
285 custom_counter_handler=self.counter_handler,
286 )
287 modified_entities = prov.generate_provenance()
288 total_modified += len(modified_entities)
290 repok = Reporter(print_sentences=False)
291 reperr = Reporter(print_sentences=True, prefix="[Storer: ERROR] ")
292 res_storer = Storer(
293 abstract_set=creator,
294 repok=repok,
295 reperr=reperr,
296 dir_split=self.dir_split_number,
297 n_file_item=self.items_per_file,
298 default_dir=self.default_dir,
299 output_format="json-ld",
300 zip_output=self.zip_output_rdf,
301 modified_entities=modified_entities,
302 )
303 prov_storer = Storer(
304 abstract_set=prov,
305 repok=repok,
306 reperr=reperr,
307 dir_split=self.dir_split_number,
308 n_file_item=self.items_per_file,
309 output_format="json-ld",
310 zip_output=self.zip_output_rdf,
311 modified_entities=modified_entities,
312 )
313 self.store_data_and_prov(res_storer, prov_storer)
314 del (
315 creator_obj,
316 creator,
317 prov,
318 res_storer,
319 prov_storer,
320 modified_entities,
321 )
323 if progress is not None and batch_task_id is not None:
324 progress.update(batch_task_id, advance=1)
326 if progress is not None and batch_task_id is not None:
327 progress.remove_task(batch_task_id)
329 self.timer.record_metric("new_entities", total_entities)
330 self.timer.record_metric("modified_entities", total_modified)
332 return {"message": "success"}, cache_path, errors_path, filename
333 except Exception as e:
334 tb = traceback.format_exc()
335 template = (
336 "An exception of type {0} occurred. Arguments:\n{1!r}\nTraceback:\n{2}"
337 )
338 message = template.format(type(e).__name__, e.args, tb)
339 return {"message": message}, cache_path, errors_path, filename
341 def _setup_output_directories(self) -> None:
342 """Create output directories for data and provenance."""
343 os.makedirs(self.data_update_dir, exist_ok=True)
344 os.makedirs(self.prov_update_dir, exist_ok=True)
346 def _upload_sparql_queries(self) -> None:
347 """Upload SPARQL queries to triplestores in parallel."""
348 data_upload_folder = os.path.join(self.data_update_dir, "to_be_uploaded")
349 prov_upload_folder = os.path.join(self.prov_update_dir, "to_be_uploaded")
351 # Use forkserver to avoid deadlocks when forking from a multi-threaded process.
352 # Libraries like Redis and rdflib create background threads, and fork() would
353 # copy locked mutexes into the child process, causing hangs.
354 ctx = multiprocessing.get_context("forkserver")
356 data_process = ctx.Process(
357 target=_upload_to_triplestore,
358 args=(
359 self.triplestore_url,
360 data_upload_folder,
361 self.redis_host,
362 self.redis_port,
363 self.redis_cache_db,
364 self.ts_failed_queries,
365 self.ts_stop_file,
366 "Uploading data SPARQL",
367 ),
368 )
370 prov_process = ctx.Process(
371 target=_upload_to_triplestore,
372 args=(
373 self.provenance_triplestore_url,
374 prov_upload_folder,
375 self.redis_host,
376 self.redis_port,
377 self.redis_cache_db,
378 self.ts_failed_queries,
379 self.ts_stop_file,
380 "Uploading prov SPARQL",
381 ),
382 )
384 data_process.start()
385 prov_process.start()
387 data_process.join()
388 prov_process.join()
390 if data_process.exitcode != 0:
391 raise RuntimeError(
392 f"Data upload failed with exit code {data_process.exitcode}"
393 )
394 if prov_process.exitcode != 0:
395 raise RuntimeError(
396 f"Provenance upload failed with exit code {prov_process.exitcode}"
397 )
399 def store_data_and_prov(self, res_storer: Storer, prov_storer: Storer) -> None:
400 """Orchestrate storage and upload."""
401 if not self.rdf_files_only:
402 self._setup_output_directories()
403 self._store_and_upload(res_storer, prov_storer, self.timer)
405 def _store_and_upload(
406 self, res_storer: Storer, prov_storer: Storer, timer: ProcessTimer
407 ) -> None:
408 """Store RDF files and upload queries to triplestore with parallel execution."""
409 with timer.timer("storage"):
410 # Use forkserver to avoid deadlocks when forking from a multi-threaded process.
411 # Libraries like rdflib create background threads, and fork() would
412 # copy locked mutexes into the child process, causing hangs.
413 ctx = multiprocessing.get_context("forkserver")
415 data_store_process = ctx.Process(
416 target=_store_rdf_worker,
417 args=(res_storer, self.output_rdf_dir, self.base_iri),
418 )
419 prov_store_process = ctx.Process(
420 target=_store_rdf_worker,
421 args=(prov_storer, self.output_rdf_dir, self.base_iri),
422 )
423 rdf_store_processes = [data_store_process, prov_store_process]
424 for p in rdf_store_processes:
425 p.start()
427 if not self.rdf_files_only:
428 data_query_process = ctx.Process(
429 target=_generate_queries_worker,
430 args=(res_storer, self.triplestore_url, self.data_update_dir),
431 )
432 prov_query_process = ctx.Process(
433 target=_generate_queries_worker,
434 args=(
435 prov_storer,
436 self.provenance_triplestore_url,
437 self.prov_update_dir,
438 ),
439 )
440 data_query_process.start()
441 prov_query_process.start()
442 data_query_process.join()
443 prov_query_process.join()
445 if data_query_process.exitcode != 0:
446 raise RuntimeError(
447 f"Data query generation failed with exit code {data_query_process.exitcode}"
448 )
449 if prov_query_process.exitcode != 0:
450 raise RuntimeError(
451 f"Prov query generation failed with exit code {prov_query_process.exitcode}"
452 )
454 self._upload_sparql_queries()
456 for p in rdf_store_processes:
457 p.join()
458 if p.exitcode != 0:
459 raise RuntimeError(
460 f"RDF storage failed with exit code {p.exitcode}"
461 )
463 def run_sparql_updates(self, endpoint: str, folder: str):
464 _upload_to_triplestore(
465 endpoint,
466 folder,
467 self.redis_host,
468 self.redis_port,
469 self.redis_cache_db,
470 self.ts_failed_queries,
471 self.ts_stop_file,
472 "Processing files",
473 True,
474 )
477def _save_incremental_report(
478 all_reports: List[Dict[str, Any]], meta_config_path: str, output_path: str
479) -> None:
480 """Save incremental timing report to JSON file."""
481 aggregate_report = {
482 "timestamp": datetime.now().isoformat(),
483 "config_path": meta_config_path,
484 "total_files_processed": len(all_reports),
485 "files": all_reports,
486 "aggregate": _compute_aggregate_metrics(all_reports),
487 }
488 with open(output_path, "wb") as f:
489 f.write(orjson.dumps(aggregate_report, option=orjson.OPT_INDENT_2))
492def _get_file_peak_memory(report: Dict[str, Any]) -> float:
493 """Get peak memory (MB) across all phases in a file report."""
494 phases = report["report"]["phases"]
495 peaks = [p["peak_memory_mb"] for p in phases if p["peak_memory_mb"]]
496 return max(peaks) if peaks else 0
499def _compute_aggregate_metrics(all_reports: List[Dict[str, Any]]) -> Dict[str, Any]:
500 """Compute aggregate statistics across all file reports."""
501 if not all_reports:
502 return {}
504 total_duration = sum(
505 r["report"]["metrics"].get("total_duration_seconds", 0) for r in all_reports
506 )
507 total_records = sum(
508 r["report"]["metrics"].get("input_records", 0) for r in all_reports
509 )
510 total_entities = sum(
511 r["report"]["metrics"].get("new_entities", 0) for r in all_reports
512 )
514 durations = [
515 r["report"]["metrics"].get("total_duration_seconds", 0) for r in all_reports
516 ]
517 throughputs = [
518 r["report"]["metrics"].get("throughput_records_per_sec", 0) for r in all_reports
519 ]
521 file_peaks = [_get_file_peak_memory(r) for r in all_reports]
522 non_zero_peaks = [p for p in file_peaks if p]
524 result: Dict[str, Any] = {
525 "total_files": len(all_reports),
526 "total_duration_seconds": round(total_duration, 3),
527 "total_records_processed": total_records,
528 "total_new_entities": total_entities,
529 "average_time_per_file": round(total_duration / len(all_reports), 3)
530 if all_reports
531 else 0,
532 "average_throughput": round(sum(throughputs) / len(throughputs), 2)
533 if throughputs
534 else 0,
535 "min_time": round(min(durations), 3) if durations else 0,
536 "max_time": round(max(durations), 3) if durations else 0,
537 "overall_throughput": round(total_records / total_duration, 2)
538 if total_duration > 0
539 else 0,
540 }
541 if non_zero_peaks:
542 result["peak_memory_mb"] = round(max(non_zero_peaks), 1)
543 result["average_peak_memory_mb"] = round(
544 sum(non_zero_peaks) / len(non_zero_peaks), 1
545 )
546 return result
549def _print_aggregate_summary(all_reports: List[Dict[str, Any]]) -> None:
550 """Print aggregate summary of all processed files."""
551 aggregate = _compute_aggregate_metrics(all_reports)
553 console.print(f"\n{'=' * 60}")
554 console.print("[bold]Aggregate Timing Summary[/bold]")
555 console.print(f"{'=' * 60}")
556 console.print(f"Total Files: {aggregate['total_files']}")
557 console.print(f"Total Duration: {aggregate['total_duration_seconds']}s")
558 console.print(f"Total Records: {aggregate['total_records_processed']}")
559 console.print(f"Total New Entities: {aggregate['total_new_entities']}")
560 console.print(f"Average Time/File: {aggregate['average_time_per_file']}s")
561 console.print(f"Min/Max Time: {aggregate['min_time']}s / {aggregate['max_time']}s")
562 console.print(f"Overall Throughput: {aggregate['overall_throughput']} rec/s")
563 if "peak_memory_mb" in aggregate:
564 console.print(f"Peak Memory (RSS): {aggregate['peak_memory_mb']} MB")
565 console.print(f"Avg Peak Memory: {aggregate['average_peak_memory_mb']} MB")
566 console.print(f"{'=' * 60}\n")
569def run_meta_process(
570 settings: dict,
571 meta_config_path: str,
572 enable_timing: bool = False,
573 timing_output: Optional[str] = None,
574) -> None:
575 is_unix = platform in {"linux", "linux2", "darwin"}
576 all_reports = []
578 meta_process_setup = MetaProcess(
579 settings=settings, meta_config_path=meta_config_path
580 )
581 files_to_be_processed = meta_process_setup.prepare_folders()
583 generate_gentle_buttons(
584 meta_process_setup.base_output_dir, meta_config_path, is_unix
585 )
587 with create_progress() as progress:
588 task_id = progress.add_task(
589 "Processing files", total=len(files_to_be_processed)
590 )
591 for idx, filename in enumerate(files_to_be_processed, 1):
592 try:
593 if os.path.exists(
594 os.path.join(meta_process_setup.base_output_dir, ".stop")
595 ):
596 console.print(
597 "\n[yellow]Stop file detected. Halting processing.[/yellow]"
598 )
599 meta_process_setup.counter_handler.flush()
600 break
602 if enable_timing:
603 console.print(
604 f"\n[cyan][{idx}/{len(files_to_be_processed)}][/cyan] Processing {filename}..."
605 )
607 on_phase_cb = None
608 if enable_timing and timing_output:
609 _chart = timing_output.replace(".json", "_chart.png")
610 _reports, _fn, _cfg, _out = (
611 all_reports,
612 filename,
613 meta_config_path,
614 timing_output,
615 )
616 _include_storage = not settings.get("rdf_files_only", False)
618 def _on_phase(timer: ProcessTimer) -> None:
619 snapshot = _reports + [
620 {"filename": _fn, "report": timer.get_report()}
621 ]
622 _save_incremental_report(snapshot, _cfg, _out)
623 plot_incremental_progress(
624 snapshot, _chart, include_storage=_include_storage
625 )
627 on_phase_cb = _on_phase
629 file_timer = ProcessTimer(
630 enabled=enable_timing,
631 verbose=enable_timing,
632 on_phase_complete=on_phase_cb,
633 )
634 meta_process_setup.timer = file_timer
636 result = meta_process_setup.curate_and_create(
637 filename,
638 meta_process_setup.cache_path,
639 meta_process_setup.errors_path,
640 settings=settings,
641 meta_config_path=meta_config_path,
642 progress=progress,
643 )
644 task_done(result)
645 meta_process_setup.counter_handler.flush()
647 if enable_timing:
648 report = file_timer.get_report()
649 all_reports.append({"filename": filename, "report": report})
650 file_timer.print_file_summary(filename)
652 except Exception as e:
653 traceback_str = traceback.format_exc()
654 console.print(
655 f"[red]Error processing file {filename}: {e}\nTraceback:\n{traceback_str}[/red]"
656 )
657 finally:
658 progress.advance(task_id)
660 meta_process_setup.counter_handler.flush()
662 if not os.path.exists(os.path.join(meta_process_setup.base_output_dir, ".stop")):
663 if os.path.exists(meta_process_setup.cache_path):
664 os.rename(
665 meta_process_setup.cache_path,
666 meta_process_setup.cache_path.replace(
667 ".txt", f"_{datetime.now().strftime('%Y-%m-%dT%H_%M_%S_%f')}.txt"
668 ),
669 )
670 if is_unix:
671 delete_lock_files(base_dir=meta_process_setup.base_output_dir)
673 if enable_timing and all_reports:
674 _print_aggregate_summary(all_reports)
675 if timing_output:
676 aggregate_report = {
677 "timestamp": datetime.now().isoformat(),
678 "config_path": meta_config_path,
679 "total_files": len(all_reports),
680 "files": all_reports,
681 "aggregate": _compute_aggregate_metrics(all_reports),
682 }
683 with open(timing_output, "wb") as f:
684 f.write(orjson.dumps(aggregate_report, option=orjson.OPT_INDENT_2))
685 console.print(f"[green][Timing] Report saved to {timing_output}[/green]")
688def _cache_sort_key(filename: str) -> int:
689 return int(filename.replace(".csv", ""))
692def task_done(task_output: tuple) -> None:
693 message, cache_path, errors_path, filename = task_output
694 if message["message"] == "skip":
695 pass
696 elif message["message"] == "success":
697 if not os.path.exists(cache_path):
698 with open(cache_path, "w", encoding="utf-8") as aux_file:
699 aux_file.write(filename + "\n")
700 else:
701 with open(cache_path, "r", encoding="utf-8") as aux_file:
702 cache_data = aux_file.read().splitlines()
703 try:
704 bisect.insort(cache_data, filename, key=_cache_sort_key)
705 except ValueError:
706 # Non-numeric filename (e.g. "data.csv"): append without ordering
707 cache_data.append(filename)
708 with open(cache_path, "w", encoding="utf-8") as aux_file:
709 aux_file.write("\n".join(cache_data))
710 else:
711 with open(errors_path, "a", encoding="utf-8") as aux_file:
712 aux_file.write(f"{filename}: {message['message']}" + "\n")
715def delete_lock_files(base_dir: str) -> None:
716 for dirpath, _, filenames in os.walk(base_dir):
717 for filename in filenames:
718 if filename.endswith(".lock"):
719 os.remove(os.path.join(dirpath, filename))
722def generate_gentle_buttons(dir: str, config: str, is_unix: bool):
723 if os.path.exists(os.path.join(dir, ".stop")):
724 os.remove(os.path.join(dir, ".stop"))
725 ext = "sh" if is_unix else "bat"
726 with open(f"gently_run.{ext}", "w") as rsh:
727 rsh.write(
728 f'{executable} -m oc_meta.lib.stopper -t "{dir}" --remove\n{executable} -m oc_meta.run.meta_process -c {config}'
729 )
730 with open(f"gently_stop.{ext}", "w") as rsh:
731 rsh.write(f'{executable} -m oc_meta.lib.stopper -t "{dir}" --add')
734if __name__ == "__main__": # pragma: no cover
735 arg_parser = ArgumentParser(
736 "meta_process.py",
737 description="This script runs the OCMeta data processing workflow",
738 formatter_class=RichHelpFormatter,
739 )
740 arg_parser.add_argument(
741 "-c",
742 "--config",
743 dest="config",
744 required=True,
745 help="Configuration file directory",
746 )
747 arg_parser.add_argument(
748 "--timing",
749 action="store_true",
750 help="Enable timing metrics collection and display summary at the end",
751 )
752 arg_parser.add_argument(
753 "--timing-output",
754 dest="timing_output",
755 default=None,
756 help="Optional path to save timing report as JSON file",
757 )
758 args = arg_parser.parse_args()
759 with open(args.config, encoding="utf-8") as file:
760 settings = yaml.full_load(file)
761 run_meta_process(
762 settings=settings,
763 meta_config_path=args.config,
764 enable_timing=args.timing,
765 timing_output=args.timing_output,
766 )