wehub-resource-sync
364 lines
14 KiB
Python
364 lines
14 KiB
Python
"""scip_ingest.py — SCIP JSON ingestion (simplified subset).
|
|
|
|
Reads a simplified SCIP-style JSON structure and converts it into
|
|
Graphify nodes and edges. NOT a full SCIP protobuf implementation —
|
|
this is a skeleton that consumes the simplified shape described below.
|
|
|
|
Not wired to the CLI in this phase.
|
|
|
|
Entry point:
|
|
ingest_scip_json(doc: object, source_file: str = "",
|
|
language: str = "python") -> dict[str, Any]
|
|
|
|
Returns {"nodes": [...], "edges": [...]} compatible with Graphify's
|
|
extraction result format. All edges emitted are endpoint-safe — the
|
|
function builds a symbol → node_id index in a first pass and either
|
|
resolves relationship targets via that index or creates a stub
|
|
external node so `build_from_json()` will keep the edge.
|
|
|
|
Supported (simplified) JSON shape:
|
|
documents[]: { relative_path, language, symbols[] }
|
|
symbols[]: { symbol, kind, display_name, documentation[],
|
|
relationships[], occurrences[] }
|
|
relationships[]: { symbol, is_reference, is_implementation,
|
|
is_type_definition, is_definition }
|
|
occurrences[]: { range[], symbol, symbol_roles }
|
|
|
|
This shape diverges from the official SCIP protobuf (where occurrences
|
|
live on the document, not on each symbol). We consume the simplified
|
|
shape that LLM-generated SCIP-style JSON commonly produces. Future
|
|
cycles may add document-level occurrence support.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
from typing import Any
|
|
|
|
from graphify.security import sanitize_metadata
|
|
|
|
|
|
def ingest_scip_json(
|
|
doc: object,
|
|
source_file: str = "",
|
|
language: str = "python",
|
|
) -> dict[str, Any]:
|
|
"""Convert a SCIP-style JSON document into Graphify nodes and edges.
|
|
|
|
Parameter ``doc`` is ``object`` (not ``dict[str, Any]``) because SCIP
|
|
documents come from external tools — we may be handed arbitrary
|
|
deserialized JSON. The first check rejects anything that isn't a dict
|
|
and returns the empty result.
|
|
|
|
Two-pass design:
|
|
1. Build a ``symbol_str → node_id`` index across every valid symbol
|
|
in every valid document, plus collect per-symbol metadata.
|
|
2. Emit nodes for every indexed symbol and then emit relationship
|
|
edges. Relationship targets are resolved via the index when
|
|
present; otherwise a stub ``scip_external`` node is added so
|
|
edges never dangle.
|
|
"""
|
|
nodes: list[dict[str, Any]] = []
|
|
edges: list[dict[str, Any]] = []
|
|
seen_node_ids: set[str] = set()
|
|
seen_edges: set[tuple[str, str, str, str | None]] = set()
|
|
|
|
if not isinstance(doc, dict):
|
|
return {"nodes": nodes, "edges": edges}
|
|
|
|
documents = doc.get("documents", [])
|
|
if not isinstance(documents, list):
|
|
return {"nodes": nodes, "edges": edges}
|
|
|
|
# ---- pass 1: build symbol → node_id indices -----------------------------
|
|
# Two indices so relationship resolution can be document-aware:
|
|
# per_doc: (symbol_id, doc_path) → node_id (same-document precedence)
|
|
# global: symbol_id → list[node_id] (cross-document fallback,
|
|
# used only when unambiguous)
|
|
per_doc_index: dict[tuple[str, str], str] = {}
|
|
global_index: dict[str, list[str]] = {}
|
|
# Per-symbol metadata kept for pass-2 node emission (avoids re-walking
|
|
# the document tree).
|
|
symbol_records: list[dict[str, Any]] = []
|
|
for document in documents:
|
|
if not isinstance(document, dict):
|
|
continue
|
|
doc_path = _coerce_str(document.get("relative_path"), source_file)
|
|
doc_language = _coerce_str(document.get("language"), language)
|
|
symbols = document.get("symbols", [])
|
|
if not isinstance(symbols, list):
|
|
continue
|
|
for symbol in symbols:
|
|
if not isinstance(symbol, dict):
|
|
continue
|
|
symbol_id = _coerce_str(symbol.get("symbol"), "")
|
|
if not symbol_id:
|
|
continue
|
|
node_id = _make_scip_node_id(symbol_id, doc_path)
|
|
per_doc_index.setdefault((symbol_id, doc_path), node_id)
|
|
# Dedupe node_ids in the global index — duplicate symbol records
|
|
# within the SAME document produce identical node_ids, and we
|
|
# don't want them to look like cross-document ambiguity.
|
|
candidates = global_index.setdefault(symbol_id, [])
|
|
if node_id not in candidates:
|
|
candidates.append(node_id)
|
|
symbol_records.append(
|
|
{
|
|
"node_id": node_id,
|
|
"symbol_id": symbol_id,
|
|
"doc_path": doc_path,
|
|
"language": doc_language,
|
|
"raw": symbol,
|
|
}
|
|
)
|
|
|
|
# ---- pass 2: emit nodes + relationship edges -----------------------------
|
|
for record in symbol_records:
|
|
_emit_symbol_node(record, nodes, seen_node_ids)
|
|
_emit_relationships(
|
|
record,
|
|
per_doc_index,
|
|
global_index,
|
|
nodes,
|
|
edges,
|
|
seen_node_ids,
|
|
seen_edges,
|
|
)
|
|
|
|
return {"nodes": nodes, "edges": edges}
|
|
|
|
|
|
def _emit_symbol_node(
|
|
record: dict[str, Any],
|
|
nodes: list[dict[str, Any]],
|
|
seen_node_ids: set[str],
|
|
) -> None:
|
|
"""Append the canonical node for a SCIP symbol record."""
|
|
node_id = record["node_id"]
|
|
if node_id in seen_node_ids:
|
|
return
|
|
raw = record["raw"]
|
|
symbol_id = record["symbol_id"]
|
|
doc_path = record["doc_path"]
|
|
kind = _coerce_str(raw.get("kind"), "unknown")
|
|
display_name = _coerce_str(raw.get("display_name"), "")
|
|
documentation = raw.get("documentation", [])
|
|
description = ""
|
|
if isinstance(documentation, list) and documentation:
|
|
first = documentation[0]
|
|
if isinstance(first, str):
|
|
description = first
|
|
occurrences = raw.get("occurrences", [])
|
|
sourceline = _first_occurrence_line(occurrences)
|
|
suffix = symbol_id.split("#")[-1] if "#" in symbol_id else symbol_id
|
|
label = display_name or suffix or symbol_id
|
|
seen_node_ids.add(node_id) # label uses display_name or suffix (never empty for valid symbols)
|
|
nodes.append(
|
|
{
|
|
"id": node_id,
|
|
"label": label,
|
|
"file_type": _scip_kind_to_file_type(kind),
|
|
"source_file": doc_path,
|
|
"source_location": f"L{sourceline}" if sourceline else "",
|
|
"metadata": sanitize_metadata(_build_scip_metadata(symbol_id, kind, description)),
|
|
}
|
|
)
|
|
|
|
|
|
def _emit_relationships(
|
|
record: dict[str, Any],
|
|
per_doc_index: dict[tuple[str, str], str],
|
|
global_index: dict[str, list[str]],
|
|
nodes: list[dict[str, Any]],
|
|
edges: list[dict[str, Any]],
|
|
seen_node_ids: set[str],
|
|
seen_edges: set[tuple[str, str, str, str | None]],
|
|
) -> None:
|
|
"""Append edges (and stub nodes when needed) for a symbol's relationships.
|
|
|
|
Relationship target resolution order:
|
|
1. Same-document `(target_symbol, doc_path)` — duplicate local symbol
|
|
names across files route to THIS file's symbol, not another's.
|
|
2. Unique cross-document match — when the symbol exists in exactly
|
|
one document and that document is different from the source.
|
|
3. Stub external node — for symbols not declared in any document
|
|
OR ambiguous duplicates across multiple documents (refusing to
|
|
guess silently).
|
|
"""
|
|
raw = record["raw"]
|
|
source_node_id = record["node_id"]
|
|
doc_path = record["doc_path"]
|
|
occurrences = raw.get("occurrences", [])
|
|
sourceline = _first_occurrence_line(occurrences)
|
|
relationships = raw.get("relationships")
|
|
if not isinstance(relationships, list):
|
|
return
|
|
for rel in relationships:
|
|
if not isinstance(rel, dict):
|
|
continue
|
|
target_symbol = _coerce_str(rel.get("symbol"), "")
|
|
if not target_symbol:
|
|
continue
|
|
target_node_id = _resolve_relationship_target(
|
|
target_symbol,
|
|
doc_path,
|
|
per_doc_index,
|
|
global_index,
|
|
)
|
|
if target_node_id is None:
|
|
# External relationship target: emit a stub node so the edge
|
|
# is never dangling. The stub uses the source document's path
|
|
# as its host context.
|
|
target_node_id = _make_scip_node_id(target_symbol, doc_path)
|
|
if target_node_id not in seen_node_ids:
|
|
seen_node_ids.add(target_node_id)
|
|
suffix = target_symbol.split("#")[-1] if "#" in target_symbol else target_symbol
|
|
nodes.append(
|
|
{
|
|
"id": target_node_id,
|
|
"label": suffix or target_symbol,
|
|
"file_type": "code",
|
|
"source_file": doc_path,
|
|
"source_location": "",
|
|
"metadata": sanitize_metadata(
|
|
_build_scip_metadata(target_symbol, "external", "")
|
|
),
|
|
}
|
|
)
|
|
relation = _scip_relation_for(rel)
|
|
source_location = f"L{sourceline}" if sourceline else ""
|
|
key = (source_node_id, target_node_id, relation, source_location)
|
|
if key in seen_edges:
|
|
continue
|
|
seen_edges.add(key)
|
|
edges.append(
|
|
{
|
|
"source": source_node_id,
|
|
"target": target_node_id,
|
|
"relation": relation,
|
|
"confidence": "EXTRACTED",
|
|
"confidence_score": 1.0,
|
|
"source_file": doc_path,
|
|
"source_location": source_location,
|
|
"weight": 1.0,
|
|
"context": "scip",
|
|
"metadata": sanitize_metadata({"scip_relationship": rel}),
|
|
}
|
|
)
|
|
|
|
|
|
def _resolve_relationship_target(
|
|
target_symbol: str,
|
|
source_doc_path: str,
|
|
per_doc_index: dict[tuple[str, str], str],
|
|
global_index: dict[str, list[str]],
|
|
) -> str | None:
|
|
"""Resolve a SCIP relationship target to an emitted node id, or None.
|
|
|
|
Resolution order:
|
|
1. Same-document match — `(target_symbol, source_doc_path)`.
|
|
2. Unique cross-document match — exactly one node id in the global
|
|
index for this symbol AND it isn't the same document we already
|
|
tried.
|
|
3. None — symbol is either absent globally OR ambiguous (defined in
|
|
multiple documents). The caller emits a stub external node.
|
|
"""
|
|
same_doc = per_doc_index.get((target_symbol, source_doc_path))
|
|
if same_doc is not None:
|
|
return same_doc
|
|
candidates = global_index.get(target_symbol, [])
|
|
if len(candidates) == 1:
|
|
return candidates[0]
|
|
return None
|
|
|
|
|
|
def _is_true(value: object) -> bool:
|
|
"""Return True only when value is exactly the boolean True.
|
|
|
|
Used for SCIP relationship flags. Truthy strings like ``"false"`` are
|
|
common in untrusted external JSON and must NOT count as a set flag.
|
|
"""
|
|
return value is True
|
|
|
|
|
|
def _scip_relation_for(rel: dict[str, Any]) -> str:
|
|
"""Pick the Graphify relation tag for a SCIP relationship dict.
|
|
|
|
Flags are accepted only when the value is exactly ``True`` — protects
|
|
against truthy-but-misleading values like ``"false"`` in external JSON.
|
|
"""
|
|
if _is_true(rel.get("is_implementation")):
|
|
return "scip_impl"
|
|
if _is_true(rel.get("is_type_definition")):
|
|
return "scip_typed"
|
|
if _is_true(rel.get("is_definition")):
|
|
return "scip_def"
|
|
return "scip_ref"
|
|
|
|
|
|
def _first_occurrence_line(occurrences: object) -> int:
|
|
"""Read the 1-based line number from the first occurrence range, defensively.
|
|
|
|
Note: ``bool`` is a subclass of ``int`` in Python — ``isinstance(True, int)``
|
|
is True. We explicitly exclude booleans so a malformed ``range: [True, …]``
|
|
cannot produce ``source_location = "LTrue"``.
|
|
"""
|
|
if not isinstance(occurrences, list) or not occurrences:
|
|
return 0
|
|
first = occurrences[0]
|
|
if not isinstance(first, dict):
|
|
return 0
|
|
rng = first.get("range", [])
|
|
if not isinstance(rng, list) or len(rng) < 1:
|
|
return 0
|
|
line = rng[0]
|
|
if isinstance(line, bool) or not isinstance(line, int) or line < 0:
|
|
return 0
|
|
return line
|
|
|
|
|
|
def _coerce_str(value: object, default: str) -> str:
|
|
"""Return ``value`` if it is a string, else the ``default`` (also a string)."""
|
|
if isinstance(value, str):
|
|
return value
|
|
if isinstance(default, str):
|
|
return default
|
|
return ""
|
|
|
|
|
|
def _make_scip_node_id(symbol: str, source_file: str) -> str:
|
|
"""Derive a stable Graphify node ID from a SCIP symbol identifier.
|
|
|
|
Uses SHA-1 truncated to 12 hex chars (48 bits). This is an identifier,
|
|
not a security boundary — collision risk is acceptable at this scale
|
|
given the per-document scoping prefix.
|
|
"""
|
|
raw = f"{source_file}:{symbol}"
|
|
h = hashlib.sha1(raw.encode(), usedforsecurity=False).hexdigest()[:12]
|
|
parts = symbol.split("#")
|
|
suffix = parts[-1] if parts else symbol
|
|
suffix = re.sub(r"[^a-zA-Z0-9_]", "_", suffix).strip("_").lower()
|
|
if suffix:
|
|
return f"scip_{suffix}_{h}"
|
|
return f"scip_{h}"
|
|
|
|
|
|
def _scip_kind_to_file_type(kind: str) -> str:
|
|
"""Map SCIP symbol kind to a Graphify file_type."""
|
|
# All SCIP symbols are code entities (functions, methods, classes, …);
|
|
# the `kind` is preserved in metadata for downstream consumers.
|
|
_ = kind # acknowledged but not currently used for file_type routing
|
|
return "code"
|
|
|
|
|
|
def _build_scip_metadata(symbol_id: str, kind: str, description: str) -> dict[str, str]:
|
|
"""Build metadata for a SCIP node."""
|
|
meta: dict[str, str] = {
|
|
"scip_symbol": symbol_id,
|
|
"scip_kind": kind,
|
|
}
|
|
if description:
|
|
meta["scip_description"] = description
|
|
return meta
|