# -*- coding: utf-8 -*-
"""๐ฑ Graphify ์ง์ ๊ทธ๋ํ ์์ฑ ๋ฐ ์๊ฐํ ๋น๋
์ ๊ตญ ํ๋ก์ ํธ(shadow_brain_core)์ ์ฝ๋ ๊ตฌ์กฐ(AST)์ ๋งํฌ๋ค์ด ๋ฌธ์๋ฅผ ๋ถ์ํ์ฌ
๊ตฌ์กฐ์ ์์กด๋ง์ ๋ด์ graph.json ๋ฐ ์น ์๊ฐํ graph.html์ ๋น๋ํฉ๋๋ค.
๋น๋๋ ํ์ผ์ ๋ก์ปฌ ๋ฐ R2 Live ์คํ ๋ฆฌ์ง์ ์
๋ก๋๋์ด AI์ ๋ง์๋๊ป ์ง๋๋ก ์ ๊ณต๋ฉ๋๋ค.
์คํ ๋ฐฉ๋ฒ:
python scripts/build_graph.py [--upload]
python scripts/build_graph.py --upload --full-upload # ๋ผ์ดํธ + ํ ๊ทธ๋ํ ๋ชจ๋
"""
import os
import sys
import json
import ast
import argparse
from pathlib import Path
# UTF-8 ์ธ์ฝ๋ฉ ๊ฐ์
sys.stdout.reconfigure(encoding='utf-8')
class ProjectGraphBuilder:
def __init__(self, root_dir: Path):
self.root_dir = root_dir.resolve()
self.nodes = {}
self.edges = []
self.scanned_files = set()
def add_node(self, node_id: str, label: str, node_type: str, details: dict = None):
if node_id not in self.nodes:
self.nodes[node_id] = {
"id": node_id,
"label": label,
"type": node_type,
"details": details or {}
}
def add_edge(self, source: str, target: str, rel_type: str):
# ์ค๋ณต ์ฃ์ง ๋ฐฉ์ง
edge = {"source": source, "target": target, "type": rel_type}
if edge not in self.edges:
self.edges.append(edge)
def scan_python_files(self):
"""Python ํ์ผ์ AST ํ์ฑ์ ํตํด ํด๋์ค, ํจ์, import ๊ด๊ณ ๋ถ์"""
for root, _, files in os.walk(self.root_dir):
# ์ ์ธ ํด๋ ํํฐ๋ง
if any(p in root for p in [".git", "venv", "__pycache__", "node_modules", "dist", "build"]):
continue
for file in files:
if not file.endswith(".py"):
continue
file_path = Path(root) / file
rel_path = file_path.relative_to(self.root_dir).as_posix()
self.scanned_files.add(rel_path)
# ํ์ผ ๋
ธ๋ ์ถ๊ฐ
self.add_node(rel_path, file, "file", {"size": file_path.stat().st_size})
try:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
code = f.read()
# AST ๋ถ์
tree = ast.parse(code, filename=str(file_path))
self._analyze_ast(tree, rel_path)
except Exception as e:
print(f"โ ๏ธ AST ํ์ฑ ์คํจ [{rel_path}]: {e}")
def _analyze_ast(self, tree: ast.AST, file_rel_path: str):
"""AST ๋
ธ๋๋ฅผ ์ํํ๋ฉฐ ํด๋์ค, ํจ์, import ๊ด๊ณ ์ ์"""
for node in ast.walk(tree):
# 1. Imports ๋ถ์
if isinstance(node, ast.Import):
for alias in node.names:
# ํ๋ก์ ํธ ๋ด๋ถ ๋ชจ๋๋ก ์ถ์ ๋๋ import๋ง ์ฐ๊ฒฐ
target_module = alias.name.split('.')[0]
self._try_add_import_edge(file_rel_path, target_module)
elif isinstance(node, ast.ImportFrom):
if node.module:
target_module = node.module.split('.')[0]
self._try_add_import_edge(file_rel_path, target_module)
# 2. ํด๋์ค ์ ์ ๋ถ์
elif isinstance(node, ast.ClassDef):
class_id = f"{file_rel_path}::{node.name}"
self.add_node(class_id, node.name, "class", {
"file": file_rel_path,
"methods": [n.name for n in node.body if isinstance(n, ast.FunctionDef)]
})
# ํ์ผ -> ํด๋์ค ๋ถ๋ชจ-์์ ์ฐ๊ฒฐ
self.add_edge(file_rel_path, class_id, "defines")
# 3. ์ต์์ ํจ์ ์ ์ ๋ถ์
elif isinstance(node, ast.FunctionDef):
# ํด๋์ค ๋ฉ์๋๊ฐ ์๋ ๋
๋ฆฝ ํจ์๋ง ์ถ๊ฐ (ํด๋์ค ๋ฉ์๋๋ ํด๋์ค ๋
ธ๋ ์์ธ ์ ๋ณด์ ํฌํจ)
# ๋ถ๋ชจ๊ฐ ClassDef์ธ์ง ๊ฐ๋จํ ํ๋จํ๊ธฐ ์ํด walk๊ฐ ์๋ ๋จ๋จ๊ณ ๋งคํ์ด ์ข์ผ๋, ๊ฐ์ํ๋ฅผ ์ํด ์์ง
pass
def _try_add_import_edge(self, source_file: str, target_module: str):
"""์ํฌํธ ๋์ ๋ชจ๋์ด ํ๋ก์ ํธ ๋ด๋ถ์ ์ค์กดํ๋ ํ์ผ์ธ์ง ํ๋ณ ํ ๊ฐ์ ์ฐ๊ฒฐ"""
# 1. module_name.py ํํ ๊ฒ์ฌ
potential_py = f"{target_module}.py"
potential_dir_py = f"{target_module}/__init__.py"
for scanned in self.scanned_files:
if scanned == potential_py or scanned.endswith("/" + potential_py) or scanned == potential_dir_py:
self.add_edge(source_file, scanned, "imports")
return
def scan_markdown_docs(self):
"""ํ๋ก์ ํธ ๋ฃจํธ ๋ด Markdown ๋ฌธ์์ ๊ทธ ์์ ์ฐ๊ฒฐ๊ณ ๋ฆฌ(wiki ๋งํฌ, ํ์ผ ์ฐธ์กฐ) ๋ถ์"""
for root, _, files in os.walk(self.root_dir):
if any(p in root for p in [".git", "venv", "node_modules", ".agent"]):
continue
for file in files:
if not file.endswith(".md"):
continue
file_path = Path(root) / file
rel_path = file_path.relative_to(self.root_dir).as_posix()
# ๋ฌธ์ ๋
ธ๋ ์ถ๊ฐ
self.add_node(rel_path, file, "document", {"size": file_path.stat().st_size})
try:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
content = f.read()
# ๋ฌธ์ ๋ด๋ถ์ ํ ํ์ผ ์ฐธ์กฐ([[๋งํฌ]] ๋๋ ๋งํฌ๋ค์ด ํ์ผ ๋งํฌ) ํ์
# ๊ฐ๋จํ ํ์ผ๋ช
๋งค์นญ
for other_file in self.scanned_files:
other_name = Path(other_file).name
if other_name in content:
self.add_edge(rel_path, other_file, "references")
except Exception as e:
print(f"โ ๏ธ ๋ฌธ์ ์ค์บ ์คํจ [{rel_path}]: {e}")
def build(self) -> dict:
self.scan_python_files()
self.scan_markdown_docs()
return {
"nodes": list(self.nodes.values()),
"links": self.edges
}
def generate_interactive_html(graph_data: dict) -> str:
"""D3.js ๊ธฐ๋ฐ์ ๋ฉ์ง Force-Directed Graph UI ํ
ํ๋ฆฟ ์์ฑ"""
data_json = json.dumps(graph_data, ensure_ascii=False, indent=2)
html_template = f"""
๐ฑ Imperial Guild Knowledge Graph Portal
๋งํฌ๋ค์ด ๋ฌธ์ (Document)
๊ตฌ์กฐ์ ํด๋์ค (Class)
"""
return html_template
def upload_to_r2_live(json_path: Path, html_path: Path):
"""R2 Live ๋ฒํท์ ์ง์ ๊ทธ๋ํ ์ง๋ ์
๋ก๋"""
try:
# shadow_brain_core/web_server.py์ R2 Live ๋ก๋ ํจํด ์ฐธ์กฐ
core_dir = Path(__file__).parent.parent
project_root = core_dir.parent
sys.path.append(str(core_dir))
sys.path.append(str(project_root / "scripts" / "tools"))
from cloud_tool import R2_ACCOUNTS, _get_bw_credential
import boto3
from botocore.config import Config
print("๐ฎ R2 Live ์๊ฒฉ์ฆ๋ช
ํ๋ณด ์์...")
# live ์ค์ ์ถ์ถ
live_cfg = R2_ACCOUNTS.get("live")
if not live_cfg:
print("โ R2 Live ๊ณ์ ์ด cloud_tool.py์ ์ ์๋์ด ์์ง ์์ต๋๋ค.")
return
account_id = _get_bw_credential(live_cfg["account_id"])
access_key = _get_bw_credential(live_cfg["access_key_id"])
secret_key = _get_bw_credential(live_cfg["secret_access_key"])
# ๐ฑ [Fix] R2_ACCOUNTS["live"] ์ ๋ฒํท ํค๋ "bucket"(="cache") ์ด๋ค.
# ์๋ชป๋ ํค "bucket_name"(๊ธฐ๋ณธ "taemingames")์ ์ฐ๋ฉด ๊ณต๊ฐ ๋๋ฉ์ธ
# r2.taemingames.com(=cache ๋ฒํท)๊ณผ ๋ค๋ฅธ ๋ฒํท์ ์ฌ๋ผ๊ฐ 404 ๊ฐ ๋๋ค.
bucket_name = live_cfg.get("bucket", "cache")
if not all([account_id, access_key, secret_key]):
print("โ R2 Live ์๊ฒฉ์ฆ๋ช
์ด ๋น์ด ์์ด ์
๋ก๋๋ฅผ ๊ฑด๋๋๋๋ค.")
return
print("โก R2 Live S3 ํด๋ผ์ด์ธํธ ์ด๊ธฐํ ์ค...")
s3 = boto3.client(
"s3",
endpoint_url=f"https://{account_id}.r2.cloudflarestorage.com",
aws_access_key_id=access_key,
aws_secret_access_key=secret_key,
config=Config(signature_version="s3v4")
)
prefix = "cloud-obsidian/graph"
# json ์
๋ก๋
print(f"๐ {json_path.name} ์
๋ก๋ ์ค -> R2 Live/{prefix}/{json_path.name}")
s3.upload_file(
Filename=str(json_path),
Bucket=bucket_name,
Key=f"{prefix}/{json_path.name}",
ExtraArgs={"ContentType": "application/json"}
)
# html ์
๋ก๋
print(f"๐ {html_path.name} ์
๋ก๋ ์ค -> R2 Live/{prefix}/{html_path.name}")
s3.upload_file(
Filename=str(html_path),
Bucket=bucket_name,
Key=f"{prefix}/{html_path.name}",
ExtraArgs={"ContentType": "text/html"}
)
print("โ
R2 Live ์ง์ ๊ทธ๋ํ ์
๋ก๋ ์๋ฃ!")
except ImportError:
print("โ ๏ธ cloud_tool ๋๋ boto3 ๋ชจ๋์ ์ฐพ์ ์ ์์ด R2 ์
๋ก๋๋ฅผ ์๋ตํฉ๋๋ค.")
except Exception as e:
print(f"โ ๏ธ R2 Live ์
๋ก๋ ์คํจ: {e}")
def _capture_full_graph(project_root: Path) -> dict:
"""๐ฑ ํ ๊ทธ๋ํ (9749 UI ํธํ) ์บก์ฒ โ portal-heimdall/assets/knowledge_graph_9749/
9749 ๋ฐ์ด๋๋ฆฌ(`codebase-memory-mcp-ui-bin`)์ ๋ฐฑ์๋ API๋ฅผ ํธ์ถํด:
- /api/layout โ 13.5 MB ํ ๋
ธ๋/์ฃ์ง ์ขํ (137k ๋
ธ๋)
- /api/project-health โ ๋
ธ๋/์ฃ์ง/ํฌ๊ธฐ ๋ฉํ
- /api/adr โ ADR (Architecture Decision Records)
- /api/project โ ํ๋ก์ ํธ ๋ฉํ
- list_projects (RPC) โ ํ๋ก์ ํธ ๋ชฉ๋ก
- get_graph_schema โ ๋
ธ๋/์ฃ์ง ์คํค๋ง (์์, ๋ผ๋ฒจ, ์์ฑ)
Returns dict: {"success": bool, "message": str, "captures": {file: bytes}}
"""
import json as _json
import urllib.request
import urllib.error
import urllib.parse as _u
import subprocess as _sp
def _hermes_python():
"""boto3์ด ๋ณด์ฅ๋ hermes venv์ python3 ๊ฒฝ๋ก. ์์ผ๋ฉด ํ์ฌ ์ธํฐํ๋ฆฌํฐ."""
from pathlib import Path as _P
candidate = _P.home() / ".hermes" / "hermes-agent" / "venv" / "bin" / "python3"
return str(candidate) if candidate.exists() else sys.executable
result = {"success": False, "message": "", "captures": {}}
# 1) 9749 ๋ฐฑ์๋ ์ด์์๋์ง ํ์ธ
try:
urllib.request.urlopen("http://localhost:9749/api/processes", timeout=2).read()
except Exception as e:
result["message"] = f"9749 ๋ฐฑ์๋ ๋ฏธ๊ฐ๋ (localhost:9749): {e}"
return result
# 2) ํ๋ก์ ํธ ์ด๋ฆ ์กฐํ (list_projects MCP RPC)
def _list_projects():
proc = _sp.run(
["codebase-memory-mcp", "cli", "list_projects"],
capture_output=True, text=True, timeout=30,
)
out = proc.stdout.strip() # stderr log(level=infoโฆ)๊ฐ ์์ ๋ถ์ ์ ์์
start = out.find("{")
if start == -1:
raise ValueError(f"list_projects JSON ํ์ฑ ์คํจ: {out[:200]}")
return _json.loads(out[start:])["projects"]
try:
projects = _list_projects()
except FileNotFoundError:
result["message"] = "codebase-memory-mcp ๋ฐ์ด๋๋ฆฌ๋ฅผ PATH์์ ์ฐพ์ ์ ์์"
return result
except Exception as e:
result["message"] = f"list_projects ์คํจ: {e}"
return result
# ๐ฑ [์๊ฐ ์น์ ] ์ธ๋ฑ์ค๊ฐ ๋น์ด ์์ผ๋ฉด(9749 ์ฌ์์ ๋ฑ) ์บก์ฒ ์ ์ ์๋ ์ธ๋ฑ์ฑํ๋ค.
# ๊ณผ๊ฑฐ์ ์ฌ๊ธฐ์ 'ํ๋ก์ ํธ ์์'์ผ๋ก ์ค๋จ๋๋ค โ ์ง๋ ๊ฐฑ์ ๋ฒํผ์ด ์กฐ์ฉํ ์คํจํ๋ ๊ทผ๋ณธ ์์ธ.
if not projects:
print("โ ๏ธ 9749 ์ธ๋ฑ์ค๊ฐ ๋น์ด ์์ โ ์๋ ์ธ๋ฑ์ฑ ๊ฐ์ (์ ๋ถ ์์, persistence๋ก ์ฌ์์ ํ์๋ ์ ์ง)...")
try:
idx = _sp.run(
["codebase-memory-mcp", "cli", "index_repository",
_json.dumps({"repo_path": str(project_root), "mode": "full", "persistence": True})],
capture_output=True, text=True, timeout=1800,
)
tail = (idx.stdout or "").strip()[-200:]
print(f" ์ธ๋ฑ์ฑ ๊ฒฐ๊ณผ: {tail}")
projects = _list_projects()
except Exception as e:
result["message"] = f"์๋ ์ธ๋ฑ์ฑ ์คํจ: {e} (์๋: codebase-memory-mcp cli index_repository)"
return result
if not projects:
result["message"] = "์๋ ์ธ๋ฑ์ฑ ํ์๋ ํ๋ก์ ํธ๊ฐ ๋น์ด ์์ โ 9749 ๋ฐฑ์๋/์ ์ฅ์ ์ํ ํ์ธ ํ์"
return result
# ์ฐ์ ์์: project_root์ ์ผ์นํ๋ ํ๋ก์ ํธ
target = None
target_name = None
target_root = None
for p in projects:
if Path(p["root_path"]).resolve() == project_root.resolve():
target = p
target_name = p["name"]
target_root = p["root_path"]
break
if target is None:
# ์ผ์น ์์ผ๋ฉด ์ฒซ ํ๋ก์ ํธ
target = projects[0]
target_name = target["name"]
target_root = target["root_path"]
print(f"๐ฎ ํ ๊ทธ๋ํ ์บก์ฒ ๋์: {target_name} ({target['nodes']:,} ๋
ธ๋ / {target['edges']:,} ์ฃ์ง)")
# 3) ์บก์ฒ ๋๋ ํฐ๋ฆฌ
capture_dir = project_root / "portal-heimdall" / "assets" / "knowledge_graph_9749" / "data"
capture_dir.mkdir(parents=True, exist_ok=True)
captures = {}
def _fetch_json(path: str, dest_name: str):
"""9749 API ํ ์ค GET โ ํ์ผ ์ ์ฅ."""
try:
with urllib.request.urlopen(f"http://localhost:9749{path}", timeout=60) as r:
data = r.read()
(capture_dir / dest_name).write_bytes(data)
captures[dest_name] = len(data)
print(f" โ {dest_name:24} {len(data):>10,} bytes")
except Exception as e:
print(f" โ {dest_name} ์บก์ฒ ์คํจ: {e}")
enc = _u.quote(target_name, safe="")
enc_root = _u.quote(target_root, safe="")
_fetch_json(f"/api/layout?project={enc}&max_nodes=50000", "layout.json")
_fetch_json(f"/api/project-health?name={enc}", "health.json")
_fetch_json(f"/api/adr?project={enc}", "adr.json")
_fetch_json(f"/api/project?name={enc}", "project.json")
# list_projects.json๋ ๋๊ธฐํ (UI ์ฒซ ํธ์ถ์ฉ)
(capture_dir / "list_projects.json").write_text(_json.dumps({"projects": projects}, ensure_ascii=False), encoding="utf-8")
captures["list_projects.json"] = (capture_dir / "list_projects.json").stat().st_size
print(f" โ list_projects.json {captures['list_projects.json']:>10,} bytes")
# 4) get_graph_schema โ CLI ํธ์ถ
try:
proc = _sp.run(
["codebase-memory-mcp", "cli", "get_graph_schema", f'{{"project":"{target_name}"}}'],
capture_output=True, text=True, timeout=30,
)
out = proc.stdout.strip()
start = out.find("{")
if start != -1:
(capture_dir / "schema.json").write_text(out[start:], encoding="utf-8")
captures["schema.json"] = (capture_dir / "schema.json").stat().st_size
print(f" โ schema.json {captures['schema.json']:>10,} bytes")
except Exception as e:
print(f" โ schema ์บก์ฒ ์คํจ: {e}")
if not captures.get("layout.json"):
result["message"] = "layout.json ์บก์ฒ ์คํจ (9749 ์ธ๋ฑ์ค ๋ฏธ์์ฑ ๊ฐ๋ฅ์ฑ)"
return result
# 5) R2 ํธ์ โ hermes venv์ python์ผ๋ก ํธ์ถ (boto3 ๋ณด์ฅ)
project_root_for_sync = project_root
scripts_dir = project_root_for_sync / "scripts"
sync_script = scripts_dir / "r2_sync_full_graph.py"
if sync_script.exists():
print(f"๐ R2 ํ ๊ทธ๋ํ ํธ์ ์์: {sync_script} --force")
proc = _sp.run(
[_hermes_python(), str(sync_script), "--force"],
cwd=str(project_root_for_sync),
capture_output=True, text=True, timeout=300,
)
if proc.returncode == 0:
print("โ
R2 ํ ๊ทธ๋ํ ํธ์ ์๋ฃ")
else:
print(f"โ R2 ํธ์ ์ผ๋ถ ์คํจ (returncode={proc.returncode}):")
print(proc.stdout[-1000:])
print(proc.stderr[-500:])
else:
print(f"โ R2 ๋๊ธฐํ ์คํฌ๋ฆฝํธ ์์: {sync_script}")
result["success"] = True
result["message"] = f"{len(captures)} ํ์ผ ์บก์ฒ, R2 ํธ์ ์๋ฃ"
result["captures"] = captures
result["project_name"] = target_name
result["project_root"] = target_root
return result
def main():
parser = argparse.ArgumentParser(description="๐ฑ Graphify ์ง์ ๊ทธ๋ํ ์์ง")
parser.add_argument("--upload", action="store_true", help="๋ผ์ดํธ ๊ทธ๋ํ R2 Live ๋ฒํท ์
๋ก๋")
parser.add_argument("--full-upload", action="store_true", help="ํ ๊ทธ๋ํ (9749 UI) 137k ๋
ธ๋ ์บก์ฒ + R2 ์
๋ก๋")
args = parser.parse_args()
# ์คํฌ๋ฆฝํธ ์คํ ์์น๊ฐ scripts/ ์ฌ๋ ๋ถ๋ชจ shadow_brain_core๋ฅผ ๊ฐ๋ฆฌํค๋๋ก ์ค์
script_dir = Path(__file__).parent.resolve()
core_dir = script_dir.parent
print(f"๐ฑ Graphify: [{core_dir}] ์์ค ๋ถ์์ ๊ฐ์ํฉ๋๋ค...")
builder = ProjectGraphBuilder(core_dir)
graph_data = builder.build()
# ๋ก์ปฌ ์ ์ฅ ๊ฒฝ๋ก
output_dir = core_dir / "static" / "graph"
output_dir.mkdir(parents=True, exist_ok=True)
json_path = output_dir / "graph.json"
html_path = output_dir / "graph.html"
# json ์ฐ๊ธฐ
with open(json_path, "w", encoding="utf-8") as f:
json.dump(graph_data, f, ensure_ascii=False, indent=2)
# html ์ฐ๊ธฐ
html_content = generate_interactive_html(graph_data)
with open(html_path, "w", encoding="utf-8") as f:
f.write(html_content)
print(f"โ
๋ถ์ ์๋ฃ! ์ด ๋
ธ๋: {len(graph_data['nodes'])}, ๊ฐ์ : {len(graph_data['links'])}")
print(f"๐พ ๋ก์ปฌ ๋ฐ์ดํฐ ์ ์ฅ: {json_path}")
print(f"๐พ ๋ก์ปฌ ์๊ฐํ ์ ์ฅ: {html_path}")
# ์
๋ก๋ ์ฒ๋ฆฌ
if args.upload:
upload_to_r2_live(json_path, html_path)
# ํ ๊ทธ๋ํ ์บก์ฒ + R2 ํธ์
if args.full_upload:
# graphify๋ก ์ธ๋ฑ์ฑํ ํ๋ก์ ํธ ๋ฃจํธ = shadow_brain_core์ ๋ถ๋ชจ (jarvis_taemin)
jarvis_root = core_dir.parent
print()
print("๐ฑ ํ ๊ทธ๋ํ (9749 UI ํธํ) ์บก์ฒ ์์...")
result = _capture_full_graph(jarvis_root)
if result["success"]:
print(f"โ
ํ ๊ทธ๋ํ ๊ฐฑ์ ์๋ฃ: {result['message']}")
else:
print(f"โ ํ ๊ทธ๋ํ ๊ฐฑ์ ์คํจ: {result['message']}")
if __name__ == "__main__":
main()