#!/usr/bin/env python3 """Generate an OKF bundle visualization. This is a project-level helper, not an OKF protocol requirement. Modes: - auto: try the official `enrichment_agent visualize` first, then fall back to the local extended viewer. - official: call only the official visualizer and fail if unavailable. - extended: generate a local HTML viewer with official-style Cytoscape/marked visualization plus project additions. """ from __future__ import annotations import argparse import json import subprocess import sys from pathlib import Path from typing import Any from okf_bundle_common import OKFDocumentData, parse_log, safe_string, scan_bundle PALETTE = [ "#4f46e5", "#0891b2", "#16a34a", "#ca8a04", "#dc2626", "#9333ea", "#0f766e", "#be185d", "#475569", "#ea580c", ] def official_command(bundle: Path, output: Path, name: str) -> list[str]: return [sys.executable, "-m", "enrichment_agent", "visualize", "--bundle", str(bundle), "--out", str(output), "--name", name] def run_official_visualizer(bundle: Path, output: Path, name: str) -> tuple[bool, str]: command = official_command(bundle, output, name) try: completed = subprocess.run(command, check=False, text=True, capture_output=True) except OSError as exc: return False, f"failed to start official visualizer: {exc}" if completed.returncode != 0: detail = (completed.stderr or completed.stdout or "").strip() return False, f"official visualizer failed with exit code {completed.returncode}: {detail}" return True, (completed.stderr or completed.stdout or "official visualizer completed").strip() def normalize_tags(value: Any) -> list[str]: if isinstance(value, list): return [str(item) for item in value if str(item).strip()] if isinstance(value, str) and value.strip(): return [value.strip()] return [] def jsonable(value: Any) -> Any: try: json.dumps(value, ensure_ascii=False) return value except TypeError: if isinstance(value, dict): return {str(key): jsonable(item) for key, item in value.items()} if isinstance(value, list): return [jsonable(item) for item in value] return safe_string(value) def concept_docs(docs: list[OKFDocumentData]) -> list[OKFDocumentData]: return [doc for doc in docs if not doc.is_reserved] def collect_bundle_data(root: Path, name: str) -> dict[str, Any]: docs = scan_bundle(root) concepts = concept_docs(docs) types = sorted({doc.type or "Concept" for doc in concepts}) colors = {concept_type: PALETTE[index % len(PALETTE)] for index, concept_type in enumerate(types)} known_ids = {doc.id for doc in concepts if doc.id} nodes: list[dict[str, Any]] = [] edges: list[dict[str, Any]] = [] seen_edges: set[tuple[str, str]] = set() broken_links: list[dict[str, Any]] = [] external_links: list[dict[str, Any]] = [] for doc in concepts: concept_type = doc.type or "Concept" tags = normalize_tags(doc.tags) node_id = doc.id or doc.path.stem body_size = len(doc.body or "") nodes.append( { "data": { "id": node_id, "label": doc.title or node_id, "type": concept_type, "description": doc.description, "resource": jsonable(doc.resource), "tags": tags, "color": colors.get(concept_type, "#64748b"), "size": 30 + min(50, body_size // 280), "timestamp": safe_string(doc.timestamp), "frontmatter": jsonable(doc.frontmatter), "body": doc.body, "citations": [citation.__dict__ for citation in doc.citations], "path": doc.rel_path, } } ) for doc in concepts: if not doc.id: continue for link in doc.links: if link.is_external: external_links.append({"source": doc.id, "label": link.label, "target": link.target_raw}) continue if not link.is_internal_markdown: continue if link.exists and link.target_id in known_ids and link.target_id != doc.id: edge_key = (doc.id, link.target_id) if edge_key not in seen_edges: seen_edges.add(edge_key) edges.append( { "data": { "id": f"{doc.id}__{link.target_id}", "source": doc.id, "target": link.target_id, "label": link.label, "raw": link.target_raw, } } ) else: broken_links.append( { "source": doc.id, "sourcePath": doc.rel_path, "label": link.label, "target": link.target_raw, "resolvedId": link.target_id, "reason": "outside bundle" if link.outside_bundle else "missing target or non-concept target", } ) logs: list[dict[str, Any]] = [] for doc in docs: if doc.reserved_name == "log.md": text = doc.path.read_text(encoding="utf-8") logs.extend({"path": doc.rel_path, "date": group.date, "entries": group.entries} for group in parse_log(text)) index_links: list[dict[str, Any]] = [] for doc in docs: if doc.reserved_name == "index.md": for link in doc.links: if link.is_internal_markdown: index_links.append( { "source": doc.rel_path, "label": link.label, "target": link.target_raw, "targetId": link.target_id, "exists": link.exists, } ) return { "name": name, "nodes": nodes, "edges": edges, "types": types, "colors": colors, "bodies": {node["data"]["id"]: node["data"].get("body", "") for node in nodes}, "brokenLinks": broken_links, "externalLinks": external_links, "logs": logs, "indexLinks": index_links, "meta": { "nodeCount": len(nodes), "edgeCount": len(edges), "brokenLinkCount": len(broken_links), "note": "Local extended viewer using the same Cytoscape.js and marked libraries as the official OKF viewer.", }, } def render_html(bundle_data: dict[str, Any]) -> str: data_json = json.dumps(bundle_data, ensure_ascii=False) return HTML_TEMPLATE.replace("__BUNDLE_DATA__", data_json) HTML_TEMPLATE = r''' OKF Bundle Visualization
未能加载 Cytoscape.js 或 marked。请检查网络/CDN，或改用官方已生成文件。
OKF bundle · local extended viewer
Click a node to see its details.
''' def generate_extended(bundle: Path, output: Path, name: str) -> tuple[int, str]: data = collect_bundle_data(bundle, name) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(render_html(data), encoding="utf-8") return data["meta"]["nodeCount"], f"generated {output} with {data['meta']['nodeCount']} concept nodes, {data['meta']['edgeCount']} links, {data['meta']['brokenLinkCount']} broken links" def main() -> int: parser = argparse.ArgumentParser(description="Visualize an OKF bundle using official visualizer or local extended fallback") parser.add_argument("bundle", type=Path, help="Path to OKF bundle directory") parser.add_argument("--out", type=Path, help="Output HTML path; defaults to /viz.html") parser.add_argument("--name", default="OKF Bundle Visualization", help="HTML title") parser.add_argument("--mode", choices=("auto", "official", "extended"), default="auto", help="Visualization mode; default: auto") args = parser.parse_args() root = args.bundle.resolve() if not root.exists() or not root.is_dir(): raise SystemExit(f"bundle path does not exist or is not a directory: {root}") output = args.out.resolve() if args.out else root / "viz.html" if args.mode in {"auto", "official"}: ok, message = run_official_visualizer(root, output, args.name) if ok: print(message) return 0 if args.mode == "official": print(message, file=sys.stderr) return 1 print(f"official visualizer unavailable; falling back to local extended viewer: {message}", file=sys.stderr) _, message = generate_extended(root, output, args.name) print(message) return 0 if __name__ == "__main__": raise SystemExit(main())