mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-30 04:40:16 +00:00
Merge pull request #1123 from fabio-rovai/owl-exporter-ontology-schema
Read the ontology shape the generator actually emits, and stop minting empty class IRIs (#1103)
This commit is contained in:
@@ -22,8 +22,10 @@ Author: Semantica Contributors
|
||||
License: MIT
|
||||
"""
|
||||
|
||||
import re
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
|
||||
from ..utils.exceptions import ProcessingError, ValidationError
|
||||
@@ -37,6 +39,9 @@ from ..utils.progress_tracker import get_progress_tracker
|
||||
# PROV-exported URIs co-resolve under one shared namespace by default.
|
||||
from ..provenance.manager import DEFAULT_BASE_URI
|
||||
|
||||
#: Module-level logger, for the classmethod helpers that have no instance.
|
||||
logger = get_logger("owl_exporter")
|
||||
|
||||
|
||||
class OWLExporter:
|
||||
"""
|
||||
@@ -225,6 +230,7 @@ class OWLExporter:
|
||||
Returns:
|
||||
String containing OWL-XML serialization
|
||||
"""
|
||||
esc_xml = self._escape_xml
|
||||
ontology_uri = ontology.get("uri") or self.ontology_uri
|
||||
ontology_name = ontology.get("name", "SemanticaOntology")
|
||||
version = ontology.get("version") or self.version
|
||||
@@ -237,97 +243,125 @@ class OWLExporter:
|
||||
lines.append("")
|
||||
|
||||
# Ontology declaration
|
||||
lines.append(f' <owl:Ontology rdf:about="{ontology_uri}">')
|
||||
lines.append(f" <rdfs:label>{ontology_name}</rdfs:label>")
|
||||
lines.append(f" <owl:versionInfo>{version}</owl:versionInfo>")
|
||||
lines.append(f' <owl:Ontology rdf:about="{esc_xml(ontology_uri)}">')
|
||||
lines.append(f" <rdfs:label>{esc_xml(ontology_name)}</rdfs:label>")
|
||||
lines.append(f" <owl:versionInfo>{esc_xml(version)}</owl:versionInfo>")
|
||||
if ontology.get("description"):
|
||||
lines.append(
|
||||
f' <rdfs:comment>{ontology.get("description")}</rdfs:comment>'
|
||||
f' <rdfs:comment>{esc_xml(ontology.get("description"))}</rdfs:comment>'
|
||||
)
|
||||
lines.append(" </owl:Ontology>")
|
||||
lines.append("")
|
||||
|
||||
class_index = self._class_iri_index(ontology, ontology_uri)
|
||||
object_properties, data_properties = self._split_properties(ontology)
|
||||
|
||||
def _as_list(value):
|
||||
if value is None:
|
||||
return []
|
||||
return value if isinstance(value, list) else [value]
|
||||
|
||||
# Classes
|
||||
classes = ontology.get("classes", [])
|
||||
for cls in classes:
|
||||
class_uri = cls.get("uri") or cls.get("id", "")
|
||||
for cls in ontology.get("classes", []) or []:
|
||||
if not isinstance(cls, dict):
|
||||
continue
|
||||
class_uri = self._term_iri(cls, ontology_uri)
|
||||
if not class_uri:
|
||||
self.logger.warning(
|
||||
"Skipping a class with no name, uri or id: it would serialise "
|
||||
"as an empty rdf:about"
|
||||
)
|
||||
continue
|
||||
class_name = cls.get("name") or cls.get("label", "")
|
||||
|
||||
lines.append(f' <owl:Class rdf:about="{class_uri}">')
|
||||
lines.append(f" <rdfs:label>{class_name}</rdfs:label>")
|
||||
lines.append(f' <owl:Class rdf:about="{esc_xml(class_uri)}">')
|
||||
lines.append(f" <rdfs:label>{esc_xml(class_name)}</rdfs:label>")
|
||||
|
||||
if cls.get("comment"):
|
||||
lines.append(f' <rdfs:comment>{cls.get("comment")}</rdfs:comment>')
|
||||
comment = cls.get("comment") or cls.get("description")
|
||||
if comment:
|
||||
lines.append(f" <rdfs:comment>{esc_xml(comment)}</rdfs:comment>")
|
||||
|
||||
# Subclass relationships
|
||||
if cls.get("subClassOf"):
|
||||
parent = cls.get("subClassOf")
|
||||
lines.append(f' <rdfs:subClassOf rdf:resource="{parent}"/>')
|
||||
for parent in _as_list(cls.get("subClassOf") or cls.get("parent")):
|
||||
parent_iri = self._resolve_class_ref(parent, ontology_uri, class_index)
|
||||
if parent_iri:
|
||||
lines.append(
|
||||
f' <rdfs:subClassOf rdf:resource="{esc_xml(parent_iri)}"/>'
|
||||
)
|
||||
|
||||
# Equivalent classes
|
||||
if cls.get("equivalentClass"):
|
||||
equiv = cls.get("equivalentClass")
|
||||
lines.append(f' <owl:equivalentClass rdf:resource="{equiv}"/>')
|
||||
for equiv in _as_list(cls.get("equivalentClass")):
|
||||
equiv_iri = self._resolve_class_ref(equiv, ontology_uri, class_index)
|
||||
if equiv_iri:
|
||||
lines.append(
|
||||
f' <owl:equivalentClass rdf:resource="{esc_xml(equiv_iri)}"/>'
|
||||
)
|
||||
|
||||
lines.append(" </owl:Class>")
|
||||
lines.append("")
|
||||
|
||||
# Object properties
|
||||
object_properties = ontology.get("object_properties", [])
|
||||
for prop in object_properties:
|
||||
prop_uri = prop.get("uri") or prop.get("id", "")
|
||||
prop_uri = self._term_iri(prop, ontology_uri)
|
||||
if not prop_uri:
|
||||
self.logger.warning(
|
||||
"Skipping an object property with no name, uri or id"
|
||||
)
|
||||
continue
|
||||
prop_name = prop.get("name") or prop.get("label", "")
|
||||
|
||||
lines.append(f' <owl:ObjectProperty rdf:about="{prop_uri}">')
|
||||
lines.append(f" <rdfs:label>{prop_name}</rdfs:label>")
|
||||
lines.append(f' <owl:ObjectProperty rdf:about="{esc_xml(prop_uri)}">')
|
||||
lines.append(f" <rdfs:label>{esc_xml(prop_name)}</rdfs:label>")
|
||||
|
||||
if prop.get("comment"):
|
||||
lines.append(f' <rdfs:comment>{prop.get("comment")}</rdfs:comment>')
|
||||
comment = prop.get("comment") or prop.get("description")
|
||||
if comment:
|
||||
lines.append(f" <rdfs:comment>{esc_xml(comment)}</rdfs:comment>")
|
||||
|
||||
# Domain
|
||||
if prop.get("domain"):
|
||||
domain = prop.get("domain")
|
||||
if isinstance(domain, list):
|
||||
for d in domain:
|
||||
lines.append(f' <rdfs:domain rdf:resource="{d}"/>')
|
||||
else:
|
||||
lines.append(f' <rdfs:domain rdf:resource="{domain}"/>')
|
||||
for domain in _as_list(prop.get("domain")):
|
||||
domain_iri = self._resolve_class_ref(domain, ontology_uri, class_index)
|
||||
if domain_iri:
|
||||
lines.append(
|
||||
f' <rdfs:domain rdf:resource="{esc_xml(domain_iri)}"/>'
|
||||
)
|
||||
|
||||
# Range
|
||||
if prop.get("range"):
|
||||
range_val = prop.get("range")
|
||||
if isinstance(range_val, list):
|
||||
for r in range_val:
|
||||
lines.append(f' <rdfs:range rdf:resource="{r}"/>')
|
||||
else:
|
||||
lines.append(f' <rdfs:range rdf:resource="{range_val}"/>')
|
||||
for range_val in _as_list(prop.get("range")):
|
||||
range_iri = self._resolve_class_ref(range_val, ontology_uri, class_index)
|
||||
if range_iri:
|
||||
lines.append(
|
||||
f' <rdfs:range rdf:resource="{esc_xml(range_iri)}"/>'
|
||||
)
|
||||
|
||||
lines.append(" </owl:ObjectProperty>")
|
||||
lines.append("")
|
||||
|
||||
# Data properties
|
||||
data_properties = ontology.get("data_properties", [])
|
||||
for prop in data_properties:
|
||||
prop_uri = prop.get("uri") or prop.get("id", "")
|
||||
prop_uri = self._term_iri(prop, ontology_uri)
|
||||
if not prop_uri:
|
||||
self.logger.warning("Skipping a data property with no name, uri or id")
|
||||
continue
|
||||
prop_name = prop.get("name") or prop.get("label", "")
|
||||
|
||||
lines.append(f' <owl:DatatypeProperty rdf:about="{prop_uri}">')
|
||||
lines.append(f" <rdfs:label>{prop_name}</rdfs:label>")
|
||||
lines.append(f' <owl:DatatypeProperty rdf:about="{esc_xml(prop_uri)}">')
|
||||
lines.append(f" <rdfs:label>{esc_xml(prop_name)}</rdfs:label>")
|
||||
|
||||
if prop.get("comment"):
|
||||
lines.append(f' <rdfs:comment>{prop.get("comment")}</rdfs:comment>')
|
||||
comment = prop.get("comment") or prop.get("description")
|
||||
if comment:
|
||||
lines.append(f" <rdfs:comment>{esc_xml(comment)}</rdfs:comment>")
|
||||
|
||||
# Domain
|
||||
if prop.get("domain"):
|
||||
domain = prop.get("domain")
|
||||
lines.append(f' <rdfs:domain rdf:resource="{domain}"/>')
|
||||
for domain in _as_list(prop.get("domain")):
|
||||
domain_iri = self._resolve_class_ref(domain, ontology_uri, class_index)
|
||||
if domain_iri:
|
||||
lines.append(
|
||||
f' <rdfs:domain rdf:resource="{esc_xml(domain_iri)}"/>'
|
||||
)
|
||||
|
||||
# Range
|
||||
if prop.get("range"):
|
||||
range_type = prop.get("range", "xsd:string")
|
||||
lines.append(
|
||||
f' <rdfs:range rdf:resource="http://www.w3.org/2001/XMLSchema#{range_type}"/>'
|
||||
)
|
||||
for range_val in _as_list(prop.get("range")):
|
||||
range_iri = self._resolve_datatype_iri(range_val)
|
||||
if range_iri:
|
||||
lines.append(
|
||||
f' <rdfs:range rdf:resource="{esc_xml(range_iri)}"/>'
|
||||
)
|
||||
|
||||
lines.append(" </owl:DatatypeProperty>")
|
||||
lines.append("")
|
||||
@@ -335,6 +369,223 @@ class OWLExporter:
|
||||
lines.append("</rdf:RDF>")
|
||||
return "\n".join(lines)
|
||||
|
||||
# ── Ontology-dict normalisation ───────────────────────────────────────────
|
||||
#
|
||||
# OntologyGenerator emits a single `properties` list tagged with
|
||||
# type/@type, while hand-authored ontologies use `object_properties` and
|
||||
# `data_properties`. Both shapes are accepted; everything below works from
|
||||
# the normalised view so the two cannot drift apart again (#1103).
|
||||
|
||||
_XSD_NS = "http://www.w3.org/2001/XMLSchema#"
|
||||
|
||||
#: Prefixes the generator and hand-authored ontologies actually use. A
|
||||
#: prefixed name is not an absolute IRI: `owl:Thing` matches the generic
|
||||
#: scheme grammar, so treating it as one produced <owl:Thing> as a domain,
|
||||
#: which is a different term from http://www.w3.org/2002/07/owl#Thing.
|
||||
_KNOWN_PREFIXES = {
|
||||
"owl": "http://www.w3.org/2002/07/owl#",
|
||||
"rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
|
||||
"rdfs": "http://www.w3.org/2000/01/rdf-schema#",
|
||||
"xsd": _XSD_NS,
|
||||
"skos": "http://www.w3.org/2004/02/skos/core#",
|
||||
"dc": "http://purl.org/dc/elements/1.1/",
|
||||
"dcterms": "http://purl.org/dc/terms/",
|
||||
"foaf": "http://xmlns.com/foaf/0.1/",
|
||||
"sem": "https://semantica.dev/ns#",
|
||||
"semantica": "https://semantica.dev/ns#",
|
||||
}
|
||||
|
||||
#: Schemes that really do introduce an absolute IRI without `//`.
|
||||
_ABSOLUTE_SCHEMES = ("urn:", "doi:", "mailto:", "tag:", "uuid:")
|
||||
|
||||
@classmethod
|
||||
def _is_absolute_iri(cls, value: str) -> bool:
|
||||
if not isinstance(value, str):
|
||||
return False
|
||||
value = value.strip()
|
||||
if "://" in value:
|
||||
return bool(re.match(r"^[A-Za-z][A-Za-z0-9+.\-]*://", value))
|
||||
return value.lower().startswith(cls._ABSOLUTE_SCHEMES)
|
||||
|
||||
@classmethod
|
||||
def _expand_prefixed_name(cls, value: str) -> str:
|
||||
"""Expand a known prefixed name, or return "" when it cannot be expanded."""
|
||||
prefix, _, local = value.partition(":")
|
||||
namespace = cls._KNOWN_PREFIXES.get(prefix)
|
||||
return f"{namespace}{local}" if namespace and local else ""
|
||||
|
||||
@staticmethod
|
||||
def _iri_safe(local: str) -> str:
|
||||
"""
|
||||
Percent-encode a local name so it can sit inside <>.
|
||||
|
||||
A name is free text. "Customer Account" pasted onto a base gives an IRI
|
||||
with a space in it, which rdflib only warns about and Oxigraph rejects
|
||||
with "Invalid IRI code point".
|
||||
"""
|
||||
return quote(local.strip(), safe="~._-!$&'()*+,;=:@/?")
|
||||
|
||||
@classmethod
|
||||
def _join_iri(cls, base: str, local: str) -> str:
|
||||
"""Append a local name to a base IRI, respecting hash and slash bases."""
|
||||
if not base:
|
||||
return ""
|
||||
local = cls._iri_safe(local)
|
||||
if not local:
|
||||
return ""
|
||||
separator = "" if base.endswith(("#", "/", ":")) else "#"
|
||||
return f"{base}{separator}{local}"
|
||||
|
||||
@classmethod
|
||||
def _term_iri(cls, term: Dict[str, Any], base: str) -> str:
|
||||
"""
|
||||
Resolve the IRI of a class or property.
|
||||
|
||||
Returns "" when the term carries nothing usable, so the caller can skip
|
||||
it. Interpolating an empty string into <> silently resolves against the
|
||||
parser's base — under rdflib that is the current working directory — and
|
||||
collapses every such term onto one subject.
|
||||
"""
|
||||
for key in ("uri", "iri", "id"):
|
||||
value = term.get(key)
|
||||
if isinstance(value, str) and value.strip():
|
||||
value = value.strip()
|
||||
return value if cls._is_absolute_iri(value) else cls._join_iri(base, value)
|
||||
|
||||
name = term.get("name") or term.get("label")
|
||||
if isinstance(name, str) and name.strip():
|
||||
return cls._join_iri(base, name.strip())
|
||||
return ""
|
||||
|
||||
@classmethod
|
||||
def _class_iri_index(cls, ontology: Dict[str, Any], base: str) -> Dict[str, str]:
|
||||
"""Map class name and label to the IRI that class is actually exported under."""
|
||||
index: Dict[str, str] = {}
|
||||
for class_def in ontology.get("classes", []) or []:
|
||||
if not isinstance(class_def, dict):
|
||||
continue
|
||||
iri = cls._term_iri(class_def, base)
|
||||
if not iri:
|
||||
continue
|
||||
for key in (class_def.get("name"), class_def.get("label")):
|
||||
if isinstance(key, str) and key.strip():
|
||||
index.setdefault(key.strip(), iri)
|
||||
return index
|
||||
|
||||
@classmethod
|
||||
def _resolve_class_ref(cls, value: Any, base: str, index: Dict[str, str]) -> str:
|
||||
"""
|
||||
Resolve a domain/range reference to an absolute IRI.
|
||||
|
||||
The generator writes bare class names here. Looking the name up in the
|
||||
class index first means a reference always lands on the IRI that class
|
||||
was exported under, rather than on a re-derived guess.
|
||||
"""
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
return ""
|
||||
value = value.strip()
|
||||
if cls._is_absolute_iri(value):
|
||||
return value
|
||||
if value in index:
|
||||
return index[value]
|
||||
if ":" in value:
|
||||
return cls._expand_prefixed_name(value)
|
||||
return cls._join_iri(base, value)
|
||||
|
||||
@classmethod
|
||||
def _resolve_datatype_iri(cls, value: Any) -> str:
|
||||
"""
|
||||
Resolve a data property range to an absolute datatype IRI.
|
||||
|
||||
Accepts "string", "xsd:string" and a full IRI alike. The previous
|
||||
`xsd:{range}` interpolation doubled the prefix whenever the generator
|
||||
had already written "xsd:string".
|
||||
"""
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
return ""
|
||||
value = value.strip()
|
||||
if value.startswith(("xsd:", "XSD:")):
|
||||
return cls._XSD_NS + value.split(":", 1)[1]
|
||||
if cls._is_absolute_iri(value):
|
||||
return value
|
||||
return cls._XSD_NS + value
|
||||
|
||||
@classmethod
|
||||
def _ttl_datatype_ref(cls, value: Any) -> str:
|
||||
"""
|
||||
Render a data property range for Turtle.
|
||||
|
||||
XSD datatypes are written with the xsd: prefix the header already
|
||||
declares; anything else is written as a full IRI. Both are the same
|
||||
term, this only keeps the compact style the module was written in.
|
||||
"""
|
||||
iri = cls._resolve_datatype_iri(value)
|
||||
if not iri:
|
||||
return ""
|
||||
if iri.startswith(cls._XSD_NS):
|
||||
return f"xsd:{iri[len(cls._XSD_NS):]}"
|
||||
return f"<{iri}>"
|
||||
|
||||
@classmethod
|
||||
def _split_properties(
|
||||
cls, ontology: Dict[str, Any]
|
||||
) -> "tuple[List[Dict[str, Any]], List[Dict[str, Any]]]":
|
||||
"""
|
||||
Return (object_properties, data_properties) across both dict shapes.
|
||||
|
||||
A property listed under an explicit key keeps that key's kind. A
|
||||
property from the generator's combined `properties` list is classified
|
||||
by its own type/@type, defaulting to a data property.
|
||||
"""
|
||||
object_props: List[Dict[str, Any]] = []
|
||||
data_props: List[Dict[str, Any]] = []
|
||||
|
||||
for prop in ontology.get("object_properties", []) or []:
|
||||
if isinstance(prop, dict):
|
||||
object_props.append(prop)
|
||||
for prop in ontology.get("data_properties", []) or []:
|
||||
if isinstance(prop, dict):
|
||||
data_props.append(prop)
|
||||
|
||||
skipped = 0
|
||||
untyped = []
|
||||
for prop in ontology.get("properties", []) or []:
|
||||
if not isinstance(prop, dict):
|
||||
skipped += 1
|
||||
continue
|
||||
kind = str(prop.get("type") or "").strip().lower()
|
||||
owl_type = str(prop.get("@type") or "").strip().lower()
|
||||
if kind in ("object", "objectproperty") or owl_type.endswith("objectproperty"):
|
||||
object_props.append(prop)
|
||||
else:
|
||||
if not kind and not owl_type:
|
||||
untyped.append(prop.get("name") or prop.get("uri") or "<unnamed>")
|
||||
data_props.append(prop)
|
||||
|
||||
if skipped:
|
||||
logger.warning(
|
||||
f"Skipped {skipped} entr(y/ies) in 'properties' that are not "
|
||||
"dictionaries and cannot be exported"
|
||||
)
|
||||
if untyped:
|
||||
logger.warning(
|
||||
"Exported as data properties because they declare no type or "
|
||||
f"@type: {', '.join(str(name) for name in untyped)}"
|
||||
)
|
||||
|
||||
return object_props, data_props
|
||||
|
||||
@staticmethod
|
||||
def _escape_xml(value: Any) -> str:
|
||||
"""Escape a value for safe embedding in XML text or an attribute value."""
|
||||
return (
|
||||
str(value)
|
||||
.replace("&", "&")
|
||||
.replace("<", "<")
|
||||
.replace(">", ">")
|
||||
.replace('"', """)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _escape_ttl_str(value: str) -> str:
|
||||
"""Escape a string value for safe embedding in a Turtle string literal."""
|
||||
@@ -387,62 +638,83 @@ class OWLExporter:
|
||||
lines.append(self._ttl_block(ontology_uri, "owl:Ontology", onto_predicates))
|
||||
lines.append("")
|
||||
|
||||
class_index = self._class_iri_index(ontology, ontology_uri)
|
||||
object_properties, data_properties = self._split_properties(ontology)
|
||||
|
||||
def _as_list(value):
|
||||
if value is None:
|
||||
return []
|
||||
return value if isinstance(value, list) else [value]
|
||||
|
||||
# Classes
|
||||
for cls in ontology.get("classes", []):
|
||||
class_uri = cls.get("uri") or cls.get("id", "")
|
||||
for cls in ontology.get("classes", []) or []:
|
||||
if not isinstance(cls, dict):
|
||||
continue
|
||||
class_uri = self._term_iri(cls, ontology_uri)
|
||||
if not class_uri:
|
||||
self.logger.warning(
|
||||
"Skipping a class with no name, uri or id: it would serialise as <>"
|
||||
)
|
||||
continue
|
||||
class_name = cls.get("name") or cls.get("label", "")
|
||||
predicates = [f'rdfs:label "{esc(class_name)}"']
|
||||
comment = cls.get("comment")
|
||||
comment = cls.get("comment") or cls.get("description")
|
||||
if comment:
|
||||
predicates.append(f'rdfs:comment "{esc(comment)}"')
|
||||
sub_class = cls.get("subClassOf")
|
||||
if sub_class:
|
||||
predicates.append(f"rdfs:subClassOf <{sub_class}>")
|
||||
equiv = cls.get("equivalentClass")
|
||||
if equiv:
|
||||
predicates.append(f"owl:equivalentClass <{equiv}>")
|
||||
for parent in _as_list(cls.get("subClassOf") or cls.get("parent")):
|
||||
parent_iri = self._resolve_class_ref(parent, ontology_uri, class_index)
|
||||
if parent_iri:
|
||||
predicates.append(f"rdfs:subClassOf <{parent_iri}>")
|
||||
for equiv in _as_list(cls.get("equivalentClass")):
|
||||
equiv_iri = self._resolve_class_ref(equiv, ontology_uri, class_index)
|
||||
if equiv_iri:
|
||||
predicates.append(f"owl:equivalentClass <{equiv_iri}>")
|
||||
lines.append(self._ttl_block(class_uri, "owl:Class", predicates))
|
||||
lines.append("")
|
||||
|
||||
# Object properties
|
||||
for prop in ontology.get("object_properties", []):
|
||||
prop_uri = prop.get("uri") or prop.get("id", "")
|
||||
for prop in object_properties:
|
||||
prop_uri = self._term_iri(prop, ontology_uri)
|
||||
if not prop_uri:
|
||||
self.logger.warning(
|
||||
"Skipping an object property with no name, uri or id"
|
||||
)
|
||||
continue
|
||||
prop_name = prop.get("name") or prop.get("label", "")
|
||||
predicates = [f'rdfs:label "{esc(prop_name)}"']
|
||||
comment = prop.get("comment")
|
||||
comment = prop.get("comment") or prop.get("description")
|
||||
if comment:
|
||||
predicates.append(f'rdfs:comment "{esc(comment)}"')
|
||||
domain = prop.get("domain")
|
||||
if domain:
|
||||
if isinstance(domain, list):
|
||||
for d in domain:
|
||||
predicates.append(f"rdfs:domain <{d}>")
|
||||
else:
|
||||
predicates.append(f"rdfs:domain <{domain}>")
|
||||
range_val = prop.get("range")
|
||||
if range_val:
|
||||
if isinstance(range_val, list):
|
||||
for r in range_val:
|
||||
predicates.append(f"rdfs:range <{r}>")
|
||||
else:
|
||||
predicates.append(f"rdfs:range <{range_val}>")
|
||||
for domain in _as_list(prop.get("domain")):
|
||||
domain_iri = self._resolve_class_ref(domain, ontology_uri, class_index)
|
||||
if domain_iri:
|
||||
predicates.append(f"rdfs:domain <{domain_iri}>")
|
||||
for range_val in _as_list(prop.get("range")):
|
||||
range_iri = self._resolve_class_ref(range_val, ontology_uri, class_index)
|
||||
if range_iri:
|
||||
predicates.append(f"rdfs:range <{range_iri}>")
|
||||
lines.append(self._ttl_block(prop_uri, "owl:ObjectProperty", predicates))
|
||||
lines.append("")
|
||||
|
||||
# Data properties
|
||||
for prop in ontology.get("data_properties", []):
|
||||
prop_uri = prop.get("uri") or prop.get("id", "")
|
||||
for prop in data_properties:
|
||||
prop_uri = self._term_iri(prop, ontology_uri)
|
||||
if not prop_uri:
|
||||
self.logger.warning("Skipping a data property with no name, uri or id")
|
||||
continue
|
||||
prop_name = prop.get("name") or prop.get("label", "")
|
||||
predicates = [f'rdfs:label "{esc(prop_name)}"']
|
||||
comment = prop.get("comment")
|
||||
comment = prop.get("comment") or prop.get("description")
|
||||
if comment:
|
||||
predicates.append(f'rdfs:comment "{esc(comment)}"')
|
||||
domain = prop.get("domain")
|
||||
if domain:
|
||||
predicates.append(f"rdfs:domain <{domain}>")
|
||||
range_type = prop.get("range")
|
||||
if range_type:
|
||||
predicates.append(f"rdfs:range xsd:{range_type}")
|
||||
for domain in _as_list(prop.get("domain")):
|
||||
domain_iri = self._resolve_class_ref(domain, ontology_uri, class_index)
|
||||
if domain_iri:
|
||||
predicates.append(f"rdfs:domain <{domain_iri}>")
|
||||
for range_val in _as_list(prop.get("range")):
|
||||
range_ref = self._ttl_datatype_ref(range_val)
|
||||
if range_ref:
|
||||
predicates.append(f"rdfs:range {range_ref}")
|
||||
lines.append(self._ttl_block(prop_uri, "owl:DatatypeProperty", predicates))
|
||||
lines.append("")
|
||||
|
||||
|
||||
@@ -33,6 +33,7 @@ License: MIT
|
||||
from dataclasses import dataclass, field, replace as dataclass_replace
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
from ..utils.exceptions import ProcessingError, ValidationError
|
||||
from ..utils.logging import get_logger
|
||||
@@ -438,10 +439,13 @@ class OntologyGenerator:
|
||||
entities=entities, relationships=relationships, classes=classes, **prop_options
|
||||
)
|
||||
|
||||
# Add types to classes
|
||||
# Add types to classes.
|
||||
# ClassInferrer sets "uri": None when it was given no namespace manager,
|
||||
# so the key is present and a `not in` guard never fires: every class
|
||||
# then reached the exporters with no IRI at all (#1103).
|
||||
for cls in classes:
|
||||
cls["@type"] = "owl:Class"
|
||||
if "uri" not in cls:
|
||||
if not cls.get("uri"):
|
||||
cls["uri"] = self.namespace_manager.generate_class_iri(cls["name"])
|
||||
|
||||
# Add types to properties
|
||||
@@ -451,7 +455,7 @@ class OntologyGenerator:
|
||||
else:
|
||||
prop["@type"] = "owl:DatatypeProperty"
|
||||
|
||||
if "uri" not in prop:
|
||||
if not prop.get("uri"):
|
||||
prop["uri"] = self.namespace_manager.generate_property_iri(prop["name"])
|
||||
|
||||
return {
|
||||
@@ -692,12 +696,19 @@ class OntologyOptimizer:
|
||||
Returns:
|
||||
Improved ontology
|
||||
"""
|
||||
# Ensure all classes have required fields
|
||||
# Ensure all classes have required fields. Both guards used `not in`,
|
||||
# which misses a key that is present and None, and the URI fallback
|
||||
# assigned a bare class name where an absolute IRI is required (#1103).
|
||||
#
|
||||
# The base comes from the ontology being optimized. OntologyOptimizer
|
||||
# holds no namespace manager, so reaching for one here would raise
|
||||
# AttributeError on every ontology carrying a class with no URI.
|
||||
base_uri = ontology.get("uri") or DEFAULT_ONTOLOGY_BASE_URI
|
||||
classes = ontology.get("classes", [])
|
||||
for cls in classes:
|
||||
if "uri" not in cls:
|
||||
cls["uri"] = cls.get("name", "Entity")
|
||||
if "label" not in cls:
|
||||
if not cls.get("uri"):
|
||||
cls["uri"] = _mint_term_iri(base_uri, cls.get("name", "Entity"))
|
||||
if not cls.get("label"):
|
||||
cls["label"] = cls.get("name", "Entity")
|
||||
|
||||
# Ensure all properties have domains and ranges
|
||||
@@ -744,6 +755,25 @@ class NodeShape:
|
||||
severity: str = "Violation"
|
||||
|
||||
|
||||
#: Used when an ontology carries no URI of its own.
|
||||
DEFAULT_ONTOLOGY_BASE_URI = "https://semantica.dev/ontology/"
|
||||
|
||||
|
||||
def _mint_term_iri(base_uri: str, name: str) -> str:
|
||||
"""
|
||||
Mint an absolute IRI for a term from a base and a name.
|
||||
|
||||
The name is percent-encoded: names are free text, and "Customer Account"
|
||||
pasted onto a base gives an IRI with a space in it, which strict parsers
|
||||
reject outright.
|
||||
"""
|
||||
local = quote(str(name).strip(), safe="~._-!$&'()*+,;=:@")
|
||||
if not local:
|
||||
local = "Entity"
|
||||
separator = "" if base_uri.endswith(("#", "/", ":")) else "#"
|
||||
return f"{base_uri}{separator}{local}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class SHACLGraph:
|
||||
"""Internal model representing the complete SHACL shapes graph."""
|
||||
|
||||
@@ -0,0 +1,291 @@
|
||||
"""
|
||||
Regression tests for #1103.
|
||||
|
||||
OWLExporter read `object_properties` / `data_properties` while OntologyGenerator
|
||||
emits a single `properties` list, so every generated property was dropped. Class
|
||||
IRIs arrived as None and were interpolated into `<>`, collapsing every class onto
|
||||
the empty relative IRI, so an ontology of N classes serialised as one node
|
||||
carrying N labels.
|
||||
|
||||
These tests drive the exporter with what the generator actually produces, and
|
||||
assert on the parsed graph rather than on the serialised text.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
rdflib = pytest.importorskip("rdflib")
|
||||
from rdflib import Graph, RDF, RDFS, OWL, URIRef, Literal # noqa: E402
|
||||
|
||||
from semantica.export.owl_exporter import OWLExporter # noqa: E402
|
||||
from semantica.ontology.ontology_generator import OntologyGenerator # noqa: E402
|
||||
|
||||
XSD = "http://www.w3.org/2001/XMLSchema#"
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def generated_ontology():
|
||||
"""A real OntologyGenerator run, not a hand-written stand-in."""
|
||||
data = {
|
||||
"entities": [
|
||||
{"type": "Person", "name": "John", "age": 30},
|
||||
{"type": "Person", "name": "Jane", "age": 25},
|
||||
{"type": "Organization", "name": "Acme"},
|
||||
{"type": "Organization", "name": "Globex"},
|
||||
],
|
||||
"relationships": [
|
||||
{"source": "John", "target": "Acme", "type": "works_at"},
|
||||
{"source": "Jane", "target": "Globex", "type": "works_at"},
|
||||
],
|
||||
}
|
||||
return OntologyGenerator().generate_ontology(data)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def turtle_graph(generated_ontology):
|
||||
ttl = OWLExporter()._export_owl_turtle(generated_ontology)
|
||||
graph = Graph()
|
||||
graph.parse(data=ttl, format="turtle")
|
||||
return graph
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def xml_graph(generated_ontology):
|
||||
xml = OWLExporter()._export_owl_xml(generated_ontology)
|
||||
graph = Graph()
|
||||
graph.parse(data=xml, format="xml")
|
||||
return graph
|
||||
|
||||
|
||||
def test_generator_mints_class_iris(generated_ontology):
|
||||
"""The `uri` key is present with a None value, so a `not in` guard misses it."""
|
||||
classes = generated_ontology["classes"]
|
||||
assert classes, "fixture produced no classes"
|
||||
for cls in classes:
|
||||
assert cls.get("uri"), f"class {cls.get('name')!r} has no URI: {cls.get('uri')!r}"
|
||||
assert str(cls["uri"]).startswith("http"), cls["uri"]
|
||||
|
||||
|
||||
def test_every_class_is_a_distinct_absolute_iri(turtle_graph, generated_ontology):
|
||||
subjects = set(turtle_graph.subjects(RDF.type, OWL.Class))
|
||||
assert len(subjects) == len(generated_ontology["classes"])
|
||||
for subject in subjects:
|
||||
assert isinstance(subject, URIRef)
|
||||
assert str(subject) != "", "class collapsed onto the empty relative IRI"
|
||||
assert str(subject).startswith("http"), subject
|
||||
|
||||
|
||||
def test_class_labels_are_not_stacked_on_one_node(turtle_graph):
|
||||
"""Two classes must not share a subject and pile up two rdfs:label values."""
|
||||
for subject in set(turtle_graph.subjects(RDF.type, OWL.Class)):
|
||||
labels = list(turtle_graph.objects(subject, RDFS.label))
|
||||
assert len(labels) == 1, f"{subject} carries {len(labels)} labels: {labels}"
|
||||
|
||||
|
||||
def test_no_generated_property_is_dropped(turtle_graph, generated_ontology):
|
||||
declared = set(turtle_graph.subjects(RDF.type, OWL.ObjectProperty)) | set(
|
||||
turtle_graph.subjects(RDF.type, OWL.DatatypeProperty)
|
||||
)
|
||||
expected = {URIRef(p["uri"]) for p in generated_ontology["properties"]}
|
||||
assert expected, "fixture produced no properties"
|
||||
assert expected <= declared, f"dropped: {expected - declared}"
|
||||
|
||||
|
||||
def test_properties_keep_their_owl_type(turtle_graph, generated_ontology):
|
||||
by_uri = {p["uri"]: p for p in generated_ontology["properties"]}
|
||||
for uri, prop in by_uri.items():
|
||||
expected = OWL.ObjectProperty if prop["type"] == "object" else OWL.DatatypeProperty
|
||||
assert (URIRef(uri), RDF.type, expected) in turtle_graph, (
|
||||
f"{prop['name']} ({prop['type']}) is not typed {expected}"
|
||||
)
|
||||
|
||||
|
||||
def test_object_property_domain_and_range_are_class_iris(turtle_graph, generated_ontology):
|
||||
"""The generator emits bare class names; they must resolve, not stay relative."""
|
||||
class_iris = {URIRef(c["uri"]) for c in generated_ontology["classes"]}
|
||||
obj_props = [p for p in generated_ontology["properties"] if p["type"] == "object"]
|
||||
assert obj_props, "fixture produced no object properties"
|
||||
for prop in obj_props:
|
||||
subject = URIRef(prop["uri"])
|
||||
for predicate in (RDFS.domain, RDFS.range):
|
||||
values = list(turtle_graph.objects(subject, predicate))
|
||||
assert values, f"{prop['name']} has no {predicate}"
|
||||
for value in values:
|
||||
assert value in class_iris, f"{prop['name']} {predicate} = {value!r}"
|
||||
|
||||
|
||||
def test_data_property_range_is_a_single_well_formed_xsd_iri(turtle_graph, generated_ontology):
|
||||
"""`rdfs:range xsd:{range}` doubled the prefix when range was already 'xsd:string'."""
|
||||
data_props = [p for p in generated_ontology["properties"] if p["type"] != "object"]
|
||||
assert data_props, "fixture produced no data properties"
|
||||
for prop in data_props:
|
||||
ranges = list(turtle_graph.objects(URIRef(prop["uri"]), RDFS.range))
|
||||
assert ranges, f"{prop['name']} has no range"
|
||||
for value in ranges:
|
||||
assert str(value).startswith(XSD), f"{prop['name']} range = {value!r}"
|
||||
assert "xsd:" not in str(value), f"doubled prefix: {value!r}"
|
||||
|
||||
|
||||
def test_xml_and_turtle_describe_the_same_ontology(turtle_graph, xml_graph):
|
||||
"""The two serialisations of one ontology must not be different graphs."""
|
||||
def summary(graph):
|
||||
return {
|
||||
"classes": set(graph.subjects(RDF.type, OWL.Class)),
|
||||
"object_properties": set(graph.subjects(RDF.type, OWL.ObjectProperty)),
|
||||
"data_properties": set(graph.subjects(RDF.type, OWL.DatatypeProperty)),
|
||||
}
|
||||
|
||||
assert summary(turtle_graph) == summary(xml_graph)
|
||||
|
||||
|
||||
def test_explicit_object_and_data_property_keys_still_work():
|
||||
"""The pre-existing hand-authored shape must keep working."""
|
||||
ontology = {
|
||||
"uri": "https://example.org/onto/",
|
||||
"name": "Hand",
|
||||
"classes": [{"uri": "https://example.org/onto/Person", "name": "Person"}],
|
||||
"object_properties": [
|
||||
{
|
||||
"uri": "https://example.org/onto/knows",
|
||||
"name": "knows",
|
||||
"domain": "https://example.org/onto/Person",
|
||||
"range": "https://example.org/onto/Person",
|
||||
}
|
||||
],
|
||||
"data_properties": [
|
||||
{"uri": "https://example.org/onto/age", "name": "age", "range": "integer"}
|
||||
],
|
||||
}
|
||||
graph = Graph()
|
||||
graph.parse(data=OWLExporter()._export_owl_turtle(ontology), format="turtle")
|
||||
|
||||
assert (URIRef("https://example.org/onto/knows"), RDF.type, OWL.ObjectProperty) in graph
|
||||
assert (URIRef("https://example.org/onto/age"), RDF.type, OWL.DatatypeProperty) in graph
|
||||
assert (
|
||||
URIRef("https://example.org/onto/age"),
|
||||
RDFS.range,
|
||||
URIRef(XSD + "integer"),
|
||||
) in graph
|
||||
|
||||
|
||||
def test_a_class_without_any_identifier_is_skipped_not_emitted_as_empty():
|
||||
"""An unusable class must not become `<>` and swallow the document IRI."""
|
||||
ontology = {
|
||||
"uri": "https://example.org/onto/",
|
||||
"name": "Partial",
|
||||
"classes": [{"comment": "no name, no uri, no id"}],
|
||||
}
|
||||
graph = Graph()
|
||||
graph.parse(data=OWLExporter()._export_owl_turtle(ontology), format="turtle")
|
||||
|
||||
assert set(graph.subjects(RDF.type, OWL.Class)) == set()
|
||||
assert (URIRef("https://example.org/onto/"), RDF.type, OWL.Ontology) in graph
|
||||
|
||||
|
||||
# ── Review findings on the first revision of this fix ────────────────────────
|
||||
|
||||
def test_a_name_with_a_space_still_mints_a_valid_iri():
|
||||
"""The name fallback pasted free text onto a base, producing `<... ...>`."""
|
||||
import pyoxigraph
|
||||
|
||||
ontology = {
|
||||
"uri": "https://example.org/onto/",
|
||||
"name": "Spaces",
|
||||
"classes": [{"name": "Customer Account"}],
|
||||
}
|
||||
turtle = OWLExporter()._export_owl_turtle(ontology)
|
||||
|
||||
graph = Graph()
|
||||
graph.parse(data=turtle, format="turtle")
|
||||
subjects = [str(s) for s in graph.subjects(RDF.type, OWL.Class)]
|
||||
assert subjects, "the class was dropped entirely"
|
||||
assert " " not in subjects[0], subjects[0]
|
||||
|
||||
# rdflib only warns about a space in an IRI; a strict parser refuses it.
|
||||
pyoxigraph.Store().load(
|
||||
turtle.encode(), format=pyoxigraph.RdfFormat.TURTLE, base_iri=None
|
||||
)
|
||||
|
||||
|
||||
def test_owl_thing_expands_instead_of_becoming_its_own_scheme():
|
||||
"""`owl:Thing` matches the generic scheme grammar but is a prefixed name."""
|
||||
ontology = {
|
||||
"uri": "https://example.org/onto/",
|
||||
"name": "Thing",
|
||||
"classes": [{"name": "Person", "uri": "https://example.org/onto/Person"}],
|
||||
"object_properties": [
|
||||
{
|
||||
"name": "relatedTo",
|
||||
"uri": "https://example.org/onto/relatedTo",
|
||||
"domain": ["owl:Thing"],
|
||||
"range": ["owl:Thing"],
|
||||
}
|
||||
],
|
||||
}
|
||||
graph = Graph()
|
||||
graph.parse(data=OWLExporter()._export_owl_turtle(ontology), format="turtle")
|
||||
|
||||
subject = URIRef("https://example.org/onto/relatedTo")
|
||||
for predicate in (RDFS.domain, RDFS.range):
|
||||
values = [str(v) for v in graph.objects(subject, predicate)]
|
||||
assert values == ["http://www.w3.org/2002/07/owl#Thing"], values
|
||||
|
||||
|
||||
def test_the_generators_owl_thing_fallback_round_trips():
|
||||
"""stage 4 assigns domain/range of ["owl:Thing"], so this is the live path."""
|
||||
ontology = {
|
||||
"uri": "https://example.org/onto/",
|
||||
"name": "Generated",
|
||||
"classes": [{"name": "Person", "uri": "https://example.org/onto/Person"}],
|
||||
"properties": [
|
||||
{"name": "linkedTo", "type": "object", "uri": "https://example.org/onto/linkedTo",
|
||||
"domain": ["owl:Thing"], "range": ["owl:Thing"], "@type": "owl:ObjectProperty"}
|
||||
],
|
||||
}
|
||||
graph = Graph()
|
||||
graph.parse(data=OWLExporter()._export_owl_turtle(ontology), format="turtle")
|
||||
|
||||
assert (
|
||||
URIRef("https://example.org/onto/linkedTo"),
|
||||
RDFS.domain,
|
||||
URIRef("http://www.w3.org/2002/07/owl#Thing"),
|
||||
) in graph
|
||||
|
||||
|
||||
def test_optimizing_a_class_without_a_uri_does_not_raise():
|
||||
"""improve_coherence lives on OntologyOptimizer, which owns no namespace manager."""
|
||||
from semantica.ontology.ontology_generator import OntologyOptimizer
|
||||
|
||||
result = OntologyOptimizer().improve_coherence(
|
||||
{"uri": "https://example.org/onto/", "classes": [{"name": "Person"}], "properties": []}
|
||||
)
|
||||
minted = result["classes"][0]["uri"]
|
||||
assert minted.startswith("https://example.org/onto/"), minted
|
||||
assert " " not in minted
|
||||
|
||||
|
||||
def test_optimizing_falls_back_to_a_base_when_the_ontology_has_no_uri():
|
||||
from semantica.ontology.ontology_generator import OntologyOptimizer
|
||||
|
||||
result = OntologyOptimizer().improve_coherence(
|
||||
{"classes": [{"name": "Customer Account"}], "properties": []}
|
||||
)
|
||||
minted = result["classes"][0]["uri"]
|
||||
assert minted.startswith("http"), minted
|
||||
assert " " not in minted, minted
|
||||
|
||||
|
||||
def test_unusable_property_entries_are_reported_not_silently_dropped(caplog):
|
||||
import logging
|
||||
|
||||
ontology = {
|
||||
"uri": "https://example.org/onto/",
|
||||
"name": "Malformed",
|
||||
"classes": [],
|
||||
"properties": ["not a dict", {"name": "untyped", "uri": "https://example.org/onto/untyped"}],
|
||||
}
|
||||
with caplog.at_level(logging.WARNING):
|
||||
OWLExporter()._export_owl_turtle(ontology)
|
||||
|
||||
messages = " ".join(record.getMessage() for record in caplog.records)
|
||||
assert "not dictionaries" in messages or "not\ndictionaries" in messages or "dictionaries" in messages
|
||||
assert "untyped" in messages
|
||||
Reference in New Issue
Block a user