mirror of
https://github.com/safishamsi/graphify.git
synced 2026-07-31 03:31:07 +00:00
Safi
5db8f7ce39
style: replace all em dashes with hyphens fix: explain hidden .graphify/ folder in skill output and README fix: rename .graphify/ to graphify-out/ so output is visible by default
674 lines
25 KiB
Python
674 lines
25 KiB
Python
# write graph to HTML, JSON, SVG, GraphML, Obsidian vault, and Neo4j Cypher
|
|
from __future__ import annotations
|
|
import json
|
|
import math
|
|
import re
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
import networkx as nx
|
|
from networkx.readwrite import json_graph
|
|
from graphify.security import sanitize_label
|
|
|
|
COMMUNITY_COLORS = [
|
|
"#4E79A7", "#F28E2B", "#E15759", "#76B7B2", "#59A14F",
|
|
"#EDC948", "#B07AA1", "#FF9DA7", "#9C755F", "#BAB0AC",
|
|
]
|
|
|
|
MAX_NODES_FOR_VIZ = 5_000
|
|
|
|
|
|
def to_json(G: nx.Graph, communities: dict[int, list[str]], output_path: str) -> None:
|
|
node_community = {n: cid for cid, nodes in communities.items() for n in nodes}
|
|
data = json_graph.node_link_data(G, edges="links")
|
|
for node in data["nodes"]:
|
|
node["community"] = node_community.get(node["id"])
|
|
with open(output_path, "w") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
|
|
def to_cypher(G: nx.Graph, output_path: str) -> None:
|
|
lines = ["// Neo4j Cypher import - generated by /graphify", ""]
|
|
for node_id, data in G.nodes(data=True):
|
|
label = data.get("label", node_id).replace("'", "\\'")
|
|
ftype = data.get("file_type", "unknown").capitalize()
|
|
lines.append(f"MERGE (n:{ftype} {{id: '{node_id}', label: '{label}'}});")
|
|
lines.append("")
|
|
for u, v, data in G.edges(data=True):
|
|
rel = data.get("relation", "RELATES_TO").upper().replace(" ", "_").replace("-", "_")
|
|
conf = data.get("confidence", "EXTRACTED")
|
|
lines.append(
|
|
f"MATCH (a {{id: '{u}'}}), (b {{id: '{v}'}}) "
|
|
f"MERGE (a)-[:{rel} {{confidence: '{conf}'}}]->(b);"
|
|
)
|
|
with open(output_path, "w") as f:
|
|
f.write("\n".join(lines))
|
|
|
|
|
|
def to_html(
|
|
G: nx.Graph,
|
|
communities: dict[int, list[str]],
|
|
output_path: str,
|
|
community_labels: dict[int, str] | None = None,
|
|
) -> None:
|
|
"""Generate an interactive pyvis HTML visualization of the graph.
|
|
|
|
Merged from visualizer.py. Raises ValueError if graph exceeds MAX_NODES_FOR_VIZ.
|
|
"""
|
|
from pyvis.network import Network
|
|
|
|
if G.number_of_nodes() > MAX_NODES_FOR_VIZ:
|
|
raise ValueError(
|
|
f"Graph has {G.number_of_nodes()} nodes - too large for pyvis. "
|
|
f"Use --no-viz or reduce input size."
|
|
)
|
|
|
|
node_community = {n: cid for cid, nodes in communities.items() for n in nodes}
|
|
|
|
net = Network(height="800px", width="100%", bgcolor="#1a1a2e", font_color="white")
|
|
net.barnes_hut()
|
|
|
|
for node_id, data in G.nodes(data=True):
|
|
cid = node_community.get(node_id, 0)
|
|
color = COMMUNITY_COLORS[cid % len(COMMUNITY_COLORS)]
|
|
net.add_node(
|
|
node_id,
|
|
label=sanitize_label(data.get("label", node_id)),
|
|
color=color,
|
|
title=sanitize_label(
|
|
f"Source: {data.get('source_file', 'unknown')}\n"
|
|
f"Type: {data.get('file_type', 'unknown')}\n"
|
|
f"Community: {community_labels.get(cid, str(cid)) if community_labels else cid}"
|
|
),
|
|
)
|
|
|
|
for u, v, data in G.edges(data=True):
|
|
confidence = data.get("confidence", "EXTRACTED")
|
|
width = {"EXTRACTED": 2, "INFERRED": 1, "AMBIGUOUS": 1}.get(confidence, 1)
|
|
net.add_edge(
|
|
u, v,
|
|
title=f"{data.get('relation', '')} [{confidence}]",
|
|
width=width,
|
|
dashes=(confidence != "EXTRACTED"),
|
|
)
|
|
|
|
net.save_graph(output_path)
|
|
|
|
# Inject community legend into saved HTML
|
|
if community_labels:
|
|
legend_items = ""
|
|
for cid in sorted(community_labels.keys()):
|
|
color = COMMUNITY_COLORS[cid % len(COMMUNITY_COLORS)]
|
|
label = community_labels[cid]
|
|
n_nodes = len(communities.get(cid, []))
|
|
legend_items += (
|
|
f'<div style="margin:4px 0">'
|
|
f'<span style="color:{color};font-size:18px">■</span> '
|
|
f'<span style="font-size:13px">{label} ({n_nodes})</span>'
|
|
f'</div>'
|
|
)
|
|
legend_html = (
|
|
'<div style="position:fixed;top:10px;right:10px;background:#2a2a4e;'
|
|
'padding:12px 16px;border-radius:8px;font-family:sans-serif;color:white;'
|
|
'z-index:9999;min-width:180px;">'
|
|
'<b style="font-size:14px">Communities</b><br>'
|
|
+ legend_items +
|
|
'</div>'
|
|
)
|
|
content = Path(output_path).read_text()
|
|
content = content.replace("</body>", legend_html + "\n</body>")
|
|
Path(output_path).write_text(content)
|
|
|
|
|
|
# Keep backward-compatible alias - skill.md calls generate_html
|
|
generate_html = to_html
|
|
|
|
|
|
def to_obsidian(
|
|
G: nx.Graph,
|
|
communities: dict[int, list[str]],
|
|
output_dir: str,
|
|
community_labels: dict[int, str] | None = None,
|
|
cohesion: dict[int, float] | None = None,
|
|
) -> int:
|
|
"""Export graph as an Obsidian vault - one .md file per node with [[wikilinks]],
|
|
plus one _COMMUNITY_name.md overview note per community (sorted to top by underscore prefix).
|
|
|
|
Open the output directory as a vault in Obsidian to get an interactive
|
|
graph view with community colors and full-text search over node metadata.
|
|
|
|
Returns the number of node notes + community notes written.
|
|
"""
|
|
out = Path(output_dir)
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
|
|
node_community = {n: cid for cid, nodes in communities.items() for n in nodes}
|
|
|
|
# Map node_id → safe filename so wikilinks stay consistent.
|
|
# Deduplicate: if two nodes produce the same filename, append a numeric suffix.
|
|
def safe_name(label: str) -> str:
|
|
return re.sub(r'[\\/*?:"<>|#^[\]]', "", label).strip() or "unnamed"
|
|
|
|
node_filename: dict[str, str] = {}
|
|
seen_names: dict[str, int] = {}
|
|
for node_id, data in G.nodes(data=True):
|
|
base = safe_name(data.get("label", node_id))
|
|
if base in seen_names:
|
|
seen_names[base] += 1
|
|
node_filename[node_id] = f"{base}_{seen_names[base]}"
|
|
else:
|
|
seen_names[base] = 0
|
|
node_filename[node_id] = base
|
|
|
|
# Helper: compute dominant confidence for a node across all its edges
|
|
def _dominant_confidence(node_id: str) -> str:
|
|
confs = []
|
|
for u, v, edata in G.edges(node_id, data=True):
|
|
confs.append(edata.get("confidence", "EXTRACTED"))
|
|
if not confs:
|
|
return "EXTRACTED"
|
|
return Counter(confs).most_common(1)[0][0]
|
|
|
|
# Map file_type → graphify tag
|
|
_FTYPE_TAG = {
|
|
"code": "graphify/code",
|
|
"document": "graphify/document",
|
|
"paper": "graphify/paper",
|
|
"image": "graphify/image",
|
|
}
|
|
|
|
# Write one .md file per node
|
|
for node_id, data in G.nodes(data=True):
|
|
label = data.get("label", node_id)
|
|
cid = node_community.get(node_id)
|
|
community_name = (
|
|
community_labels.get(cid, f"Community {cid}")
|
|
if community_labels and cid is not None
|
|
else f"Community {cid}"
|
|
)
|
|
|
|
# Build tags for this node
|
|
ftype = data.get("file_type", "")
|
|
ftype_tag = _FTYPE_TAG.get(ftype, f"graphify/{ftype}" if ftype else "graphify/document")
|
|
dom_conf = _dominant_confidence(node_id)
|
|
conf_tag = f"graphify/{dom_conf}"
|
|
comm_tag = f"community/{community_name.replace(' ', '_')}"
|
|
node_tags = [ftype_tag, conf_tag, comm_tag]
|
|
|
|
lines: list[str] = []
|
|
|
|
# YAML frontmatter - readable in Obsidian's properties panel
|
|
lines += [
|
|
"---",
|
|
f'source_file: "{data.get("source_file", "")}"',
|
|
f'type: "{ftype}"',
|
|
f'community: "{community_name}"',
|
|
]
|
|
if data.get("source_location"):
|
|
lines.append(f'location: "{data["source_location"]}"')
|
|
# Add tags list to frontmatter
|
|
lines.append("tags:")
|
|
for tag in node_tags:
|
|
lines.append(f" - {tag}")
|
|
lines += ["---", "", f"# {label}", ""]
|
|
|
|
# Outgoing edges as wikilinks
|
|
neighbors = list(G.neighbors(node_id))
|
|
if neighbors:
|
|
lines.append("## Connections")
|
|
for neighbor in sorted(neighbors, key=lambda n: G.nodes[n].get("label", n)):
|
|
edge_data = G.edges[node_id, neighbor]
|
|
neighbor_label = node_filename[neighbor]
|
|
relation = edge_data.get("relation", "")
|
|
confidence = edge_data.get("confidence", "EXTRACTED")
|
|
lines.append(f"- [[{neighbor_label}]] - `{relation}` [{confidence}]")
|
|
lines.append("")
|
|
|
|
# Inline tags at bottom of note body (for Obsidian tag panel)
|
|
inline_tags = " ".join(f"#{t}" for t in node_tags)
|
|
lines.append(inline_tags)
|
|
|
|
fname = node_filename[node_id] + ".md"
|
|
(out / fname).write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
# Write one _COMMUNITY_name.md overview note per community
|
|
# Build inter-community edge counts for "Connections to other communities"
|
|
inter_community_edges: dict[int, dict[int, int]] = {}
|
|
for cid in communities:
|
|
inter_community_edges[cid] = {}
|
|
for u, v in G.edges():
|
|
cu = node_community.get(u)
|
|
cv = node_community.get(v)
|
|
if cu is not None and cv is not None and cu != cv:
|
|
inter_community_edges.setdefault(cu, {})
|
|
inter_community_edges.setdefault(cv, {})
|
|
inter_community_edges[cu][cv] = inter_community_edges[cu].get(cv, 0) + 1
|
|
inter_community_edges[cv][cu] = inter_community_edges[cv].get(cu, 0) + 1
|
|
|
|
# Precompute per-node community reach (number of distinct communities a node connects to)
|
|
def _community_reach(node_id: str) -> int:
|
|
neighbor_cids = {
|
|
node_community[nb]
|
|
for nb in G.neighbors(node_id)
|
|
if nb in node_community and node_community[nb] != node_community.get(node_id)
|
|
}
|
|
return len(neighbor_cids)
|
|
|
|
community_notes_written = 0
|
|
for cid, members in communities.items():
|
|
community_name = (
|
|
community_labels.get(cid, f"Community {cid}")
|
|
if community_labels and cid is not None
|
|
else f"Community {cid}"
|
|
)
|
|
n_members = len(members)
|
|
coh_value = cohesion.get(cid) if cohesion else None
|
|
|
|
lines: list[str] = []
|
|
|
|
# YAML frontmatter
|
|
lines.append("---")
|
|
lines.append("type: community")
|
|
if coh_value is not None:
|
|
lines.append(f"cohesion: {coh_value:.2f}")
|
|
lines.append(f"members: {n_members}")
|
|
lines.append("---")
|
|
lines.append("")
|
|
lines.append(f"# {community_name}")
|
|
lines.append("")
|
|
|
|
# Cohesion + member count summary
|
|
if coh_value is not None:
|
|
cohesion_desc = (
|
|
"tightly connected" if coh_value >= 0.7
|
|
else "moderately connected" if coh_value >= 0.4
|
|
else "loosely connected"
|
|
)
|
|
lines.append(f"**Cohesion:** {coh_value:.2f} - {cohesion_desc}")
|
|
lines.append(f"**Members:** {n_members} nodes")
|
|
lines.append("")
|
|
|
|
# Members section
|
|
lines.append("## Members")
|
|
for node_id in sorted(members, key=lambda n: G.nodes[n].get("label", n)):
|
|
data = G.nodes[node_id]
|
|
node_label = node_filename[node_id]
|
|
ftype = data.get("file_type", "")
|
|
source = data.get("source_file", "")
|
|
entry = f"- [[{node_label}]]"
|
|
if ftype:
|
|
entry += f" - {ftype}"
|
|
if source:
|
|
entry += f" - {source}"
|
|
lines.append(entry)
|
|
lines.append("")
|
|
|
|
# Dataview live query (improvement 2)
|
|
comm_tag_name = community_name.replace(" ", "_")
|
|
lines.append("## Live Query (requires Dataview plugin)")
|
|
lines.append("")
|
|
lines.append("```dataview")
|
|
lines.append(f"TABLE source_file, type FROM #community/{comm_tag_name}")
|
|
lines.append("SORT file.name ASC")
|
|
lines.append("```")
|
|
lines.append("")
|
|
|
|
# Connections to other communities
|
|
cross = inter_community_edges.get(cid, {})
|
|
if cross:
|
|
lines.append("## Connections to other communities")
|
|
for other_cid, edge_count in sorted(cross.items(), key=lambda x: -x[1]):
|
|
other_name = (
|
|
community_labels.get(other_cid, f"Community {other_cid}")
|
|
if community_labels and other_cid is not None
|
|
else f"Community {other_cid}"
|
|
)
|
|
other_safe = safe_name(other_name)
|
|
lines.append(f"- {edge_count} edge{'s' if edge_count != 1 else ''} to [[_COMMUNITY_{other_safe}]]")
|
|
lines.append("")
|
|
|
|
# Top bridge nodes - highest degree nodes that connect to other communities
|
|
bridge_nodes = [
|
|
(node_id, G.degree(node_id), _community_reach(node_id))
|
|
for node_id in members
|
|
if _community_reach(node_id) > 0
|
|
]
|
|
bridge_nodes.sort(key=lambda x: (-x[2], -x[1]))
|
|
top_bridges = bridge_nodes[:5]
|
|
if top_bridges:
|
|
lines.append("## Top bridge nodes")
|
|
for node_id, degree, reach in top_bridges:
|
|
node_label = node_filename[node_id]
|
|
lines.append(
|
|
f"- [[{node_label}]] - degree {degree}, connects to {reach} "
|
|
f"{'community' if reach == 1 else 'communities'}"
|
|
)
|
|
|
|
community_safe = safe_name(community_name)
|
|
fname = f"_COMMUNITY_{community_safe}.md"
|
|
(out / fname).write_text("\n".join(lines), encoding="utf-8")
|
|
community_notes_written += 1
|
|
|
|
# Improvement 4: write .obsidian/graph.json to color nodes by community in graph view
|
|
obsidian_dir = out / ".obsidian"
|
|
obsidian_dir.mkdir(exist_ok=True)
|
|
graph_config = {
|
|
"colorGroups": [
|
|
{
|
|
"query": f"tag:#community/{label.replace(' ', '_')}",
|
|
"color": {"a": 1, "rgb": int(COMMUNITY_COLORS[cid % len(COMMUNITY_COLORS)].lstrip('#'), 16)}
|
|
}
|
|
for cid, label in sorted((community_labels or {}).items())
|
|
]
|
|
}
|
|
(obsidian_dir / "graph.json").write_text(json.dumps(graph_config, indent=2))
|
|
|
|
return G.number_of_nodes() + community_notes_written
|
|
|
|
|
|
def to_canvas(
|
|
G: nx.Graph,
|
|
communities: dict[int, list[str]],
|
|
output_path: str,
|
|
community_labels: dict[int, str] | None = None,
|
|
node_filenames: dict[str, str] | None = None,
|
|
) -> None:
|
|
"""Export graph as an Obsidian Canvas file - communities as groups, nodes as cards.
|
|
|
|
Generates a structured layout: communities arranged in a grid, nodes within
|
|
each community arranged in rows. Edges shown between connected nodes.
|
|
Opens in Obsidian as an infinite canvas with community groupings visible.
|
|
"""
|
|
# Obsidian canvas color codes (cycle through for communities)
|
|
CANVAS_COLORS = ["1", "2", "3", "4", "5", "6"] # red, orange, yellow, green, cyan, purple
|
|
|
|
def safe_name(label: str) -> str:
|
|
return re.sub(r'[\\/*?:"<>|#^[\]]', "", label).strip() or "unnamed"
|
|
|
|
# Build node_filenames if not provided (same dedup logic as to_obsidian)
|
|
if node_filenames is None:
|
|
node_filenames = {}
|
|
seen_names: dict[str, int] = {}
|
|
for node_id, data in G.nodes(data=True):
|
|
base = safe_name(data.get("label", node_id))
|
|
if base in seen_names:
|
|
seen_names[base] += 1
|
|
node_filenames[node_id] = f"{base}_{seen_names[base]}"
|
|
else:
|
|
seen_names[base] = 0
|
|
node_filenames[node_id] = base
|
|
|
|
num_communities = len(communities)
|
|
cols = math.ceil(math.sqrt(num_communities)) if num_communities > 0 else 1
|
|
rows = math.ceil(num_communities / cols) if num_communities > 0 else 1
|
|
|
|
canvas_nodes: list[dict] = []
|
|
canvas_edges: list[dict] = []
|
|
|
|
# Lay out communities in a grid
|
|
gap = 80
|
|
group_x_offsets: list[int] = []
|
|
group_y_offsets: list[int] = []
|
|
|
|
# Precompute group sizes so we can calculate offsets
|
|
sorted_cids = sorted(communities.keys())
|
|
group_sizes: dict[int, tuple[int, int]] = {}
|
|
for cid in sorted_cids:
|
|
members = communities[cid]
|
|
n = len(members)
|
|
w = max(600, 220 * math.ceil(math.sqrt(n)) if n > 0 else 600)
|
|
h = max(400, 100 * math.ceil(n / 3) + 120 if n > 0 else 400)
|
|
group_sizes[cid] = (w, h)
|
|
|
|
# Compute cumulative row heights and col widths for grid placement
|
|
# Each grid cell uses the max width/height in its col/row
|
|
col_widths: list[int] = []
|
|
row_heights: list[int] = []
|
|
for col_idx in range(cols):
|
|
max_w = 0
|
|
for row_idx in range(rows):
|
|
linear = row_idx * cols + col_idx
|
|
if linear < len(sorted_cids):
|
|
cid = sorted_cids[linear]
|
|
w, _ = group_sizes[cid]
|
|
max_w = max(max_w, w)
|
|
col_widths.append(max_w)
|
|
|
|
for row_idx in range(rows):
|
|
max_h = 0
|
|
for col_idx in range(cols):
|
|
linear = row_idx * cols + col_idx
|
|
if linear < len(sorted_cids):
|
|
cid = sorted_cids[linear]
|
|
_, h = group_sizes[cid]
|
|
max_h = max(max_h, h)
|
|
row_heights.append(max_h)
|
|
|
|
# Map from cid → (group_x, group_y, group_w, group_h)
|
|
group_layout: dict[int, tuple[int, int, int, int]] = {}
|
|
for idx, cid in enumerate(sorted_cids):
|
|
col_idx = idx % cols
|
|
row_idx = idx // cols
|
|
gx = sum(col_widths[:col_idx]) + col_idx * gap
|
|
gy = sum(row_heights[:row_idx]) + row_idx * gap
|
|
gw, gh = group_sizes[cid]
|
|
group_layout[cid] = (gx, gy, gw, gh)
|
|
|
|
# Build set of all node_ids in canvas for edge filtering
|
|
all_canvas_nodes: set[str] = set()
|
|
for members in communities.values():
|
|
all_canvas_nodes.update(members)
|
|
|
|
# Generate group and node canvas entries
|
|
for idx, cid in enumerate(sorted_cids):
|
|
members = communities[cid]
|
|
community_name = (
|
|
community_labels.get(cid, f"Community {cid}")
|
|
if community_labels and cid is not None
|
|
else f"Community {cid}"
|
|
)
|
|
gx, gy, gw, gh = group_layout[cid]
|
|
canvas_color = CANVAS_COLORS[idx % len(CANVAS_COLORS)]
|
|
|
|
# Group node
|
|
canvas_nodes.append({
|
|
"id": f"g{cid}",
|
|
"type": "group",
|
|
"label": community_name,
|
|
"x": gx,
|
|
"y": gy,
|
|
"width": gw,
|
|
"height": gh,
|
|
"color": canvas_color,
|
|
})
|
|
|
|
# Node cards inside the group - rows of 3
|
|
sorted_members = sorted(members, key=lambda n: G.nodes[n].get("label", n))
|
|
for m_idx, node_id in enumerate(sorted_members):
|
|
col = m_idx % 3
|
|
row = m_idx // 3
|
|
nx_x = gx + 20 + col * (180 + 20)
|
|
nx_y = gy + 80 + row * (60 + 20)
|
|
fname = node_filenames.get(node_id, safe_name(G.nodes[node_id].get("label", node_id)))
|
|
canvas_nodes.append({
|
|
"id": f"n_{node_id}",
|
|
"type": "file",
|
|
"file": f"graphify/obsidian/{fname}.md",
|
|
"x": nx_x,
|
|
"y": nx_y,
|
|
"width": 180,
|
|
"height": 60,
|
|
})
|
|
|
|
# Generate edges - only between nodes both in canvas, cap at 200 highest-weight
|
|
all_edges_weighted: list[tuple[float, str, str, str]] = []
|
|
for u, v, edata in G.edges(data=True):
|
|
if u in all_canvas_nodes and v in all_canvas_nodes:
|
|
weight = edata.get("weight", 1.0)
|
|
relation = edata.get("relation", "")
|
|
conf = edata.get("confidence", "EXTRACTED")
|
|
label = f"{relation} [{conf}]" if relation else f"[{conf}]"
|
|
all_edges_weighted.append((weight, u, v, label))
|
|
|
|
all_edges_weighted.sort(key=lambda x: -x[0])
|
|
for weight, u, v, label in all_edges_weighted[:200]:
|
|
canvas_edges.append({
|
|
"id": f"e_{u}_{v}",
|
|
"fromNode": f"n_{u}",
|
|
"toNode": f"n_{v}",
|
|
"label": label,
|
|
})
|
|
|
|
canvas_data = {"nodes": canvas_nodes, "edges": canvas_edges}
|
|
Path(output_path).write_text(json.dumps(canvas_data, indent=2), encoding="utf-8")
|
|
|
|
|
|
def push_to_neo4j(
|
|
G: nx.Graph,
|
|
uri: str,
|
|
user: str,
|
|
password: str,
|
|
communities: dict[int, list[str]] | None = None,
|
|
) -> dict[str, int]:
|
|
"""Push graph directly to a running Neo4j instance via the Python driver.
|
|
|
|
Requires: pip install neo4j
|
|
|
|
Uses MERGE so re-running is safe - nodes and edges are upserted, not duplicated.
|
|
Returns a dict with counts of nodes and edges pushed.
|
|
"""
|
|
try:
|
|
from neo4j import GraphDatabase
|
|
except ImportError as e:
|
|
raise ImportError(
|
|
"neo4j driver not installed. Run: pip install neo4j"
|
|
) from e
|
|
|
|
node_community = (
|
|
{n: cid for cid, nodes in communities.items() for n in nodes}
|
|
if communities else {}
|
|
)
|
|
|
|
def _safe_rel(relation: str) -> str:
|
|
return re.sub(r"[^A-Z0-9_]", "_", relation.upper().replace(" ", "_").replace("-", "_")) or "RELATED_TO"
|
|
|
|
driver = GraphDatabase.driver(uri, auth=(user, password))
|
|
nodes_pushed = 0
|
|
edges_pushed = 0
|
|
|
|
with driver.session() as session:
|
|
for node_id, data in G.nodes(data=True):
|
|
props = {k: v for k, v in data.items() if isinstance(v, (str, int, float, bool))}
|
|
props["id"] = node_id
|
|
cid = node_community.get(node_id)
|
|
if cid is not None:
|
|
props["community"] = cid
|
|
ftype = data.get("file_type", "Entity").capitalize()
|
|
session.run(
|
|
f"MERGE (n:{ftype} {{id: $id}}) SET n += $props",
|
|
id=node_id,
|
|
props=props,
|
|
)
|
|
nodes_pushed += 1
|
|
|
|
for u, v, data in G.edges(data=True):
|
|
rel = _safe_rel(data.get("relation", "RELATED_TO"))
|
|
props = {k: v for k, v in data.items() if isinstance(v, (str, int, float, bool))}
|
|
session.run(
|
|
f"MATCH (a {{id: $src}}), (b {{id: $tgt}}) "
|
|
f"MERGE (a)-[r:{rel}]->(b) SET r += $props",
|
|
src=u,
|
|
tgt=v,
|
|
props=props,
|
|
)
|
|
edges_pushed += 1
|
|
|
|
driver.close()
|
|
return {"nodes": nodes_pushed, "edges": edges_pushed}
|
|
|
|
|
|
def to_graphml(
|
|
G: nx.Graph,
|
|
communities: dict[int, list[str]],
|
|
output_path: str,
|
|
) -> None:
|
|
"""Export graph as GraphML - opens in Gephi, yEd, and any GraphML-compatible tool.
|
|
|
|
Community IDs are written as a node attribute so Gephi can colour by community.
|
|
Edge confidence (EXTRACTED/INFERRED/AMBIGUOUS) is preserved as an edge attribute.
|
|
"""
|
|
H = G.copy()
|
|
node_community = {n: cid for cid, nodes in communities.items() for n in nodes}
|
|
for node_id in H.nodes():
|
|
H.nodes[node_id]["community"] = node_community.get(node_id, -1)
|
|
nx.write_graphml(H, output_path)
|
|
|
|
|
|
def to_svg(
|
|
G: nx.Graph,
|
|
communities: dict[int, list[str]],
|
|
output_path: str,
|
|
community_labels: dict[int, str] | None = None,
|
|
figsize: tuple[int, int] = (20, 14),
|
|
) -> None:
|
|
"""Export graph as an SVG file using matplotlib + spring layout.
|
|
|
|
Lightweight and embeddable - works in Obsidian notes, Notion, GitHub READMEs,
|
|
and any markdown renderer. No JavaScript required.
|
|
|
|
Node size scales with degree. Community colors match the pyvis HTML output.
|
|
"""
|
|
try:
|
|
import matplotlib
|
|
matplotlib.use("Agg")
|
|
import matplotlib.pyplot as plt
|
|
import matplotlib.patches as mpatches
|
|
except ImportError as e:
|
|
raise ImportError("matplotlib not installed. Run: pip install matplotlib") from e
|
|
|
|
node_community = {n: cid for cid, nodes in communities.items() for n in nodes}
|
|
|
|
fig, ax = plt.subplots(figsize=figsize, facecolor="#1a1a2e")
|
|
ax.set_facecolor("#1a1a2e")
|
|
ax.axis("off")
|
|
|
|
pos = nx.spring_layout(G, seed=42, k=2.0 / (G.number_of_nodes() ** 0.5 + 1))
|
|
|
|
degree = dict(G.degree())
|
|
max_deg = max(degree.values()) if degree else 1
|
|
|
|
node_colors = [COMMUNITY_COLORS[node_community.get(n, 0) % len(COMMUNITY_COLORS)] for n in G.nodes()]
|
|
node_sizes = [300 + 1200 * (degree.get(n, 1) / max_deg) for n in G.nodes()]
|
|
|
|
# Draw edges - dashed for non-EXTRACTED
|
|
for u, v, data in G.edges(data=True):
|
|
conf = data.get("confidence", "EXTRACTED")
|
|
style = "solid" if conf == "EXTRACTED" else "dashed"
|
|
alpha = 0.6 if conf == "EXTRACTED" else 0.3
|
|
x0, y0 = pos[u]
|
|
x1, y1 = pos[v]
|
|
ax.plot([x0, x1], [y0, y1], color="#aaaaaa", linewidth=0.8,
|
|
linestyle=style, alpha=alpha, zorder=1)
|
|
|
|
nx.draw_networkx_nodes(G, pos, ax=ax, node_color=node_colors,
|
|
node_size=node_sizes, alpha=0.9)
|
|
nx.draw_networkx_labels(G, pos, ax=ax,
|
|
labels={n: G.nodes[n].get("label", n) for n in G.nodes()},
|
|
font_size=7, font_color="white")
|
|
|
|
# Legend
|
|
if community_labels:
|
|
patches = [
|
|
mpatches.Patch(
|
|
color=COMMUNITY_COLORS[cid % len(COMMUNITY_COLORS)],
|
|
label=f"{label} ({len(communities.get(cid, []))})",
|
|
)
|
|
for cid, label in sorted(community_labels.items())
|
|
]
|
|
ax.legend(handles=patches, loc="upper left", framealpha=0.7,
|
|
facecolor="#2a2a4e", labelcolor="white", fontsize=8)
|
|
|
|
plt.tight_layout()
|
|
plt.savefig(output_path, format="svg", bbox_inches="tight",
|
|
facecolor=fig.get_facecolor())
|
|
plt.close(fig)
|