feat(utils): implement new type rdf importer

apis_core/apis_entities/triple_configs/E21_PersonFromDNB.toml

@@ -0,0 +1,13 @@
+[[filters]]
+"rdf:type" = "gndo:DifferentiatedPerson"
+
+[attributes]
+forename = ["gndo:forename", "gndo:preferredNameEntityForThePerson/gndo:forename"]
+alternative_names = "gndo:variantNameForThePerson"
+surname = ["gndo:surname", "gndo:preferredNameEntityForThePerson/gndo:surname"]
+start_date_written = "gndo:dateOfBirth"
+end_date_written = "gndo:dateOfDeath"
+same_as = "owl:sameAs"
+profession = "gndo:professionOrOccupation"
+
+relations = ["gndo:placeOfDeath", "gndo:placeOfBirth"]

apis_core/utils/rdf2.py

@@ -0,0 +1,70 @@
+# SPDX-FileCopyrightText: 2025 Birger Schacht
+# SPDX-License-Identifier: MIT
+
+import logging
+from collections import defaultdict
+
+from AcdhArcheAssets.uri_norm_rules import get_normalized_uri
+from rdflib import BNode, Graph
+
+from apis_core.utils.settings import dict_from_toml_directory
+
+logger = logging.getLogger(__name__)
+
+
+def find_matching_config(graph: Graph) -> dict | None:
+    triple_configs = dict_from_toml_directory("triple_configs")
+    for path, config in triple_configs.items():
+        for _filter in config.get("filters", []):
+            triples = [
+                (
+                    None,
+                    graph.namespace_manager.expand_curie(predicate),
+                    graph.namespace_manager.expand_curie(obj),
+                )
+                for predicate, obj in _filter.items()
+            ]
+            triples = [triple in graph for triple in triples]
+            if all(triples):
+                logger.debug("Using %s for parsing graph", path)
+                return config
+    return None
+
+
+def get_something_from_uri(uri: str) -> dict | None:
+    uri = get_normalized_uri(uri)
+    graph = Graph()
+    graph.parse(uri)
+
+    if config := find_matching_config(graph):
+        result = defaultdict(list)
+        result["relations"] = defaultdict(list)
+
+        for attribute, curies in config.get("attributes", {}).items():
+            if isinstance(curies, str):
+                curies = [curies]
+            for curie in curies:
+                predicate = None
+                if "/" in curie:
+                    curie, predicate = curie.split("/")
+                    predicate = graph.namespace_manager.expand_curie(predicate)
+
+                matches = graph[: graph.namespace_manager.expand_curie(curie)]
+                for pred, obj in matches:
+                    if isinstance(obj, BNode):
+                        value = [
+                            x.toPython()
+                            for x in graph.objects(subject=obj, predicate=predicate)
+                        ]
+                    else:
+                        value = obj.toPython()
+
+                    if not isinstance(value, list):
+                        value = [value]
+
+                    if attribute == "relations":
+                        result["relations"][curie].extend(value)
+                    else:
+                        result[attribute].extend(value)
+        return dict(result)
+    return None