From db66b8727bc39467769d01122f1712829079242b Mon Sep 17 00:00:00 2001 From: Porun <63705582+PorunC@users.noreply.github.com> Date: Sun, 10 May 2026 06:26:01 +0800 Subject: [PATCH] feat: add callflow HTML export with Mermaid architecture diagrams --- ARCHITECTURE.md | 1 + README.md | 12 + graphify/__main__.py | 94 +- graphify/callflow_html.py | 2014 +++++++++++++++++++++++++++++++++++ tests/test_callflow_html.py | 170 +++ 5 files changed, 2288 insertions(+), 3 deletions(-) create mode 100644 graphify/callflow_html.py create mode 100644 tests/test_callflow_html.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 51ad580f1..5672bf0df 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -21,6 +21,7 @@ Each stage is a single function in its own module. They communicate through plai | `analyze.py` | `analyze(G)` | graph → analysis dict (god nodes, surprises, questions) | | `report.py` | `render_report(G, analysis)` | graph + analysis → GRAPH_REPORT.md string | | `export.py` | `export(G, out_dir, ...)` | graph → Obsidian vault, graph.json, graph.html, graph.svg | +| `callflow_html.py` | `write_callflow_html(...)` | graphify-out files → Mermaid architecture/call-flow HTML | | `ingest.py` | `ingest(url, ...)` | URL → file saved to corpus dir | | `cache.py` | `check_semantic_cache / save_semantic_cache` | files → (cached, uncached) split | | `security.py` | validation helpers | URL / path / label → validated or raises | diff --git a/README.md b/README.md index e48d042c9..037e7b4e6 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,12 @@ graphify-out/ └── graph.json the full graph — query it anytime without re-reading your files ``` +For a readable architecture page with Mermaid call-flow diagrams, run: + +```bash +graphify export callflow-html +``` + --- ## Install @@ -164,6 +170,7 @@ You can also set `GRAPHIFY_GOOGLE_WORKSPACE=1`. Graphify exports shortcuts into /graphify . --cluster-only # rerun clustering without re-extracting /graphify . --no-viz # skip the HTML, just the report + JSON /graphify . --wiki # build a markdown wiki from the graph +graphify export callflow-html # architecture/call-flow HTML from graphify-out/ /graphify query "what connects auth to the database?" /graphify path "UserService" "DatabasePool" @@ -314,6 +321,11 @@ graphify extract ./docs --dedup-llm # LLM tiebreaker for ambiguous en graphify extract ./docs --global --as myrepo # extract and register into the cross-project global graph GRAPHIFY_MAX_OUTPUT_TOKENS=32768 graphify extract ./docs --backend claude # raise output cap for dense corpora +graphify export callflow-html # graphify-out/-callflow.html +graphify export callflow-html --max-sections 8 # cap generated architecture sections +graphify export callflow-html --output docs/arch.html +graphify export callflow-html ./some-repo/graphify-out + graphify global add graphify-out/graph.json myrepo # register a project graph into ~/.graphify/global.json graphify global remove myrepo # remove a project from the global graph graphify global list # show all registered repos + node/edge counts diff --git a/graphify/__main__.py b/graphify/__main__.py index 500511b0e..8f79c5ae4 100644 --- a/graphify/__main__.py +++ b/graphify/__main__.py @@ -1185,6 +1185,7 @@ def main() -> None: print(" global list list repos in the global graph") print(" global path print path to the global graph file") print(" benchmark [graph.json] measure token reduction vs naive full-corpus approach") + print(" export callflow-html emit Mermaid-based architecture/call-flow HTML") print(" hook install install post-commit/post-checkout git hooks (all platforms)") print(" hook uninstall remove git hooks") print(" hook status check if git hooks are installed") @@ -1923,9 +1924,11 @@ def main() -> None: elif cmd == "export": subcmd = sys.argv[2] if len(sys.argv) > 2 else "" - if subcmd not in ("html", "obsidian", "wiki", "svg", "graphml", "neo4j"): + if subcmd not in ("html", "callflow-html", "obsidian", "wiki", "svg", "graphml", "neo4j"): print("Usage: graphify export ", file=sys.stderr) print(" html [--graph PATH] [--labels PATH] [--node-limit N] [--no-viz]", file=sys.stderr) + print(" callflow-html [GRAPH|DIR] [--graph PATH] [--labels PATH] [--report PATH] [--sections PATH] [--output HTML]", file=sys.stderr) + print(" [--lang auto|zh-CN|en] [--max-sections N] [--diagram-scale N]", file=sys.stderr) print(" obsidian [--graph PATH] [--labels PATH] [--dir PATH]", file=sys.stderr) print(" wiki [--graph PATH] [--labels PATH]", file=sys.stderr) print(" svg [--graph PATH] [--labels PATH]", file=sys.stderr) @@ -1937,7 +1940,18 @@ def main() -> None: # Parse shared args args = sys.argv[3:] graph_path = Path(_GRAPHIFY_OUT) / "graph.json" + graph_path_explicit = False labels_path = Path(_GRAPHIFY_OUT) / ".graphify_labels.json" + labels_path_explicit = False + report_path = Path(_GRAPHIFY_OUT) / "GRAPH_REPORT.md" + report_path_explicit = False + sections_path: Path | None = None + callflow_output: Path | None = None + callflow_lang = "auto" + callflow_max_sections = 15 + callflow_diagram_scale = 1.0 + callflow_max_diagram_nodes = 18 + callflow_max_diagram_edges = 24 analysis_path = Path(_GRAPHIFY_OUT) / ".graphify_analysis.json" node_limit = 5000 no_viz = False @@ -1952,9 +1966,45 @@ def main() -> None: while i < len(args): a = args[i] if a == "--graph" and i + 1 < len(args): - graph_path = Path(args[i + 1]); i += 2 + graph_path = Path(args[i + 1]) + graph_path_explicit = True + i += 2 elif a == "--labels" and i + 1 < len(args): - labels_path = Path(args[i + 1]); i += 2 + labels_path = Path(args[i + 1]) + labels_path_explicit = True + i += 2 + elif a == "--report" and i + 1 < len(args): + report_path = Path(args[i + 1]) + report_path_explicit = True + i += 2 + elif a == "--sections" and i + 1 < len(args): + sections_path = Path(args[i + 1]); i += 2 + elif a == "--output" and i + 1 < len(args): + callflow_output = Path(args[i + 1]).expanduser() + if not callflow_output.is_absolute(): + callflow_output = Path.cwd() / callflow_output + i += 2 + elif a == "--lang" and i + 1 < len(args): + callflow_lang = args[i + 1]; i += 2 + elif a == "--max-sections" and i + 1 < len(args): + callflow_max_sections = int(args[i + 1]); i += 2 + elif a == "--diagram-scale" and i + 1 < len(args): + callflow_diagram_scale = float(args[i + 1]); i += 2 + elif a == "--max-diagram-nodes" and i + 1 < len(args): + callflow_max_diagram_nodes = int(args[i + 1]); i += 2 + elif a == "--max-diagram-edges" and i + 1 < len(args): + callflow_max_diagram_edges = int(args[i + 1]); i += 2 + elif a in ("-h", "--help") and subcmd == "callflow-html": + print("Usage: graphify export callflow-html [GRAPH|DIR] [--graph PATH] [--labels PATH]") + print(" --report PATH path to GRAPH_REPORT.md") + print(" --sections PATH JSON section definitions") + print(" --output HTML output path (default graphify-out/-callflow.html)") + print(" --lang LANG auto, zh-CN, en, etc. (default auto)") + print(" --max-sections N maximum auto-derived sections (default 15)") + print(" --diagram-scale N Mermaid diagram scale (default 1.0)") + print(" --max-diagram-nodes N representative nodes per section (default 18)") + print(" --max-diagram-edges N representative edges per section (default 24)") + sys.exit(0) elif a == "--node-limit" and i + 1 < len(args): node_limit = int(args[i + 1]); i += 2 elif a == "--no-viz": @@ -1967,13 +2017,51 @@ def main() -> None: neo4j_user = args[i + 1]; i += 2 elif a == "--password" and i + 1 < len(args): neo4j_password = args[i + 1]; i += 2 + elif subcmd == "callflow-html" and not a.startswith("-") and not graph_path_explicit: + candidate = Path(a) + if candidate.name == "graph.json" or candidate.suffix.lower() == ".json": + graph_path = candidate + elif (candidate / "graph.json").exists(): + graph_path = candidate / "graph.json" + else: + graph_path = candidate / _GRAPHIFY_OUT / "graph.json" + graph_path_explicit = True + i += 1 else: i += 1 + graph_path = graph_path.expanduser() + if graph_path_explicit: + graph_out_dir = graph_path.parent + if not labels_path_explicit: + labels_path = graph_out_dir / ".graphify_labels.json" + if not report_path_explicit: + report_path = graph_out_dir / "GRAPH_REPORT.md" + labels_path = labels_path.expanduser() + report_path = report_path.expanduser() + if not graph_path.exists(): print(f"error: graph not found: {graph_path}. Run /graphify first.", file=sys.stderr) sys.exit(1) + if subcmd == "callflow-html": + from graphify.callflow_html import write_callflow_html as _write_callflow_html + out = _write_callflow_html( + graph=graph_path, + report=report_path, + labels=labels_path, + sections=sections_path, + output=callflow_output, + lang=callflow_lang, + max_sections=callflow_max_sections, + diagram_scale=callflow_diagram_scale, + max_diagram_nodes=callflow_max_diagram_nodes, + max_diagram_edges=callflow_max_diagram_edges, + verbose=True, + ) + print(f"callflow HTML written - open in any browser: {out}") + sys.exit(0) + from networkx.readwrite import json_graph as _jg from graphify.build import build_from_json as _bfj diff --git a/graphify/callflow_html.py b/graphify/callflow_html.py new file mode 100644 index 000000000..96d4f1d3b --- /dev/null +++ b/graphify/callflow_html.py @@ -0,0 +1,2014 @@ +#!/usr/bin/env python3 +""" +callflow_html.py — Generate call-flow architecture HTML from graphify knowledge graph outputs. + +Reads graph.json plus optional GRAPH_REPORT.md, .graphify_labels.json, and sections JSON, +then produces a self-contained HTML file with: + - Dark-themed CSS (fixed template) + - Navigation bar from section list + - Architecture overview flowchart LR (aggregated section-level edges) + - Per-section flowchart LR (auto-generated representative intra-section edges) + - Call detail table scaffolding (headers + representative node rows) + - Auto-generated section intros and key-file cards + +Usage: + python3 -m graphify export callflow-html + python3 -m graphify export callflow-html /path/to/project/graphify-out/graph.json + python3 -m graphify export callflow-html --graph /path/to/graph.json --output docs/architecture.html +""" + +from __future__ import annotations + +import json +import argparse +import os +import re +import sys +import hashlib +from pathlib import Path +from collections import Counter, defaultdict +from datetime import datetime, timezone +from html import escape + + +# ────────────────────────────────────────────── +# 1. CSS template (fixed, project-agnostic) +# ────────────────────────────────────────────── + +CSS = """:root { + --bg: #0f172a; --surface: #1e293b; --border: #334155; + --text: #e2e8f0; --muted: #94a3b8; --accent: #38bdf8; + --warn: #fbbf24; --err: #f87171; --ok: #34d399; +} +* { box-sizing: border-box; margin: 0; padding: 0; } +body { font-family: 'Segoe UI', system-ui, -apple-system, sans-serif; background: var(--bg); color: var(--text); line-height: 1.7; } +.container { max-width: 1200px; margin: 0 auto; padding: 40px 24px; } +h1 { font-size: 2.4rem; margin-bottom: 8px; background: linear-gradient(135deg, var(--accent), #a78bfa); -webkit-background-clip: text; -webkit-text-fill-color: transparent; } +h2 { font-size: 1.7rem; margin: 48px 0 16px; padding-bottom: 8px; border-bottom: 2px solid var(--accent); } +h3 { font-size: 1.25rem; margin: 32px 0 12px; color: var(--accent); } +h4 { font-size: 1.05rem; margin: 20px 0 8px; color: var(--warn); } +p { margin: 8px 0; color: var(--muted); } +.subtitle { color: var(--muted); font-size: 1.1rem; margin-bottom: 32px; } +.mermaid { background: var(--surface); border: 1px solid var(--border); border-radius: 12px; padding: 24px; margin: 20px 0; overflow-x: auto; position: relative; } +.mermaid.is-enhanced { padding: 0; overflow: hidden; min-height: 260px; } +.mermaid-viewport { padding: 54px 24px 24px; overflow: hidden; cursor: grab; touch-action: none; min-height: 260px; } +.mermaid-viewport.is-dragging { cursor: grabbing; } +.mermaid-viewport svg { max-width: none !important; height: auto; transform-origin: 0 0; transition: transform 120ms ease; } +.mermaid-toolbar { position: absolute; top: 10px; right: 10px; z-index: 3; display: flex; align-items: center; gap: 6px; padding: 6px; background: rgba(15,23,42,0.92); border: 1px solid var(--border); border-radius: 8px; box-shadow: 0 8px 24px rgba(0,0,0,0.28); } +.mermaid-toolbar button, .mermaid-toolbar .zoom-level { height: 28px; min-width: 32px; border: 1px solid var(--border); border-radius: 6px; background: #1e293b; color: var(--text); font: 600 0.78rem system-ui, sans-serif; display: inline-flex; align-items: center; justify-content: center; } +.mermaid-toolbar button { cursor: pointer; } +.mermaid-toolbar button:hover { border-color: var(--accent); color: var(--accent); } +.mermaid-toolbar .zoom-level { min-width: 52px; color: var(--muted); background: transparent; } +.call-table { width: 100%; border-collapse: collapse; margin: 16px 0; font-size: 0.92rem; } +.call-table th { background: #1a2744; color: var(--accent); text-align: left; padding: 10px 14px; border: 1px solid var(--border); } +.call-table td { padding: 8px 14px; border: 1px solid var(--border); vertical-align: top; } +.call-table tr:nth-child(even) { background: rgba(255,255,255,0.02); } +.tag { display: inline-block; padding: 2px 8px; border-radius: 4px; font-size: 0.8rem; font-weight: 600; } +.tag-async { background: #7c3aed33; color: #a78bfa; } +.tag-class { background: #05966933; color: var(--ok); } +.tag-func { background: #2563eb33; color: var(--accent); } +.tag-cmd { background: #d9770633; color: var(--warn); } +.tag-endpoint { background: #dc262633; color: var(--err); } +.tag-hook { background: #db277733; color: #f472b6; } +.card { background: var(--surface); border: 1px solid var(--border); border-radius: 10px; padding: 20px; margin: 16px 0; } +.grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(340px, 1fr)); gap: 16px; margin: 16px 0; } +.arrow-chain { font-family: 'Fira Code', monospace; font-size: 0.85rem; color: var(--accent); padding: 10px; background: rgba(56,189,248,0.06); border-radius: 6px; } +code { font-family: 'Fira Code', 'Cascadia Code', monospace; background: rgba(255,255,255,0.06); padding: 1px 6px; border-radius: 3px; font-size: 0.88em; } +ul, ol { margin: 8px 0 8px 24px; color: var(--muted); } +li { margin: 4px 0; } +a { color: var(--accent); } +hr { border: none; border-top: 1px solid var(--border); margin: 40px 0; } +.nav { position: sticky; top: 0; background: var(--bg); z-index: 10; padding: 12px 0; border-bottom: 1px solid var(--border); display: flex; gap: 20px; flex-wrap: wrap; font-size: 0.9rem; } +.nav a { text-decoration: none; } +.nav a:hover { text-decoration: underline; } +@media (max-width: 768px) { .container { padding: 16px; } h1 { font-size: 1.8rem; } } +""" + + +# ────────────────────────────────────────────── +# 2. Data loading and normalization helpers +# ────────────────────────────────────────────── + +def read_json(path: str | Path, default=None): + """Read JSON with a useful error message.""" + if not path: + return default + path = Path(path) + if not path.exists(): + return default + try: + return json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise SystemExit(f"ERROR: invalid JSON in {path}: {exc}") from exc + + +def first_present(mapping: dict, *keys, default=None): + """Return the first non-empty value for any candidate key.""" + for key in keys: + if key in mapping and mapping[key] not in (None, ""): + return mapping[key] + return default + + +def first_list(*values) -> list: + """Return the first list from a set of possible schema locations.""" + for value in values: + if isinstance(value, list): + return value + return [] + + +def to_float(value, default: float = 0.0) -> float: + """Convert graph numeric fields that may be serialized as strings.""" + try: + return float(value) + except (TypeError, ValueError): + return default + + +def endpoint_id(value) -> str: + """Normalize edge endpoints that may be strings or node-like objects.""" + if isinstance(value, dict): + value = first_present(value, "id", "node_id", "key", "name", "qualified_name") + return str(value or "") + + +def normalize_node(raw: dict, index: int) -> dict: + """Normalize a graphify node across common graph.json schema variants.""" + node = dict(raw) + node_id = first_present( + node, + "id", + "node_id", + "key", + "uid", + "name", + "qualified_name", + "fqname", + "symbol", + default=f"node_{index + 1}", + ) + source_file = first_present( + node, + "source_file", + "file", + "file_path", + "filepath", + "path", + "module_path", + "defined_in", + default="", + ) + label = first_present( + node, + "label", + "display_name", + "title", + "name", + "qualified_name", + "fqname", + "symbol", + default=node_id, + ) + community = first_present( + node, + "community", + "community_id", + "cluster", + "cluster_id", + "group", + "group_id", + "modularity_class", + default="unknown", + ) + node_type = first_present(node, "node_type", "kind", "type", "category", default="") + file_type = first_present(node, "file_type", "content_type", "artifact_type", default="") + if not file_type: + suffix = Path(str(source_file)).suffix.lower() + file_type = "document" if suffix in {".md", ".mdx", ".rst", ".txt"} else "code" + + node["id"] = str(node_id) + node["label"] = str(label) + node["community"] = community + node["source_file"] = str(source_file or "") + node["node_type"] = str(node_type or "") + node["file_type"] = str(file_type or "code") + return node + + +def normalize_edge(raw: dict, index: int) -> dict | None: + """Normalize graphify edges while preserving original fields.""" + edge = dict(raw) + source = endpoint_id(first_present(edge, "source", "src", "from", "from_id", "start", "u")) + target = endpoint_id(first_present(edge, "target", "dst", "to", "to_id", "end", "v")) + if not source or not target: + return None + + relation = first_present(edge, "relation", "type", "kind", "label", "predicate", default="relates") + confidence = first_present(edge, "confidence", "evidence", "provenance", default="EXTRACTED") + score = first_present(edge, "confidence_score", "score", "weight", "probability", default=1.0) + + edge["id"] = str(first_present(edge, "id", "edge_id", default=f"edge_{index + 1}")) + edge["source"] = source + edge["target"] = target + edge["relation"] = str(relation or "relates").lower() + edge["confidence"] = str(confidence or "EXTRACTED").upper() + edge["confidence_score"] = to_float(score, 1.0) + return edge + + +def _node_link_payload(data: dict) -> tuple[list, list] | None: + """Read current graphify graph.json via NetworkX's node-link parser.""" + if not isinstance(data.get("nodes"), list): + return None + if not isinstance(data.get("links"), list) and not isinstance(data.get("edges"), list): + return None + + try: + from networkx.readwrite import json_graph + + try: + graph = json_graph.node_link_graph(data, edges="links") + except TypeError: + graph = json_graph.node_link_graph(data) + except Exception: + return None + + nodes = [] + for node_id, attrs in graph.nodes(data=True): + node = dict(attrs) + node["id"] = node_id + nodes.append(node) + + edges = [] + for index, (source, target, attrs) in enumerate(graph.edges(data=True), 1): + edge = dict(attrs) + edge["source"] = edge.get("_src", edge.get("source", source)) + edge["target"] = edge.get("_tgt", edge.get("target", target)) + edge.setdefault("id", f"edge_{index}") + edges.append(edge) + return nodes, edges + + +def load_graph(path: str | Path) -> tuple: + """Load graph.json. Returns normalized (nodes, edges, hyperedges, metadata).""" + data = read_json(path) + if not isinstance(data, dict): + raise SystemExit(f"ERROR: graph file must contain a JSON object: {path}") + + graph_block = data.get("graph") if isinstance(data.get("graph"), dict) else {} + meta_block = data.get("metadata") if isinstance(data.get("metadata"), dict) else {} + + node_link = _node_link_payload(data) + if node_link: + raw_nodes, raw_edges = node_link + else: + raw_nodes = first_list(data.get("nodes"), data.get("vertices"), graph_block.get("nodes"), graph_block.get("vertices")) + raw_edges = first_list(data.get("links"), data.get("edges"), graph_block.get("links"), graph_block.get("edges")) + hyperedges = first_list(data.get("hyperedges"), graph_block.get("hyperedges"), data.get("groups"), graph_block.get("groups")) + + nodes = [normalize_node(n, i) for i, n in enumerate(raw_nodes) if isinstance(n, dict)] + edges = [] + for i, raw_edge in enumerate(raw_edges): + if not isinstance(raw_edge, dict): + continue + edge = normalize_edge(raw_edge, i) + if edge: + edges.append(edge) + + meta = dict(graph_block) + meta.update(meta_block) + for key in ("built_at_commit", "commit", "project_name", "repo", "repository", "language_breakdown"): + if data.get(key) and not meta.get(key): + meta[key] = data.get(key) + if meta.get("commit") and not meta.get("built_at_commit"): + meta["built_at_commit"] = meta["commit"] + + return nodes, edges, hyperedges, meta + + +def load_labels(path: str | Path | None) -> dict: + """Load community labels from .graphify_labels.json, tolerating wrapper keys.""" + data = read_json(path, default={}) + if not isinstance(data, dict): + return {} + if isinstance(data.get("labels"), dict): + data = data["labels"] + if isinstance(data.get("communities"), dict): + data = data["communities"] + labels = {} + for key, value in data.items(): + if isinstance(value, dict): + value = first_present(value, "label", "name", "title", default=key) + labels[str(key)] = str(value) + return labels + + +def load_sections(path: str | Path | None) -> list: + """Load section definitions from JSON file.""" + data = read_json(path, default=[]) + if isinstance(data, dict) and isinstance(data.get("sections"), list): + data = data["sections"] + if not isinstance(data, list): + raise SystemExit(f"ERROR: sections file must contain a JSON array: {path}") + return data + + +def load_report(path: str | Path | None) -> str: + """Load GRAPH_REPORT.md if it exists.""" + if path and os.path.exists(path): + return Path(path).read_text(encoding="utf-8") + return "" + + +# ────────────────────────────────────────────── +# 3. Mermaid-safe label helpers +# ────────────────────────────────────────────── + +def safe_mermaid_text(text: str) -> str: + """Sanitize text for use inside a Mermaid node label. + + Replaces characters that Mermaid interprets as syntax: + - -> (edge arrow) -> text + - # (comment) -> removed + - {} (shape syntax) -> removed + - backticks -> removed + - " -> ' + - HTML metacharacters -> entities + """ + text = str(text or "") + text = text.replace('"', "'") + text = text.replace('`', '') + text = text.replace('#', '') + text = text.replace('|', ' ') + text = text.replace('{', '').replace('}', '') + text = text.replace("->>", " to ").replace("-->", " to ").replace("->", " to ") + text = " ".join(text.split()) + return escape(text, quote=False) + + +def html_comment_text(text: str) -> str: + """Keep generated HTML comments well-formed.""" + return str(text or "").replace("--", "- -").replace("\n", " ") + + +def stable_ascii_id(raw: str, prefix: str = "node", limit: int = 48) -> str: + """Build a Mermaid-safe ASCII identifier with a hash suffix to avoid collisions.""" + raw = str(raw or "") + digest = hashlib.sha1(raw.encode("utf-8")).hexdigest()[:8] + slug = re.sub(r"[^A-Za-z0-9_]+", "_", raw) + slug = re.sub(r"_+", "_", slug).strip("_") + if not slug: + slug = prefix + if slug[0].isdigit(): + slug = f"{prefix}_{slug}" + return f"{slug[:limit].rstrip('_')}_{digest}" + + +def node_mermaid_id(node: dict) -> str: + """Generate a safe Mermaid node ID from a graph node. + + Mermaid IDs must match [a-zA-Z][a-zA-Z0-9_]* — no dots, hyphens, slashes. + """ + return stable_ascii_id(node.get("id", "unknown"), "node") + + +def mermaid_section_id(section_id: str) -> str: + """Convert a section ID (like 'cli-entry') to a safe Mermaid ID (like 'CLI_ENTRY').""" + return stable_ascii_id(section_id, "section").upper() + + +def safe_file_path(path: str) -> str: + """Return a short, safe display path.""" + # Truncate long paths for display + parts = path.split("/") + if len(parts) > 3: + return "/".join(parts[-3:]) + return path + + +def safe_filename(text: str, fallback: str = "project") -> str: + """Create a conservative filename stem from a project name.""" + stem = re.sub(r"[^A-Za-z0-9._-]+", "-", str(text or "")).strip("-._") + return stem or fallback + + +def infer_project_name(graph_path: str, meta: dict) -> str: + """Infer a display project name when graph metadata does not include one.""" + if meta.get("project_name"): + return meta["project_name"] + path = Path(graph_path).resolve() + if path.parent.name == "graphify-out" and len(path.parents) > 1: + return path.parents[1].name + return path.parent.name or "Project" + + +def resolve_graphify_paths(args) -> dict: + """Resolve project root, graphify output dir, and optional files.""" + base = Path(args.project).expanduser() if args.project else Path.cwd() + if args.graphify_out: + graphify_out = Path(args.graphify_out).expanduser() + elif args.graph: + graphify_out = Path(args.graph).expanduser().parent + elif (base / "graph.json").exists(): + graphify_out = base + else: + graphify_out = base / "graphify-out" + + project_root = graphify_out.parent if graphify_out.name == "graphify-out" else base + graph = Path(args.graph).expanduser() if args.graph else graphify_out / "graph.json" + report = Path(args.report).expanduser() if args.report else graphify_out / "GRAPH_REPORT.md" + labels = Path(args.labels).expanduser() if args.labels else graphify_out / ".graphify_labels.json" + sections = Path(args.sections).expanduser() if args.sections else None + return { + "base": project_root, + "graphify_out": graphify_out, + "graph": graph, + "report": report, + "labels": labels, + "sections": sections, + } + + +def is_zh(lang: str) -> bool: + """Return true when localized strings should be Chinese.""" + return (lang or "").lower().startswith("zh") + + +def pick_text(lang: str, zh: str, en: str) -> str: + """Small localization helper for generated copy.""" + return zh if is_zh(lang) else en + + +def detect_lang(lang: str, nodes: list, labels: dict) -> str: + """Resolve auto language from labels and node names.""" + if lang and lang.lower() != "auto": + return lang + sample = " ".join( + list(labels.values())[:50] + + [str(n.get("label", "")) for n in nodes[:200]] + + [str(n.get("source_file", "")) for n in nodes[:100]] + ) + return "zh-CN" if re.search(r"[\u4e00-\u9fff]", sample) else "en" + + +def truncate_text(text: str, limit: int) -> str: + """Truncate without splitting Mermaid syntax.""" + text = " ".join(str(text or "").split()) + if len(text) <= limit: + return text + return text[: max(0, limit - 3)].rstrip() + "..." + + +def humanize_label(label: str, source_file: str = "") -> str: + """Convert graph labels into short labels people can scan in a diagram.""" + label = str(label or "").strip() + if not label: + return Path(source_file).name if source_file else "Unknown" + if label.startswith(".") and label.endswith("()"): + return label[1:] + if label.endswith((".py", ".ts", ".tsx", ".js", ".jsx", ".go", ".rs", ".java", ".rb")): + return Path(label).name + if "_" in label and " " not in label and len(label) > 28: + parts = [p for p in label.split("_") if p] + if parts: + label = " ".join(parts[-3:]) + return truncate_text(label, 42) + + +def node_kind(node: dict) -> str: + """Classify a graph node for Mermaid styling and table tags.""" + label = str(node.get("label") or node.get("id") or "").lower() + source_file = str(node.get("source_file") or "").lower() + file_type = str(node.get("file_type") or "").lower() + node_type = str(node.get("node_type") or "").lower() + if node_type in {"class", "klass", "struct", "interface", "enum", "trait", "model"}: + return "klass" + if node_type in {"module", "file", "package", "namespace"}: + return "module" + if node_type in {"endpoint", "route", "api", "handler", "controller"}: + return "api" + if node_type in {"test", "spec"}: + return "test" + if node_type in {"component", "hook", "view", "page"}: + return "ui" + if file_type in {"rationale", "document"}: + return "concept" + if "test" in source_file or label.startswith("test_") or "spec" in source_file: + return "test" + if any(word in label for word in ("endpoint", "router", "api", "route")): + return "api" + if any(word in label for word in ("cli", "command", "click", "typer")): + return "entry" + if any(word in label for word in ("async", "await", "stream", "sse")): + return "async" + raw_label = str(node.get("label") or "") + hook_like = raw_label.startswith("use") and len(raw_label) > 3 and (raw_label[3].isupper() or raw_label[3] in "_-") + if any(word in label for word in ("component", "props", "hook", "store")) or hook_like or source_file.endswith((".tsx", ".jsx", ".vue", ".svelte")): + return "ui" + raw = raw_label + if raw[:1].isupper() and not raw.endswith("()"): + return "klass" + if raw.endswith((".py", ".ts", ".tsx", ".js", ".jsx", ".go", ".rs", ".java", ".kt", ".rb", ".php", ".cs", ".swift", ".vue", ".svelte")): + return "module" + return "function" + + +def relation_label(relation: str, lang: str) -> str: + """Map graph edge relation names to short diagram labels.""" + relation = str(relation or "").strip() + zh = { + "calls": "调用", + "uses": "使用", + "imports": "导入", + "imports_from": "导入", + "method": "方法", + "contains": "包含", + "rationale_for": "说明", + "conceptually_related_to": "相关", + "participate_in": "参与", + "form": "组成", + } + en = { + "calls": "calls", + "uses": "uses", + "imports": "imports", + "imports_from": "imports", + "method": "method", + "contains": "contains", + "rationale_for": "explains", + "conceptually_related_to": "relates", + "participate_in": "joins", + "form": "forms", + } + mapped = (zh if is_zh(lang) else en).get(relation, relation.replace("_", " ")) + return safe_mermaid_text(mapped) + + +def preferred_edges(edges: list, allow_structure: bool = False) -> list: + """Filter to edges that make a readable call-flow diagram.""" + primary = {"calls", "uses", "method", "imports", "imports_from"} + secondary = {"contains", "rationale_for", "conceptually_related_to"} + selected = [] + for edge in edges: + if not should_include_edge(edge): + continue + relation = edge.get("relation", "") + if relation in primary or (allow_structure and relation in secondary): + selected.append(edge) + if selected: + return selected + return [edge for edge in edges if should_include_edge(edge)] + + +def edge_score(edge: dict) -> float: + """Rank edges by confidence and usefulness for diagrams.""" + relation = edge.get("relation", "") + score = to_float(edge.get("confidence_score", 1.0), 1.0) + if str(edge.get("confidence", "")).upper() == "EXTRACTED": + score += 2.0 + if relation in {"calls", "uses", "method"}: + score += 1.0 + elif relation in {"imports", "imports_from"}: + score += 0.6 + elif relation == "contains": + score -= 0.2 + elif relation == "rationale_for": + score -= 0.6 + return score + + +def mermaid_init(scale: float, direction: str = "LR") -> str: + """Return a Mermaid init directive that scales diagrams using Mermaid config.""" + scale = max(0.65, min(float(scale or 1.0), 1.8)) + config = { + "theme": "dark", + "themeVariables": { + "fontSize": f"{round(15 * scale, 1)}px", + "fontFamily": "Segoe UI, system-ui, sans-serif", + "primaryColor": "#1e293b", + "primaryTextColor": "#e2e8f0", + "primaryBorderColor": "#38bdf8", + "secondaryColor": "#0f172a", + "tertiaryColor": "#334155", + "lineColor": "#64748b", + "textColor": "#e2e8f0", + }, + "flowchart": { + "htmlLabels": True, + "curve": "basis", + "nodeSpacing": round(48 * scale), + "rankSpacing": round(64 * scale), + "padding": round(14 * scale), + "diagramPadding": round(10 * scale), + "useMaxWidth": True, + }, + } + return f"%%{{init: {json.dumps(config, ensure_ascii=False)}}}%%\nflowchart {direction}" + + +def mermaid_class_defs() -> list: + """Shared Mermaid-native styles for readable diagrams.""" + return [ + " classDef entry fill:#422006,stroke:#fbbf24,color:#fde68a,stroke-width:1px;", + " classDef api fill:#450a0a,stroke:#f87171,color:#fee2e2,stroke-width:1px;", + " classDef async fill:#2e1065,stroke:#a78bfa,color:#ede9fe,stroke-width:1px;", + " classDef klass fill:#064e3b,stroke:#34d399,color:#d1fae5,stroke-width:1px;", + " classDef ui fill:#831843,stroke:#f472b6,color:#fce7f3,stroke-width:1px;", + " classDef module fill:#172554,stroke:#60a5fa,color:#dbeafe,stroke-width:1px;", + " classDef test fill:#3f3f46,stroke:#a1a1aa,color:#f4f4f5,stroke-width:1px;", + " classDef concept fill:#292524,stroke:#a8a29e,color:#fafaf9,stroke-dasharray:4 3;", + " classDef function fill:#0f172a,stroke:#38bdf8,color:#e0f2fe,stroke-width:1px;", + ] + + +# ────────────────────────────────────────────── +# 4. Community and section indexing +# ────────────────────────────────────────────── + +def build_community_index(nodes: list) -> dict: + """Map community_id (str) -> list of nodes.""" + idx = defaultdict(list) + for n in nodes: + cid = str(n.get("community", "unknown")) + idx[cid].append(n) + return idx + + +def html_anchor_id(raw: str, fallback: str, used: set) -> str: + """Generate a stable, unique HTML anchor ID.""" + raw = str(raw or fallback or "") + base = re.sub(r"[^a-z0-9]+", "-", raw.lower()).strip("-") + if not base: + base = re.sub(r"[^a-z0-9]+", "-", str(fallback or "section").lower()).strip("-") + if not base: + base = "section" + base = base[:48].strip("-") or "section" + candidate = base + if candidate in used: + candidate = f"{base}-{hashlib.sha1(raw.encode('utf-8')).hexdigest()[:6]}" + suffix = 2 + while candidate in used: + candidate = f"{base}-{suffix}" + suffix += 1 + used.add(candidate) + return candidate + + +def normalize_communities(value) -> list: + """Normalize section community lists from JSON or simple strings.""" + if isinstance(value, list): + return value + if value in (None, ""): + return [] + if isinstance(value, str): + return [part.strip() for part in value.split(",") if part.strip()] + return [value] + + +def normalize_sections(sections: list, lang: str) -> list: + """Ensure sections have safe unique IDs and an overview section first.""" + overview_name = pick_text(lang, "架构总览", "Architecture Overview") + normalized = [{"id": "overview", "name": overview_name, "communities": []}] + used = {"overview", "hyperedges", "stats"} + + for index, raw in enumerate(sections or [], 1): + if not isinstance(raw, dict): + continue + raw_id = str(raw.get("id") or raw.get("key") or raw.get("name") or f"section-{index}") + raw_name = str(raw.get("name") or raw.get("label") or raw_id) + if raw_id.lower() == "overview": + normalized[0]["name"] = raw_name or overview_name + continue + + sid = html_anchor_id(raw_id, f"section-{index}", used) + normalized.append({ + "id": sid, + "name": raw_name, + "communities": normalize_communities(raw.get("communities", raw.get("community"))), + }) + return normalized + + +def label_for_community(cid: str, labels: dict, nodes: list, lang: str) -> str: + """Choose a readable section name for a community.""" + if str(cid) in labels and labels[str(cid)]: + return labels[str(cid)] + keywords = section_keywords(nodes, 3) + if keywords: + return " ".join(word.title() for word in keywords[:3]) + return pick_text(lang, f"社区 {cid}", f"Community {cid}") + + +SECTION_ARCHETYPES = [ + ( + "extract-pipeline", + "提取管线", + "Extraction Pipeline", + { + "extract", "extractor", "tree", "sitter", "parser", "language", + "python", "javascript", "typescript", "rust", "java", "go", + "ast", "calls", "imports", "multilang", + }, + ), + ( + "build-graph", + "图谱构建", + "Graph Build", + { + "build", "graph", "merge", "dedup", "node", "edge", "hyperedge", + "json", "schema", "normalize", "confidence", + }, + ), + ( + "analysis-clustering", + "分析聚类", + "Analysis & Clustering", + { + "cluster", "community", "leiden", "cohesion", "analyze", "god", + "surprise", "question", "query", "path", "explain", "benchmark", + }, + ), + ( + "outputs-docs", + "输出文档", + "Outputs & Docs", + { + "export", "html", "wiki", "obsidian", "canvas", "svg", "graphml", + "report", "callflow", "mermaid", "tree", "documentation", + }, + ), + ( + "cli-skills", + "CLI 与技能安装", + "CLI & Skill Installers", + { + "main", "install", "uninstall", "skill", "agent", "claude", + "codex", "opencode", "aider", "copilot", "kiro", "vscode", + "hook", "command", + }, + ), + ( + "ingest-cache-update", + "摄取与增量更新", + "Ingestion & Updates", + { + "ingest", "fetch", "download", "url", "html", "markdown", + "cache", "manifest", "watch", "update", "incremental", + "transcribe", "video", "audio", "google", + }, + ), + ( + "serve-api", + "服务 API", + "Serving API", + { + "serve", "api", "request", "response", "endpoint", "router", + "handle", "upload", "search", "delete", "enrich", + }, + ), + ( + "security-global", + "安全与全局图", + "Security & Global Graph", + { + "security", "safe", "ssrf", "xss", "path", "traversal", + "global", "prefix", "prune", "repo", "clone", + }, + ), + ( + "tests-fixtures", + "测试与样例", + "Tests & Fixtures", + { + "test", "tests", "fixture", "fixtures", "sample", "assert", + "pytest", "mock", + }, + ), +] + + +def _community_text(nodes: list, label: str = "") -> str: + parts = [label] + for node in nodes[:80]: + parts.append(str(node.get("label", ""))) + parts.append(str(node.get("source_file", ""))) + parts.append(str(node.get("node_type", ""))) + parts.append(str(node.get("file_type", ""))) + return " ".join(parts).lower() + + +def _keyword_score(text: str, keywords: set[str]) -> int: + score = 0 + for keyword in keywords: + score += len(re.findall(rf"(? tuple[list, list]: + """Return selected grouped sections and overflow communities.""" + ranked = sorted( + grouped.values(), + key=lambda sec: (sec["priority"], -sec["node_count"], sec["id"]), + ) + cap = max(1, int(max_sections or 15)) + selected = ranked[:cap] + overflow = ranked[cap:] + overflow_communities = [] + for sec in overflow: + overflow_communities.extend(sec["communities"]) + return selected, overflow_communities + + +def derive_sections_from_communities(nodes: list, labels: dict, lang: str, max_sections: int) -> list: + """Derive architecture-oriented sections when no sections JSON is supplied.""" + comm_idx = build_community_index(nodes) + sections = [{"id": "overview", "name": pick_text(lang, "架构总览", "Architecture Overview"), "communities": []}] + grouped = {} + unassigned = [] + + for cid, community_nodes in sorted(comm_idx.items(), key=lambda item: (-len(item[1]), str(item[0]))): + label = label_for_community(cid, labels, community_nodes, lang) + text = _community_text(community_nodes, label) + best = None + best_score = 0 + for priority, (sid, zh_name, en_name, keywords) in enumerate(SECTION_ARCHETYPES): + score = _keyword_score(text, keywords) + if score > best_score: + best = (priority, sid, zh_name, en_name) + best_score = score + + if best and best_score >= 2: + priority, sid, zh_name, en_name = best + sec = grouped.setdefault( + sid, + { + "id": sid, + "name": pick_text(lang, zh_name, en_name), + "communities": [], + "node_count": 0, + "priority": priority, + }, + ) + sec["communities"].append(cid) + sec["node_count"] += len(community_nodes) + else: + unassigned.append((cid, community_nodes, label)) + + selected, overflow_communities = _rank_grouped_sections(grouped, max(1, int(max_sections or 15)) - 1) + sections.extend( + {"id": sec["id"], "name": sec["name"], "communities": sec["communities"]} + for sec in selected + ) + + remaining_slots = max(0, int(max_sections or 15) - (len(sections) - 1) - 1) + for cid, community_nodes, label in unassigned[:remaining_slots]: + sections.append({"id": str(label or f"community-{cid}"), "name": label, "communities": [cid]}) + + other_communities = overflow_communities + [cid for cid, _, _ in unassigned[remaining_slots:]] + if other_communities: + sections.append({ + "id": "other", + "name": pick_text(lang, "其他", "Other"), + "communities": other_communities, + }) + return sections + + +def build_section_node_map(sections: list, comm_idx: dict) -> dict: + """Map section_id -> list of nodes belonging to its communities.""" + section_nodes = {} + for sec in sections: + sid = sec["id"] + if sid == "overview": + section_nodes[sid] = [] + continue + nodes = [] + for cid in sec.get("communities", []): + nodes.extend(comm_idx.get(str(cid), [])) + section_nodes[sid] = nodes + return section_nodes + + +def node_in_section(node_id: str, section_node_ids: set) -> bool: + """Check if a node belongs to a section.""" + return node_id in section_node_ids + + +# ────────────────────────────────────────────── +# 5. Edge analysis +# ────────────────────────────────────────────── + +def classify_edges(edges: list, section_nodes_map: dict) -> dict: + """Classify edges as intra-section or inter-section. + + Returns: + { + "intra": {section_id: [edges]}, + "inter": [edges], + "orphan": [edges] # one endpoint not in any section + } + """ + # Build node -> section lookup + node_section = {} + for sid, nodes in section_nodes_map.items(): + for n in nodes: + node_section[n.get("id")] = sid + + intra = defaultdict(list) + inter = [] + orphan = [] + + for e in edges: + src = e.get("source", "") + tgt = e.get("target", "") + src_sec = node_section.get(src) + tgt_sec = node_section.get(tgt) + + if src_sec is None or tgt_sec is None: + orphan.append(e) + elif src_sec == tgt_sec: + intra[src_sec].append(e) + else: + inter.append(e) + + return {"intra": dict(intra), "inter": inter, "orphan": orphan, "node_section": node_section} + + +def should_include_edge(edge: dict) -> bool: + """Decide whether to auto-include an edge in Mermaid output.""" + conf = str(edge.get("confidence", "EXTRACTED")).upper() + score = to_float(edge.get("confidence_score", 1.0), 1.0) + + if conf == "EXTRACTED": + return True + if conf == "INFERRED" and score >= 0.85: + return True + # Low-confidence INFERRED or AMBIGUOUS: comment out for LLM review + return False + + +# ────────────────────────────────────────────── +# 6. Mermaid diagram generators +# ────────────────────────────────────────────── + +def node_degree_scores(edges: list) -> Counter: + """Score nodes by useful edge participation.""" + scores = Counter() + for edge in edges: + score = edge_score(edge) + scores[edge.get("source", "")] += score + scores[edge.get("target", "")] += score + return scores + + +def node_importance(node: dict) -> float: + """Use graphify centrality fields when available.""" + for key in ("pagerank", "page_rank", "pageRank", "rank", "centrality", "score"): + if key in node: + return to_float(node.get(key), 0.0) + return 0.0 + + +def select_diagram_nodes(nodes: list, edges: list, max_nodes: int) -> list: + """Select a compact, connected subset of nodes for readable diagrams.""" + node_by_id = {n.get("id"): n for n in nodes} + usable_edges = preferred_edges(edges, allow_structure=False) + if not usable_edges: + usable_edges = preferred_edges(edges, allow_structure=True) + scores = node_degree_scores(usable_edges) + outgoing = Counter(edge.get("source", "") for edge in usable_edges) + incoming = Counter(edge.get("target", "") for edge in usable_edges) + selected = [] + seen = set() + + def add_node(nid: str) -> bool: + node = node_by_id.get(nid) + if not node or nid in seen: + return False + kind = node_kind(node) + if kind == "concept" and len(selected) >= max(4, max_nodes // 3): + return False + selected.append(node) + seen.add(nid) + return len(selected) >= max_nodes + + # Start with likely entry points: nodes that call out more than they are called. + entry_candidates = sorted( + node_by_id, + key=lambda nid: (-(outgoing[nid] - incoming[nid]), -outgoing[nid], str(nid)), + ) + for nid in entry_candidates[: max(3, max_nodes // 3)]: + if outgoing[nid] > 0 and add_node(nid): + return selected + + # Then pull in the most useful neighbors from the strongest edges. + for edge in sorted(usable_edges, key=edge_score, reverse=True): + for nid in (edge.get("source"), edge.get("target")): + if add_node(nid): + return selected + + def fallback_key(node: dict) -> tuple: + nid = node.get("id", "") + kind_penalty = 1 if node_kind(node) == "concept" else 0 + return ( + kind_penalty, + -scores.get(nid, 0), + -node_importance(node), + safe_file_path(node.get("source_file", "")), + humanize_label(node.get("label", nid)), + ) + + for node in sorted(nodes, key=fallback_key): + nid = node.get("id") + if nid not in seen: + selected.append(node) + seen.add(nid) + if len(selected) >= max_nodes: + break + return selected + + +def node_label(node: dict) -> str: + """Build a readable Mermaid node label.""" + label = humanize_label(node.get("label") or node.get("id"), node.get("source_file", "")) + source_file = safe_file_path(node.get("source_file", "")) + if source_file and not label.endswith(Path(source_file).name): + return f"{safe_mermaid_text(label)}
{safe_mermaid_text(source_file)}" + return safe_mermaid_text(label) + + +def group_nodes_by_file(nodes: list) -> dict: + """Group selected nodes by source file for Mermaid subgraphs.""" + groups = defaultdict(list) + for node in nodes: + source_file = safe_file_path(node.get("source_file", "")) or "External / generated" + groups[source_file].append(node) + return dict(sorted(groups.items(), key=lambda item: (-len(item[1]), item[0]))) + + +def section_edge_summary(classified_edges: dict) -> dict: + """Aggregate inter-section edge counts and relation names.""" + node_section = classified_edges.get("node_section", {}) + summary = defaultdict(lambda: {"count": 0, "relations": Counter()}) + for edge in classified_edges.get("inter", []): + if not should_include_edge(edge): + continue + src_sec = node_section.get(edge.get("source")) + tgt_sec = node_section.get(edge.get("target")) + if not src_sec or not tgt_sec or src_sec == tgt_sec: + continue + key = (src_sec, tgt_sec) + summary[key]["count"] += 1 + summary[key]["relations"][edge.get("relation", "relates")] += 1 + return summary + + +def generate_overview_graph(sections: list, section_nodes_map: dict, + classified_edges: dict, labels: dict, lang: str, + diagram_scale: float) -> str: + """Generate a readable section-level architecture overview.""" + lines = [mermaid_init(diagram_scale, "LR")] + section_defs = [sec for sec in sections if sec["id"] != "overview"] + + for sec in section_defs: + sid = mermaid_section_id(sec["id"]) + node_count = len(section_nodes_map.get(sec["id"], [])) + label = ( + f"{safe_mermaid_text(sec.get('name', sec['id']))}" + f"
{node_count} {safe_mermaid_text('nodes')}" + ) + lines.append(f' {sid}("{label}")') + lines.append(f" class {sid} module;") + + aggregated = section_edge_summary(classified_edges) + for (src, tgt), data in sorted(aggregated.items(), key=lambda item: item[1]["count"], reverse=True)[:12]: + src_id = mermaid_section_id(src) + tgt_id = mermaid_section_id(tgt) + relation, _ = data["relations"].most_common(1)[0] + label = relation_label(relation, lang) + if data["count"] > 1: + label = f"{label} x{data['count']}" + lines.append(f" {src_id} -->|{label}| {tgt_id}") + + if not aggregated and len(section_defs) > 1: + for prev, cur in zip(section_defs, section_defs[1:]): + lines.append(f" {mermaid_section_id(prev['id'])} -.-> {mermaid_section_id(cur['id'])}") + + lines.extend(mermaid_class_defs()) + return "\n".join(lines) + + +def generate_section_flowchart(section_id: str, section_name: str, + nodes: list, edges: list, lang: str, + diagram_scale: float, max_nodes: int, + max_edges: int) -> str: + """Generate a compact, human-readable call-flow chart for a section.""" + lines = [mermaid_init(diagram_scale, "LR")] + lines.append(f" %% Section: {safe_mermaid_text(section_name)} ({len(nodes)} nodes, {len(edges)} edges)") + + if not nodes: + empty_label = pick_text(lang, f"{section_name} - 无节点", f"{section_name} - no nodes") + lines.append(f' empty("{safe_mermaid_text(empty_label)}")') + lines.extend(mermaid_class_defs()) + return "\n".join(lines) + + selected_nodes = select_diagram_nodes(nodes, edges, max_nodes) + selected_ids = {node.get("id") for node in selected_nodes} + visible_edges = [ + edge for edge in preferred_edges(edges, allow_structure=False) + if edge.get("source") in selected_ids and edge.get("target") in selected_ids + ] + if not visible_edges: + visible_edges = [ + edge for edge in preferred_edges(edges, allow_structure=True) + if edge.get("source") in selected_ids and edge.get("target") in selected_ids + ] + + groups = group_nodes_by_file(selected_nodes) + class_lines = [] + for source_file, group in groups.items(): + group_id = node_mermaid_id({"id": f"{section_id}_{source_file}"}) + if len(groups) > 1 and len(group) > 1: + lines.append(f' subgraph {group_id}["{safe_mermaid_text(source_file)}"]') + indent = " " + else: + indent = " " + for node in group: + mid = node_mermaid_id(node) + lines.append(f'{indent}{mid}("{node_label(node)}")') + class_lines.append(f" class {mid} {node_kind(node)};") + if len(groups) > 1 and len(group) > 1: + lines.append(" end") + + included = 0 + for edge in sorted(visible_edges, key=edge_score, reverse=True): + if included >= max_edges: + break + src_id = node_mermaid_id({"id": edge.get("source", "")}) + tgt_id = node_mermaid_id({"id": edge.get("target", "")}) + rel = relation_label(edge.get("relation", ""), lang) + lines.append(f" {src_id} -->|{rel}| {tgt_id}") + included += 1 + + omitted_nodes = max(0, len(nodes) - len(selected_nodes)) + omitted_edges = max(0, len(visible_edges) - included) + if omitted_nodes or omitted_edges: + lines.append(f" %% Omitted for readability: {omitted_nodes} nodes, {omitted_edges} edges") + lines.extend(class_lines) + lines.extend(mermaid_class_defs()) + return "\n".join(lines) + + +# ────────────────────────────────────────────── +# 7. HTML generators +# ────────────────────────────────────────────── + +def generate_nav(sections: list) -> str: + """Generate the sticky navigation bar.""" + links = [] + for sec in sections: + links.append(f' {escape(sec["name"])}') + return '
\n' + "\n".join(links) + "\n
" + + +def node_display_name(node: dict | None, fallback: str = "") -> str: + """Readable node label for tables and summaries.""" + if not node: + return str(fallback or "") + label = str(node.get("label") or node.get("id") or fallback or "") + return humanize_label(label, node.get("source_file", "")) + + +def format_node_refs(node_ids: set, node_by_id: dict, lang: str, empty_text: str, limit: int = 3) -> str: + """Render node references as readable labels instead of internal IDs.""" + if not node_ids: + return escape(empty_text) + parts = [] + for nid in sorted(node_ids, key=lambda item: node_display_name(node_by_id.get(item), item).lower())[:limit]: + node = node_by_id.get(nid) + label = node_display_name(node, nid) + source = safe_file_path((node or {}).get("source_file", "")) + if source: + parts.append(f"{escape(label)}
{escape(source)}") + else: + parts.append(f"{escape(label)}") + if len(node_ids) > limit: + parts.append(escape(pick_text(lang, f"+{len(node_ids) - limit} 个更多", f"+{len(node_ids) - limit} more"))) + return "
".join(parts) + + +def generate_call_table_rows(nodes: list, section_edges: list, lang: str) -> str: + """Generate call table row scaffolding for a section's nodes.""" + if not nodes: + return "" + + # Build source/target lookup from edges + node_by_id = {n.get("id"): n for n in nodes} + callers = defaultdict(set) + callees = defaultdict(set) + for e in section_edges: + src = e.get("source", "") + tgt = e.get("target", "") + if e.get("relation") in ("calls", "imports", "imports_from", "uses", "method"): + callers[tgt].add(src) + callees[src].add(tgt) + + rows = [] + for i, n in enumerate(nodes[:30], 1): # cap at 30 rows + nid = n.get("id", "") + label = n.get("label", nid) + source_file = safe_file_path(n.get("source_file", "")) + file_type = n.get("file_type", "code") + + # Suggest a tag type based on file_type and label heuristics + tag = _suggest_tag(label, file_type, lang, node_kind(n)) + + caller_text = format_node_refs( + callers.get(nid, set()), + node_by_id, + lang, + pick_text(lang, "外部入口 / 无直接入边", "External entry / no inbound edge"), + ) + callee_text = format_node_refs( + callees.get(nid, set()), + node_by_id, + lang, + pick_text(lang, "无直接出边", "No direct outbound edge"), + ) + + rows.append(f""" + {i} + {escape(label)}
{escape(source_file)} + {tag} + {caller_text} + {callee_text} + {escape(_describe_node(label, source_file, file_type, lang))} +""") + + return "\n".join(rows) + + +def _suggest_tag(label: str, file_type: str, lang: str, kind: str = "") -> str: + """Heuristic tag suggestion based on label name and file type.""" + lower = label.lower() + names = { + "concept": ("概念", "Concept", "tag-func"), + "entry": ("入口", "Entry", "tag-cmd"), + "api": ("API", "API", "tag-endpoint"), + "async": ("异步", "Async", "tag-async"), + "klass": ("类", "Class", "tag-class"), + "ui": ("UI", "UI", "tag-hook"), + "module": ("模块", "Module", "tag-class"), + "test": ("测试", "Test", "tag-func"), + "function": ("函数", "Function", "tag-func"), + } + if kind in names: + zh, en, cls = names[kind] + return f'{pick_text(lang, zh, en)}' + if file_type == "rationale": + return f'{pick_text(lang, "概念", "Concept")}' + if any(kw in lower for kw in ("cli", "command", "scan", "serve", "chat", "config")): + if "group" in lower or "command" in lower: + return f'{pick_text(lang, "CLI命令", "CLI")}' + if any(kw in lower for kw in ("router", "endpoint", "api", "/api/")): + return f'{pick_text(lang, "API端点", "API")}' + if any(kw in lower for kw in ("async", "await", "stream")): + return f'{pick_text(lang, "异步", "Async")}' + if any(kw in lower for kw in ("class", "model", "schema", "dataclass", "pydantic")): + return f'{pick_text(lang, "类", "Class")}' + if any(kw in lower for kw in ("hook", "usestate", "useeffect", "store")): + return 'Hook' + if any(kw in lower for kw in ("component", "props", "tsx", "jsx", "render")): + return f'{pick_text(lang, "组件", "Component")}' + return f'{pick_text(lang, "函数", "Function")}' + + +def _describe_node(label: str, source_file: str, file_type: str, lang: str) -> str: + """Generate a compact human-readable description for a graph node.""" + lower = label.lower() + source = source_file or pick_text(lang, "项目", "project") + if file_type == "rationale": + return pick_text(lang, f"设计说明：{label}", f"Design note for {label}.") + if file_type == "document": + return pick_text(lang, f"文档入口，描述 {label} 相关能力。", f"Documentation node describing {label}.") + if label.endswith(".py") or label.endswith(".tsx") or label.endswith(".ts"): + return pick_text(lang, f"{source} 中的模块文件，承载该层主要实现。", f"Module file in {source}.") + if "config" in lower: + return pick_text(lang, "读取、解析或持久化项目配置。", "Reads, resolves, or persists project configuration.") + if "scan" in lower: + return pick_text(lang, "触发项目扫描或处理扫描状态。", "Starts scanning or handles scan status.") + if "ingest" in lower or "clone" in lower or "git" in lower: + return pick_text(lang, "把本地目录或远程仓库转换为分析上下文。", "Turns a local path or remote repository into analysis context.") + if "prompt" in lower: + return pick_text(lang, "构造发送给 LLM 的结构化提示。", "Builds structured prompts for model calls.") + if "analy" in lower: + return pick_text(lang, "编排分析流程并产出结构化文档数据。", "Orchestrates analysis and returns structured documentation data.") + if "graph" in lower or "dependency" in lower: + return pick_text(lang, "构建依赖关系并提供排序或图形化数据。", "Builds dependency relationships and graph data.") + if "export" in lower or "markdown" in lower or "html" in lower: + return pick_text(lang, "将文档数据导出为目标格式。", "Exports documentation data to a target format.") + if "chat" in lower or "rag" in lower or "retrieve" in lower: + return pick_text(lang, "支撑检索增强问答或流式聊天。", "Supports retrieval-augmented Q&A or streaming chat.") + if "wiki" in lower or "page" in lower or "sidebar" in lower: + return pick_text(lang, "组织文档页面、侧边栏或内容读取。", "Organizes documentation pages, navigation, or content lookup.") + if "cache" in lower or "hash" in lower: + return pick_text(lang, "缓存分析结果或生成缓存键。", "Caches analysis results or computes cache keys.") + if "test" in lower: + return pick_text(lang, "验证导入、入口点或版本等基础行为。", "Verifies imports, entry points, or version behavior.") + return pick_text(lang, f"{source} 中的 {label} 节点。", f"{label} node in {source}.") + + +def generate_header(sections: list, meta: dict, lang: str) -> str: + """Generate the HTML header, title, subtitle, and nav.""" + project_name = str(meta.get("project_name", "Project")) + commit = str(meta.get("built_at_commit", "unknown"))[:7] + + if lang.startswith("zh"): + title = f"{project_name} — 完整调用流程与架构文档" + subtitle = ( + f"由 graphify 知识图谱生成：{meta.get('node_count', '?')} 个节点、" + f"{meta.get('edge_count', '?')} 条边、{meta.get('community_count', '?')} 个社区。" + f"Commit: {commit}" + ) + else: + title = f"{project_name} — Complete Call Flow & Architecture Documentation" + subtitle = ( + f"Generated from graphify knowledge graph: {meta.get('node_count', '?')} nodes, " + f"{meta.get('edge_count', '?')} edges, {meta.get('community_count', '?')} communities. " + f"Commit: {commit}" + ) + + return f"""

