feat(explorer): integrate API routers and add RDF parsing util

This commit is contained in:
ZohaibHassan16
2026-03-30 15:04:46 +05:00
parent e30ef6cb76
commit 73af7d5bfc
2 changed files with 168 additions and 1 deletions
+133
View File
@@ -0,0 +1,133 @@
"""
RDF / SKOS parsing utility for the knowledge Explorer
Parses `.ttl` and `.rdf` files, extracting skos:Concept and skos:ConceptScheme entities into flat dicts
compatible with ContextGraph.
"""
from typing import Any, Dict, List, Tuple
import rdflib
from rdflib.namespace import RDF, RDFS, SKOS
def _get_best_label(graph: rdflib.Graph, subject: rdflib.URIRef, predicate: rdflib.URIRef) -> str:
"""
Extracts the best available string label for a given predicate.
Prioritizes English tags ('en'), then untagged strings, then falls back to whatever
is available. Strips language tags in the process.
"""
labels = list(graph.objects(subject, predicate))
if not labels:
return ""
# priority 1: English match exact
for lbl in labels:
if getattr(lbl, "language", None) == "en":
return str(lbl)
# priority 2: English variants
for lbl in labels:
lang = getattr(lbl, "language", "")
if lang and lang.startswith("en"):
return str(lbl)
# priority 3: No lang tag
for lbl in labels:
if getattr(lbl, "language", None) is None:
return str(lbl)
# whatever is first if not any of the three above
return str(labels[0])
def _get_all_labels(graph: rdflib.Graph, subject: rdflib.URIRef, predicate: rdflib.URIRef) -> List[str]:
""" Returns a list of all string values for a predicate, stripping lang tags."""
return list({str(lbl) for lbl in graph.objects(subject, predicate)})
def parse_skos_file(file_bytes: bytes, format: str = "turtle") -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
"""
Parses RDF data and extracts SKOS concepts and relationships.
Args:
file_bytes: The raw bytes of the uploaded file.
format: The rdflib parse format (e.g., "turtle", "xml").
Returns:
A tuple of (nodes_list, edges_list) formatted for ContextGraph ingestion.
"""
g = rdflib.Graph()
try:
g.parse(data=file_bytes, format=format)
except Exception as e:
raise ValueError(f"Failed to parse RDF file as {format}. Ensure the file is valid. Details: {str(e)}")
nodes_dict: Dict[str, Dict[str, Any]] = {}
edges: List[Dict[str, Any]] = []
# extract concept schemas
for scheme in g.subjects(RDF.type, SKOS.ConceptScheme):
uri = str(scheme)
# if no prefLabel
pref_label = _get_best_label(g, scheme, SKOS.prefLabel)
if not pref_label:
pref_label = uri.split("/")[-1].split("#")[-1]
nodes_dict[uri] = {
"id": uri,
"type": "skos:ConceptScheme",
"properties": {
"content": pref_label,
"alt_labels": _get_all_labels(g, scheme, SKOS.altLabel),
"description": _get_best_label(g, scheme, SKOS.definition)
}
}
# Extract concepts
for concept in g.subjects(RDF.type, SKOS.Concept):
uri = str(concept)
pref_label = _get_best_label(g, concept, SKOS.prefLabel)
if not pref_label:
pref_label = uri.split("/")[-1].split("#")[-1]
nodes_dict[uri] = {
"id": uri,
"type": "skos:Concept",
"properties": {
"content": pref_label,
"alt_labels": _get_all_labels(g, concept, SKOS.altLabel),
"description": _get_best_label(g, concept, SKOS.definition)
}
}
# Extract Relationships aka edges
structural_preds = {
SKOS.broader: "skos:broader",
SKOS.narrower: "skos:narrower",
SKOS.inScheme: "skos:inScheme",
SKOS.related: "skos:related",
SKOS.topConceptOf: "skos:topConceptOf",
SKOS.hasTopConcept: "skos:hasTopConcept"
}
for pred, edge_type in structural_preds.items():
for source, target in g.subject_objects(pred):
# Only track edges where nodes were successfully extracted
if str(source) in nodes_dict and str(target) in nodes_dict:
edges.append({
"source_id": str(source),
"target_id": str(target),
"type": edge_type,
"weight": 1.0,
"properties": {}
})
return list(nodes_dict.values()), edges
+35 -1
View File
@@ -5,6 +5,7 @@ This module provides the REST API server for the Semantica framework
using FastAPI and uvicorn.
"""
import logging
import uvicorn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
@@ -53,9 +54,42 @@ async def build_kb(request: BuildRequest):
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# Explorer API Routers (Loaded gracefully if semantica[explorer] is installed)
try:
from .explorer.routes import (
analytics,
annotations,
decisions,
enrich,
export_import,
graph,
temporal,
vocabulary,
)
app.include_router(analytics.router)
app.include_router(annotations.router)
app.include_router(decisions.router)
app.include_router(enrich.router)
app.include_router(export_import.router)
app.include_router(graph.router)
app.include_router(temporal.router)
app.include_router(vocabulary.router)
logging.info("Explorer API routes successfully mounted.")
except ImportError as exc:
logging.warning(
f"Explorer API routes not mounted. To enable the Knowledge Explorer, "
f"install the required dependencies: pip install semantica[explorer]. "
f"Details: {exc}"
)
def main():
"""Server entry point."""
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
main()
main()