diff --git a/.gitignore b/.gitignore index 89ceb3f1..a4c94d18 100644 --- a/.gitignore +++ b/.gitignore @@ -110,3 +110,7 @@ sample_data/ # Test Results test_results.txt + +# Frontend workspace artifacts +semantica-explorer/ +node_modules/ diff --git a/semantica/explorer/app.py b/semantica/explorer/app.py index 8435b3d0..34babfeb 100644 --- a/semantica/explorer/app.py +++ b/semantica/explorer/app.py @@ -85,6 +85,7 @@ def create_app(session: Optional[GraphSession] = None) -> FastAPI: from .routes.enrich import router as enrich_router from .routes.export_import import router as export_import_router from .routes.annotations import router as annotations_router + from .routes.provenance import router as provenance_router app.include_router(graph_router) app.include_router(analytics_router) @@ -93,6 +94,7 @@ def create_app(session: Optional[GraphSession] = None) -> FastAPI: app.include_router(enrich_router) app.include_router(export_import_router) app.include_router(annotations_router) + app.include_router(provenance_router) from fastapi import WebSocket, WebSocketDisconnect diff --git a/semantica/explorer/routes/provenance.py b/semantica/explorer/routes/provenance.py new file mode 100644 index 00000000..5a3043fc --- /dev/null +++ b/semantica/explorer/routes/provenance.py @@ -0,0 +1,201 @@ +""" +Provenance routes for lineage visualization and exportable reports. +""" + +import asyncio +import json +from typing import Any, Dict, List, Optional + +import networkx as nx +from fastapi import APIRouter, Depends, Query +from fastapi.responses import PlainTextResponse, Response +from pydantic import BaseModel + +from ..dependencies import get_session +from ..session import GraphSession + +router = APIRouter(prefix="/api/provenance", tags=["Power User Tools"]) + + +class ProvenanceNode(BaseModel): + id: str + label: str + prov_type: str + parent_id: str + + +class ProvenanceEdge(BaseModel): + id: str + source: str + target: str + label: str + direction: str # "upstream" | "downstream" | "lateral" + + +class ProvenanceResponse(BaseModel): + nodes: List[ProvenanceNode] + edges: List[ProvenanceEdge] + + +_AGENT_TYPES = {"person", "organization", "system", "agent"} +_ACTIVITY_TYPES = {"action", "event", "process", "activity", "decision", "publication"} + + +def _classify_prov(node_type: str) -> tuple[str, str]: + lowered = node_type.lower() + if lowered in _AGENT_TYPES: + return "Agent", "group_agent" + if lowered in _ACTIVITY_TYPES: + return "Activity", "group_activity" + return "Entity", "group_entity" + + +def _build_provenance(session: GraphSession, node_id: Optional[str] = None) -> dict: + if not node_id or node_id not in session.graph.nodes: + return {"nodes": [], "edges": []} + + graph = nx.DiGraph() + graph.add_node(node_id) + + hop_nodes = {node_id} + for edge in session.graph.edges: + if edge.source_id == node_id or edge.target_id == node_id: + graph.add_edge(edge.source_id, edge.target_id, label=edge.edge_type) + hop_nodes.add(edge.source_id) + hop_nodes.add(edge.target_id) + + for edge in session.graph.edges: + if edge.source_id in hop_nodes or edge.target_id in hop_nodes: + graph.add_edge(edge.source_id, edge.target_id, label=edge.edge_type) + + subgraph = nx.ego_graph(graph, node_id, radius=2, undirected=True) + provenance_nodes: List[Dict[str, Any]] = [] + for graph_node_id in subgraph.nodes(): + node = session.graph.nodes.get(graph_node_id) + if node is None: + continue + prov_type, parent_id = _classify_prov(node.node_type) + provenance_nodes.append( + { + "id": graph_node_id, + "label": node.content or graph_node_id, + "prov_type": prov_type, + "parent_id": parent_id, + } + ) + + provenance_edges: List[Dict[str, Any]] = [] + for source, target, data in subgraph.edges(data=True): + if target == node_id: + direction = "upstream" + elif source == node_id: + direction = "downstream" + else: + direction = "lateral" + + provenance_edges.append( + { + "id": f"{source}-{target}", + "source": source, + "target": target, + "label": data.get("label", "related_to"), + "direction": direction, + } + ) + + return {"nodes": provenance_nodes, "edges": provenance_edges} + + +def _build_report(session: GraphSession, node_id: str) -> Dict[str, Any]: + node = session.get_node(node_id) + provenance = _build_provenance(session, node_id) + return { + "node_id": node_id, + "label": node.get("content", node_id) if node else node_id, + "type": node.get("type", "entity") if node else "entity", + "properties": node.get("properties", {}) if node else {}, + "lineage": provenance, + } + + +def _render_markdown(report: Dict[str, Any]) -> str: + lines = [ + f"# Provenance Report: {report['label']}", + "", + f"- Node ID: `{report['node_id']}`", + f"- Type: `{report['type']}`", + "", + "## Properties", + ] + properties = report.get("properties", {}) + if properties: + for key, value in properties.items(): + lines.append(f"- **{key}**: {value}") + else: + lines.append("- No properties recorded") + + lines.extend(["", "## Lineage Nodes"]) + for node in report.get("lineage", {}).get("nodes", []): + lines.append(f"- `{node['id']}` ({node['prov_type']}): {node['label']}") + + edges = report.get("lineage", {}).get("edges", []) + grouped_edges = { + "upstream": [], + "downstream": [], + "lateral": [], + } + for edge in edges: + direction = edge.get("direction", "lateral") + grouped_edges.setdefault(direction, []).append(edge) + + if grouped_edges.get("upstream"): + lines.extend(["", "## Upstream"]) + for edge in grouped_edges["upstream"]: + lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`") + + if grouped_edges.get("downstream"): + lines.extend(["", "## Downstream"]) + for edge in grouped_edges["downstream"]: + lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`") + + if grouped_edges.get("lateral"): + lines.extend(["", "## Lateral"]) + for edge in grouped_edges["lateral"]: + lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`") + + return "\n".join(lines) + + +@router.get("", response_model=ProvenanceResponse) +@router.get("/", response_model=ProvenanceResponse, include_in_schema=False) +async def get_provenance_lineage( + node_id: Optional[str] = None, + session: GraphSession = Depends(get_session), +): + data = await asyncio.to_thread(_build_provenance, session, node_id) + return ProvenanceResponse( + nodes=[ProvenanceNode(**node) for node in data["nodes"]], + edges=[ProvenanceEdge(**edge) for edge in data["edges"]], + ) + + +@router.get("/report") +async def export_provenance_report( + node_id: str = Query(..., description="Node ID to export"), + format: str = Query("json", description="json or markdown"), + session: GraphSession = Depends(get_session), +): + report = await asyncio.to_thread(_build_report, session, node_id) + if format.lower() in {"md", "markdown"}: + content = _render_markdown(report) + return PlainTextResponse( + content, + headers={"Content-Disposition": f'attachment; filename="{node_id}_provenance.md"'}, + ) + + content = json.dumps(report, indent=2, default=str) + return Response( + content=content, + media_type="application/json", + headers={"Content-Disposition": f'attachment; filename="{node_id}_provenance.json"'}, + ) diff --git a/tests/explorer/test_provenance_route.py b/tests/explorer/test_provenance_route.py new file mode 100644 index 00000000..73f59bf2 --- /dev/null +++ b/tests/explorer/test_provenance_route.py @@ -0,0 +1,74 @@ +"""Unit tests for explorer provenance route helpers.""" + +from types import SimpleNamespace + +from semantica.explorer.routes.provenance import _build_provenance, _render_markdown + + +def _make_session_with_chain() -> SimpleNamespace: + """Build a minimal session-like object for Source -> Intermediate -> node_id.""" + nodes = { + "Source": SimpleNamespace(node_type="entity", content="Source"), + "Intermediate": SimpleNamespace(node_type="entity", content="Intermediate"), + "node_id": SimpleNamespace(node_type="entity", content="Target"), + } + edges = [ + SimpleNamespace(source_id="Source", target_id="Intermediate", edge_type="related_to"), + SimpleNamespace(source_id="Intermediate", target_id="node_id", edge_type="related_to"), + ] + graph = SimpleNamespace(nodes=nodes, edges=edges) + return SimpleNamespace(graph=graph) + + +def test_build_provenance_direction_classification_chain(): + session = _make_session_with_chain() + + data = _build_provenance(session, "node_id") + + node_ids = {node["id"] for node in data["nodes"]} + assert "Source" in node_ids + assert "Intermediate" in node_ids + + edge_by_pair = {(edge["source"], edge["target"]): edge for edge in data["edges"]} + + assert edge_by_pair[("Intermediate", "node_id")]["direction"] == "upstream" + assert edge_by_pair[("Source", "Intermediate")]["direction"] != "downstream" + + +def test_render_markdown_groups_edges_by_direction(): + report = { + "node_id": "node_id", + "label": "Target", + "type": "entity", + "properties": {}, + "lineage": { + "nodes": [ + {"id": "Source", "prov_type": "Entity", "label": "Source"}, + {"id": "Intermediate", "prov_type": "Entity", "label": "Intermediate"}, + {"id": "node_id", "prov_type": "Entity", "label": "Target"}, + ], + "edges": [ + { + "id": "Intermediate-node_id", + "source": "Intermediate", + "target": "node_id", + "label": "related_to", + "direction": "upstream", + }, + { + "id": "Source-Intermediate", + "source": "Source", + "target": "Intermediate", + "label": "related_to", + "direction": "lateral", + }, + ], + }, + } + + markdown = _render_markdown(report) + + assert "## Upstream" in markdown + assert "## Lateral" in markdown + assert "`Intermediate` -[related_to]-> `node_id`" in markdown + assert "`Source` -[related_to]-> `Intermediate`" in markdown