{escape(title)}

+

{escape(subtitle)}

+ +{generate_nav(sections)} +""" + + +def derive_flow_chain(sections: list, classified_edges: dict) -> str: + """Derive a readable section flow from inter-section edges.""" + section_names = {sec["id"]: sec.get("name", sec["id"]) for sec in sections} + order = [sec["id"] for sec in sections if sec["id"] != "overview"] + if not order: + return "Graph nodes -> documentation" + + outgoing = defaultdict(Counter) + incoming = Counter() + for (src, tgt), data in section_edge_summary(classified_edges).items(): + outgoing[src][tgt] += data["count"] + incoming[tgt] += data["count"] + + start = min(order, key=lambda sid: (incoming.get(sid, 0), order.index(sid))) + chain = [start] + seen = {start} + current = start + while len(chain) < min(7, len(order)): + candidates = [(count, tgt) for tgt, count in outgoing.get(current, {}).items() if tgt not in seen] + if candidates: + _, nxt = max(candidates) + else: + remaining = [sid for sid in order if sid not in seen] + if not remaining: + break + nxt = remaining[0] + chain.append(nxt) + seen.add(nxt) + current = nxt + return " -> ".join(section_names.get(sid, sid) for sid in chain) + + +def generate_overview_cards(meta: dict, report_text: str, sections: list, + section_nodes_map: dict, classified_edges: dict, + lang: str) -> str: + """Generate generic overview cards.""" + rows = [] + for sec in sections: + if sec["id"] == "overview": + continue + communities = ", ".join(str(c) for c in sec.get("communities", [])) + node_count = len(section_nodes_map.get(sec["id"], [])) + rows.append( + f"{escape(sec['name'])}{node_count}{escape(communities)}" + ) + + flow = derive_flow_chain(sections, classified_edges) + layer_title = pick_text(lang, "架构层次", "Architecture Layers") + layer_cols = pick_text(lang, "层节点社区", "LayerNodesCommunities") + flow_title = pick_text(lang, "核心数据流", "Core Flow") + return f"""
+
+

