"""Additional export formats: GraphML, Neo4j Cypher, Obsidian vault, SVG.""" from __future__ import annotations import html import logging import re from pathlib import Path from .graph import GraphStore, _sanitize_name from .visualization import export_graph_data logger = logging.getLogger(__name__) # ------------------------------------------------------------------- # GraphML export (for Gephi, yEd, Cytoscape) # ------------------------------------------------------------------- def export_graphml(store: GraphStore, output_path: Path) -> Path: """Export the graph as GraphML XML for Gephi/yEd/Cytoscape. Returns the path to the written file. """ data = export_graph_data(store) nodes = data["nodes"] edges = data["edges"] lines = [ '', '', ' ', ' ', ' ', ' ', ' ', ' ', ] for n in nodes: nid = html.escape(n["qualified_name"], quote=True) lines.append(f' ') lines.append(f' ' f'{html.escape(n.get("kind", ""))}') lines.append(f' ' f'{html.escape(n.get("file_path", ""))}') lang = n.get("language", "") or "" lines.append(f' ' f'{html.escape(lang)}') cid = n.get("community_id") if cid is not None: lines.append(f' ' f'{cid}') lines.append(' ') for i, e in enumerate(edges): src = html.escape(e["source"], quote=True) tgt = html.escape(e["target"], quote=True) kind = html.escape(e.get("kind", ""), quote=True) lines.append( f' ' ) lines.append(f' {kind}') lines.append(' ') lines.append(' ') lines.append('') output_path.write_text("\n".join(lines), encoding="utf-8") logger.info("GraphML exported to %s (%d nodes, %d edges)", output_path, len(nodes), len(edges)) return output_path # ------------------------------------------------------------------- # Neo4j Cypher export # ------------------------------------------------------------------- def export_neo4j_cypher(store: GraphStore, output_path: Path) -> Path: """Export the graph as Neo4j Cypher CREATE statements. Returns the path to the written file. """ data = export_graph_data(store) nodes = data["nodes"] edges = data["edges"] lines = [ "// Generated by code-review-graph", "// Import: paste into Neo4j Browser or run via cypher-shell", "", ] # Create nodes for n in nodes: kind = n.get("kind", "Node") props = { "qualified_name": n["qualified_name"], "name": n.get("name", ""), "file_path": n.get("file_path", ""), "language": n.get("language", "") or "", } cid = n.get("community_id") if cid is not None: props["community_id"] = cid props_str = _cypher_props(props) lines.append(f"CREATE (:{kind} {props_str});") lines.append("") # Create edges via MATCH for e in edges: kind = e.get("kind", "RELATES_TO") src_qn = _cypher_escape(e["source"]) tgt_qn = _cypher_escape(e["target"]) lines.append( f"MATCH (a {{qualified_name: '{src_qn}'}}), " f"(b {{qualified_name: '{tgt_qn}'}}) " f"CREATE (a)-[:{kind}]->(b);" ) output_path.write_text("\n".join(lines), encoding="utf-8") logger.info("Neo4j Cypher exported to %s (%d nodes, %d edges)", output_path, len(nodes), len(edges)) return output_path def _cypher_escape(s: str) -> str: """Escape a string for Cypher single-quoted literals.""" return s.replace("\\", "\\\\").replace("'", "\\'") def _cypher_props(d: dict) -> str: """Format a dict as Cypher property map.""" parts = [] for k, v in d.items(): if isinstance(v, str): parts.append(f"{k}: '{_cypher_escape(v)}'") elif isinstance(v, (int, float)): parts.append(f"{k}: {v}") elif isinstance(v, bool): parts.append(f"{k}: {'true' if v else 'false'}") return "{" + ", ".join(parts) + "}" # ------------------------------------------------------------------- # Obsidian vault export # ------------------------------------------------------------------- def export_obsidian_vault( store: GraphStore, output_dir: Path ) -> Path: """Export the graph as an Obsidian vault with wikilinks. Creates: - One .md per node with YAML frontmatter and [[wikilinks]] - _COMMUNITY_*.md overview notes per community - _INDEX.md with links to all nodes Returns the output directory path. """ data = export_graph_data(store) nodes = data["nodes"] edges = data["edges"] communities = data.get("communities", []) output_dir.mkdir(parents=True, exist_ok=True) # Build adjacency for wikilinks neighbors: dict[str, list[dict]] = {} for e in edges: src = e["source"] tgt = e["target"] kind = e.get("kind", "RELATES_TO") neighbors.setdefault(src, []).append( {"target": tgt, "kind": kind} ) neighbors.setdefault(tgt, []).append( {"target": src, "kind": kind} ) # Node name -> slug mapping slugs: dict[str, str] = {} for n in nodes: slug = _obsidian_slug(n.get("name", n["qualified_name"])) # Handle collisions base_slug = slug counter = 1 while slug in slugs.values(): slug = f"{base_slug}-{counter}" counter += 1 slugs[n["qualified_name"]] = slug # Write node pages for n in nodes: qn = n["qualified_name"] slug = slugs[qn] name = n.get("name", qn) frontmatter = { "kind": n.get("kind", ""), "file": n.get("file_path", ""), "language": n.get("language", "") or "", "community": n.get("community_id"), "tags": [n.get("kind", "").lower()], } lines = ["---"] for k, v in frontmatter.items(): if isinstance(v, list): lines.append(f"{k}:") for item in v: lines.append(f" - {item}") elif v is not None: lines.append(f"{k}: {v}") lines.append("---") lines.append(f"# {_sanitize_name(name)}") lines.append("") lines.append(f"**Kind:** {n.get('kind', '')}") lines.append(f"**File:** `{n.get('file_path', '')}`") lines.append("") # Wikilinks to neighbors nbrs = neighbors.get(qn, []) if nbrs: lines.append("## Connections") lines.append("") seen = set() for nb in nbrs: tgt_slug = slugs.get(nb["target"]) if tgt_slug and tgt_slug not in seen: seen.add(tgt_slug) tgt_name = tgt_slug.replace("-", " ").title() lines.append( f"- {nb['kind']}: " f"[[{tgt_slug}|{tgt_name}]]" ) page_path = output_dir / f"{slug}.md" page_path.write_text("\n".join(lines), encoding="utf-8") # Write community overview pages community_map: dict[int, list[str]] = {} for n in nodes: cid = n.get("community_id") if cid is not None: community_map.setdefault(cid, []).append( n["qualified_name"] ) for c in communities: cid = c.get("id") cname = c.get("name", f"community-{cid}") members = community_map.get(cid, []) lines = [f"# Community: {_sanitize_name(cname)}", ""] lines.append(f"**Size:** {c.get('size', len(members))}") lines.append(f"**Cohesion:** {c.get('cohesion', 0):.2f}") lang = c.get("dominant_language", "") if lang: lines.append(f"**Language:** {lang}") lines.append("") lines.append("## Members") lines.append("") for qn in members[:50]: slug = slugs.get(qn) if slug: lines.append(f"- [[{slug}]]") page_path = output_dir / f"_COMMUNITY_{cid}.md" page_path.write_text("\n".join(lines), encoding="utf-8") # Write index index_lines = ["# Code Graph Index", ""] index_lines.append(f"**Nodes:** {len(nodes)}") index_lines.append(f"**Edges:** {len(edges)}") index_lines.append( f"**Communities:** {len(communities)}" ) index_lines.append("") index_lines.append("## All Nodes") index_lines.append("") for n in sorted(nodes, key=lambda x: x.get("name", "")): slug = slugs.get(n["qualified_name"]) if slug: index_lines.append( f"- [[{slug}]] ({n.get('kind', '')})" ) (output_dir / "_INDEX.md").write_text( "\n".join(index_lines), encoding="utf-8" ) logger.info( "Obsidian vault exported to %s (%d pages)", output_dir, len(nodes) ) return output_dir def _obsidian_slug(name: str) -> str: """Convert a name to an Obsidian-friendly filename slug.""" slug = re.sub(r"[^\w\s-]", "", name.lower()) slug = re.sub(r"[\s_]+", "-", slug).strip("-") return slug[:100] or "unnamed" # ------------------------------------------------------------------- # SVG export (matplotlib-based) # ------------------------------------------------------------------- def export_svg(store: GraphStore, output_path: Path) -> Path: """Export a static SVG graph visualization. Requires matplotlib (optional dependency). Returns the path to the written file. """ try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt except ImportError: raise ImportError( "matplotlib is required for SVG export. " "Install with: pip install matplotlib" ) import networkx as nx data = export_graph_data(store) nodes_data = data["nodes"] edges_data = data["edges"] nxg: nx.DiGraph = nx.DiGraph() # type: ignore[type-arg] for n in nodes_data: nxg.add_node( n["qualified_name"], label=n.get("name", ""), kind=n.get("kind", ""), ) for e in edges_data: if e["source"] in nxg and e["target"] in nxg: nxg.add_edge(e["source"], e["target"]) if nxg.number_of_nodes() == 0: raise ValueError("Graph is empty, nothing to export") # Color by kind kind_colors = { "File": "#6c757d", "Class": "#0d6efd", "Function": "#198754", "Type": "#ffc107", "Test": "#dc3545", } colors = [ kind_colors.get( nxg.nodes[n].get("kind", ""), "#adb5bd" ) for n in nxg.nodes() ] fig, ax = plt.subplots(1, 1, figsize=(16, 12)) pos = nx.spring_layout( nxg, k=2 / (nxg.number_of_nodes() ** 0.5), iterations=50, seed=42 ) # Limit labels to avoid clutter labels = {} if nxg.number_of_nodes() <= 100: labels = { n: nxg.nodes[n].get("label", n.split("::")[-1]) for n in nxg.nodes() } nx.draw_networkx_nodes( nxg, pos, ax=ax, node_color=colors, node_size=30, alpha=0.8 ) nx.draw_networkx_edges( nxg, pos, ax=ax, alpha=0.2, arrows=True, arrowsize=5 ) if labels: nx.draw_networkx_labels( nxg, pos, labels=labels, ax=ax, font_size=6 ) ax.set_title("Code Review Graph", fontsize=14) ax.axis("off") fig.savefig( str(output_path), format="svg", bbox_inches="tight", dpi=150 ) plt.close(fig) logger.info("SVG exported to %s (%d nodes)", output_path, nxg.number_of_nodes()) return output_path