Coverage for oc_meta / run / meta / generate_csv.py: 37%
380 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
7from __future__ import annotations
9import csv
10import os
11from argparse import ArgumentParser
12from functools import lru_cache
13import multiprocessing
14from typing import Dict, List, Optional, Tuple
15from zipfile import ZipFile
17import orjson
18import redis
19import yaml
21from oc_meta.lib.console import create_progress
22from oc_meta.lib.file_manager import collect_zip_files, find_rdf_file
24csv.field_size_limit(2**31 - 1)
26FIELDNAMES = [
27 "id",
28 "title",
29 "author",
30 "issue",
31 "volume",
32 "venue",
33 "page",
34 "pub_date",
35 "type",
36 "publisher",
37 "editor",
38]
40URI_TYPE_DICT = {
41 "http://purl.org/spar/doco/Abstract": "abstract",
42 "http://purl.org/spar/fabio/ArchivalDocument": "archival document",
43 "http://purl.org/spar/fabio/AudioDocument": "audio document",
44 "http://purl.org/spar/fabio/Book": "book",
45 "http://purl.org/spar/fabio/BookChapter": "book chapter",
46 "http://purl.org/spar/fabio/ExpressionCollection": "book section",
47 "http://purl.org/spar/fabio/BookSeries": "book series",
48 "http://purl.org/spar/fabio/BookSet": "book set",
49 "http://purl.org/spar/fabio/ComputerProgram": "computer program",
50 "http://purl.org/spar/doco/Part": "book part",
51 "http://purl.org/spar/fabio/Expression": "",
52 "http://purl.org/spar/fabio/DataFile": "dataset",
53 "http://purl.org/spar/fabio/DataManagementPlan": "data management plan",
54 "http://purl.org/spar/fabio/Thesis": "dissertation",
55 "http://purl.org/spar/fabio/Editorial": "editorial",
56 "http://purl.org/spar/fabio/Journal": "journal",
57 "http://purl.org/spar/fabio/JournalArticle": "journal article",
58 "http://purl.org/spar/fabio/JournalEditorial": "journal editorial",
59 "http://purl.org/spar/fabio/JournalIssue": "journal issue",
60 "http://purl.org/spar/fabio/JournalVolume": "journal volume",
61 "http://purl.org/spar/fabio/Newspaper": "newspaper",
62 "http://purl.org/spar/fabio/NewspaperArticle": "newspaper article",
63 "http://purl.org/spar/fabio/NewspaperIssue": "newspaper issue",
64 "http://purl.org/spar/fr/ReviewVersion": "peer review",
65 "http://purl.org/spar/fabio/AcademicProceedings": "proceedings",
66 "http://purl.org/spar/fabio/Preprint": "preprint",
67 "http://purl.org/spar/fabio/Presentation": "presentation",
68 "http://purl.org/spar/fabio/ProceedingsPaper": "proceedings article",
69 "http://purl.org/spar/fabio/ReferenceBook": "reference book",
70 "http://purl.org/spar/fabio/ReferenceEntry": "reference entry",
71 "http://purl.org/spar/fabio/ReportDocument": "report",
72 "http://purl.org/spar/fabio/RetractionNotice": "retraction notice",
73 "http://purl.org/spar/fabio/Series": "series",
74 "http://purl.org/spar/fabio/SpecificationDocument": "standard",
75 "http://purl.org/spar/fabio/WebContent": "web content",
76}
78_worker_redis: Optional[redis.Redis] = None
79_worker_config: Optional[Tuple[str, int, int]] = None
82def _init_worker(
83 redis_host: str,
84 redis_port: int,
85 redis_db: int,
86 input_dir: str,
87 dir_split_number: int,
88 items_per_file: int,
89) -> None:
90 global _worker_redis, _worker_config
91 _worker_redis = redis.Redis(
92 host=redis_host, port=redis_port, db=redis_db, decode_responses=True
93 )
94 _worker_config = (input_dir, dir_split_number, items_per_file)
97def _process_file_worker(filepath: str) -> Tuple[str, List[Dict[str, str]]]:
98 assert _worker_redis is not None and _worker_config is not None
99 input_dir, dir_split_number, items_per_file = _worker_config
100 results = []
101 data = load_json_from_file(filepath)
102 for graph in data:
103 for entity in graph.get("@graph", []):
104 entity_types = entity.get("@type", [])
105 if (
106 "http://purl.org/spar/fabio/JournalVolume" in entity_types
107 or "http://purl.org/spar/fabio/JournalIssue" in entity_types
108 ):
109 continue
110 entity_id = entity.get("@id", "")
111 if entity_id:
112 omid = f"omid:br/{entity_id.split('/')[-1]}"
113 if _worker_redis.sismember("processed_omids", omid):
114 continue
115 br_data = process_bibliographic_resource(
116 entity, input_dir, dir_split_number, items_per_file
117 )
118 if br_data:
119 results.append(br_data)
120 return (filepath, results)
123def init_redis_connection(
124 host: str = "localhost", port: int = 6379, db: int = 2
125) -> redis.Redis:
126 client = redis.Redis(host=host, port=port, db=db, decode_responses=True)
127 client.ping()
128 return client
131def is_omid_processed(omid: str, redis_client: redis.Redis) -> bool:
132 return bool(redis_client.sismember("processed_omids", omid))
135def load_processed_omids_to_redis(output_dir: str, redis_client: redis.Redis) -> int:
136 redis_client.delete("processed_omids")
138 if not os.path.exists(output_dir):
139 return 0
141 csv.field_size_limit(2**31 - 1)
143 count = 0
144 BATCH_SIZE = 1000
145 csv_files = [f for f in os.listdir(output_dir) if f.endswith(".csv")]
147 with create_progress() as progress:
148 task = progress.add_task("Loading existing identifiers", total=len(csv_files))
150 for filename in csv_files:
151 filepath = os.path.join(output_dir, filename)
152 with open(filepath, "r", encoding="utf-8") as f:
153 reader = csv.DictReader(f)
154 batch_pipe = redis_client.pipeline()
155 batch_count = 0
157 for row in reader:
158 omids = [
159 id_part.strip()
160 for id_part in row["id"].split()
161 if id_part.startswith("omid:br/")
162 ]
163 for omid in omids:
164 batch_pipe.sadd("processed_omids", omid)
165 batch_count += 1
166 count += 1
168 if batch_count >= BATCH_SIZE:
169 batch_pipe.execute()
170 batch_pipe = redis_client.pipeline()
171 batch_count = 0
173 if batch_count > 0:
174 batch_pipe.execute()
176 progress.update(task, advance=1)
178 return count
181def load_checkpoint(checkpoint_file: str) -> set:
182 if not os.path.exists(checkpoint_file):
183 return set()
184 with open(checkpoint_file, "r") as f:
185 return set(line.strip() for line in f if line.strip())
188def mark_file_processed(checkpoint_file: str, filepath: str) -> None:
189 with open(checkpoint_file, "a") as f:
190 f.write(filepath + "\n")
193@lru_cache(maxsize=2000)
194def load_json_from_file(filepath: str) -> list:
195 with ZipFile(filepath, "r") as zip_file:
196 json_filename = zip_file.namelist()[0]
197 with zip_file.open(json_filename) as json_file:
198 return orjson.loads(json_file.read())
201def process_identifier(id_data: dict) -> Optional[str]:
202 try:
203 id_schema = id_data["http://purl.org/spar/datacite/usesIdentifierScheme"][0][
204 "@id"
205 ].split("/datacite/")[1]
206 literal_value = id_data[
207 "http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue"
208 ][0]["@value"]
209 return f"{id_schema}:{literal_value}"
210 except (KeyError, IndexError):
211 return None
214def process_responsible_agent(
215 ra_data: dict, ra_uri: str, rdf_dir: str, dir_split_number: int, items_per_file: int
216) -> Optional[str]:
217 try:
218 family_name = ra_data.get("http://xmlns.com/foaf/0.1/familyName", [{}])[0].get(
219 "@value", ""
220 )
221 given_name = ra_data.get("http://xmlns.com/foaf/0.1/givenName", [{}])[0].get(
222 "@value", ""
223 )
224 foaf_name = ra_data.get("http://xmlns.com/foaf/0.1/name", [{}])[0].get(
225 "@value", ""
226 )
228 if family_name or given_name:
229 if family_name and given_name:
230 name = f"{family_name}, {given_name}"
231 elif family_name:
232 name = f"{family_name},"
233 else:
234 name = f", {given_name}"
235 elif foaf_name:
236 name = foaf_name
237 else:
238 return None
240 omid = ra_uri.split("/")[-1]
241 identifiers = [f"omid:ra/{omid}"]
243 if "http://purl.org/spar/datacite/hasIdentifier" in ra_data:
244 for identifier in ra_data["http://purl.org/spar/datacite/hasIdentifier"]:
245 id_uri = identifier["@id"]
246 id_file = find_rdf_file(
247 id_uri, rdf_dir, dir_split_number, items_per_file, zip_output=True
248 )
249 if os.path.exists(id_file):
250 id_data = load_json_from_file(id_file)
251 for graph in id_data:
252 for entity in graph.get("@graph", []):
253 if entity["@id"] == id_uri:
254 id_value = process_identifier(entity)
255 if id_value:
256 identifiers.append(id_value)
258 if identifiers:
259 return f"{name} [{' '.join(identifiers)}]"
260 return name
261 except (KeyError, IndexError):
262 return None
265def process_venue_title(
266 venue_data: dict,
267 venue_uri: str,
268 rdf_dir: str,
269 dir_split_number: int,
270 items_per_file: int,
271) -> str:
272 venue_title = venue_data.get("http://purl.org/dc/terms/title", [{}])[0].get(
273 "@value", ""
274 )
275 if not venue_title:
276 return ""
278 omid = venue_uri.split("/")[-1]
279 identifiers = [f"omid:br/{omid}"]
281 if "http://purl.org/spar/datacite/hasIdentifier" in venue_data:
282 for identifier in venue_data["http://purl.org/spar/datacite/hasIdentifier"]:
283 id_uri = identifier["@id"]
284 id_file = find_rdf_file(
285 id_uri, rdf_dir, dir_split_number, items_per_file, zip_output=True
286 )
287 if os.path.exists(id_file):
288 id_data = load_json_from_file(id_file)
289 for graph in id_data:
290 for entity in graph.get("@graph", []):
291 if entity["@id"] == id_uri:
292 id_value = process_identifier(entity)
293 if id_value:
294 identifiers.append(id_value)
296 return f"{venue_title} [{' '.join(identifiers)}]" if identifiers else venue_title
299def process_hierarchical_venue(
300 entity: dict,
301 rdf_dir: str,
302 dir_split_number: int,
303 items_per_file: int,
304 visited: Optional[set] = None,
305 depth: int = 0,
306) -> Dict[str, str]:
307 result = {"volume": "", "issue": "", "venue": ""}
309 if visited is None:
310 visited = set()
312 entity_id = entity.get("@id", "")
313 if entity_id in visited or depth > 5:
314 print(f"Warning: Cycle detected in venue hierarchy at: {entity_id}")
315 return result
316 visited.add(entity_id)
318 entity_types = entity.get("@type", [])
320 if "http://purl.org/spar/fabio/JournalIssue" in entity_types:
321 result["issue"] = entity.get(
322 "http://purl.org/spar/fabio/hasSequenceIdentifier", [{}]
323 )[0].get("@value", "")
324 elif "http://purl.org/spar/fabio/JournalVolume" in entity_types:
325 result["volume"] = entity.get(
326 "http://purl.org/spar/fabio/hasSequenceIdentifier", [{}]
327 )[0].get("@value", "")
328 else:
329 result["venue"] = process_venue_title(
330 entity, entity["@id"], rdf_dir, dir_split_number, items_per_file
331 )
332 return result
334 if "http://purl.org/vocab/frbr/core#partOf" in entity:
335 parent_uri = entity["http://purl.org/vocab/frbr/core#partOf"][0]["@id"]
336 parent_file = find_rdf_file(
337 parent_uri, rdf_dir, dir_split_number, items_per_file, zip_output=True
338 )
339 if os.path.exists(parent_file):
340 parent_data = load_json_from_file(parent_file)
341 for graph in parent_data:
342 for parent_entity in graph.get("@graph", []):
343 if parent_entity["@id"] == parent_uri:
344 parent_info = process_hierarchical_venue(
345 parent_entity,
346 rdf_dir,
347 dir_split_number,
348 items_per_file,
349 visited,
350 depth + 1,
351 )
352 for key, value in parent_info.items():
353 if not result[key]:
354 result[key] = value
356 return result
359def find_first_ar_by_role(
360 agent_roles: Dict, next_relations: Dict, role_type: str
361) -> Optional[str]:
362 role_ars = {
363 ar_uri: ar_data
364 for ar_uri, ar_data in agent_roles.items()
365 if role_type
366 in ar_data.get("http://purl.org/spar/pro/withRole", [{}])[0].get("@id", "")
367 }
369 role_next_relations = {
370 ar_uri: next_ar
371 for ar_uri, next_ar in next_relations.items()
372 if ar_uri in role_ars and next_ar in role_ars
373 }
375 referenced_ars = set(role_next_relations.values())
376 for ar_uri in role_ars:
377 if ar_uri not in referenced_ars:
378 return ar_uri
380 return next(iter(role_ars)) if role_ars else None
383def process_bibliographic_resource(
384 br_data: dict, rdf_dir: str, dir_split_number: int, items_per_file: int
385) -> Optional[Dict[str, str]]:
386 br_types = br_data.get("@type", [])
387 if (
388 "http://purl.org/spar/fabio/JournalVolume" in br_types
389 or "http://purl.org/spar/fabio/JournalIssue" in br_types
390 ):
391 return None
393 output = {field: "" for field in FIELDNAMES}
395 try:
396 entity_id = br_data.get("@id", "")
397 identifiers = [f"omid:br/{entity_id.split('/')[-1]}"] if entity_id else []
399 output["title"] = br_data.get("http://purl.org/dc/terms/title", [{}])[0].get(
400 "@value", ""
401 )
402 output["pub_date"] = br_data.get(
403 "http://prismstandard.org/namespaces/basic/2.0/publicationDate", [{}]
404 )[0].get("@value", "")
406 br_types = [
407 t
408 for t in br_data.get("@type", [])
409 if t != "http://purl.org/spar/fabio/Expression"
410 ]
411 output["type"] = URI_TYPE_DICT.get(br_types[0], "") if br_types else ""
413 if "http://purl.org/spar/datacite/hasIdentifier" in br_data:
414 for identifier in br_data["http://purl.org/spar/datacite/hasIdentifier"]:
415 id_uri = identifier["@id"]
416 id_file = find_rdf_file(
417 id_uri, rdf_dir, dir_split_number, items_per_file, zip_output=True
418 )
419 if os.path.exists(id_file):
420 id_data = load_json_from_file(id_file)
421 for graph in id_data:
422 for entity in graph.get("@graph", []):
423 if entity["@id"] == id_uri:
424 id_value = process_identifier(entity)
425 if id_value:
426 identifiers.append(id_value)
427 output["id"] = " ".join(identifiers)
429 authors = []
430 editors = []
431 publishers = []
432 agent_roles = {}
433 next_relations = {}
435 if "http://purl.org/spar/pro/isDocumentContextFor" in br_data:
436 for ar_data in br_data["http://purl.org/spar/pro/isDocumentContextFor"]:
437 ar_uri = ar_data["@id"]
438 ar_file = find_rdf_file(
439 ar_uri, rdf_dir, dir_split_number, items_per_file, zip_output=True
440 )
441 if os.path.exists(ar_file):
442 ar_data = load_json_from_file(ar_file)
443 for graph in ar_data:
444 for entity in graph.get("@graph", []):
445 if entity["@id"] == ar_uri:
446 agent_roles[ar_uri] = entity
447 if "https://w3id.org/oc/ontology/hasNext" in entity:
448 next_ar = entity[
449 "https://w3id.org/oc/ontology/hasNext"
450 ][0]["@id"]
451 next_relations[ar_uri] = next_ar
453 for role_type, role_list in [
454 ("author", authors),
455 ("editor", editors),
456 ("publisher", publishers),
457 ]:
458 first_ar = find_first_ar_by_role(agent_roles, next_relations, role_type)
459 if not first_ar:
460 continue
462 current_ar = first_ar
463 processed_ars = set()
464 max_iterations = len(agent_roles)
465 iterations = 0
467 while current_ar and current_ar in agent_roles:
468 if current_ar in processed_ars or iterations >= max_iterations:
469 print(
470 f"Warning: Detected cycle in hasNext relations or exceeded maximum iterations at AR: {current_ar}"
471 )
472 break
474 processed_ars.add(current_ar)
475 iterations += 1
477 entity = agent_roles[current_ar]
478 role = entity.get("http://purl.org/spar/pro/withRole", [{}])[0].get(
479 "@id", ""
480 )
482 if role_type in role:
483 if "http://purl.org/spar/pro/isHeldBy" in entity:
484 ra_uri = entity["http://purl.org/spar/pro/isHeldBy"][0][
485 "@id"
486 ]
487 ra_file = find_rdf_file(
488 ra_uri,
489 rdf_dir,
490 dir_split_number,
491 items_per_file,
492 zip_output=True,
493 )
494 if os.path.exists(ra_file):
495 ra_data = load_json_from_file(ra_file)
496 for ra_graph in ra_data:
497 for ra_entity in ra_graph.get("@graph", []):
498 if ra_entity["@id"] == ra_uri:
499 agent_name = process_responsible_agent(
500 ra_entity,
501 ra_uri,
502 rdf_dir,
503 dir_split_number,
504 items_per_file,
505 )
506 if agent_name:
507 role_list.append(agent_name)
509 current_ar = next_relations.get(current_ar)
511 output["author"] = "; ".join(authors)
512 output["editor"] = "; ".join(editors)
513 output["publisher"] = "; ".join(publishers)
515 if "http://purl.org/vocab/frbr/core#partOf" in br_data:
516 venue_uri = br_data["http://purl.org/vocab/frbr/core#partOf"][0]["@id"]
517 venue_file = find_rdf_file(
518 venue_uri, rdf_dir, dir_split_number, items_per_file, zip_output=True
519 )
520 if os.path.exists(venue_file):
521 venue_data = load_json_from_file(venue_file)
522 for graph in venue_data:
523 for entity in graph.get("@graph", []):
524 if entity["@id"] == venue_uri:
525 venue_info = process_hierarchical_venue(
526 entity, rdf_dir, dir_split_number, items_per_file
527 )
528 output.update(venue_info)
530 if "http://purl.org/vocab/frbr/core#embodiment" in br_data:
531 page_uri = br_data["http://purl.org/vocab/frbr/core#embodiment"][0]["@id"]
532 page_file = find_rdf_file(
533 page_uri, rdf_dir, dir_split_number, items_per_file, zip_output=True
534 )
535 if os.path.exists(page_file):
536 page_data = load_json_from_file(page_file)
537 for graph in page_data:
538 for entity in graph.get("@graph", []):
539 if entity["@id"] == page_uri:
540 start_page = entity.get(
541 "http://prismstandard.org/namespaces/basic/2.0/startingPage",
542 [{}],
543 )[0].get("@value", "")
544 end_page = entity.get(
545 "http://prismstandard.org/namespaces/basic/2.0/endingPage",
546 [{}],
547 )[0].get("@value", "")
548 if start_page or end_page:
549 output["page"] = f"{start_page}-{end_page}"
551 except Exception as e:
552 print(f"Error processing bibliographic resource: {type(e).__name__}: {e}")
554 return output
557class ResultBuffer:
558 def __init__(self, output_dir: str, max_rows: int = 3000):
559 self.buffer = []
560 self.output_dir = output_dir
561 self.max_rows = max_rows
562 self.file_counter = self._get_last_file_number() + 1
564 def _get_last_file_number(self) -> int:
565 if not os.path.exists(self.output_dir):
566 return -1
568 max_number = -1
569 for filename in os.listdir(self.output_dir):
570 if filename.startswith("output_") and filename.endswith(".csv"):
571 try:
572 number = int(filename[7:-4])
573 max_number = max(max_number, number)
574 except ValueError:
575 continue
576 return max_number
578 def add_results(self, results: List[Dict[str, str]]) -> None:
579 self.buffer.extend(results)
580 while len(self.buffer) >= self.max_rows:
581 self._write_buffer_chunk()
583 def _write_buffer_chunk(self) -> None:
584 chunk = self.buffer[: self.max_rows]
585 output_file = os.path.join(self.output_dir, f"output_{self.file_counter}.csv")
586 write_csv(output_file, chunk)
587 self.buffer = self.buffer[self.max_rows :]
588 self.file_counter += 1
590 def flush(self) -> None:
591 if self.buffer:
592 output_file = os.path.join(
593 self.output_dir, f"output_{self.file_counter}.csv"
594 )
595 write_csv(output_file, self.buffer)
596 self.buffer = []
597 self.file_counter += 1
600def generate_csv(
601 input_dir: str,
602 output_dir: str,
603 dir_split_number: int,
604 items_per_file: int,
605 redis_host: str = "localhost",
606 redis_port: int = 6379,
607 redis_db: int = 2,
608 workers: int = 4,
609) -> None:
610 if not os.path.exists(output_dir):
611 os.makedirs(output_dir)
613 checkpoint_file = os.path.join(output_dir, "processed_br_files.txt")
614 processed_br_files = load_checkpoint(checkpoint_file)
616 redis_client = init_redis_connection(redis_host, redis_port, redis_db)
617 load_processed_omids_to_redis(output_dir, redis_client)
619 br_dir = os.path.join(input_dir, "br")
620 if not os.path.exists(br_dir):
621 print(f"Error: directory not found at {br_dir}")
622 return
624 all_files = collect_zip_files(br_dir, only_data=True)
625 files_to_process = [f for f in all_files if f not in processed_br_files]
627 if not files_to_process:
628 print("All files already processed")
629 return
631 print(f"Skipping {len(processed_br_files)} already processed files")
632 print(
633 f"Processing {len(files_to_process)} remaining files with {workers} workers..."
634 )
636 result_buffer = ResultBuffer(output_dir)
638 # Use forkserver to avoid deadlocks when forking in a multi-threaded environment
639 ctx = multiprocessing.get_context("forkserver")
640 with ctx.Pool(
641 workers,
642 _init_worker,
643 (redis_host, redis_port, redis_db, input_dir, dir_split_number, items_per_file),
644 ) as pool:
645 with create_progress() as progress:
646 task = progress.add_task("Processing files", total=len(files_to_process))
648 for filepath, results in pool.imap_unordered(
649 _process_file_worker, files_to_process
650 ):
651 if results:
652 result_buffer.add_results(results)
653 mark_file_processed(checkpoint_file, filepath)
654 progress.update(task, advance=1)
656 result_buffer.flush()
657 print("Processing complete.")
660def write_csv(filepath: str, data: List[Dict[str, str]]) -> None:
661 with open(filepath, "w", newline="", encoding="utf-8") as f:
662 writer = csv.DictWriter(f, fieldnames=FIELDNAMES)
663 writer.writeheader()
664 writer.writerows(data)
667if __name__ == "__main__":
668 parser = ArgumentParser(
669 "generate_csv.py",
670 description="Generate CSV files from OpenCitations Meta RDF dump",
671 )
672 parser.add_argument(
673 "-c",
674 "--config",
675 required=True,
676 help="OpenCitations Meta configuration file location",
677 )
678 parser.add_argument(
679 "-o",
680 "--output_dir",
681 required=True,
682 help="Directory where CSV files will be stored",
683 )
684 parser.add_argument(
685 "--redis-host", default="localhost", help="Redis host (default: localhost)"
686 )
687 parser.add_argument(
688 "--redis-port", type=int, default=6379, help="Redis port (default: 6379)"
689 )
690 parser.add_argument(
691 "--redis-db", type=int, default=2, help="Redis database number (default: 2)"
692 )
693 parser.add_argument(
694 "--workers", type=int, default=4, help="Number of parallel workers (default: 4)"
695 )
696 parser.add_argument(
697 "--clean",
698 action="store_true",
699 help="Clear checkpoint file and Redis cache before starting",
700 )
701 args = parser.parse_args()
703 with open(args.config, encoding="utf-8") as f:
704 settings = yaml.full_load(f)
706 rdf_dir = os.path.join(settings["output_rdf_dir"], "rdf")
707 dir_split_number = settings["dir_split_number"]
708 items_per_file = settings["items_per_file"]
710 if args.clean:
711 checkpoint_file = "processed_br_files.txt"
712 if os.path.exists(checkpoint_file):
713 os.remove(checkpoint_file)
714 print(f"Removed checkpoint file: {checkpoint_file}")
715 redis_client = redis.Redis(
716 host=args.redis_host, port=args.redis_port, db=args.redis_db
717 )
718 deleted = redis_client.delete("processed_omids")
719 if deleted:
720 print("Cleared Redis processed_omids cache")
722 generate_csv(
723 input_dir=rdf_dir,
724 output_dir=args.output_dir,
725 dir_split_number=dir_split_number,
726 items_per_file=items_per_file,
727 redis_host=args.redis_host,
728 redis_port=args.redis_port,
729 redis_db=args.redis_db,
730 workers=args.workers,
731 )