{layer_title}

+ + {layer_cols} + {''.join(rows)} +
+
+
+

{flow_title}

+
{escape(flow)}
+
+
""" + + +def section_keywords(nodes: list, limit: int = 5) -> list: + """Pick representative words from labels and file names.""" + counts = Counter() + stopwords = { + "the", "and", "for", "with", "from", "this", "that", "class", "function", + "method", "file", "src", "lib", "core", "index", "main", "init", "py", + "ts", "tsx", "js", "jsx", "go", "rs", "java", "html", "css", + } + for node in nodes: + text = f"{node.get('label', '')} {node.get('source_file', '')}".replace("/", " ").replace("_", " ").replace("-", " ") + for raw in text.split(): + word = "".join(ch for ch in raw.lower() if ch.isalnum()) + if len(word) < 3 or word in stopwords: + continue + counts[word] += 1 + return [word for word, _ in counts.most_common(limit)] + + +def generate_section_intro(sec: dict, nodes: list, edge_count: int, lang: str) -> str: + """Generate the section introductory paragraph.""" + file_counts = Counter(n.get("source_file") for n in nodes if n.get("source_file")) + files = [safe_file_path(path) for path, _ in file_counts.most_common(3)] + keywords = section_keywords(nodes, 4) + if is_zh(lang): + file_text = "、".join(files) if files else "未标注源文件" + keyword_text = "、".join(keywords) if keywords else sec.get("name", sec["id"]) + text = ( + f"{sec.get('name', sec['id'])} 汇集了与 {keyword_text} 相关的实现，" + f"主要分布在 {file_text}。本节覆盖 {len(nodes)} 个节点、{edge_count} 条内部边，" + "图中只展示最有代表性的调用关系以保持可读性。" + ) + else: + file_text = ", ".join(files) if files else "unmapped files" + keyword_text = ", ".join(keywords) if keywords else sec.get("name", sec["id"]) + text = ( + f"{sec.get('name', sec['id'])} groups implementation around {keyword_text}, " + f"mostly in {file_text}. This section covers {len(nodes)} nodes and {edge_count} internal edges; " + "the diagram shows only representative relationships to stay readable." + ) + return f"

{escape(text)}

" + + +def generate_section_cards(sec: dict, nodes: list, section_edges: list, lang: str) -> str: + """Generate key file and design-note cards for a section.""" + file_counts = defaultdict(int) + for n in nodes: + source_file = n.get("source_file") or "" + if source_file: + file_counts[source_file] += 1 + top_files = sorted(file_counts.items(), key=lambda item: (-item[1], item[0]))[:8] + if top_files: + file_rows = "\n".join( + f"{escape(safe_file_path(path))}{count} {escape(pick_text(lang, '个节点', 'nodes'))}" + for path, count in top_files + ) + else: + file_rows = f'{escape(pick_text(lang, "无源文件映射", "No source file mapping"))}' + + relation_counts = Counter(edge.get("relation", "relates") for edge in section_edges if should_include_edge(edge)) + relation_text = ", ".join(f"{relation_label(rel, lang)} x{count}" for rel, count in relation_counts.most_common(4)) + if not relation_text: + relation_text = pick_text(lang, "未检测到高置信调用边", "No high-confidence call edges detected") + note = pick_text( + lang, + f"本节由 graphify 社区聚类生成。关系概况：{relation_text}。图表优先展示高置信、跨节点调用或使用关系，完整节点清单位于表格中。", + f"This section comes from graphify community clustering. Relationship summary: {relation_text}. The diagram prioritizes high-confidence calls or usage relationships; the table keeps the broader node inventory.", + ) + key_files = pick_text(lang, "关键文件", "Key Files") + role = pick_text(lang, "覆盖节点", "Coverage") + design_notes = pick_text(lang, "设计备注", "Design Notes") + return f"""
+
+

