Coverage for oc_meta / core / creator.py: 99%
415 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
10from typing import TYPE_CHECKING, List
12from oc_meta.constants import BR_ID_SCHEMAS, RA_ID_SCHEMAS
13from oc_meta.core.curator import get_edited_br_metaid
14from oc_meta.lib.finder import ResourceFinder
15from oc_meta.lib.master_of_regex import (
16 RE_COMMA_AND_SPACES,
17 RE_ONE_OR_MORE_SPACES,
18 RE_SEMICOLON_IN_PEOPLE_FIELD,
19 split_name_and_ids,
20)
21from oc_ocdm.counter_handler.counter_handler import CounterHandler
22from oc_ocdm.graph import GraphSet
23from oc_ocdm.graph.entities.bibliographic import BibliographicResource
24from oc_ocdm.graph.entities.bibliographic_entity import BibliographicEntity
25from oc_ocdm.support import create_date
27if TYPE_CHECKING:
28 from rich.progress import Progress
31class Creator(object):
32 def __init__(
33 self,
34 data: list,
35 finder: ResourceFinder,
36 base_iri: str,
37 counter_handler: CounterHandler,
38 supplier_prefix: str,
39 resp_agent: str,
40 ra_index: list,
41 br_index: list,
42 re_index_csv: list,
43 ar_index_csv: list,
44 vi_index: dict,
45 silencer: list | None = None,
46 progress: Progress | None = None,
47 ):
48 self.url = base_iri
49 self.progress = progress
50 self.setgraph = GraphSet(
51 self.url,
52 supplier_prefix=supplier_prefix,
53 wanted_label=False,
54 custom_counter_handler=counter_handler,
55 )
56 self.resp_agent = resp_agent
57 self.finder = finder
59 self.ra_id_schemas = RA_ID_SCHEMAS
60 self.br_id_schemas = BR_ID_SCHEMAS
61 self.schemas = RA_ID_SCHEMAS | BR_ID_SCHEMAS
63 self.ra_index = self.indexer_id(ra_index)
64 self.br_index = self.indexer_id(br_index)
65 self.re_index = self.index_re(re_index_csv)
66 self.ar_index = self.index_ar(ar_index_csv)
67 self.vi_index = vi_index
68 self.data = data
69 self.counter_handler = counter_handler
70 self.silencer = silencer or []
72 _PRO_IS_DOC_CONTEXT_FOR = "http://purl.org/spar/pro/isDocumentContextFor"
73 _PRO_WITH_ROLE = "http://purl.org/spar/pro/withRole"
74 _PRO_AUTHOR = "http://purl.org/spar/pro/author"
75 _PRO_EDITOR = "http://purl.org/spar/pro/editor"
76 _PRO_PUBLISHER = "http://purl.org/spar/pro/publisher"
78 def _has_existing_roles(self, br_uri: str) -> dict:
79 has_roles = {"author": False, "editor": False, "publisher": False}
81 for ar_uri in self.finder._get_objects(br_uri, self._PRO_IS_DOC_CONTEXT_FOR):
82 for role in self.finder._get_objects(ar_uri, self._PRO_WITH_ROLE):
83 if role == self._PRO_AUTHOR:
84 has_roles["author"] = True
85 elif role == self._PRO_EDITOR:
86 has_roles["editor"] = True
87 elif role == self._PRO_PUBLISHER:
88 has_roles["publisher"] = True
90 return has_roles
92 def creator(self, source=None):
93 self.src = source
94 task_id = None
95 if self.progress:
96 task_id = self.progress.add_task(
97 " [dim]Creating RDF entities[/dim]", total=len(self.data)
98 )
99 for row in self.data:
100 self.row_meta = ""
101 self.venue_meta = ""
102 ids = row["id"]
103 title = row["title"]
104 authors = row["author"]
105 pub_date = row["pub_date"]
106 venue = row["venue"]
107 vol = row["volume"]
108 issue = row["issue"]
109 page = row["page"]
110 self.type = row["type"]
111 publisher = row["publisher"]
112 editor = row["editor"]
113 self.venue_graph = None
114 self.vol_graph = None
115 self.issue_graph = None
116 self.id_action(ids)
117 self.vvi_action(venue, vol, issue)
118 self.title_action(title)
120 br_uri = f"{self.url}{self.row_meta}"
121 br_is_preexisting = br_uri in self.finder
123 skip_author = False
124 skip_publisher = False
125 skip_editor = False
127 if br_is_preexisting:
128 existing_roles = self._has_existing_roles(br_uri)
130 skip_author = "author" in self.silencer and existing_roles["author"]
131 skip_publisher = (
132 "publisher" in self.silencer and existing_roles["publisher"]
133 )
135 editor_br_key = get_edited_br_metaid(
136 row, self.row_meta, self.venue_meta
137 )
138 editor_br_uri = f"{self.url}{editor_br_key}"
139 editor_existing_roles = (
140 existing_roles
141 if editor_br_uri == br_uri
142 else self._has_existing_roles(editor_br_uri)
143 )
144 skip_editor = (
145 "editor" in self.silencer and editor_existing_roles["editor"]
146 )
148 if not skip_author:
149 self.author_action(authors)
151 self.pub_date_action(pub_date)
152 self.page_action(page)
153 self.type_action(self.type)
155 if not skip_publisher:
156 self.publisher_action(publisher)
158 if not skip_editor:
159 self.editor_action(editor, row)
161 if self.progress and task_id is not None:
162 self.progress.advance(task_id)
164 if self.progress and task_id is not None:
165 self.progress.remove_task(task_id)
166 return self.setgraph
168 @staticmethod
169 def index_re(id_index):
170 index = dict()
171 for row in id_index:
172 index[row["br"]] = row["re"]
173 return index
175 @staticmethod
176 def index_ar(id_index):
177 index = dict()
178 for row in id_index:
179 index[row["meta"]] = dict()
180 index[row["meta"]]["author"] = Creator.__ar_worker(row["author"])
181 index[row["meta"]]["editor"] = Creator.__ar_worker(row["editor"])
182 index[row["meta"]]["publisher"] = Creator.__ar_worker(row["publisher"])
183 return index
185 @staticmethod
186 def __ar_worker(s: str) -> dict:
187 if s:
188 ar_dict = dict()
189 couples = s.split("; ")
190 for c in couples:
191 cou = c.split(", ")
192 ar_dict[cou[1]] = cou[0]
193 return ar_dict
194 else:
195 return dict()
197 def indexer_id(self, csv_index):
198 index = {}
199 for row in csv_index:
200 if row_id := row["id"]:
201 schema, value = row_id.split(":", 1)
202 if schema in self.schemas:
203 if schema not in index:
204 index[schema] = {}
205 index[schema][value] = row["meta"]
206 return index
208 def id_action(self, ids):
209 idslist = RE_ONE_OR_MORE_SPACES.split(ids)
210 # publication id
211 for identifier in idslist:
212 if "omid:" in identifier:
213 identifier = identifier.replace("omid:", "")
214 url = self.url + identifier
215 preexisting_entity = url in self.finder
216 self.row_meta = identifier
217 preexisting_graph = (
218 self.finder.graph.subgraph(url) if preexisting_entity else None
219 )
220 self.br_graph = self.setgraph.add_br(
221 self.resp_agent,
222 source=self.src,
223 res=url,
224 preexisting_graph=preexisting_graph,
225 )
226 for identifier in idslist:
227 self.id_creator(self.br_graph, identifier, ra=False)
229 def title_action(self, title):
230 if title:
231 self.br_graph.has_title(title)
233 def author_action(self, authors):
234 if authors:
235 authorslist = RE_SEMICOLON_IN_PEOPLE_FIELD.split(authors)
236 aut_role_list = list()
237 seen_ar = set()
238 for aut in authorslist:
239 author_name, aut_id = split_name_and_ids(aut)
240 aut_id_list = aut_id.split(" ")
241 author_ra = None
242 aut_meta = ""
243 for identifier in aut_id_list:
244 if "omid:" in identifier:
245 identifier = str(identifier).replace("omid:", "")
246 url = self.url + identifier
247 preexisting_entity = url in self.finder
248 aut_meta = identifier
249 preexisting_graph = (
250 self.finder.graph.subgraph(url)
251 if preexisting_entity
252 else None
253 )
254 author_ra = self.setgraph.add_ra(
255 self.resp_agent,
256 source=self.src,
257 res=url,
258 preexisting_graph=preexisting_graph,
259 )
260 if "," in author_name:
261 author_name_splitted = RE_COMMA_AND_SPACES.split(
262 author_name
263 )
264 first_name = author_name_splitted[1]
265 last_name = author_name_splitted[0]
266 if first_name.strip():
267 author_ra.has_given_name(first_name)
268 author_ra.has_family_name(last_name)
269 else:
270 author_ra.has_name(author_name)
271 assert author_ra is not None
272 for identifier in aut_id_list:
273 self.id_creator(author_ra, identifier, ra=True)
274 ar_meta = self.ar_index[self.row_meta]["author"][aut_meta]
275 if ar_meta in seen_ar:
276 continue
277 seen_ar.add(ar_meta)
278 ar_url = self.url + ar_meta
279 preexisting_entity = ar_url in self.finder
280 preexisting_graph = (
281 self.finder.graph.subgraph(ar_url) if preexisting_entity else None
282 )
283 author_ra_role = self.setgraph.add_ar(
284 self.resp_agent,
285 source=self.src,
286 res=ar_url,
287 preexisting_graph=preexisting_graph,
288 )
289 author_ra_role.create_author()
290 self.br_graph.has_contributor(author_ra_role)
291 author_ra_role.is_held_by(author_ra)
292 aut_role_list.append(author_ra_role)
293 if len(aut_role_list) > 1:
294 aut_role_list[-2].has_next(author_ra_role)
296 def pub_date_action(self, pub_date):
297 if pub_date:
298 datelist: list[int | None] = [int(x) for x in pub_date.split("-")]
299 str_date = create_date(datelist)
300 if str_date:
301 self.br_graph.has_pub_date(str_date)
303 def vvi_action(self, venue, vol, issue):
304 if venue:
305 venue_title, venue_ids = split_name_and_ids(venue)
306 venue_ids_list = venue_ids.split()
307 for identifier in venue_ids_list:
308 if "omid:" in identifier:
309 ven_id = str(identifier).replace("omid:", "")
310 self.venue_meta = ven_id
311 url = self.url + ven_id
312 preexisting_entity = url in self.finder
313 preexisting_graph = (
314 self.finder.graph.subgraph(url) if preexisting_entity else None
315 )
316 self.venue_graph = self.setgraph.add_br(
317 self.resp_agent,
318 source=self.src,
319 res=url,
320 preexisting_graph=preexisting_graph,
321 )
322 venue_type = self.get_venue_type(self.type, venue_ids_list)
323 if venue_type:
324 venue_type = venue_type.replace(" ", "_")
325 getattr(self.venue_graph, f"create_{venue_type}")()
326 self.venue_graph.has_title(venue_title)
327 assert self.venue_graph is not None
328 for identifier in venue_ids_list:
329 self.id_creator(self.venue_graph, identifier, ra=False)
330 if self.type in {"journal article", "journal volume", "journal issue"}:
331 if vol:
332 vol_meta = self.vi_index[self.venue_meta]["volume"][vol]["id"]
333 vol_url = self.url + vol_meta
334 preexisting_entity = vol_url in self.finder
335 preexisting_graph = (
336 self.finder.graph.subgraph(vol_url)
337 if preexisting_entity
338 else None
339 )
340 self.vol_graph = self.setgraph.add_br(
341 self.resp_agent,
342 source=self.src,
343 res=vol_url,
344 preexisting_graph=preexisting_graph,
345 )
346 self.vol_graph.create_volume()
347 self.vol_graph.has_number(vol)
348 if issue:
349 if vol:
350 issue_meta = self.vi_index[self.venue_meta]["volume"][vol][
351 "issue"
352 ][issue]["id"]
353 else:
354 issue_meta = self.vi_index[self.venue_meta]["issue"][issue][
355 "id"
356 ]
357 issue_url = self.url + issue_meta
358 preexisting_entity = issue_url in self.finder
359 preexisting_graph = (
360 self.finder.graph.subgraph(issue_url)
361 if preexisting_entity
362 else None
363 )
364 self.issue_graph = self.setgraph.add_br(
365 self.resp_agent,
366 source=self.src,
367 res=issue_url,
368 preexisting_graph=preexisting_graph,
369 )
370 self.issue_graph.create_issue()
371 self.issue_graph.has_number(issue)
372 if venue and vol and issue:
373 assert self.issue_graph is not None
374 assert self.vol_graph is not None
375 assert self.venue_graph is not None
376 self.br_graph.is_part_of(self.issue_graph)
377 self.issue_graph.is_part_of(self.vol_graph)
378 self.vol_graph.is_part_of(self.venue_graph)
379 elif venue and vol and not issue:
380 assert self.vol_graph is not None
381 assert self.venue_graph is not None
382 self.br_graph.is_part_of(self.vol_graph)
383 self.vol_graph.is_part_of(self.venue_graph)
384 elif venue and not vol and not issue:
385 assert self.venue_graph is not None
386 self.br_graph.is_part_of(self.venue_graph)
387 elif venue and not vol and issue:
388 assert self.issue_graph is not None
389 assert self.venue_graph is not None
390 self.br_graph.is_part_of(self.issue_graph)
391 self.issue_graph.is_part_of(self.venue_graph)
393 @classmethod
394 def get_venue_type(cls, br_type: str, venue_ids: list) -> str:
395 schemas = {venue_id.split(":", maxsplit=1)[0] for venue_id in venue_ids}
396 venue_type = ""
397 if br_type in {"journal article", "journal volume", "journal issue"}:
398 venue_type = "journal"
399 elif br_type in {"book chapter", "book part", "book section", "book track"}:
400 venue_type = "book"
401 elif br_type in {"book", "edited book", "monograph", "reference book"}:
402 venue_type = "book series"
403 elif br_type == "proceedings article":
404 venue_type = "proceedings"
405 elif br_type in {"proceedings", "report", "standard", "series"}:
406 venue_type = "series"
407 elif br_type == "reference entry":
408 venue_type = "reference book"
409 elif br_type == "report series":
410 venue_type = "report series"
411 elif not br_type or br_type in {"dataset", "data file"}:
412 venue_type = ""
413 # Check the type based on the identifier scheme
414 if any(
415 identifier for identifier in venue_ids if not identifier.startswith("omid:")
416 ):
417 if venue_type in {"journal", "book series", "series", "report series"}:
418 if "isbn" in schemas or "issn" not in schemas:
419 # It is undecidable
420 venue_type = ""
421 elif venue_type in {"book", "proceedings"}:
422 if "issn" in schemas or "isbn" not in schemas:
423 venue_type = ""
424 elif venue_type == "reference book":
425 if "isbn" in schemas and "issn" not in schemas:
426 venue_type = "reference book"
427 elif "issn" in schemas and "isbn" not in schemas:
428 venue_type = "journal"
429 elif "issn" in schemas and "isbn" in schemas:
430 venue_type = ""
431 return venue_type
433 def page_action(self, page):
434 if page:
435 re_meta = self.re_index[self.row_meta]
436 re_url = self.url + re_meta
437 preexisting_entity = re_url in self.finder
438 preexisting_graph = (
439 self.finder.graph.subgraph(re_url) if preexisting_entity else None
440 )
441 form = self.setgraph.add_re(
442 self.resp_agent,
443 source=self.src,
444 res=re_url,
445 preexisting_graph=preexisting_graph,
446 )
447 form.has_starting_page(page)
448 form.has_ending_page(page)
449 self.br_graph.has_format(form)
451 _TYPE_TO_METHOD = {
452 "abstract": "create_abstract",
453 "archival document": "create_archival_document",
454 "audio document": "create_audio_document",
455 "book": "create_book",
456 "book chapter": "create_book_chapter",
457 "book part": "create_book_part",
458 "book section": "create_book_section",
459 "book series": "create_book_series",
460 "book set": "create_book_set",
461 "computer program": "create_computer_program",
462 "data file": "create_dataset",
463 "dataset": "create_dataset",
464 "data management plan": "create_data_management_plan",
465 "dissertation": "create_dissertation",
466 "editorial": "create_editorial",
467 "journal": "create_journal",
468 "journal article": "create_journal_article",
469 "journal editorial": "create_journal_editorial",
470 "journal issue": "create_issue",
471 "journal volume": "create_volume",
472 "newspaper": "create_newspaper",
473 "newspaper article": "create_newspaper_article",
474 "newspaper issue": "create_newspaper_issue",
475 "peer review": "create_peer_review",
476 "preprint": "create_preprint",
477 "presentation": "create_presentation",
478 "proceedings": "create_proceedings",
479 "proceedings article": "create_proceedings_article",
480 "reference book": "create_reference_book",
481 "reference entry": "create_reference_entry",
482 "report": "create_report",
483 "report series": "create_report_series",
484 "retraction notice": "create_retraction_notice",
485 "standard": "create_standard",
486 "series": "create_series",
487 "web content": "create_web_content",
488 }
490 def type_action(self, entity_type):
491 method_name = self._TYPE_TO_METHOD.get(entity_type)
492 if method_name:
493 getattr(self.br_graph, method_name)()
495 def publisher_action(self, publisher):
496 if publisher:
497 publishers_list = RE_SEMICOLON_IN_PEOPLE_FIELD.split(publisher)
498 pub_role_list = list()
499 seen_ar = set()
500 for pub in publishers_list:
501 publ_name, publ_id = split_name_and_ids(pub)
502 publ_id_list = publ_id.split()
503 publisher_ra = None
504 pub_meta = ""
505 for identifier in publ_id_list:
506 if "omid:" in identifier:
507 identifier = str(identifier).replace("omid:", "")
508 pub_meta = identifier
509 url = self.url + identifier
510 preexisting_entity = url in self.finder
511 preexisting_graph = (
512 self.finder.graph.subgraph(url)
513 if preexisting_entity
514 else None
515 )
516 publisher_ra = self.setgraph.add_ra(
517 self.resp_agent,
518 source=self.src,
519 res=url,
520 preexisting_graph=preexisting_graph,
521 )
522 publisher_ra.has_name(publ_name)
523 assert publisher_ra is not None
524 for identifier in publ_id_list:
525 self.id_creator(publisher_ra, identifier, ra=True)
526 ar_meta = self.ar_index[self.row_meta]["publisher"][pub_meta]
527 if ar_meta in seen_ar:
528 continue
529 seen_ar.add(ar_meta)
530 ar_url = self.url + ar_meta
531 preexisting_entity = ar_url in self.finder
532 preexisting_graph = (
533 self.finder.graph.subgraph(ar_url) if preexisting_entity else None
534 )
535 publ_role = self.setgraph.add_ar(
536 self.resp_agent,
537 source=self.src,
538 res=ar_url,
539 preexisting_graph=preexisting_graph,
540 )
541 publ_role.create_publisher()
542 self.br_graph.has_contributor(publ_role)
543 publ_role.is_held_by(publisher_ra)
544 pub_role_list.append(publ_role)
545 if len(pub_role_list) > 1:
546 pub_role_list[-2].has_next(publ_role)
548 def editor_action(self, editor, row):
549 if editor:
550 editorslist = RE_SEMICOLON_IN_PEOPLE_FIELD.split(editor)
551 edit_role_list = list()
552 seen_ar = set()
553 for ed in editorslist:
554 editor_name, ed_id = split_name_and_ids(ed)
555 ed_id_list = ed_id.split(" ")
556 editor_ra = None
557 ed_meta = ""
558 for identifier in ed_id_list:
559 if "omid:" in identifier:
560 identifier = str(identifier).replace("omid:", "")
561 ed_meta = identifier
562 url = self.url + identifier
563 preexisting_entity = url in self.finder
564 preexisting_graph = (
565 self.finder.graph.subgraph(url)
566 if preexisting_entity
567 else None
568 )
569 editor_ra = self.setgraph.add_ra(
570 self.resp_agent,
571 source=self.src,
572 res=url,
573 preexisting_graph=preexisting_graph,
574 )
575 if "," in editor_name:
576 editor_name_splitted = RE_COMMA_AND_SPACES.split(
577 editor_name
578 )
579 firstName = editor_name_splitted[1]
580 lastName = editor_name_splitted[0]
581 if firstName.strip():
582 editor_ra.has_given_name(firstName)
583 editor_ra.has_family_name(lastName)
584 else:
585 editor_ra.has_name(editor_name)
586 assert editor_ra is not None
587 for identifier in ed_id_list:
588 self.id_creator(editor_ra, identifier, ra=True)
589 br_key = get_edited_br_metaid(row, self.row_meta, self.venue_meta)
590 ar_meta = self.ar_index[br_key]["editor"][ed_meta]
591 if ar_meta in seen_ar:
592 continue
593 seen_ar.add(ar_meta)
594 ar_url = self.url + ar_meta
595 preexisting_entity = ar_url in self.finder
596 preexisting_graph = (
597 self.finder.graph.subgraph(ar_url) if preexisting_entity else None
598 )
599 editor_ra_role = self.setgraph.add_ar(
600 self.resp_agent,
601 source=self.src,
602 res=ar_url,
603 preexisting_graph=preexisting_graph,
604 )
605 editor_ra_role.create_editor()
606 br_graphs: List[BibliographicResource] = [
607 g
608 for g in [
609 self.br_graph,
610 self.issue_graph,
611 self.vol_graph,
612 self.venue_graph,
613 ]
614 if g is not None
615 ]
616 for graph in br_graphs:
617 if br_key == self.__res_metaid(graph):
618 graph.has_contributor(editor_ra_role)
619 editor_ra_role.is_held_by(editor_ra)
620 edit_role_list.append(editor_ra_role)
621 if len(edit_role_list) > 1:
622 edit_role_list[-2].has_next(editor_ra_role)
624 def __res_metaid(self, graph: BibliographicResource):
625 return graph.res.replace(self.url, "")
627 def id_creator(self, graph: BibliographicEntity, identifier: str, ra: bool) -> None:
628 new_id = None
629 # Skip temporary identifiers - they should not be saved in the final dataset
630 if identifier.startswith("temp:"):
631 return
633 if ra:
634 for ra_id_schema in self.ra_id_schemas:
635 if identifier.startswith(f"{ra_id_schema}:"):
636 identifier = identifier.split(":", 1)[1]
637 res = self.ra_index[ra_id_schema][identifier]
638 url = self.url + res
639 preexisting_entity = url in self.finder
640 preexisting_graph = (
641 self.finder.graph.subgraph(url) if preexisting_entity else None
642 )
643 new_id = self.setgraph.add_id(
644 self.resp_agent,
645 source=self.src,
646 res=url,
647 preexisting_graph=preexisting_graph,
648 )
649 getattr(new_id, f"create_{ra_id_schema}")(identifier)
650 else:
651 for br_id_schema in self.br_id_schemas:
652 if identifier.startswith(f"{br_id_schema}:"):
653 identifier = identifier.split(":", 1)[1]
654 res = self.br_index[br_id_schema][identifier]
655 url = self.url + res
656 preexisting_entity = url in self.finder
657 preexisting_graph = (
658 self.finder.graph.subgraph(url) if preexisting_entity else None
659 )
660 new_id = self.setgraph.add_id(
661 self.resp_agent,
662 source=self.src,
663 res=url,
664 preexisting_graph=preexisting_graph,
665 )
666 getattr(new_id, f"create_{br_id_schema}")(identifier)
667 if new_id:
668 graph.has_identifier(new_id)