mirror of
https://github.com/safishamsi/graphify.git
synced 2026-07-14 11:27:10 +00:00
Safi
8c17230586
- build/validate: accept NetworkX <=3.1 "links" key alongside "edges" (#212) - __main__: skip version check during install/uninstall, deduplicate paths (#220) - all file IO: explicit encoding="utf-8" to prevent crashes on Windows CJK locales (#204) - hooks: add newline="\n" on write to prevent CRLF shebang breakage on Windows (#204) - export: strip trailing .md from safe_name so "CLAUDE.md" doesn't become "CLAUDE.md.md" (#221) - report: add Community Hubs navigation block so Obsidian vault stays connected (#221) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
88 lines
4.3 KiB
Python
88 lines
4.3 KiB
Python
# assemble node+edge dicts into a NetworkX graph, preserving edge direction
|
|
#
|
|
# Node deduplication — three layers:
|
|
#
|
|
# 1. Within a file (AST): each extractor tracks a `seen_ids` set. A node ID is
|
|
# emitted at most once per file, so duplicate class/function definitions in
|
|
# the same source file are collapsed to the first occurrence.
|
|
#
|
|
# 2. Between files (build): NetworkX G.add_node() is idempotent — calling it
|
|
# twice with the same ID overwrites the attributes with the second call's
|
|
# values. Nodes are added in extraction order (AST first, then semantic),
|
|
# so if the same entity is extracted by both passes the semantic node
|
|
# silently overwrites the AST node. This is intentional: semantic nodes
|
|
# carry richer labels and cross-file context, while AST nodes have precise
|
|
# source_location. If you need to change the priority, reorder extractions
|
|
# passed to build().
|
|
#
|
|
# 3. Semantic merge (skill): before calling build(), the skill merges cached
|
|
# and new semantic results using an explicit `seen` set keyed on node["id"],
|
|
# so duplicates across cache hits and new extractions are resolved there
|
|
# before any graph construction happens.
|
|
#
|
|
from __future__ import annotations
|
|
import sys
|
|
import networkx as nx
|
|
from .validate import validate_extraction
|
|
|
|
|
|
def build_from_json(extraction: dict, *, directed: bool = False) -> nx.Graph:
|
|
"""Build a NetworkX graph from an extraction dict.
|
|
|
|
directed=True produces a DiGraph that preserves edge direction (source→target).
|
|
directed=False (default) produces an undirected Graph for backward compatibility.
|
|
"""
|
|
# NetworkX <= 3.1 serialised edges as "links"; remap to "edges" for compatibility.
|
|
if "edges" not in extraction and "links" in extraction:
|
|
extraction = dict(extraction, edges=extraction["links"])
|
|
errors = validate_extraction(extraction)
|
|
# Dangling edges (stdlib/external imports) are expected - only warn about real schema errors.
|
|
real_errors = [e for e in errors if "does not match any node id" not in e]
|
|
if real_errors:
|
|
print(f"[graphify] Extraction warning ({len(real_errors)} issues): {real_errors[0]}", file=sys.stderr)
|
|
G: nx.Graph = nx.DiGraph() if directed else nx.Graph()
|
|
for node in extraction.get("nodes", []):
|
|
G.add_node(node["id"], **{k: v for k, v in node.items() if k != "id"})
|
|
node_set = set(G.nodes())
|
|
for edge in extraction.get("edges", []):
|
|
if "source" not in edge and "from" in edge:
|
|
edge["source"] = edge["from"]
|
|
if "target" not in edge and "to" in edge:
|
|
edge["target"] = edge["to"]
|
|
if "source" not in edge or "target" not in edge:
|
|
continue
|
|
src, tgt = edge["source"], edge["target"]
|
|
if src not in node_set or tgt not in node_set:
|
|
continue # skip edges to external/stdlib nodes - expected, not an error
|
|
attrs = {k: v for k, v in edge.items() if k not in ("source", "target")}
|
|
# Preserve original edge direction - undirected graphs lose it otherwise,
|
|
# causing display functions to show edges backwards.
|
|
attrs["_src"] = src
|
|
attrs["_tgt"] = tgt
|
|
G.add_edge(src, tgt, **attrs)
|
|
hyperedges = extraction.get("hyperedges", [])
|
|
if hyperedges:
|
|
G.graph["hyperedges"] = hyperedges
|
|
return G
|
|
|
|
|
|
def build(extractions: list[dict], *, directed: bool = False) -> nx.Graph:
|
|
"""Merge multiple extraction results into one graph.
|
|
|
|
directed=True produces a DiGraph that preserves edge direction (source→target).
|
|
directed=False (default) produces an undirected Graph for backward compatibility.
|
|
|
|
Extractions are merged in order. For nodes with the same ID, the last
|
|
extraction's attributes win (NetworkX add_node overwrites). Pass AST
|
|
results before semantic results so semantic labels take precedence, or
|
|
reverse the order if you prefer AST source_location precision to win.
|
|
"""
|
|
combined: dict = {"nodes": [], "edges": [], "hyperedges": [], "input_tokens": 0, "output_tokens": 0}
|
|
for ext in extractions:
|
|
combined["nodes"].extend(ext.get("nodes", []))
|
|
combined["edges"].extend(ext.get("edges", []))
|
|
combined["hyperedges"].extend(ext.get("hyperedges", []))
|
|
combined["input_tokens"] += ext.get("input_tokens", 0)
|
|
combined["output_tokens"] += ext.get("output_tokens", 0)
|
|
return build_from_json(combined, directed=directed)
|