{key_files}

+ + + {file_rows} +
File{role}
+
+
+

{design_notes}

+

{escape(note)}

+
+
""" + + +# ────────────────────────────────────────────── +# 8. Main entry point +# ────────────────────────────────────────────── + +class CallflowOptions: + """Options for call-flow architecture HTML generation.""" + + def __init__( + self, + project: str | Path | None = None, + *, + graphify_out: str | Path | None = None, + graph: str | Path | None = None, + report: str | Path | None = None, + labels: str | Path | None = None, + sections: str | Path | None = None, + output: str | Path | None = None, + lang: str = "auto", + max_sections: int = 15, + diagram_scale: float = 1.0, + max_diagram_nodes: int = 18, + max_diagram_edges: int = 24, + ): + self.project = str(project) if project is not None else None + self.graphify_out = str(graphify_out) if graphify_out is not None else None + self.graph = str(graph) if graph is not None else None + self.report = str(report) if report is not None else None + self.labels = str(labels) if labels is not None else None + self.sections = str(sections) if sections is not None else None + self.output = str(output) if output is not None else None + self.lang = lang + self.max_sections = max_sections + self.diagram_scale = diagram_scale + self.max_diagram_nodes = max_diagram_nodes + self.max_diagram_edges = max_diagram_edges + + +def _report_highlights(report_text: str, lang: str) -> str: + """Extract a compact highlights card from GRAPH_REPORT.md.""" + if not report_text.strip(): + return "" + + lines = report_text.splitlines() + keep: list[str] = [] + in_gods = False + in_summary = False + for line in lines: + stripped = line.strip() + if stripped.startswith("## "): + in_summary = stripped == "## Summary" + in_gods = stripped.startswith("## God Nodes") + continue + if in_summary and stripped.startswith("- "): + keep.append(stripped[2:]) + elif in_gods and re.match(r"^\d+\.", stripped): + keep.append(stripped) + if len(keep) >= 6: + break + + if not keep: + return "" + + title = pick_text(lang, "图谱报告摘要", "Graph Report Highlights") + items = "\n".join(f"
  • {escape(item)}
    • " for item in keep) + return f"""
    +

    {title}

    +
      +{items} +
    +
    """ + + +def write_callflow_html( + project: str | Path | None = None, + *, + graphify_out: str | Path | None = None, + graph: str | Path | None = None, + report: str | Path | None = None, + labels: str | Path | None = None, + sections: str | Path | None = None, + output: str | Path | None = None, + lang: str = "auto", + max_sections: int = 15, + diagram_scale: float = 1.0, + max_diagram_nodes: int = 18, + max_diagram_edges: int = 24, + verbose: bool = False, +) -> Path: + """Generate call-flow architecture HTML from graphify output files.""" + args = CallflowOptions( + project, + graphify_out=graphify_out, + graph=graph, + report=report, + labels=labels, + sections=sections, + output=output, + lang=lang, + max_sections=max_sections, + diagram_scale=diagram_scale, + max_diagram_nodes=max_diagram_nodes, + max_diagram_edges=max_diagram_edges, + ) + + paths = resolve_graphify_paths(args) + if not paths["graph"].exists(): + raise FileNotFoundError( + f"graphify output not found: {paths['graph']}. " + "Run graphify first or pass --graph /path/to/graph.json." + ) + + # Load data + nodes, edges, hyperedges, meta = load_graph(paths["graph"]) + labels = load_labels(paths["labels"]) + lang = detect_lang(args.lang, nodes, labels) + if paths["sections"]: + sections = load_sections(paths["sections"]) + else: + sections = derive_sections_from_communities(nodes, labels, lang, args.max_sections) + sections = normalize_sections(sections, lang) + report_text = load_report(paths["report"]) + + if not nodes: + raise ValueError("graph.json contains 0 nodes") + if len(sections) <= 1: + raise ValueError("no sections defined") + + if verbose and len(nodes) >= 5000: + print("WARNING: Large graph -- Mermaid rendering may be slow. Consider --max-sections 5.", file=sys.stderr) + + node_ids = {node.get("id") for node in nodes} + missing_endpoint_edges = [edge for edge in edges if edge.get("source") not in node_ids or edge.get("target") not in node_ids] + if verbose and missing_endpoint_edges: + print(f"WARNING: {len(missing_endpoint_edges)} edges reference nodes not present in graph.json.", file=sys.stderr) + + meta["project_name"] = infer_project_name(str(paths["graph"]), meta) + meta["node_count"] = len(nodes) + meta["edge_count"] = len(edges) + meta["hyperedge_count"] = len(hyperedges) + + if args.output: + output_path = Path(args.output).expanduser() + if not output_path.is_absolute(): + output_path = paths["base"] / output_path + else: + output_path = paths["graphify_out"] / f"{safe_filename(meta['project_name'])}-callflow.html" + + if verbose: + print(f"Loaded: {len(nodes)} nodes, {len(edges)} edges, {len(sections)} sections") + print(f"Graph: {paths['graph']}") + + # Build index + comm_idx = build_community_index(nodes) + meta["community_count"] = len(comm_idx) + section_nodes_map = build_section_node_map(sections, comm_idx) + classified = classify_edges(edges, section_nodes_map) + + # Build HTML + html = [] + doc_title = ( + f"{meta.get('project_name', 'Project')} — 完整调用流程与架构文档" + if lang.startswith("zh") + else f"{meta.get('project_name', 'Project')} — Complete Call Flow & Architecture Documentation" + ) + + # Doctype and head + html.append(f""" + + + + +{escape(doc_title)} + + + + +
    +""") + + # Header + nav + html.append(generate_header(sections, meta, lang)) + + # ── Architecture Overview (Section "overview") ── + overview_name = sections[0].get("name", "Architecture Overview") if sections else "Architecture Overview" + html.append(f""" +

    1. {escape(str(overview_name))}

    + +
    +""") + html.append(generate_overview_graph(sections, section_nodes_map, classified, labels, lang, args.diagram_scale)) + html.append("""
    +""") + html.append(generate_overview_cards(meta, report_text, sections, section_nodes_map, classified, lang)) + report_card = _report_highlights(report_text, lang) + if report_card: + html.append(f'
    \n {report_card}\n
    ') + html.append("
    ") + + # ── Per-section content ── + section_num = 1 # overview was #1 + for sec in sections: + if sec["id"] == "overview": + continue + section_num += 1 + sid = sec["id"] + name = sec.get("name", sid) + sec_nodes = section_nodes_map.get(sid, []) + sec_edges = classified.get("intra", {}).get(sid, []) + + edge_count = len(sec_edges) + h3_title = pick_text(lang, "调用明细", "Call Details") + number_header = "#" + function_header = pick_text(lang, "节点", "Node") + type_header = pick_text(lang, "类型", "Type") + caller_header = pick_text(lang, "调用方", "Caller") + callee_header = pick_text(lang, "被调用/依赖", "Callees") + desc_header = pick_text(lang, "说明", "Description") + + html.append(f""" +

    {section_num}. {escape(str(name))}

    +{generate_section_intro(sec, sec_nodes, edge_count, lang)} + +
    +{generate_section_flowchart(sid, name, sec_nodes, sec_edges, lang, args.diagram_scale, args.max_diagram_nodes, args.max_diagram_edges)} +
    + +

    {h3_title}

    + + + + + + + + + +{generate_call_table_rows(sec_nodes, sec_edges, lang)} +
    {number_header}{function_header}{type_header}{caller_header}{callee_header}{desc_header}
    + +{generate_section_cards(sec, sec_nodes, sec_edges, lang)} +
    +""") + + # ── Section: Hyperedges (if any) ── + if hyperedges: + html.append("""

    Group Relationships (Hyperedges)

    +
    +""") + for he in hyperedges[:9]: + hid = he.get("id", "?") + hlabel = he.get("label", hid) + hnodes = he.get("nodes", []) + hrel = he.get("relation", "") + html.append(f"""
    +

    {escape(str(hlabel))}

    +

    {escape(str(hrel))} — {len(hnodes)} participants

    +
      """) + for hn in hnodes[:5]: + html.append(f"
    • {escape(str(hn))}
      • ") + if len(hnodes) > 5: + html.append(f"
    • ... and {len(hnodes) - 5} more
      • ") + html.append("
    \n
    ") + html.append("
    \n
    ") + + # ── Section: Statistics ── + total_sections = sum(1 for s in sections if s["id"] != "overview") + html.append(f"""

    Project Statistics

    + +
    +
    +

    Graph

    + + + + + + +
    Nodes{len(nodes)}
    Edges{len(edges)}
    Hyperedges{len(hyperedges)}
    Communities{len(comm_idx)}
    Documented Sections{total_sections}
    +
    +
    +

    Edge Confidence

    + + + + +
    EXTRACTED{sum(1 for e in edges if e.get('confidence') == 'EXTRACTED')}
    INFERRED{sum(1 for e in edges if e.get('confidence') == 'INFERRED')}
    AMBIGUOUS{sum(1 for e in edges if e.get('confidence') == 'AMBIGUOUS')}
    +
    +
    +""") + + # ── Footer ── + html.append(f"""
    +

    {escape(str(meta.get('project_name', 'Project')))} — Architecture Documentation

    +

    Generated: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')} · graphify callflow-html

    +
    +""") + + # Close + html.append("""
    + + + + +""") + + # Write output + output = "\n".join(html) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(output, encoding="utf-8") + + # Summary + mermaid_count = output.count('
    ') + table_count = output.count('') + section_count = output.count('