diff --git a/graphify/export.py b/graphify/export.py
index 36e4f99..bf27eed 100644
--- a/graphify/export.py
+++ b/graphify/export.py
@@ -17,6 +17,8 @@ from graphify.security import sanitize_label
from graphify.analyze import _node_community_map
from graphify.build import edge_data
+from graphify.exporters.graphdb import push_to_falkordb, push_to_neo4j # noqa: E402,F401
+
# Artifacts worth preserving across rebuilds (non-regenerable without LLM or curation).
_BACKUP_ARTIFACTS = [
@@ -149,313 +151,9 @@ def _yaml_str(s: str) -> str:
return "".join(out)
-COMMUNITY_COLORS = [
- "#4E79A7", "#F28E2B", "#E15759", "#76B7B2", "#59A14F",
- "#EDC948", "#B07AA1", "#FF9DA7", "#9C755F", "#BAB0AC",
-]
+from graphify.exporters.base import COMMUNITY_COLORS # noqa: E402,F401
-MAX_NODES_FOR_VIZ = 5_000
-
-
-def _viz_node_limit() -> int:
- """Return the effective viz node limit, honoring GRAPHIFY_VIZ_NODE_LIMIT env var.
-
- Falls back to MAX_NODES_FOR_VIZ when the env var is unset, empty, or non-integer.
- Set to 0 to disable HTML viz unconditionally (useful for CI runners).
- """
- import os
- raw = os.environ.get("GRAPHIFY_VIZ_NODE_LIMIT")
- if raw is None or not raw.strip():
- return MAX_NODES_FOR_VIZ
- try:
- return int(raw)
- except ValueError:
- return MAX_NODES_FOR_VIZ
-
-
-def _html_styles() -> str:
- return """"""
-
-
-def _hyperedge_script(hyperedges_json: str) -> str:
- return f""""""
-
-
-def _html_script(nodes_json: str, edges_json: str, legend_json: str) -> str:
- return f""""""
+from graphify.exporters.html import to_html # noqa: E402,F401
_CONFIDENCE_SCORE_DEFAULTS = {"EXTRACTED": 1.0, "INFERRED": 0.5, "AMBIGUOUS": 0.2}
@@ -629,244 +327,6 @@ def to_cypher(G: nx.Graph, output_path: str) -> None:
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,
- member_counts: dict[int, int] | None = None,
- node_limit: int | None = None,
- learning_overlay: dict | None = None,
-) -> None:
- """Generate an interactive vis.js HTML visualization of the graph.
-
- Features: node size by degree, click-to-inspect panel, search box,
- community filter, physics clustering by community, confidence-styled edges.
- Raises ValueError if graph exceeds MAX_NODES_FOR_VIZ.
-
- If member_counts is provided (aggregated community view), node sizes are
- based on community member counts rather than graph degree.
-
- If node_limit is set and the graph exceeds it, automatically builds an
- aggregated community-level meta-graph instead of raising ValueError.
- """
- limit = node_limit if node_limit is not None else _viz_node_limit()
- if G.number_of_nodes() > limit:
- if node_limit is not None:
- # Build aggregated community meta-graph
- from collections import Counter as _Counter
- import networkx as _nx
- print(f"Graph has {G.number_of_nodes()} nodes (above {limit} limit). Building aggregated community view...")
- node_to_community = {nid: cid for cid, members in communities.items() for nid in members}
- meta = _nx.Graph()
- for cid, members in communities.items():
- meta.add_node(str(cid), label=(community_labels or {}).get(cid, f"Community {cid}"))
- edge_counts = _Counter()
- for u, v in G.edges():
- cu, cv = node_to_community.get(u), node_to_community.get(v)
- if cu is not None and cv is not None and cu != cv:
- edge_counts[(min(cu, cv), max(cu, cv))] += 1
- for (cu, cv), w in edge_counts.items():
- meta.add_edge(str(cu), str(cv), weight=w,
- relation=f"{w} cross-community edges", confidence="AGGREGATED")
- if meta.number_of_nodes() <= 1:
- print("Single community - aggregated view not useful. Skipping graph.html.")
- return
- meta_communities = {cid: [str(cid)] for cid in communities}
- mc = {cid: len(members) for cid, members in communities.items()}
- # Remap hyperedges from semantic node IDs to community IDs
- raw_hyperedges = G.graph.get("hyperedges", [])
- if raw_hyperedges:
- remapped = []
- for he in raw_hyperedges:
- he_members = he.get("nodes", [])
- comm_ids, seen = [], set()
- for nid in he_members:
- c = node_to_community.get(nid)
- if c is None:
- continue
- s = str(c)
- if s in seen:
- continue
- seen.add(s)
- comm_ids.append(s)
- if len(comm_ids) < 2:
- continue
- remapped.append({
- "id": he.get("id", ""),
- "label": he.get("label") or he.get("relation", "").replace("_", " "),
- "nodes": comm_ids,
- })
- meta.graph["hyperedges"] = remapped
- to_html(meta, meta_communities, output_path,
- community_labels=community_labels, member_counts=mc)
- print(f"graph.html written (aggregated: {meta.number_of_nodes()} community nodes, {meta.number_of_edges()} cross-community edges)")
- print("Tip: run with --obsidian for full node-level detail.")
- return
- raise ValueError(
- f"Graph has {G.number_of_nodes()} nodes - too large for HTML viz "
- f"(limit: {limit}). Use --no-viz, raise GRAPHIFY_VIZ_NODE_LIMIT, "
- f"or reduce input size."
- )
-
- node_community = _node_community_map(communities)
- degree = dict(G.degree())
- max_deg = max(degree.values(), default=1) or 1
- max_mc = (max(member_counts.values(), default=1) or 1) if member_counts else 1
-
- # Work-memory overlay (derived sidecar). When not passed explicitly, load it
- # best-effort from the sibling .graphify_learning.json next to the output
- # graph.html (which lives beside graph.json). Empty/missing => no learning
- # fields, so the un-annotated render is byte-identical to pre-feature.
- if learning_overlay is None:
- learning_overlay = {}
- try:
- from graphify.reflect import load_learning_overlay as _llo
- learning_overlay = _llo(Path(output_path))
- except Exception:
- learning_overlay = {}
- # Status -> ring color. preferred=green, contested=amber. Tentative gets no
- # ring (it's not yet trustworthy enough to highlight in the map).
- _RING = {"preferred": "#22c55e", "contested": "#f59e0b"}
-
- # Build nodes list for vis.js
- vis_nodes = []
- for node_id, data in G.nodes(data=True):
- cid = node_community.get(node_id, 0)
- color = COMMUNITY_COLORS[cid % len(COMMUNITY_COLORS)]
- label = sanitize_label(data.get("label", node_id))
- deg = degree.get(node_id, 1)
- if member_counts:
- mc = member_counts.get(cid, 1)
- size = 10 + 30 * (mc / max_mc)
- font_size = 12
- else:
- size = 10 + 30 * (deg / max_deg)
- # Only show label for high-degree nodes by default; others show on hover
- font_size = 12 if deg >= max_deg * 0.15 else 0
- node = {
- "id": node_id,
- "label": label,
- "color": {"background": color, "border": color, "highlight": {"background": "#ffffff", "border": color}},
- "size": round(size, 1),
- "font": {"size": font_size, "color": "#ffffff"},
- "title": _html.escape(label),
- "community": cid,
- "community_name": sanitize_label((community_labels or {}).get(cid, f"Community {cid}")),
- "source_file": sanitize_label(str(data.get("source_file") or "")),
- "file_type": data.get("file_type", ""),
- "degree": deg,
- }
- # Conditional learning fields — only present for annotated nodes, so
- # un-annotated output keeps the exact pre-feature node dict shape.
- entry = learning_overlay.get(str(node_id)) if learning_overlay else None
- if entry:
- status = sanitize_label(str(entry.get("status", "")))
- stale = bool(entry.get("stale"))
- node["learning_status"] = status
- node["learning_stale"] = stale
- ring = _RING.get(status)
- if ring:
- # Status-colored ring via the border; stale => desaturated +
- # dashed (vis.js supports per-node `shapeProperties.borderDashes`).
- if stale:
- ring = "#9ca3af"
- node["shapeProperties"] = {"borderDashes": [4, 4]}
- node["borderWidth"] = 3
- node["color"] = {
- "background": color, "border": ring,
- "highlight": {"background": "#ffffff", "border": ring},
- }
- # Lesson line appended to the hover title.
- if status == "contested":
- lesson = f"Lesson: contested (useful {entry.get('uses', 0)} / dead-end {entry.get('neg', 0)})"
- elif status == "preferred":
- lesson = f"Lesson: preferred source ({entry.get('uses', 0)} useful, score={entry.get('score', 0)})"
- else:
- lesson = f"Lesson: {status} ({entry.get('uses', 0)} useful)"
- if stale:
- lesson += " [code changed — re-verify]"
- node["title"] = _html.escape(label) + "\n" + _html.escape(sanitize_label(lesson))
- vis_nodes.append(node)
-
- # Build edges list. Restore original edge direction from _src/_tgt
- # (stashed by build.py for exactly this reason): undirected NetworkX
- # canonicalizes endpoint order, which would otherwise flip the arrow
- # for `calls` and `rationale_for` in the rendered graph (#563).
- vis_edges = []
- for u, v, data in G.edges(data=True):
- confidence = data.get("confidence", "EXTRACTED")
- relation = data.get("relation", "")
- true_src = data.get("_src", u)
- true_tgt = data.get("_tgt", v)
- vis_edges.append({
- "from": true_src,
- "to": true_tgt,
- "label": relation,
- "title": _html.escape(f"{relation} [{confidence}]"),
- "dashes": confidence != "EXTRACTED",
- "width": 2 if confidence == "EXTRACTED" else 1,
- "color": {"opacity": 0.7 if confidence == "EXTRACTED" else 0.35},
- "confidence": confidence,
- })
-
- # Build community legend data
- legend_data = []
- for cid in sorted((community_labels or {}).keys()):
- color = COMMUNITY_COLORS[cid % len(COMMUNITY_COLORS)]
- lbl = _html.escape(sanitize_label((community_labels or {}).get(cid, f"Community {cid}")))
- n = member_counts.get(cid, len(communities.get(cid, []))) if member_counts else len(communities.get(cid, []))
- legend_data.append({"cid": cid, "color": color, "label": lbl, "count": n})
-
- # Escape sequences so embedded JSON cannot break out of the script tag
- def _js_safe(obj) -> str:
- return json.dumps(obj).replace("
-
-
-
-graphify - {title}
-
-{_html_styles()}
-
-
-
-
-{_html_script(nodes_json, edges_json, legend_json)}
-{_hyperedge_script(hyperedges_json)}
-
-"""
-
- Path(output_path).write_text(html, encoding="utf-8") # nosec
-
-
# Keep backward-compatible alias - skill.md calls generate_html
generate_html = to_html
@@ -1397,174 +857,6 @@ def to_canvas(
Path(output_path).write_text(json.dumps(canvas_data, indent=2), encoding="utf-8") # nosec
-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 = _node_community_map(communities) if communities else {}
-
- def _safe_rel(relation: str) -> str:
- return re.sub(r"[^A-Z0-9_]", "_", relation.upper().replace(" ", "_").replace("-", "_")) or "RELATED_TO"
-
- def _safe_label(label: str) -> str:
- """Sanitize a Neo4j node label to prevent Cypher injection."""
- sanitized = re.sub(r"[^A-Za-z0-9_]", "", label)
- return sanitized if sanitized else "Entity"
-
- 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)) and not k.startswith("_")
- }
- props["id"] = node_id
- cid = node_community.get(node_id)
- if cid is not None:
- props["community"] = cid
- ftype = _safe_label(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)) and not k.startswith("_")
- }
- 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 push_to_falkordb(
- G: nx.Graph,
- uri: str,
- user: str | None = None,
- password: str | None = None,
- communities: dict[int, list[str]] | None = None,
- graph_name: str = "graphify",
-) -> dict[str, int]:
- """Push graph directly to a running FalkorDB instance via the Python SDK.
-
- Requires: pip install falkordb
-
- FalkorDB is OpenCypher-compatible, so the MERGE/SET upsert queries are
- identical to push_to_neo4j. Differences from the Neo4j path:
- - connects with FalkorDB(host, port, username, password) instead of a bolt
- driver; only the host/port are read from the URI, so the scheme is
- informational - "falkordb://localhost:6379", "redis://localhost:6379"
- and a bare "localhost:6379" are all equivalent (default port 6379).
- - a named graph is selected via db.select_graph(graph_name) (default
- "graphify"); FalkorDB keys each graph by name in the same instance.
- - queries run via graph.query(cypher, params) - there is no session object.
- - auth is optional (FalkorDB runs without credentials by default), so user
- and password may be None.
- - no APOC: the Neo4j path does not use APOC either, so nothing to port.
-
- 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 falkordb import FalkorDB
- except ImportError as e:
- raise ImportError(
- "falkordb SDK not installed. Run: pip install falkordb"
- ) from e
-
- from urllib.parse import urlparse
-
- node_community = _node_community_map(communities) if communities else {}
-
- def _safe_rel(relation: str) -> str:
- return re.sub(r"[^A-Z0-9_]", "_", relation.upper().replace(" ", "_").replace("-", "_")) or "RELATED_TO"
-
- def _safe_label(label: str) -> str:
- """Sanitize a FalkorDB node label to prevent Cypher injection."""
- sanitized = re.sub(r"[^A-Za-z0-9_]", "", label)
- return sanitized if sanitized else "Entity"
-
- parsed = urlparse(uri if "://" in uri else f"redis://{uri}")
- # FalkorDB auth is optional. Only send credentials when a password is
- # provided; otherwise connect anonymously and ignore any bolt-style default
- # username (e.g. Neo4j's "neo4j"), which FalkorDB rejects as an unknown ACL
- # user. Credentials embedded in the URI take precedence over the args.
- connect_user = parsed.username or (user if password else None)
- connect_password = parsed.password or (password or None)
- db = FalkorDB(
- host=parsed.hostname or "localhost",
- port=parsed.port or 6379,
- username=connect_user,
- password=connect_password,
- )
- graph = db.select_graph(graph_name)
- nodes_pushed = 0
- edges_pushed = 0
-
- 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)) and not k.startswith("_")
- }
- props["id"] = node_id
- cid = node_community.get(node_id)
- if cid is not None:
- props["community"] = cid
- ftype = _safe_label(data.get("file_type", "Entity").capitalize())
- graph.query(
- 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)) and not k.startswith("_")
- }
- graph.query(
- 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
-
- return {"nodes": nodes_pushed, "edges": edges_pushed}
-
-
def to_graphml(
G: nx.Graph,
communities: dict[int, list[str]],
diff --git a/graphify/exporters/__init__.py b/graphify/exporters/__init__.py
new file mode 100644
index 0000000..41e1aef
--- /dev/null
+++ b/graphify/exporters/__init__.py
@@ -0,0 +1 @@
+"""exporters package."""
diff --git a/graphify/exporters/base.py b/graphify/exporters/base.py
new file mode 100644
index 0000000..9c94196
--- /dev/null
+++ b/graphify/exporters/base.py
@@ -0,0 +1,14 @@
+"""Shared constants/helpers for the graphify exporters package.
+
+Symbols used by more than one exporter live here so each exporter module can be
+split out of graphify/export.py without a circular import (export.py and the
+per-format modules both import from here, never from each other).
+"""
+from __future__ import annotations
+
+# Categorical palette for community coloring, shared by the HTML, SVG, and
+# Obsidian exporters. Moved verbatim from graphify/export.py.
+COMMUNITY_COLORS = [
+ "#4E79A7", "#F28E2B", "#E15759", "#76B7B2", "#59A14F",
+ "#EDC948", "#B07AA1", "#FF9DA7", "#9C755F", "#BAB0AC",
+]
diff --git a/graphify/exporters/graphdb.py b/graphify/exporters/graphdb.py
new file mode 100644
index 0000000..14c47f0
--- /dev/null
+++ b/graphify/exporters/graphdb.py
@@ -0,0 +1,173 @@
+"""graphdb — moved verbatim from graphify/export.py."""
+from __future__ import annotations
+
+from graphify.analyze import _node_community_map
+import networkx as nx
+import re
+
+
+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 = _node_community_map(communities) if communities else {}
+
+ def _safe_rel(relation: str) -> str:
+ return re.sub(r"[^A-Z0-9_]", "_", relation.upper().replace(" ", "_").replace("-", "_")) or "RELATED_TO"
+
+ def _safe_label(label: str) -> str:
+ """Sanitize a Neo4j node label to prevent Cypher injection."""
+ sanitized = re.sub(r"[^A-Za-z0-9_]", "", label)
+ return sanitized if sanitized else "Entity"
+
+ 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)) and not k.startswith("_")
+ }
+ props["id"] = node_id
+ cid = node_community.get(node_id)
+ if cid is not None:
+ props["community"] = cid
+ ftype = _safe_label(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)) and not k.startswith("_")
+ }
+ 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 push_to_falkordb(
+ G: nx.Graph,
+ uri: str,
+ user: str | None = None,
+ password: str | None = None,
+ communities: dict[int, list[str]] | None = None,
+ graph_name: str = "graphify",
+) -> dict[str, int]:
+ """Push graph directly to a running FalkorDB instance via the Python SDK.
+
+ Requires: pip install falkordb
+
+ FalkorDB is OpenCypher-compatible, so the MERGE/SET upsert queries are
+ identical to push_to_neo4j. Differences from the Neo4j path:
+ - connects with FalkorDB(host, port, username, password) instead of a bolt
+ driver; only the host/port are read from the URI, so the scheme is
+ informational - "falkordb://localhost:6379", "redis://localhost:6379"
+ and a bare "localhost:6379" are all equivalent (default port 6379).
+ - a named graph is selected via db.select_graph(graph_name) (default
+ "graphify"); FalkorDB keys each graph by name in the same instance.
+ - queries run via graph.query(cypher, params) - there is no session object.
+ - auth is optional (FalkorDB runs without credentials by default), so user
+ and password may be None.
+ - no APOC: the Neo4j path does not use APOC either, so nothing to port.
+
+ 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 falkordb import FalkorDB
+ except ImportError as e:
+ raise ImportError(
+ "falkordb SDK not installed. Run: pip install falkordb"
+ ) from e
+
+ from urllib.parse import urlparse
+
+ node_community = _node_community_map(communities) if communities else {}
+
+ def _safe_rel(relation: str) -> str:
+ return re.sub(r"[^A-Z0-9_]", "_", relation.upper().replace(" ", "_").replace("-", "_")) or "RELATED_TO"
+
+ def _safe_label(label: str) -> str:
+ """Sanitize a FalkorDB node label to prevent Cypher injection."""
+ sanitized = re.sub(r"[^A-Za-z0-9_]", "", label)
+ return sanitized if sanitized else "Entity"
+
+ parsed = urlparse(uri if "://" in uri else f"redis://{uri}")
+ # FalkorDB auth is optional. Only send credentials when a password is
+ # provided; otherwise connect anonymously and ignore any bolt-style default
+ # username (e.g. Neo4j's "neo4j"), which FalkorDB rejects as an unknown ACL
+ # user. Credentials embedded in the URI take precedence over the args.
+ connect_user = parsed.username or (user if password else None)
+ connect_password = parsed.password or (password or None)
+ db = FalkorDB(
+ host=parsed.hostname or "localhost",
+ port=parsed.port or 6379,
+ username=connect_user,
+ password=connect_password,
+ )
+ graph = db.select_graph(graph_name)
+ nodes_pushed = 0
+ edges_pushed = 0
+
+ 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)) and not k.startswith("_")
+ }
+ props["id"] = node_id
+ cid = node_community.get(node_id)
+ if cid is not None:
+ props["community"] = cid
+ ftype = _safe_label(data.get("file_type", "Entity").capitalize())
+ graph.query(
+ 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)) and not k.startswith("_")
+ }
+ graph.query(
+ 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
+
+ return {"nodes": nodes_pushed, "edges": edges_pushed}
diff --git a/graphify/exporters/html.py b/graphify/exporters/html.py
new file mode 100644
index 0000000..5ba61ce
--- /dev/null
+++ b/graphify/exporters/html.py
@@ -0,0 +1,547 @@
+"""html — moved verbatim from graphify/export.py."""
+from __future__ import annotations
+
+from graphify.exporters.base import COMMUNITY_COLORS # noqa: E402,F401
+from pathlib import Path
+import html as _html
+from graphify.analyze import _node_community_map
+import json
+import networkx as nx
+from graphify.security import sanitize_label
+
+
+MAX_NODES_FOR_VIZ = 5_000
+
+def _viz_node_limit() -> int:
+ """Return the effective viz node limit, honoring GRAPHIFY_VIZ_NODE_LIMIT env var.
+
+ Falls back to MAX_NODES_FOR_VIZ when the env var is unset, empty, or non-integer.
+ Set to 0 to disable HTML viz unconditionally (useful for CI runners).
+ """
+ import os
+ raw = os.environ.get("GRAPHIFY_VIZ_NODE_LIMIT")
+ if raw is None or not raw.strip():
+ return MAX_NODES_FOR_VIZ
+ try:
+ return int(raw)
+ except ValueError:
+ return MAX_NODES_FOR_VIZ
+
+def _html_styles() -> str:
+ return """"""
+
+def _hyperedge_script(hyperedges_json: str) -> str:
+ return f""""""
+
+def _html_script(nodes_json: str, edges_json: str, legend_json: str) -> str:
+ return f""""""
+
+def to_html(
+ G: nx.Graph,
+ communities: dict[int, list[str]],
+ output_path: str,
+ community_labels: dict[int, str] | None = None,
+ member_counts: dict[int, int] | None = None,
+ node_limit: int | None = None,
+ learning_overlay: dict | None = None,
+) -> None:
+ """Generate an interactive vis.js HTML visualization of the graph.
+
+ Features: node size by degree, click-to-inspect panel, search box,
+ community filter, physics clustering by community, confidence-styled edges.
+ Raises ValueError if graph exceeds MAX_NODES_FOR_VIZ.
+
+ If member_counts is provided (aggregated community view), node sizes are
+ based on community member counts rather than graph degree.
+
+ If node_limit is set and the graph exceeds it, automatically builds an
+ aggregated community-level meta-graph instead of raising ValueError.
+ """
+ limit = node_limit if node_limit is not None else _viz_node_limit()
+ if G.number_of_nodes() > limit:
+ if node_limit is not None:
+ # Build aggregated community meta-graph
+ from collections import Counter as _Counter
+ import networkx as _nx
+ print(f"Graph has {G.number_of_nodes()} nodes (above {limit} limit). Building aggregated community view...")
+ node_to_community = {nid: cid for cid, members in communities.items() for nid in members}
+ meta = _nx.Graph()
+ for cid, members in communities.items():
+ meta.add_node(str(cid), label=(community_labels or {}).get(cid, f"Community {cid}"))
+ edge_counts = _Counter()
+ for u, v in G.edges():
+ cu, cv = node_to_community.get(u), node_to_community.get(v)
+ if cu is not None and cv is not None and cu != cv:
+ edge_counts[(min(cu, cv), max(cu, cv))] += 1
+ for (cu, cv), w in edge_counts.items():
+ meta.add_edge(str(cu), str(cv), weight=w,
+ relation=f"{w} cross-community edges", confidence="AGGREGATED")
+ if meta.number_of_nodes() <= 1:
+ print("Single community - aggregated view not useful. Skipping graph.html.")
+ return
+ meta_communities = {cid: [str(cid)] for cid in communities}
+ mc = {cid: len(members) for cid, members in communities.items()}
+ # Remap hyperedges from semantic node IDs to community IDs
+ raw_hyperedges = G.graph.get("hyperedges", [])
+ if raw_hyperedges:
+ remapped = []
+ for he in raw_hyperedges:
+ he_members = he.get("nodes", [])
+ comm_ids, seen = [], set()
+ for nid in he_members:
+ c = node_to_community.get(nid)
+ if c is None:
+ continue
+ s = str(c)
+ if s in seen:
+ continue
+ seen.add(s)
+ comm_ids.append(s)
+ if len(comm_ids) < 2:
+ continue
+ remapped.append({
+ "id": he.get("id", ""),
+ "label": he.get("label") or he.get("relation", "").replace("_", " "),
+ "nodes": comm_ids,
+ })
+ meta.graph["hyperedges"] = remapped
+ to_html(meta, meta_communities, output_path,
+ community_labels=community_labels, member_counts=mc)
+ print(f"graph.html written (aggregated: {meta.number_of_nodes()} community nodes, {meta.number_of_edges()} cross-community edges)")
+ print("Tip: run with --obsidian for full node-level detail.")
+ return
+ raise ValueError(
+ f"Graph has {G.number_of_nodes()} nodes - too large for HTML viz "
+ f"(limit: {limit}). Use --no-viz, raise GRAPHIFY_VIZ_NODE_LIMIT, "
+ f"or reduce input size."
+ )
+
+ node_community = _node_community_map(communities)
+ degree = dict(G.degree())
+ max_deg = max(degree.values(), default=1) or 1
+ max_mc = (max(member_counts.values(), default=1) or 1) if member_counts else 1
+
+ # Work-memory overlay (derived sidecar). When not passed explicitly, load it
+ # best-effort from the sibling .graphify_learning.json next to the output
+ # graph.html (which lives beside graph.json). Empty/missing => no learning
+ # fields, so the un-annotated render is byte-identical to pre-feature.
+ if learning_overlay is None:
+ learning_overlay = {}
+ try:
+ from graphify.reflect import load_learning_overlay as _llo
+ learning_overlay = _llo(Path(output_path))
+ except Exception:
+ learning_overlay = {}
+ # Status -> ring color. preferred=green, contested=amber. Tentative gets no
+ # ring (it's not yet trustworthy enough to highlight in the map).
+ _RING = {"preferred": "#22c55e", "contested": "#f59e0b"}
+
+ # Build nodes list for vis.js
+ vis_nodes = []
+ for node_id, data in G.nodes(data=True):
+ cid = node_community.get(node_id, 0)
+ color = COMMUNITY_COLORS[cid % len(COMMUNITY_COLORS)]
+ label = sanitize_label(data.get("label", node_id))
+ deg = degree.get(node_id, 1)
+ if member_counts:
+ mc = member_counts.get(cid, 1)
+ size = 10 + 30 * (mc / max_mc)
+ font_size = 12
+ else:
+ size = 10 + 30 * (deg / max_deg)
+ # Only show label for high-degree nodes by default; others show on hover
+ font_size = 12 if deg >= max_deg * 0.15 else 0
+ node = {
+ "id": node_id,
+ "label": label,
+ "color": {"background": color, "border": color, "highlight": {"background": "#ffffff", "border": color}},
+ "size": round(size, 1),
+ "font": {"size": font_size, "color": "#ffffff"},
+ "title": _html.escape(label),
+ "community": cid,
+ "community_name": sanitize_label((community_labels or {}).get(cid, f"Community {cid}")),
+ "source_file": sanitize_label(str(data.get("source_file") or "")),
+ "file_type": data.get("file_type", ""),
+ "degree": deg,
+ }
+ # Conditional learning fields — only present for annotated nodes, so
+ # un-annotated output keeps the exact pre-feature node dict shape.
+ entry = learning_overlay.get(str(node_id)) if learning_overlay else None
+ if entry:
+ status = sanitize_label(str(entry.get("status", "")))
+ stale = bool(entry.get("stale"))
+ node["learning_status"] = status
+ node["learning_stale"] = stale
+ ring = _RING.get(status)
+ if ring:
+ # Status-colored ring via the border; stale => desaturated +
+ # dashed (vis.js supports per-node `shapeProperties.borderDashes`).
+ if stale:
+ ring = "#9ca3af"
+ node["shapeProperties"] = {"borderDashes": [4, 4]}
+ node["borderWidth"] = 3
+ node["color"] = {
+ "background": color, "border": ring,
+ "highlight": {"background": "#ffffff", "border": ring},
+ }
+ # Lesson line appended to the hover title.
+ if status == "contested":
+ lesson = f"Lesson: contested (useful {entry.get('uses', 0)} / dead-end {entry.get('neg', 0)})"
+ elif status == "preferred":
+ lesson = f"Lesson: preferred source ({entry.get('uses', 0)} useful, score={entry.get('score', 0)})"
+ else:
+ lesson = f"Lesson: {status} ({entry.get('uses', 0)} useful)"
+ if stale:
+ lesson += " [code changed — re-verify]"
+ node["title"] = _html.escape(label) + "\n" + _html.escape(sanitize_label(lesson))
+ vis_nodes.append(node)
+
+ # Build edges list. Restore original edge direction from _src/_tgt
+ # (stashed by build.py for exactly this reason): undirected NetworkX
+ # canonicalizes endpoint order, which would otherwise flip the arrow
+ # for `calls` and `rationale_for` in the rendered graph (#563).
+ vis_edges = []
+ for u, v, data in G.edges(data=True):
+ confidence = data.get("confidence", "EXTRACTED")
+ relation = data.get("relation", "")
+ true_src = data.get("_src", u)
+ true_tgt = data.get("_tgt", v)
+ vis_edges.append({
+ "from": true_src,
+ "to": true_tgt,
+ "label": relation,
+ "title": _html.escape(f"{relation} [{confidence}]"),
+ "dashes": confidence != "EXTRACTED",
+ "width": 2 if confidence == "EXTRACTED" else 1,
+ "color": {"opacity": 0.7 if confidence == "EXTRACTED" else 0.35},
+ "confidence": confidence,
+ })
+
+ # Build community legend data
+ legend_data = []
+ for cid in sorted((community_labels or {}).keys()):
+ color = COMMUNITY_COLORS[cid % len(COMMUNITY_COLORS)]
+ lbl = _html.escape(sanitize_label((community_labels or {}).get(cid, f"Community {cid}")))
+ n = member_counts.get(cid, len(communities.get(cid, []))) if member_counts else len(communities.get(cid, []))
+ legend_data.append({"cid": cid, "color": color, "label": lbl, "count": n})
+
+ # Escape sequences so embedded JSON cannot break out of the script tag
+ def _js_safe(obj) -> str:
+ return json.dumps(obj).replace("
+
+
+
+graphify - {title}
+
+{_html_styles()}
+
+
+
+
+{_html_script(nodes_json, edges_json, legend_json)}
+{_hyperedge_script(hyperedges_json)}
+
+"""
+
+ Path(output_path).write_text(html, encoding="utf-8") # nosec