Coverage for ramose/skg_if/_base.py: 86%
732 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-24 21:59 +0000
1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import csv
6import json
7from collections.abc import Callable
8from dataclasses import dataclass
9from dataclasses import field as dataclass_field
10from io import StringIO
11from math import ceil
12from re import sub
13from typing import NoReturn
14from urllib.parse import parse_qs, urlencode, urlsplit
16from ramose import ConversionResult, HttpError
18_YEAR_MONTH_PART_COUNT = 2
19_UNPROCESSABLE_CONTENT = 422
21_PRODUCT_COLUMNS: dict[str, str] = dict.fromkeys(
22 (
23 "local_identifier",
24 "identifier_scheme",
25 "identifier_value",
26 "title",
27 "title_lang",
28 "abstract",
29 "abstract_lang",
30 "product_type",
31 "affiliation_name",
32 "affiliation_short_name",
33 "affiliation_country",
34 "affiliation_local_identifier",
35 "affiliation_identifier_scheme",
36 "affiliation_identifier_value",
37 "affiliation_type",
38 "affiliation_website",
39 "affiliation_other_name",
40 "affiliation_role",
41 "affiliation_period_start",
42 "affiliation_period_end",
43 "_affiliation_key",
44 "topic_term",
45 "topic_identifier_scheme",
46 "topic_identifier_value",
47 "topic_label",
48 "topic_label_lang",
49 "topic_provenance_associated_with",
50 "topic_provenance_trust",
51 "contribution_by_family_name",
52 "contribution_by_given_name",
53 "contribution_by_name",
54 "contribution_by_identifier_scheme",
55 "contribution_by_identifier_value",
56 "contribution_by_local_identifier",
57 "contribution_role",
58 "contribution_type",
59 "_contribution_key",
60 "_contribution_next_key",
61 "contribution_declared_affiliation_name",
62 "contribution_declared_affiliation_short_name",
63 "contribution_declared_affiliation_country",
64 "contribution_declared_affiliation_local_identifier",
65 "contribution_declared_affiliation_identifier_scheme",
66 "contribution_declared_affiliation_identifier_value",
67 "contribution_declared_affiliation_type",
68 "contribution_declared_affiliation_website",
69 "contribution_declared_affiliation_other_name",
70 "manifestation_type_class",
71 "manifestation_type_label",
72 "manifestation_type_label_lang",
73 "manifestation_type_defined_in",
74 "_manifestation_key",
75 "manifestation_identifier_scheme",
76 "manifestation_identifier_value",
77 "manifestation_dates_type",
78 "manifestation_dates_value",
79 "manifestation_peer_review_status",
80 "manifestation_peer_review_description",
81 "manifestation_access_rights_status",
82 "manifestation_access_rights_description",
83 "manifestation_licence",
84 "manifestation_version",
85 "manifestation_biblio_volume",
86 "manifestation_biblio_issue",
87 "manifestation_biblio_edition",
88 "manifestation_biblio_number",
89 "manifestation_biblio_pages_first",
90 "manifestation_biblio_pages_last",
91 "manifestation_biblio_in_name",
92 "manifestation_biblio_in_local_identifier",
93 "manifestation_biblio_in_identifier_scheme",
94 "manifestation_biblio_in_identifier_value",
95 "manifestation_biblio_in_acronym",
96 "manifestation_biblio_hosting_data_source_local_identifier",
97 "manifestation_biblio_hosting_data_source_name",
98 "manifestation_biblio_hosting_data_source_identifier_scheme",
99 "manifestation_biblio_hosting_data_source_identifier_value",
100 "related_products_cites",
101 "related_products_is_supplemented_by",
102 "related_products_is_documented_by",
103 "related_products_is_new_version_of",
104 "related_products_is_part_of",
105 "funding_local_identifier",
106 "funding_grant_number",
107 "funding_title",
108 "funding_title_lang",
109 "funding_abstract",
110 "funding_abstract_lang",
111 "funding_acronym",
112 "funding_identifier_scheme",
113 "funding_identifier_value",
114 "funding_stream",
115 "funding_agency_name",
116 "funding_agency_short_name",
117 "funding_agency_country",
118 "funding_agency_local_identifier",
119 "funding_agency_identifier_scheme",
120 "funding_agency_identifier_value",
121 "funding_agency_type",
122 "funding_agency_website",
123 "beneficiaries_name",
124 "beneficiaries_short_name",
125 "beneficiaries_country",
126 "beneficiaries_local_identifier",
127 "beneficiaries_identifier_scheme",
128 "beneficiaries_identifier_value",
129 "beneficiaries_type",
130 "beneficiaries_website",
131 "relevant_organisation_name",
132 "relevant_organisation_short_name",
133 "relevant_organisation_country",
134 "relevant_organisation_local_identifier",
135 "relevant_organisation_identifier_scheme",
136 "relevant_organisation_identifier_value",
137 "relevant_organisation_type",
138 "relevant_organisation_website",
139 "relevant_organisation_other_name",
140 "research_product_types",
141 "persistent_identity_system_for",
142 "persistent_identity_system_pid_scheme",
143 "_persistent_identity_system_key",
144 "audience_type",
145 "disciplines",
146 "policy_about",
147 "policy_target",
148 "policy_documented_at",
149 "policy_description",
150 "_policy_key",
151 ),
152 "",
153)
155SKGIF_CONTEXT_URLS = [
156 "https://w3id.org/skg-if/context/1.1.0/skg-if.json",
157 "https://w3id.org/skg-if/context/1.0.0/skg-if-api.json",
158]
161def _build_context(base_url: str) -> list:
162 return [*SKGIF_CONTEXT_URLS, {"@base": f"{base_url.rstrip('/')}/"}]
165def _collect_identifiers(
166 rows: list[dict],
167 scheme_col: str = "identifier_scheme",
168 value_col: str = "identifier_value",
169) -> list[dict]:
170 seen = set()
171 identifiers = []
172 for row in rows:
173 scheme = row[scheme_col]
174 value = row[value_col]
175 if scheme and value and (scheme, value) not in seen:
176 seen.add((scheme, value))
177 identifiers.append({"value": value, "scheme": scheme})
178 return identifiers
181def _order_linked_list(items: dict[str, dict], next_map: dict[str, str | None]) -> list[dict]:
182 if not items:
183 return []
185 next_values = set(next_map.values()) - {None}
186 start_candidates = [key for key in items if key not in next_values]
187 if not start_candidates:
188 return list(items.values())
190 ordered = []
191 current = start_candidates[0]
192 visited = set()
193 while current and current in items and current not in visited:
194 visited.add(current)
195 ordered.append(items[current])
196 current = next_map.get(current)
198 for key, contributor in items.items():
199 if key not in visited:
200 ordered.append(contributor)
202 return ordered
205def _merge_agent_identifier(agent: dict, row: dict) -> None:
206 scheme = row["contribution_by_identifier_scheme"]
207 value = row["contribution_by_identifier_value"]
208 if not scheme or not value:
209 return
210 identifier = {"value": value, "scheme": scheme}
211 identifiers = agent.setdefault("identifiers", [])
212 if identifier not in identifiers:
213 identifiers.append(identifier)
216def _build_agent(row: dict, non_person_entity_type: str) -> dict | None:
217 family_name = row["contribution_by_family_name"]
218 given_name = row["contribution_by_given_name"]
219 full_name = row["contribution_by_name"]
220 agent_local_id = row["contribution_by_local_identifier"]
222 is_person = bool(family_name or given_name)
224 if is_person:
225 display_name = f"{family_name}, {given_name}" if family_name and given_name else (family_name or given_name)
226 entity_type = "person"
227 elif full_name:
228 display_name = full_name
229 entity_type = non_person_entity_type
230 else:
231 return None
233 agent: dict = {"name": display_name, "entity_type": entity_type}
234 if is_person:
235 if family_name:
236 agent["family_name"] = family_name
237 if given_name:
238 agent["given_name"] = given_name
239 _merge_agent_identifier(agent, row)
240 if not agent_local_id:
241 msg = f"Missing required local_identifier for {entity_type} '{display_name}'"
242 raise ValueError(msg)
243 agent["local_identifier"] = agent_local_id
244 return agent
247def _build_org(row: dict, prefix: str) -> dict:
248 org: dict = {"entity_type": "organisation"}
249 name = row[f"{prefix}_name"]
250 if name:
251 org["name"] = name
252 for field in ("short_name", "country", "website"):
253 val = row[f"{prefix}_{field}"]
254 if val:
255 org[field] = val
256 local_id = row[f"{prefix}_local_identifier"]
257 if not local_id:
258 msg = f"Missing required local_identifier for organisation '{row[f'{prefix}_name']}'"
259 raise ValueError(msg)
260 org["local_identifier"] = local_id
261 return org
264def _org_entry(row: dict, prefix: str) -> dict:
265 return {"obj": _build_org(row, prefix), "seen_ids": set(), "seen_types": set(), "seen_other_names": set()}
268def _merge_org_multivalued(entry: dict, row: dict, prefix: str) -> None:
269 id_scheme = row[f"{prefix}_identifier_scheme"]
270 id_value = row[f"{prefix}_identifier_value"]
271 if id_scheme and id_value and (id_scheme, id_value) not in entry["seen_ids"]:
272 entry["seen_ids"].add((id_scheme, id_value))
273 entry["obj"].setdefault("identifiers", []).append({"value": id_value, "scheme": id_scheme})
274 org_type = row[f"{prefix}_type"]
275 if org_type and org_type not in entry["seen_types"]:
276 entry["seen_types"].add(org_type)
277 entry["obj"].setdefault("types", []).append(org_type)
278 if f"{prefix}_other_name" in row:
279 other_name = row[f"{prefix}_other_name"]
280 if other_name and other_name not in entry["seen_other_names"]:
281 entry["seen_other_names"].add(other_name)
282 entry["obj"].setdefault("other_names", []).append(other_name)
285_DECLARED_AFFILIATION_PREFIX = "contribution_declared_affiliation"
288def _merge_declared_affiliation(store: dict, row: dict) -> None:
289 prefix = _DECLARED_AFFILIATION_PREFIX
290 aff_name = row[f"{prefix}_name"]
291 aff_local_id = row[f"{prefix}_local_identifier"]
292 if not aff_name and not aff_local_id:
293 return
294 if aff_local_id not in store:
295 store[aff_local_id] = _org_entry(row, prefix)
296 _merge_org_multivalued(store[aff_local_id], row, prefix)
299def _collect_declared_affiliations(rows: list[dict], role: str, key: str, store: dict) -> None:
300 role_store = store.setdefault(role, {}).setdefault(key, {})
301 for row in rows:
302 if row["contribution_role"] == role and row["_contribution_key"] == key:
303 _merge_declared_affiliation(role_store, row)
306def _enrich_contributor(
307 contributor: dict,
308 key: str,
309 role_type: str,
310 contribution_types: dict,
311 affiliations: dict,
312) -> None:
313 types = contribution_types.get(role_type, {}).get(key)
314 if types:
315 contributor["contribution_types"] = types
316 affs = affiliations.get(role_type, {}).get(key)
317 if affs:
318 contributor["declared_affiliations"] = [entry["obj"] for entry in affs.values()]
321@dataclass
322class _ContributorAccumulator:
323 by_role_type: dict[str, dict[str, dict]] = dataclass_field(default_factory=dict)
324 next_map: dict[str, dict[str, str | None]] = dataclass_field(default_factory=dict)
325 types: dict[str, dict[str, list[str]]] = dataclass_field(default_factory=dict)
326 affiliations: dict[str, dict[str, dict]] = dataclass_field(default_factory=dict)
329def _process_contributor_row(
330 row: dict,
331 rows: list[dict],
332 acc: _ContributorAccumulator,
333) -> None:
334 role = row["contribution_role"]
335 key = row["_contribution_key"]
336 if not role or not key:
337 return
339 if role not in acc.by_role_type:
340 acc.by_role_type[role] = {}
341 acc.next_map[role] = {}
342 acc.types[role] = {}
344 contribution_type = row["contribution_type"]
345 if contribution_type:
346 type_list = acc.types[role].setdefault(key, [])
347 if contribution_type not in type_list:
348 type_list.append(contribution_type)
350 if key in acc.by_role_type[role]:
351 existing = acc.by_role_type[role][key]
352 _merge_agent_identifier(existing["by"], row)
353 return
355 _collect_declared_affiliations(rows, role, key, acc.affiliations)
356 agent = _build_agent(row, "organisation" if role == "publisher" else "agent")
357 if not agent:
358 return
359 acc.by_role_type[role][key] = {"role": role, "by": agent}
360 acc.next_map[role][key] = row["_contribution_next_key"] or None
363def _collect_contributors(rows: list[dict]) -> list[dict]:
364 acc = _ContributorAccumulator()
366 for row in rows:
367 _process_contributor_row(row, rows, acc)
369 result = []
370 for role_type in ["author", "editor", "publisher"]:
371 if role_type not in acc.by_role_type:
372 continue
373 ordered = _order_linked_list(acc.by_role_type[role_type], acc.next_map[role_type])
374 for rank, contributor in enumerate(ordered, start=1):
375 contributor["rank"] = rank
376 key = next(k for k, v in acc.by_role_type[role_type].items() if v is contributor)
377 _enrich_contributor(contributor, key, role_type, acc.types, acc.affiliations)
378 result.append(contributor)
380 return result
383def _collect_grouped_contributors(rows: list[dict]) -> list[dict]:
384 contributions: dict[str, dict] = {}
385 affiliations: dict[str, dict] = {}
387 for row in rows:
388 key = row["_contribution_key"]
389 if not key:
390 continue
391 if key not in contributions:
392 agent = _build_agent(row, "organisation")
393 if not agent:
394 continue
395 contributions[key] = {"by": agent, "roles": []}
396 affiliations[key] = {}
397 role = row["contribution_role"]
398 if role and role not in contributions[key]["roles"]:
399 contributions[key]["roles"].append(role)
400 _merge_agent_identifier(contributions[key]["by"], row)
401 _merge_declared_affiliation(affiliations[key], row)
403 for key, contribution in contributions.items():
404 if not contribution["roles"]:
405 del contribution["roles"]
406 if affiliations[key]:
407 contribution["declared_affiliations"] = [entry["obj"] for entry in affiliations[key].values()]
408 return list(contributions.values())
411def _build_venue(rows: list[dict], venue_name: str, venue_local_id: str) -> dict:
412 venue: dict = {"name": venue_name, "entity_type": "venue"}
413 if not venue_local_id:
414 msg = f"Missing required local_identifier for venue '{venue_name}'"
415 raise ValueError(msg)
416 venue["local_identifier"] = venue_local_id
417 acronym = rows[0]["manifestation_biblio_in_acronym"]
418 if acronym:
419 venue["acronym"] = acronym
421 venue_ids_seen = set()
422 venue_identifiers = []
423 for row in rows:
424 venue_scheme = row["manifestation_biblio_in_identifier_scheme"]
425 venue_value = row["manifestation_biblio_in_identifier_value"]
426 if venue_scheme and venue_value and (venue_scheme, venue_value) not in venue_ids_seen:
427 venue_ids_seen.add((venue_scheme, venue_value))
428 venue_identifiers.append({"value": venue_value, "scheme": venue_scheme})
429 if venue_identifiers:
430 venue["identifiers"] = venue_identifiers
431 return venue
434def _normalize_datetime(date_str: str) -> str:
435 parts = date_str.split("-")
436 if len(parts) == 1:
437 return f"{parts[0]}-01-01T00:00:00"
438 if len(parts) == _YEAR_MONTH_PART_COUNT:
439 return f"{parts[0]}-{parts[1]}-01T00:00:00"
440 if "T" not in date_str:
441 return f"{date_str}T00:00:00"
442 return date_str
445def _collect_manifestation_dates(rows: list[dict]) -> dict[str, list[str]]:
446 dates: dict[str, list[str]] = {}
447 seen: set[tuple[str, str]] = set()
448 for row in rows:
449 date_type = row["manifestation_dates_type"]
450 date_value = row["manifestation_dates_value"]
451 if date_type and date_value and (date_type, date_value) not in seen:
452 seen.add((date_type, date_value))
453 dates.setdefault(date_type, []).append(_normalize_datetime(date_value))
454 return dates
457_BIBLIO_SIMPLE_FIELDS: tuple[tuple[str, str], ...] = (
458 ("manifestation_biblio_volume", "volume"),
459 ("manifestation_biblio_issue", "issue"),
460 ("manifestation_biblio_edition", "edition"),
461 ("manifestation_biblio_number", "number"),
462)
465def _build_biblio_venue(rows: list[dict], first_row: dict) -> dict | None:
466 venue_name = first_row["manifestation_biblio_in_name"]
467 if not venue_name:
468 return None
469 venue_local_id = first_row["manifestation_biblio_in_local_identifier"]
470 return _build_venue(rows, venue_name, venue_local_id)
473def _build_biblio_hosting(rows: list[dict], first_row: dict) -> dict | None:
474 hosting_local_id = first_row["manifestation_biblio_hosting_data_source_local_identifier"]
475 if not hosting_local_id:
476 return None
477 hosting: dict = {"local_identifier": hosting_local_id, "entity_type": "datasource"}
478 hosting_name = first_row["manifestation_biblio_hosting_data_source_name"]
479 if hosting_name:
480 hosting["name"] = hosting_name
481 hosting_identifiers = _collect_identifiers(
482 rows,
483 "manifestation_biblio_hosting_data_source_identifier_scheme",
484 "manifestation_biblio_hosting_data_source_identifier_value",
485 )
486 if hosting_identifiers:
487 hosting["identifiers"] = hosting_identifiers
488 return hosting
491def _build_biblio(rows: list[dict]) -> dict:
492 first_row = rows[0]
493 biblio: dict = {}
494 for sparql_var, json_key in _BIBLIO_SIMPLE_FIELDS:
495 value = first_row[sparql_var]
496 if value:
497 biblio[json_key] = value
498 first_page = first_row["manifestation_biblio_pages_first"]
499 last_page = first_row["manifestation_biblio_pages_last"]
500 if first_page and last_page:
501 biblio["pages"] = {"first": first_page, "last": last_page}
502 venue = _build_biblio_venue(rows, first_row)
503 if venue:
504 biblio["in"] = venue
505 hosting = _build_biblio_hosting(rows, first_row)
506 if hosting:
507 biblio["hosting_data_source"] = hosting
508 return biblio
511def _build_manifestation_type(first_row: dict) -> dict | None:
512 type_class = first_row["manifestation_type_class"]
513 type_label = first_row["manifestation_type_label"]
514 defined_in = first_row["manifestation_type_defined_in"]
515 if not type_class and not type_label and not defined_in:
516 return None
517 manifestation_type: dict = {}
518 if type_class:
519 manifestation_type["class"] = type_class
520 if type_label:
521 label_lang = first_row["manifestation_type_label_lang"] or "none"
522 manifestation_type["labels"] = {label_lang: type_label}
523 if defined_in:
524 manifestation_type["defined_in"] = defined_in
525 return manifestation_type
528def _build_status_with_description(first_row: dict, status_field: str, desc_field: str) -> dict | None:
529 status = first_row[status_field]
530 if not status:
531 return None
532 result: dict = {"status": status}
533 desc = first_row[desc_field]
534 if desc:
535 result["description"] = desc
536 return result
539def _build_manifestation(rows: list[dict]) -> dict | None:
540 first_row = rows[0]
541 manifestation: dict = {}
543 manifestation_type = _build_manifestation_type(first_row)
544 if manifestation_type:
545 manifestation["type"] = manifestation_type
547 dates = _collect_manifestation_dates(rows)
548 if dates:
549 manifestation["dates"] = dates
551 identifiers = _collect_identifiers(rows, "manifestation_identifier_scheme", "manifestation_identifier_value")
552 if identifiers:
553 manifestation["identifiers"] = identifiers
555 peer_review = _build_status_with_description(
556 first_row, "manifestation_peer_review_status", "manifestation_peer_review_description"
557 )
558 if peer_review:
559 manifestation["peer_review"] = peer_review
561 access_rights = _build_status_with_description(
562 first_row, "manifestation_access_rights_status", "manifestation_access_rights_description"
563 )
564 if access_rights:
565 manifestation["access_rights"] = access_rights
567 licence = first_row["manifestation_licence"]
568 if licence:
569 manifestation["licence"] = licence
571 version = first_row["manifestation_version"]
572 if version:
573 manifestation["version"] = version
575 biblio = _build_biblio(rows)
576 if biblio:
577 manifestation["biblio"] = biblio
579 return manifestation or None
582def _collect_manifestations(rows: list[dict]) -> list[dict]:
583 groups: dict[str, list[dict]] = {}
584 for row in rows:
585 if not any(
586 value
587 for column, value in row.items()
588 if column.startswith("manifestation_") and column != "_manifestation_key"
589 ):
590 continue
591 groups.setdefault(row["_manifestation_key"], []).append(row)
592 return [manifestation for group in groups.values() if (manifestation := _build_manifestation(group))]
595_RELATED_PRODUCT_COLUMNS = [
596 "related_products_cites",
597 "related_products_is_supplemented_by",
598 "related_products_is_documented_by",
599 "related_products_is_new_version_of",
600 "related_products_is_part_of",
601]
604def _collect_related_products(rows: list[dict]) -> dict:
605 result: dict[str, list[str]] = {}
606 for column in _RELATED_PRODUCT_COLUMNS:
607 key = column.replace("related_products_", "")
608 seen: set[str] = set()
609 values: list[str] = []
610 for row in rows:
611 val = row[column]
612 if val and val not in seen:
613 seen.add(val)
614 values.append(val)
615 if values:
616 result[key] = values
617 return result
620def _collect_topics(rows: list[dict]) -> list[dict]:
621 topics_by_uri: dict[str, dict] = {}
622 seen_identifiers: dict[str, set] = {}
623 seen_provenance: dict[str, set] = {}
625 for row in rows:
626 uri = row["topic_term"]
627 if not uri:
628 continue
630 if uri not in topics_by_uri:
631 topics_by_uri[uri] = {"term": {"local_identifier": uri, "entity_type": "topic"}}
632 seen_identifiers[uri] = set()
633 seen_provenance[uri] = set()
635 topic = topics_by_uri[uri]
636 term = topic["term"]
638 label = row["topic_label"]
639 if label:
640 lang = row["topic_label_lang"] or "none"
641 term.setdefault("labels", {})[lang] = label
643 id_scheme = row["topic_identifier_scheme"]
644 id_value = row["topic_identifier_value"]
645 if id_scheme and id_value and (id_scheme, id_value) not in seen_identifiers[uri]:
646 seen_identifiers[uri].add((id_scheme, id_value))
647 term.setdefault("identifiers", []).append({"scheme": id_scheme, "value": id_value})
649 prov_agent = row["topic_provenance_associated_with"]
650 prov_trust = row["topic_provenance_trust"]
651 if prov_agent and prov_trust and prov_agent not in seen_provenance[uri]:
652 seen_provenance[uri].add(prov_agent)
653 topic.setdefault("provenance", []).append({"associated_with": prov_agent, "trust": float(prov_trust)})
655 return list(topics_by_uri.values())
658def _collect_organisation(rows: list[dict], prefix: str) -> list[dict]:
659 entries: dict[str, dict] = {}
660 for row in rows:
661 name = row[f"{prefix}_name"]
662 local_id = row[f"{prefix}_local_identifier"]
663 if not name and not local_id:
664 continue
665 if local_id not in entries:
666 entries[local_id] = _org_entry(row, prefix)
667 _merge_org_multivalued(entries[local_id], row, prefix)
668 return [entry["obj"] for entry in entries.values()]
671def _collect_single_organisation(rows: list[dict], prefix: str) -> dict | None:
672 base_row = next((row for row in rows if row[f"{prefix}_name"]), None)
673 if base_row is None:
674 return None
675 entry = _org_entry(base_row, prefix)
676 for row in rows:
677 _merge_org_multivalued(entry, row, prefix)
678 return entry["obj"]
681def _build_person_affiliation(rows: list[dict]) -> dict:
682 affiliation = {"affiliation": _collect_organisation(rows, "affiliation")[0]}
683 first_row = rows[0]
684 if role := first_row["affiliation_role"]:
685 affiliation["role"] = role
686 period = {
687 key: first_row[column]
688 for key, column in (("start", "affiliation_period_start"), ("end", "affiliation_period_end"))
689 if first_row[column]
690 }
691 if period:
692 affiliation["period"] = period
693 return affiliation
696def _collect_person_affiliations(rows: list[dict]) -> list[dict]:
697 groups: dict[str, list[dict]] = {}
698 for row in rows:
699 local_identifier = row["affiliation_local_identifier"]
700 if not local_identifier:
701 continue
702 key = row["_affiliation_key"] or local_identifier
703 groups.setdefault(key, []).append(row)
704 return [_build_person_affiliation(group) for group in groups.values()]
707def _build_grant(row: dict) -> dict:
708 funding_local_id = row["funding_local_identifier"]
709 if not funding_local_id:
710 msg = "Missing required local_identifier for grant"
711 raise ValueError(msg)
712 grant: dict = {"local_identifier": funding_local_id, "entity_type": "grant"}
713 for field, csv_col in (
714 ("grant_number", "funding_grant_number"),
715 ("acronym", "funding_acronym"),
716 ("funding_stream", "funding_stream"),
717 ):
718 val = row[csv_col]
719 if val:
720 grant[field] = val
721 title = row["funding_title"]
722 if title:
723 grant["titles"] = {row["funding_title_lang"] or "none": title}
724 abstract = row["funding_abstract"]
725 if abstract:
726 grant["abstracts"] = {row["funding_abstract_lang"] or "none": abstract}
727 agency_name = row["funding_agency_name"]
728 if agency_name:
729 grant["funding_agency"] = _build_org(row, "funding_agency")
730 return grant
733def _collect_funding(rows: list[dict]) -> list[dict]:
734 funding_by_key: dict[str, dict] = {}
735 seen_ids: dict[str, set[tuple[str, str]]] = {}
736 agency_trackers: dict[str, dict] = {}
738 for row in rows:
739 local_id = row["funding_local_identifier"]
740 if not local_id:
741 continue
742 if local_id not in funding_by_key:
743 funding_by_key[local_id] = _build_grant(row)
744 seen_ids[local_id] = set()
745 agency_trackers[local_id] = {"seen_ids": set(), "seen_types": set()}
747 id_scheme = row["funding_identifier_scheme"]
748 id_value = row["funding_identifier_value"]
749 if id_scheme and id_value and (id_scheme, id_value) not in seen_ids[local_id]:
750 seen_ids[local_id].add((id_scheme, id_value))
751 funding_by_key[local_id].setdefault("identifiers", []).append({"value": id_value, "scheme": id_scheme})
753 if "funding_agency" not in funding_by_key[local_id]:
754 continue
755 tracker = agency_trackers[local_id]
756 agency = funding_by_key[local_id]["funding_agency"]
757 a_scheme = row["funding_agency_identifier_scheme"]
758 a_value = row["funding_agency_identifier_value"]
759 if a_scheme and a_value and (a_scheme, a_value) not in tracker["seen_ids"]:
760 tracker["seen_ids"].add((a_scheme, a_value))
761 agency.setdefault("identifiers", []).append({"value": a_value, "scheme": a_scheme})
762 a_type = row["funding_agency_type"]
763 if a_type and a_type not in tracker["seen_types"]:
764 tracker["seen_types"].add(a_type)
765 agency.setdefault("types", []).append(a_type)
767 return list(funding_by_key.values())
770def normalize_local_identifier_url(local_identifier: str) -> tuple[str]:
771 # Reverse proxies (e.g. Traefik) merge duplicate slashes in request paths,
772 # turning "https://example.org/..." into "https:/example.org/...": restore the scheme separator.
773 return (sub(r"^(https?):/+", r"\1://", local_identifier),)
776def _canonical_path(path: str) -> str:
777 segments = [segment for segment in path.split("/") if segment]
778 for index, segment in enumerate(segments):
779 if segment in ENTITY_TYPES and index + 1 < len(segments):
780 identifier = normalize_local_identifier_url("/".join(segments[index + 1 :]))[0]
781 return "/" + "/".join(segments[: index + 1]) + "/" + identifier
782 return path
785def _build_search_result_page(url: str) -> dict:
786 return {"local_identifier": url, "entity_type": "search_result_page"}
789def _meta_base_url(request_url: str) -> str:
790 parsed = urlsplit(request_url)
791 path = _canonical_path(parsed.path)
792 if parsed.scheme and parsed.netloc:
793 return f"{parsed.scheme}://{parsed.netloc}{path}"
794 return path
797def _page_url(base_path: str, params: dict[str, list[str]], page: int) -> str:
798 page_params = {**params, "page": [str(page)]}
799 return f"{base_path}?{urlencode(page_params, doseq=True, safe=':,')}"
802def _raise_unprocessable(message: str) -> NoReturn:
803 raise HttpError(_UNPROCESSABLE_CONTENT, f"HTTP status code {_UNPROCESSABLE_CONTENT}: {message}")
806def _parse_positive_int_param(params: dict[str, list[str]], name: str) -> int:
807 raw_value = params[name][0]
808 try:
809 value = int(raw_value)
810 except ValueError:
811 _raise_unprocessable(f"{name} must be an integer, got {raw_value!r}")
812 if value < 1:
813 _raise_unprocessable(f"{name} must be >= 1, got {value}")
814 return value
817def _validate_page_range(page: int, total_items: int, total_pages: int) -> None:
818 if total_items and page > total_pages:
819 _raise_unprocessable(f"page {page} exceeds total pages {total_pages}")
822def _build_meta(request_url: str, graph_size: int) -> dict:
823 parsed = urlsplit(request_url)
824 base_url = _meta_base_url(request_url)
825 if _is_single_entity_request(request_url):
826 return {"local_identifier": base_url, "entity_type": "single_entity"}
827 params = parse_qs(parsed.query)
828 if "total_items" in params:
829 total_items = int(params["total_items"][0])
830 page = _parse_positive_int_param(params, "page")
831 page_size = _parse_positive_int_param(params, "page_size")
832 elif "page_size" in params:
833 total_items = graph_size
834 page = _parse_positive_int_param(params, "page") if "page" in params else 1
835 page_size = _parse_positive_int_param(params, "page_size")
836 elif "page" in params:
837 _raise_unprocessable("page requires page_size")
838 else:
839 total_items = graph_size
840 page = 1
841 page_size = max(graph_size, 1)
842 total_pages = ceil(total_items / page_size) if page_size > 0 else 0
843 non_pagination_params = {k: v for k, v in params.items() if k not in ("page", "page_size", "total_items")}
844 clean_params = {**non_pagination_params, "page": [str(page)], "page_size": [str(page_size)]}
845 self_url = f"{base_url}?{urlencode(clean_params, doseq=True, safe=':,')}"
846 meta = _build_search_result_page(self_url)
847 if page < total_pages:
848 meta["next_page"] = _build_search_result_page(_page_url(base_url, clean_params, page + 1))
849 if page > 1:
850 meta["prev_page"] = _build_search_result_page(_page_url(base_url, clean_params, page - 1))
851 base_params = {k: v for k, v in clean_params.items() if k not in ("page", "page_size")}
852 search_result_url = f"{base_url}?{urlencode(base_params, doseq=True, safe=':,')}" if base_params else base_url
853 meta["part_of"] = {
854 "local_identifier": search_result_url,
855 "entity_type": "search_result",
856 "total_items": total_items,
857 "first_page": _build_search_result_page(_page_url(base_url, clean_params, 1)),
858 "last_page": _build_search_result_page(_page_url(base_url, clean_params, max(total_pages, 1))),
859 }
860 return meta
863_BUILDER_COLUMN_PREFIXES = (
864 "identifier_",
865 "affiliation_",
866 "contribution_",
867 "manifestation_",
868 "related_products_",
869 "topic_",
870 "funding_",
871 "relevant_organisation_",
872 "beneficiaries_",
873 "access_rights_",
874 "duration_",
875 "persistent_identity_system_",
876 "audience_",
877 "policy_",
878)
881_MULTI_VALUED_COLUMNS: tuple[str, ...] = (
882 "types",
883 "other_names",
884 "keywords",
885 "research_product_types",
886 "disciplines",
887)
890def _collect_passthrough_fields(first_row: dict, excluded: set[str]) -> dict:
891 entity: dict = {}
892 for col, val in first_row.items():
893 if col.startswith("_") or col in excluded:
894 continue
895 if any(col.startswith(prefix) for prefix in _BUILDER_COLUMN_PREFIXES):
896 continue
897 if val:
898 entity[col] = val
899 return entity
902def _add_formatted_text(entity: dict, rows: list[dict], text_field: tuple[str, str, str], *, as_list: bool) -> None:
903 field, lang_field, output_key = text_field
904 if as_list:
905 values: dict[str, list[str]] = {}
906 for row in rows:
907 value = row.get(field)
908 if value:
909 language_values = values.setdefault(row.get(lang_field) or "none", [])
910 if value not in language_values:
911 language_values.append(value)
912 if values:
913 entity[output_key] = values
914 return
915 first_row = rows[0]
916 if first_row.get(field):
917 entity[output_key] = {first_row.get(lang_field) or "none": first_row[field]}
920_SECTION_BUILDERS: tuple[tuple[str, str, Callable], ...] = (
921 ("identifier_scheme", "identifiers", _collect_identifiers),
922 ("topic_term", "topics", _collect_topics),
923 ("funding_local_identifier", "funding", _collect_funding),
924)
927def _collect_contributions(rows: list[dict], entity_type: str | None) -> list[dict]:
928 if entity_type in {"grant", "venue"}:
929 return _collect_grouped_contributors(rows)
930 return _collect_contributors(rows)
933def _parse_number(value: str) -> int | float:
934 number = float(value)
935 return int(number) if number.is_integer() else number
938def _apply_grant_fields(entity: dict, rows: list[dict], first_row: dict, columns: set[str]) -> None:
939 if "funded_amount" in entity:
940 if "currency" not in entity:
941 msg = "Missing required currency for funded_amount"
942 raise ValueError(msg)
943 entity["funded_amount"] = _parse_number(entity["funded_amount"])
944 if "duration_start" in columns and (start := first_row["duration_start"]):
945 duration = {"start": start}
946 if "duration_end" in columns and (end := first_row["duration_end"]):
947 duration["end"] = end
948 entity["duration"] = duration
949 if "funding_stream" in columns and (funding_stream := first_row["funding_stream"]):
950 entity["funding_stream"] = funding_stream
951 if "funding_agency_name" in columns and (agency := _collect_single_organisation(rows, "funding_agency")):
952 entity["funding_agency"] = agency
955_FORMATTED_TEXT_FIELDS: tuple[tuple[str, str, str], ...] = (
956 ("title", "title_lang", "titles"),
957 ("abstract", "abstract_lang", "abstracts"),
958 ("label", "label_lang", "labels"),
959)
961# Research products hold a list of values per language; every other entity holds a single string.
962_MULTI_VALUED_TEXT_ENTITY_TYPE = "product"
965def _active_formatted_columns(columns: set[str]) -> set[str]:
966 active: set[str] = set()
967 for field, lang_field, _ in _FORMATTED_TEXT_FIELDS:
968 if field in columns:
969 active.update((field, lang_field))
970 return active
973def _attach_multi_valued_columns(entity: dict, rows: list[dict], columns: set[str]) -> None:
974 for column in _MULTI_VALUED_COLUMNS:
975 if column not in columns:
976 continue
977 values = list(dict.fromkeys(row[column] for row in rows if row[column]))
978 if values:
979 entity[column] = values
982def _attach_organisation_lists(entity: dict, rows: list[dict], columns: set[str]) -> None:
983 if "relevant_organisation_name" in columns and (
984 organisations := _collect_organisation(rows, "relevant_organisation")
985 ):
986 entity["relevant_organisations"] = organisations
987 if "beneficiaries_name" in columns and (beneficiaries := _collect_organisation(rows, "beneficiaries")):
988 entity["beneficiaries"] = beneficiaries
991def _collect_persistent_identity_systems(rows: list[dict]) -> list[dict]:
992 systems: dict[str, dict] = {}
993 for row in rows:
994 product_type = row["persistent_identity_system_for"]
995 pid_scheme = row["persistent_identity_system_pid_scheme"]
996 if not product_type or not pid_scheme:
997 continue
998 key = row["_persistent_identity_system_key"] or product_type
999 system = systems.setdefault(key, {"for": product_type, "pid_schemes": []})
1000 if pid_scheme not in system["pid_schemes"]:
1001 system["pid_schemes"].append(pid_scheme)
1002 return list(systems.values())
1005def _collect_audience(rows: list[dict]) -> list[dict]:
1006 audience_types = list(dict.fromkeys(row["audience_type"] for row in rows if row["audience_type"]))
1007 return [{"audience_type": audience_type} for audience_type in audience_types]
1010def _collect_policies(rows: list[dict]) -> list[dict]:
1011 policies: dict[str, dict] = {}
1012 for row in rows:
1013 about = row["policy_about"]
1014 if not about:
1015 continue
1016 key = row["_policy_key"] or about
1017 policy = policies.setdefault(key, {"about": about})
1018 target = row["policy_target"]
1019 if target:
1020 targets = policy.setdefault("targets", [])
1021 if target not in targets:
1022 targets.append(target)
1023 if row["policy_documented_at"]:
1024 policy["documented_at"] = row["policy_documented_at"]
1025 if row["policy_description"]:
1026 policy["description"] = row["policy_description"]
1027 return list(policies.values())
1030def _apply_data_source_fields(entity: dict, rows: list[dict], columns: set[str]) -> None:
1031 if "persistent_identity_system_for" in columns and (systems := _collect_persistent_identity_systems(rows)):
1032 entity["persistent_identity_systems"] = systems
1033 if "audience_type" in columns and (audience := _collect_audience(rows)):
1034 entity["audience"] = audience
1035 if "policy_about" in columns and (policies := _collect_policies(rows)):
1036 entity["policies"] = policies
1039def _apply_entity_type(
1040 entity: dict, rows: list[dict], first_row: dict, columns: set[str], entity_type: str | None
1041) -> None:
1042 if entity_type == "grant":
1043 _apply_grant_fields(entity, rows, first_row, columns)
1044 if entity_type == "venue" and "access_rights_status" in columns:
1045 access_rights = _build_status_with_description(first_row, "access_rights_status", "access_rights_description")
1046 if access_rights:
1047 entity["access_rights"] = access_rights
1048 if (
1049 entity_type == "person"
1050 and "affiliation_local_identifier" in columns
1051 and (affiliations := _collect_person_affiliations(rows))
1052 ):
1053 entity["affiliations"] = affiliations
1054 if entity_type == "datasource":
1055 _apply_data_source_fields(entity, rows, columns)
1056 if entity_type is not None:
1057 entity["entity_type"] = entity_type
1060def _build_entity(rows: list[dict], entity_type: str | None) -> dict:
1061 first_row = rows[0]
1062 columns = set(first_row)
1064 entity = _collect_passthrough_fields(first_row, _active_formatted_columns(columns) | set(_MULTI_VALUED_COLUMNS))
1065 as_list = entity_type == _MULTI_VALUED_TEXT_ENTITY_TYPE
1066 for text_field in _FORMATTED_TEXT_FIELDS:
1067 _add_formatted_text(entity, rows, text_field, as_list=as_list)
1069 _attach_multi_valued_columns(entity, rows, columns)
1071 for anchor, key, builder in _SECTION_BUILDERS:
1072 if anchor in columns and (section := builder(rows)):
1073 entity[key] = section
1074 if "contribution_role" in columns and (contributions := _collect_contributions(rows, entity_type)):
1075 entity["contributions"] = contributions
1076 if any(column.startswith("manifestation_") for column in columns) and (
1077 manifestations := _collect_manifestations(rows)
1078 ):
1079 entity["manifestations"] = manifestations
1080 if columns & set(_RELATED_PRODUCT_COLUMNS) and (related := _collect_related_products(rows)):
1081 entity["related_products"] = related
1082 _attach_organisation_lists(entity, rows, columns)
1083 _apply_entity_type(entity, rows, first_row, columns, entity_type)
1085 return entity
1088def _build_entities(rows: list[dict], entity_type: str | None) -> list[dict]:
1089 if not rows:
1090 return []
1091 groups: dict[str, list[dict]] = {}
1092 for row in rows:
1093 groups.setdefault(row["local_identifier"], []).append(row)
1094 return [_build_entity(group, entity_type) for group in groups.values()]
1097ENTITY_TYPES = frozenset({"products", "persons", "organisations", "grants", "venues", "topics", "datasources"})
1099_ENTITY_TYPE_MAP: dict[str, str] = {
1100 "products": "product",
1101 "persons": "person",
1102 "organisations": "organisation",
1103 "grants": "grant",
1104 "venues": "venue",
1105 "topics": "topic",
1106 "datasources": "datasource",
1107}
1110def _extract_entity_type(request_url: str) -> str | None:
1111 for segment in urlsplit(request_url).path.split("/"):
1112 if segment in _ENTITY_TYPE_MAP:
1113 return _ENTITY_TYPE_MAP[segment]
1114 return None
1117def _is_single_entity_request(request_url: str) -> bool:
1118 segments = [s for s in urlsplit(request_url).path.split("/") if s]
1119 for i, segment in enumerate(segments):
1120 if segment in ENTITY_TYPES:
1121 return i + 1 < len(segments)
1122 return False
1125_COLUMN_GROUP_PREFIXES = (
1126 "identifier_",
1127 "affiliation_",
1128 "_affiliation_",
1129 "contribution_",
1130 "_contribution_",
1131 "manifestation_",
1132 "_manifestation_",
1133 "related_products_",
1134 "topic_",
1135 "funding_",
1136 "relevant_organisation_",
1137 "beneficiaries_",
1138 "persistent_identity_system_",
1139 "_persistent_identity_system_",
1140 "audience_",
1141 "policy_",
1142 "_policy_",
1143)
1145_CORE_COLUMNS = frozenset(col for col in _PRODUCT_COLUMNS if not any(col.startswith(p) for p in _COLUMN_GROUP_PREFIXES))
1148def _fill_missing_columns(rows: list[dict[str, str]]) -> list[dict[str, str]]:
1149 if not rows:
1150 return rows
1151 present = set(rows[0])
1152 active_prefixes = {prefix for prefix in _COLUMN_GROUP_PREFIXES if any(col.startswith(prefix) for col in present)}
1153 missing = {
1154 col: ""
1155 for col in _PRODUCT_COLUMNS
1156 if col not in present
1157 and (
1158 (active_prefixes and col in _CORE_COLUMNS)
1159 or any(col.startswith(p) for p in active_prefixes)
1160 or (col == "_manifestation_key" and "manifestation_" in active_prefixes)
1161 or (col == "_affiliation_key" and "affiliation_" in active_prefixes)
1162 or (col == "_persistent_identity_system_key" and "persistent_identity_system_" in active_prefixes)
1163 or (col == "_policy_key" and "policy_" in active_prefixes)
1164 )
1165 }
1166 if not missing:
1167 return rows
1168 return [missing | row for row in rows]
1171def to_skg_if(csv_str: str, request_url: str = "", base_url: str = "") -> str | ConversionResult:
1172 context = _build_context(base_url)
1173 rows = _fill_missing_columns(list(csv.DictReader(StringIO(csv_str))))
1174 entity_type = _extract_entity_type(request_url)
1175 graph = _build_entities(rows, entity_type)
1177 total_entities = len(graph)
1179 parsed = urlsplit(request_url)
1180 params = parse_qs(parsed.query)
1181 if "page_size" in params and "total_items" not in params:
1182 page_size = _parse_positive_int_param(params, "page_size")
1183 page = _parse_positive_int_param(params, "page") if "page" in params else 1
1184 total_pages = ceil(total_entities / page_size) if page_size > 0 else 0
1185 _validate_page_range(page, total_entities, total_pages)
1186 start = (page - 1) * page_size
1187 graph = graph[start : start + page_size]
1188 elif "page" in params and "total_items" not in params:
1189 _raise_unprocessable("page requires page_size")
1191 result = {
1192 "@context": context,
1193 "meta": _build_meta(request_url, total_entities),
1194 "@graph": graph,
1195 }
1196 body = json.dumps(result, ensure_ascii=False, indent=4)
1197 if not graph:
1198 return ConversionResult(status_code=404, body=body)
1199 return body