Initial commit: 25 scripts Python sauvegardés
Date: 2026-08-28 11:43:34 EDT
This commit is contained in:
@@ -0,0 +1,435 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
extract-domain-context.py — Lightweight codebase scanner for domain knowledge extraction.
|
||||
|
||||
Scans a project directory and produces a structured JSON context file that the
|
||||
domain-analyzer agent uses to identify business domains, flows, and steps.
|
||||
|
||||
Usage:
|
||||
python extract-domain-context.py <project-root>
|
||||
|
||||
Output:
|
||||
<ua-dir>/intermediate/domain-context.json, where <ua-dir> is `.ua/` (or
|
||||
legacy `.understand-anything/` when that directory already exists).
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def resolve_ua_dir(root: Path) -> Path:
|
||||
"""Mirror core resolveUaDir: legacy .understand-anything/ wins if present."""
|
||||
legacy = root / ".understand-anything"
|
||||
return legacy if legacy.is_dir() else root / ".ua"
|
||||
|
||||
# ── Configuration ──────────────────────────────────────────────────────────
|
||||
|
||||
MAX_FILE_TREE_DEPTH = 6
|
||||
MAX_FILES_PER_DIR = 50
|
||||
MAX_FILES_TOTAL = 5000
|
||||
MAX_SAMPLED_FILES = 40
|
||||
MAX_LINES_PER_FILE = 80
|
||||
MAX_ENTRY_POINTS = 200
|
||||
MAX_OUTPUT_BYTES = 512 * 1024 # 512 KB — keeps output within agent context limits
|
||||
|
||||
# File extensions we care about for domain analysis
|
||||
SOURCE_EXTENSIONS = {
|
||||
".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs",
|
||||
".py", ".pyi",
|
||||
".go",
|
||||
".rs",
|
||||
".java", ".kt", ".scala",
|
||||
".rb",
|
||||
".cs",
|
||||
".php",
|
||||
".swift",
|
||||
".c", ".cpp", ".h", ".hpp",
|
||||
".ex", ".exs",
|
||||
".hs",
|
||||
".lua",
|
||||
".r", ".R",
|
||||
}
|
||||
|
||||
# Directories to always skip
|
||||
SKIP_DIRS = {
|
||||
"node_modules", ".git", ".svn", ".hg", "__pycache__", ".tox",
|
||||
"venv", ".venv", "env", ".env", "dist", "build", "out", ".next",
|
||||
".nuxt", "target", "vendor", ".idea", ".vscode", "coverage",
|
||||
".understand-anything", ".ua", ".pytest_cache", ".mypy_cache",
|
||||
"Pods", "DerivedData", ".gradle", "bin", "obj",
|
||||
}
|
||||
|
||||
# Files that reveal project metadata
|
||||
METADATA_FILES = [
|
||||
"package.json", "Cargo.toml", "go.mod", "pyproject.toml",
|
||||
"setup.py", "setup.cfg", "pom.xml", "build.gradle",
|
||||
"Gemfile", "composer.json", "mix.exs", "Makefile",
|
||||
"docker-compose.yml", "docker-compose.yaml",
|
||||
"README.md", "README.rst", "README.txt", "README",
|
||||
]
|
||||
|
||||
# ── Entry point detection patterns ─────────────────────────────────────────
|
||||
|
||||
ENTRY_POINT_PATTERNS: list[tuple[str, str, re.Pattern[str]]] = [
|
||||
# HTTP routes
|
||||
("http", "Express/Koa route", re.compile(
|
||||
r"""(?:app|router|server)\s*\.\s*(?:get|post|put|patch|delete|all|use)\s*\(\s*['"](/[^'"]*?)['"]""",
|
||||
re.IGNORECASE,
|
||||
)),
|
||||
("http", "Decorator route (Flask/FastAPI/NestJS)", re.compile(
|
||||
r"""@(?:app\.)?(?:route|get|post|put|patch|delete|api_view|RequestMapping|GetMapping|PostMapping)\s*\(\s*['"](/[^'"]*?)['"]""",
|
||||
re.IGNORECASE,
|
||||
)),
|
||||
("http", "Next.js/Remix route handler", re.compile(
|
||||
r"""export\s+(?:async\s+)?function\s+(GET|POST|PUT|PATCH|DELETE|HEAD|OPTIONS)\b""",
|
||||
)),
|
||||
# CLI
|
||||
("cli", "CLI command", re.compile(
|
||||
r"""\.command\s*\(\s*['"]([\w\-:]+)['"]""",
|
||||
)),
|
||||
("cli", "argparse subparser", re.compile(
|
||||
r"""add_parser\s*\(\s*['"]([\w\-]+)['"]""",
|
||||
)),
|
||||
# Event handlers
|
||||
("event", "Event listener", re.compile(
|
||||
r"""\.on\s*\(\s*['"]([\w\-:.]+)['"]""",
|
||||
)),
|
||||
("event", "Event subscriber decorator", re.compile(
|
||||
r"""@(?:EventHandler|Subscribe|Listener|on_event)\s*\(\s*['"]([\w\-:.]+)['"]""",
|
||||
)),
|
||||
# Cron / scheduled
|
||||
("cron", "Cron schedule", re.compile(
|
||||
r"""@?(?:Cron|Schedule|Scheduled|crontab)\s*\(\s*['"]([^'"]+)['"]""",
|
||||
re.IGNORECASE,
|
||||
)),
|
||||
# GraphQL
|
||||
("http", "GraphQL resolver", re.compile(
|
||||
r"""@(?:Query|Mutation|Subscription|Resolver)\s*\(""",
|
||||
)),
|
||||
# gRPC (only in .proto files — handled by file extension check below)
|
||||
("http", "gRPC service", re.compile(
|
||||
r"""^service\s+(\w+)\s*\{""", re.MULTILINE,
|
||||
)),
|
||||
# Exported handlers (generic)
|
||||
("manual", "Exported handler", re.compile(
|
||||
r"""export\s+(?:async\s+)?function\s+(handle\w+|process\w+|on\w+)\b""",
|
||||
)),
|
||||
]
|
||||
|
||||
|
||||
# ── Gitignore support ──────────────────────────────────────────────────────
|
||||
|
||||
def parse_gitignore(project_root: Path) -> list[re.Pattern[str]]:
|
||||
"""Parse .gitignore into a list of compiled regex patterns."""
|
||||
gitignore = project_root / ".gitignore"
|
||||
patterns: list[re.Pattern[str]] = []
|
||||
if not gitignore.exists():
|
||||
return patterns
|
||||
|
||||
for line in gitignore.read_text(encoding="utf-8", errors="replace").splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
# Convert glob to regex (simplified)
|
||||
regex = line.replace(".", r"\.").replace("**/", "(.*/)?").replace("*", "[^/]*").replace("?", "[^/]")
|
||||
if line.endswith("/"):
|
||||
regex = regex.rstrip("/") + "(/|$)"
|
||||
try:
|
||||
patterns.append(re.compile(regex))
|
||||
except re.error as e:
|
||||
print(f"Warning: skipping invalid gitignore pattern '{line}': {e}", file=sys.stderr)
|
||||
return patterns
|
||||
|
||||
|
||||
def is_ignored(rel_path: str, gitignore_patterns: list[re.Pattern[str]]) -> bool:
|
||||
"""Check if a relative path matches any gitignore pattern."""
|
||||
for pattern in gitignore_patterns:
|
||||
if pattern.search(rel_path):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# ── File tree scanner ──────────────────────────────────────────────────────
|
||||
|
||||
def scan_file_tree(
|
||||
root: Path,
|
||||
gitignore_patterns: list[re.Pattern[str]],
|
||||
max_depth: int = MAX_FILE_TREE_DEPTH,
|
||||
) -> list[str]:
|
||||
"""Return a flat list of relative file paths (source files only)."""
|
||||
result: list[str] = []
|
||||
|
||||
def _walk(dir_path: Path, depth: int) -> None:
|
||||
if depth > max_depth or len(result) >= MAX_FILES_TOTAL:
|
||||
return
|
||||
try:
|
||||
entries = sorted(dir_path.iterdir(), key=lambda e: (not e.is_dir(), e.name.lower()))
|
||||
except PermissionError:
|
||||
return
|
||||
|
||||
file_count = 0
|
||||
for entry in entries:
|
||||
if len(result) >= MAX_FILES_TOTAL:
|
||||
break
|
||||
# Skip symlinks to avoid infinite loops
|
||||
if entry.is_symlink():
|
||||
continue
|
||||
rel = str(entry.relative_to(root))
|
||||
if entry.is_dir():
|
||||
if entry.name in SKIP_DIRS:
|
||||
continue
|
||||
if is_ignored(rel + "/", gitignore_patterns):
|
||||
continue
|
||||
_walk(entry, depth + 1)
|
||||
elif entry.is_file():
|
||||
if file_count >= MAX_FILES_PER_DIR:
|
||||
break
|
||||
if entry.suffix not in SOURCE_EXTENSIONS:
|
||||
continue
|
||||
if is_ignored(rel, gitignore_patterns):
|
||||
continue
|
||||
result.append(rel)
|
||||
file_count += 1
|
||||
|
||||
_walk(root, 0)
|
||||
return result
|
||||
|
||||
|
||||
# ── Entry point detection ──────────────────────────────────────────────────
|
||||
|
||||
def detect_entry_points(root: Path, file_paths: list[str]) -> list[dict[str, Any]]:
|
||||
"""Scan source files for entry point patterns."""
|
||||
entry_points: list[dict[str, Any]] = []
|
||||
|
||||
# Skip test files and the extraction script itself
|
||||
test_patterns = re.compile(r"(?:\.test\.|\.spec\.|__tests__|_test\.py|test_\w+\.py|extract-domain-context\.py)")
|
||||
|
||||
for rel_path in file_paths:
|
||||
if len(entry_points) >= MAX_ENTRY_POINTS:
|
||||
break
|
||||
if test_patterns.search(rel_path):
|
||||
continue
|
||||
full_path = root / rel_path
|
||||
try:
|
||||
content = full_path.read_text(encoding="utf-8", errors="replace")
|
||||
except (OSError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
lines = content.splitlines()
|
||||
for entry_type, description, pattern in ENTRY_POINT_PATTERNS:
|
||||
for match in pattern.finditer(content):
|
||||
# Find line number
|
||||
line_no = content[:match.start()].count("\n") + 1
|
||||
# Extract a snippet (signature + a few lines)
|
||||
start = max(0, line_no - 1)
|
||||
end = min(len(lines), start + 5)
|
||||
snippet = "\n".join(lines[start:end])
|
||||
|
||||
entry_points.append({
|
||||
"file": rel_path,
|
||||
"line": line_no,
|
||||
"type": entry_type,
|
||||
"description": description,
|
||||
"match": match.group(0)[:120],
|
||||
"snippet": snippet[:300],
|
||||
})
|
||||
|
||||
if len(entry_points) >= MAX_ENTRY_POINTS:
|
||||
break
|
||||
if len(entry_points) >= MAX_ENTRY_POINTS:
|
||||
break
|
||||
|
||||
return entry_points
|
||||
|
||||
|
||||
# ── File signatures ────────────────────────────────────────────────────────
|
||||
|
||||
def extract_file_signatures(root: Path, file_paths: list[str]) -> list[dict[str, Any]]:
|
||||
"""Extract exports and imports from each file (lightweight)."""
|
||||
signatures: list[dict[str, Any]] = []
|
||||
|
||||
# Prioritize files likely to contain business logic
|
||||
priority_keywords = [
|
||||
"controller", "service", "handler", "router", "route", "api",
|
||||
"model", "entity", "repository", "usecase", "use_case",
|
||||
"command", "query", "event", "subscriber", "listener",
|
||||
"middleware", "guard", "interceptor", "resolver",
|
||||
"workflow", "flow", "process", "pipeline", "job", "task",
|
||||
]
|
||||
|
||||
def priority_score(path: str) -> int:
|
||||
lower = path.lower()
|
||||
score = 0
|
||||
for kw in priority_keywords:
|
||||
if kw in lower:
|
||||
score += 1
|
||||
return score
|
||||
|
||||
sorted_paths = sorted(file_paths, key=priority_score, reverse=True)
|
||||
|
||||
for rel_path in sorted_paths[:MAX_SAMPLED_FILES]:
|
||||
full_path = root / rel_path
|
||||
try:
|
||||
content = full_path.read_text(encoding="utf-8", errors="replace")
|
||||
except (OSError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
lines = content.splitlines()[:MAX_LINES_PER_FILE]
|
||||
truncated = "\n".join(lines)
|
||||
|
||||
# Extract exports (JS/TS)
|
||||
exports = re.findall(
|
||||
r"export\s+(?:default\s+)?(?:async\s+)?(?:function|class|const|let|var|interface|type|enum)\s+(\w+)",
|
||||
truncated,
|
||||
)
|
||||
# Extract exports (Python)
|
||||
if not exports:
|
||||
exports = re.findall(r"^(?:def|class)\s+(\w+)", truncated, re.MULTILINE)
|
||||
|
||||
# Extract imports (first 20)
|
||||
imports = re.findall(
|
||||
r"""(?:import\s+.*?from\s+['"]([^'"]+)['"]|from\s+([\w.]+)\s+import)""",
|
||||
truncated,
|
||||
)
|
||||
import_list = [m[0] or m[1] for m in imports][:20]
|
||||
|
||||
signatures.append({
|
||||
"file": rel_path,
|
||||
"exports": exports[:20],
|
||||
"imports": import_list,
|
||||
"lines": len(content.splitlines()),
|
||||
"preview": truncated[:500],
|
||||
})
|
||||
|
||||
return signatures
|
||||
|
||||
|
||||
# ── Metadata extraction ────────────────────────────────────────────────────
|
||||
|
||||
def extract_metadata(root: Path) -> dict[str, Any]:
|
||||
"""Read project metadata files."""
|
||||
metadata: dict[str, Any] = {}
|
||||
|
||||
for filename in METADATA_FILES:
|
||||
filepath = root / filename
|
||||
if not filepath.exists():
|
||||
continue
|
||||
try:
|
||||
content = filepath.read_text(encoding="utf-8", errors="replace")
|
||||
except (OSError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
if filename == "package.json":
|
||||
try:
|
||||
pkg = json.loads(content)
|
||||
metadata["package.json"] = {
|
||||
"name": pkg.get("name"),
|
||||
"description": pkg.get("description"),
|
||||
"scripts": list((pkg.get("scripts") or {}).keys()),
|
||||
"dependencies": list((pkg.get("dependencies") or {}).keys()),
|
||||
"devDependencies": list((pkg.get("devDependencies") or {}).keys()),
|
||||
}
|
||||
except json.JSONDecodeError:
|
||||
metadata["package.json"] = content[:500]
|
||||
elif filename.endswith((".md", ".rst", ".txt")) or filename == "README":
|
||||
metadata[filename] = content[:2000]
|
||||
elif filename.endswith((".toml", ".cfg", ".mod")):
|
||||
metadata[filename] = content[:1000]
|
||||
elif filename.endswith((".json", ".yml", ".yaml", ".xml", ".gradle")):
|
||||
metadata[filename] = content[:1000]
|
||||
|
||||
return metadata
|
||||
|
||||
|
||||
# ── Main ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def _truncate_to_fit(context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Progressively trim context sections to stay under MAX_OUTPUT_BYTES."""
|
||||
output = json.dumps(context, indent=2)
|
||||
if len(output.encode()) <= MAX_OUTPUT_BYTES:
|
||||
return context
|
||||
|
||||
# 1. Trim file tree to just a count
|
||||
context["fileTree"] = context["fileTree"][:200]
|
||||
output = json.dumps(context, indent=2)
|
||||
if len(output.encode()) <= MAX_OUTPUT_BYTES:
|
||||
return context
|
||||
|
||||
# 2. Trim previews in signatures
|
||||
for sig in context.get("fileSignatures", []):
|
||||
sig["preview"] = sig["preview"][:200]
|
||||
output = json.dumps(context, indent=2)
|
||||
if len(output.encode()) <= MAX_OUTPUT_BYTES:
|
||||
return context
|
||||
|
||||
# 3. Trim snippets in entry points
|
||||
for ep in context.get("entryPoints", []):
|
||||
ep["snippet"] = ep["snippet"][:100]
|
||||
output = json.dumps(context, indent=2)
|
||||
if len(output.encode()) <= MAX_OUTPUT_BYTES:
|
||||
return context
|
||||
|
||||
# 4. Reduce number of signatures and entry points
|
||||
context["fileSignatures"] = context["fileSignatures"][:20]
|
||||
context["entryPoints"] = context["entryPoints"][:100]
|
||||
|
||||
return context
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: python extract-domain-context.py <project-root>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
project_root = Path(sys.argv[1]).resolve()
|
||||
if not project_root.is_dir():
|
||||
print(f"Error: {project_root} is not a directory", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
# Ensure output directory exists
|
||||
output_dir = resolve_ua_dir(project_root) / "intermediate"
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
output_path = output_dir / "domain-context.json"
|
||||
|
||||
print(f"Scanning {project_root} ...", file=sys.stderr)
|
||||
|
||||
gitignore_patterns = parse_gitignore(project_root)
|
||||
file_tree = scan_file_tree(project_root, gitignore_patterns)
|
||||
print(f" Found {len(file_tree)} source files", file=sys.stderr)
|
||||
|
||||
entry_points = detect_entry_points(project_root, file_tree)
|
||||
print(f" Detected {len(entry_points)} entry points", file=sys.stderr)
|
||||
|
||||
signatures = extract_file_signatures(project_root, file_tree)
|
||||
print(f" Extracted {len(signatures)} file signatures", file=sys.stderr)
|
||||
|
||||
metadata = extract_metadata(project_root)
|
||||
print(f" Read {len(metadata)} metadata files", file=sys.stderr)
|
||||
|
||||
context = {
|
||||
"projectRoot": str(project_root),
|
||||
"fileCount": len(file_tree),
|
||||
"fileTree": file_tree,
|
||||
"entryPoints": entry_points,
|
||||
"fileSignatures": signatures,
|
||||
"metadata": metadata,
|
||||
}
|
||||
|
||||
context = _truncate_to_fit(context)
|
||||
output = json.dumps(context, indent=2)
|
||||
output_path.write_text(output, encoding="utf-8")
|
||||
size_kb = len(output.encode()) / 1024
|
||||
print(f" Wrote {output_path} ({size_kb:.0f} KB)", file=sys.stderr)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,426 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Merge script for Karpathy-pattern knowledge graphs.
|
||||
|
||||
Combines the deterministic scan-manifest.json with LLM analysis batches
|
||||
(analysis-batch-*.json) into a final assembled knowledge graph.
|
||||
|
||||
Handles: entity deduplication, edge normalization, layer building from
|
||||
index.md categories, tour generation from index.md section ordering.
|
||||
|
||||
Usage:
|
||||
python merge-knowledge-graph.py <wiki-directory>
|
||||
|
||||
Output:
|
||||
Writes assembled-graph.json to <wiki-directory>/<ua-dir>/intermediate/, where
|
||||
<ua-dir> is `.ua/` (or legacy `.understand-anything/` when that directory
|
||||
already exists).
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def resolve_ua_dir(root: Path) -> Path:
|
||||
"""Mirror core resolveUaDir: legacy .understand-anything/ wins if present."""
|
||||
legacy = root / ".understand-anything"
|
||||
return legacy if legacy.is_dir() else root / ".ua"
|
||||
|
||||
|
||||
def _find_markdown_case_insensitive(parent: Path, name: str) -> Path:
|
||||
"""Resolve a known markdown filename case-insensitively within one directory.
|
||||
|
||||
Mirrors find_markdown_case_insensitive in parse-knowledge-base.py (the
|
||||
scripts are standalone, so the helper is duplicated): an exact match
|
||||
always wins; otherwise the first case-insensitive sibling is returned.
|
||||
"""
|
||||
candidate = parent / name
|
||||
if candidate.is_file():
|
||||
return candidate
|
||||
if not parent.is_dir():
|
||||
return candidate
|
||||
wanted = name.lower()
|
||||
for child in sorted(parent.iterdir()):
|
||||
if child.is_file() and child.name.lower() == wanted:
|
||||
return child
|
||||
return candidate
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Canonical type sets (must match core/src/types.ts)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
VALID_NODE_TYPES = {
|
||||
"article", "entity", "topic", "claim", "source",
|
||||
# Codebase types (for cross-compatibility)
|
||||
"file", "function", "class", "module", "concept",
|
||||
"config", "document", "service", "table", "endpoint",
|
||||
"pipeline", "schema", "resource", "domain", "flow", "step",
|
||||
}
|
||||
|
||||
VALID_EDGE_TYPES = {
|
||||
"cites", "contradicts", "builds_on", "exemplifies",
|
||||
"categorized_under", "authored_by", "related", "similar_to",
|
||||
# Codebase types
|
||||
"imports", "exports", "contains", "inherits", "implements",
|
||||
"calls", "subscribes", "publishes", "middleware",
|
||||
"reads_from", "writes_to", "transforms", "validates",
|
||||
"depends_on", "tested_by", "configures",
|
||||
"deploys", "serves", "provisions", "triggers",
|
||||
"migrates", "documents", "routes", "defines_schema",
|
||||
"contains_flow", "flow_step", "cross_domain",
|
||||
}
|
||||
|
||||
NODE_TYPE_ALIASES = {
|
||||
"note": "article", "page": "article", "wiki_page": "article",
|
||||
"person": "entity", "actor": "entity", "organization": "entity",
|
||||
"tag": "topic", "category": "topic", "theme": "topic",
|
||||
"assertion": "claim", "decision": "claim", "thesis": "claim",
|
||||
"reference": "source", "raw": "source", "paper": "source",
|
||||
}
|
||||
|
||||
EDGE_TYPE_ALIASES = {
|
||||
"references": "cites", "cites_source": "cites",
|
||||
"conflicts_with": "contradicts", "disagrees_with": "contradicts",
|
||||
"refines": "builds_on", "elaborates": "builds_on",
|
||||
"illustrates": "exemplifies", "instance_of": "exemplifies", "example_of": "exemplifies",
|
||||
"belongs_to": "categorized_under", "tagged_with": "categorized_under",
|
||||
"written_by": "authored_by", "created_by": "authored_by",
|
||||
"relates_to": "related", "related_to": "related",
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Normalization
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def normalize_node_type(t: str) -> str:
|
||||
t = t.lower().strip()
|
||||
return NODE_TYPE_ALIASES.get(t, t)
|
||||
|
||||
|
||||
def normalize_edge_type(t: str) -> str:
|
||||
t = t.lower().strip()
|
||||
return EDGE_TYPE_ALIASES.get(t, t)
|
||||
|
||||
|
||||
def normalize_entity_name(name: str) -> str:
|
||||
"""Normalize entity names for deduplication."""
|
||||
return re.sub(r'\s+', ' ', name.strip().lower())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Merge pipeline
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def merge(root: Path) -> dict:
|
||||
intermediate = resolve_ua_dir(root) / "intermediate"
|
||||
manifest_path = intermediate / "scan-manifest.json"
|
||||
|
||||
if not manifest_path.is_file():
|
||||
print(f"Error: {manifest_path} not found. Run parse-knowledge-base.py first.",
|
||||
file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Load scan manifest (deterministic base)
|
||||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||||
nodes = {n["id"]: n for n in manifest["nodes"]}
|
||||
edges = list(manifest["edges"])
|
||||
|
||||
report = {"base_nodes": len(nodes), "base_edges": len(edges),
|
||||
"batches": 0, "new_entities": 0, "new_claims": 0,
|
||||
"new_edges": 0, "deduped_entities": 0, "dropped_edges": 0}
|
||||
|
||||
# Load analysis batches
|
||||
batch_files = sorted(intermediate.glob("analysis-batch-*.json"))
|
||||
entity_name_map: dict[str, str] = {} # normalized_name → entity_id
|
||||
dedup_remap: dict[str, str] = {} # duplicate_id → canonical_id
|
||||
|
||||
for bf in batch_files:
|
||||
report["batches"] += 1
|
||||
try:
|
||||
batch = json.loads(bf.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, OSError) as e:
|
||||
print(f"[merge] Warning: Failed to load {bf.name}: {e}", file=sys.stderr)
|
||||
continue
|
||||
|
||||
# Process new nodes from LLM analysis
|
||||
for node in batch.get("nodes", []):
|
||||
node_type = normalize_node_type(node.get("type", ""))
|
||||
if node_type not in VALID_NODE_TYPES:
|
||||
print(f"[merge] Warning: Unknown node type '{node.get('type')}' — skipping",
|
||||
file=sys.stderr)
|
||||
continue
|
||||
|
||||
node["type"] = node_type
|
||||
node_id = node.get("id", "")
|
||||
|
||||
# Entity deduplication — track remapping for edge fixup
|
||||
if node_type == "entity":
|
||||
norm_name = normalize_entity_name(node.get("name", ""))
|
||||
if norm_name in entity_name_map:
|
||||
# Map duplicate ID → canonical ID for edge remapping
|
||||
dedup_remap[node_id] = entity_name_map[norm_name]
|
||||
report["deduped_entities"] += 1
|
||||
continue
|
||||
entity_name_map[norm_name] = node_id
|
||||
report["new_entities"] += 1
|
||||
elif node_type == "claim":
|
||||
report["new_claims"] += 1
|
||||
|
||||
# Ensure required fields
|
||||
node.setdefault("summary", node.get("name", ""))
|
||||
node.setdefault("tags", [])
|
||||
node.setdefault("complexity", "simple")
|
||||
|
||||
nodes[node_id] = node
|
||||
|
||||
# Process new edges from LLM analysis
|
||||
for edge in batch.get("edges", []):
|
||||
edge_type = normalize_edge_type(edge.get("type", ""))
|
||||
if edge_type not in VALID_EDGE_TYPES:
|
||||
print(f"[merge] Warning: Unknown edge type '{edge.get('type')}' — "
|
||||
f"mapped to 'related'", file=sys.stderr)
|
||||
edge_type = "related"
|
||||
|
||||
edge["type"] = edge_type
|
||||
edge.setdefault("direction", "forward")
|
||||
edge.setdefault("weight", 0.5)
|
||||
|
||||
# Remap deduped entity IDs, then validate source/target exist
|
||||
src = dedup_remap.get(edge.get("source", ""), edge.get("source", ""))
|
||||
tgt = dedup_remap.get(edge.get("target", ""), edge.get("target", ""))
|
||||
edge["source"] = src
|
||||
edge["target"] = tgt
|
||||
if src in nodes and tgt in nodes:
|
||||
edges.append(edge)
|
||||
report["new_edges"] += 1
|
||||
else:
|
||||
report["dropped_edges"] += 1
|
||||
|
||||
# --- Deduplicate edges ---
|
||||
seen: set[tuple[str, str, str]] = set()
|
||||
final_edges = []
|
||||
for edge in edges:
|
||||
key = (edge["source"], edge["target"], edge["type"])
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
final_edges.append(edge)
|
||||
|
||||
# --- Build article→layer map from categories ---
|
||||
categories = manifest.get("categories", [])
|
||||
article_layer_map: dict[str, str] = {} # article_id → layer_id
|
||||
layer_members: dict[str, list[str]] = {} # layer_id → [node_ids]
|
||||
|
||||
for cat in categories:
|
||||
cat_name = cat["name"]
|
||||
cat_slug = cat_name.lower().replace(" ", "-")
|
||||
layer_id = f"layer:{cat_slug}"
|
||||
topic_id = f"topic:{cat_slug}"
|
||||
members = [e["source"] for e in final_edges
|
||||
if e["type"] == "categorized_under" and e["target"] == topic_id]
|
||||
if topic_id in nodes:
|
||||
members.append(topic_id)
|
||||
layer_members[layer_id] = members
|
||||
for mid in members:
|
||||
article_layer_map[mid] = layer_id
|
||||
|
||||
# --- Assign entity/claim nodes to their parent article's layer ---
|
||||
# Step 1: Build entity/claim → article mapping from edges
|
||||
child_to_article: dict[str, str] = {}
|
||||
for edge in final_edges:
|
||||
src_type = nodes.get(edge["source"], {}).get("type", "")
|
||||
tgt_type = nodes.get(edge["target"], {}).get("type", "")
|
||||
# If an article connects to an entity/claim, map the child to the article
|
||||
if src_type == "article" and tgt_type in ("entity", "claim"):
|
||||
child_to_article.setdefault(edge["target"], edge["source"])
|
||||
elif tgt_type == "article" and src_type in ("entity", "claim"):
|
||||
child_to_article.setdefault(edge["source"], edge["target"])
|
||||
|
||||
# Step 2: For orphan entities/claims, try to match by ID prefix
|
||||
# Build a reverse lookup: bare article name → full article ID
|
||||
# e.g., "concept-aaak-compression" → "article:concepts/concept-aaak-compression"
|
||||
bare_to_article: dict[str, str] = {}
|
||||
for nid in nodes:
|
||||
if nid.startswith("article:"):
|
||||
# Extract the bare filename from paths like "article:concepts/concept-foo"
|
||||
bare = nid.split("/")[-1] if "/" in nid else nid.replace("article:", "")
|
||||
bare_to_article[bare] = nid
|
||||
|
||||
for nid, node in nodes.items():
|
||||
if node["type"] in ("entity", "claim") and nid not in child_to_article:
|
||||
# e.g., "claim:concept-aaak-compression:not-zero-loss" → stem "concept-aaak-compression"
|
||||
# e.g., "entity:brain" → stem "brain"
|
||||
raw = nid.split(":", 1)[1] if ":" in nid else nid # "concept-aaak-compression:not-zero-loss"
|
||||
stem = raw.split(":")[0] # "concept-aaak-compression"
|
||||
|
||||
# Try exact bare name match first
|
||||
if stem in bare_to_article:
|
||||
child_to_article[nid] = bare_to_article[stem]
|
||||
else:
|
||||
# Try suffix/substring match against bare names
|
||||
# e.g., entity:brain → segment-brain, entity:mempalace → tool-mempalace
|
||||
matched = False
|
||||
for bare, aid in bare_to_article.items():
|
||||
if stem in bare or bare in stem:
|
||||
child_to_article[nid] = aid
|
||||
matched = True
|
||||
break
|
||||
# Also try: bare ends with -stem (e.g., "segment-brain" ends with "-brain")
|
||||
if bare.endswith(f"-{stem}") or bare.endswith(f"/{stem}"):
|
||||
child_to_article[nid] = aid
|
||||
matched = True
|
||||
break
|
||||
# Last resort: check if the node's name appears in any article's
|
||||
# name OR content (knowledgeMeta.content)
|
||||
if not matched and node.get("name"):
|
||||
node_name_lower = node["name"].lower()
|
||||
for aid, anode in nodes.items():
|
||||
if not aid.startswith("article:"):
|
||||
continue
|
||||
# Match against article name
|
||||
if node_name_lower in anode.get("name", "").lower():
|
||||
child_to_article[nid] = aid
|
||||
matched = True
|
||||
break
|
||||
# Match against article content (wikilinks or text)
|
||||
meta = anode.get("knowledgeMeta", {})
|
||||
content = (meta.get("content") or "").lower()
|
||||
if len(node_name_lower) >= 3 and node_name_lower in content:
|
||||
child_to_article[nid] = aid
|
||||
matched = True
|
||||
break
|
||||
|
||||
# Step 3: Place children into their parent article's layer
|
||||
for child_id, article_id in child_to_article.items():
|
||||
layer_id = article_layer_map.get(article_id)
|
||||
if layer_id and layer_id in layer_members:
|
||||
layer_members[layer_id].append(child_id)
|
||||
article_layer_map[child_id] = layer_id
|
||||
|
||||
# --- Build layers ---
|
||||
layers = []
|
||||
for cat in categories:
|
||||
cat_name = cat["name"]
|
||||
cat_slug = cat_name.lower().replace(" ", "-")
|
||||
layer_id = f"layer:{cat_slug}"
|
||||
members = list(dict.fromkeys(layer_members.get(layer_id, []))) # Deduplicate preserving order
|
||||
layers.append({
|
||||
"id": layer_id,
|
||||
"name": cat_name,
|
||||
"description": f"{cat_name} ({len(members)} nodes)",
|
||||
"nodeIds": members,
|
||||
})
|
||||
|
||||
# Assign uncategorized nodes to an "Other" layer
|
||||
categorized_ids = set()
|
||||
for layer in layers:
|
||||
categorized_ids.update(layer["nodeIds"])
|
||||
uncategorized = [nid for nid in nodes if nid not in categorized_ids]
|
||||
if uncategorized:
|
||||
layers.append({
|
||||
"id": "layer:other",
|
||||
"name": "Other",
|
||||
"description": f"Uncategorized nodes ({len(uncategorized)})",
|
||||
"nodeIds": uncategorized,
|
||||
})
|
||||
|
||||
# --- Build tour from index.md category ordering ---
|
||||
tour = []
|
||||
for i, cat in enumerate(categories):
|
||||
cat_slug = cat["name"].lower().replace(" ", "-")
|
||||
topic_id = f"topic:{cat_slug}"
|
||||
# Pick representative articles (up to 3 per category)
|
||||
members = [e["source"] for e in final_edges
|
||||
if e["type"] == "categorized_under" and e["target"] == topic_id][:3]
|
||||
if not members and topic_id in nodes:
|
||||
members = [topic_id]
|
||||
if members:
|
||||
tour.append({
|
||||
"order": i + 1,
|
||||
"title": cat["name"],
|
||||
"description": f"Explore the {cat['name']} section ({cat['count']} articles)",
|
||||
"nodeIds": members,
|
||||
})
|
||||
|
||||
# --- Detect project name ---
|
||||
project_name = root.name
|
||||
# Try to find a better name from index.md H1 (case-insensitively —
|
||||
# Index.md is a reasonable convention; exact lowercase wins if both exist)
|
||||
index_path = _find_markdown_case_insensitive(root / "wiki", "index.md")
|
||||
if not index_path.is_file():
|
||||
index_path = _find_markdown_case_insensitive(root, "index.md")
|
||||
if index_path.is_file():
|
||||
text = index_path.read_text(encoding="utf-8", errors="replace")
|
||||
h1_match = re.search(r"^#\s+(.+)$", text, re.MULTILINE)
|
||||
if h1_match:
|
||||
project_name = h1_match.group(1).strip()
|
||||
|
||||
# --- Assemble final graph ---
|
||||
graph = {
|
||||
"version": "1.0.0",
|
||||
"kind": "knowledge",
|
||||
"project": {
|
||||
"name": project_name,
|
||||
"languages": ["markdown"],
|
||||
"frameworks": ["karpathy-wiki"],
|
||||
"description": f"Knowledge graph for {project_name}",
|
||||
"analyzedAt": datetime.now(timezone.utc).isoformat(),
|
||||
"gitCommitHash": "",
|
||||
},
|
||||
"nodes": list(nodes.values()),
|
||||
"edges": final_edges,
|
||||
"layers": layers,
|
||||
"tour": tour,
|
||||
}
|
||||
|
||||
# Try to get git commit hash
|
||||
try:
|
||||
import subprocess
|
||||
result = subprocess.run(
|
||||
["git", "rev-parse", "HEAD"],
|
||||
capture_output=True, text=True, cwd=str(root), timeout=5
|
||||
)
|
||||
if result.returncode == 0:
|
||||
graph["project"]["gitCommitHash"] = result.stdout.strip()
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
|
||||
# Write output
|
||||
out_path = intermediate / "assembled-graph.json"
|
||||
out_path.write_text(json.dumps(graph, indent=2), encoding="utf-8")
|
||||
|
||||
# Report
|
||||
print(f"[merge] Input: {report['base_nodes']} scan nodes, "
|
||||
f"{report['base_edges']} scan edges, {report['batches']} analysis batches",
|
||||
file=sys.stderr)
|
||||
print(f"[merge] Added: {report['new_entities']} entities, "
|
||||
f"{report['new_claims']} claims, {report['new_edges']} edges "
|
||||
f"({report['deduped_entities']} deduped entities, "
|
||||
f"{report['dropped_edges']} dropped dangling edges)", file=sys.stderr)
|
||||
print(f"[merge] Output: {len(graph['nodes'])} nodes, {len(final_edges)} edges, "
|
||||
f"{len(layers)} layers, {len(tour)} tour steps", file=sys.stderr)
|
||||
print(f"[merge] Written: {out_path}", file=sys.stderr)
|
||||
|
||||
return graph
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: merge-knowledge-graph.py <wiki-directory>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
root = Path(sys.argv[1]).resolve()
|
||||
if not root.is_dir():
|
||||
print(f"Error: {root} is not a directory", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
merge(root)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,563 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Deterministic parser for Karpathy-pattern LLM wikis.
|
||||
|
||||
Detects the three-layer pattern (raw sources + wiki markdown + schema),
|
||||
extracts structure from markdown files, resolves wikilinks, and derives
|
||||
categories from index.md section headings.
|
||||
|
||||
Usage:
|
||||
python parse-knowledge-base.py <wiki-directory>
|
||||
|
||||
Output:
|
||||
Writes scan-manifest.json to <wiki-directory>/<ua-dir>/intermediate/, where
|
||||
<ua-dir> is `.ua/` (or legacy `.understand-anything/` when that directory
|
||||
already exists).
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def resolve_ua_dir(root: Path) -> Path:
|
||||
"""Mirror core resolveUaDir: legacy .understand-anything/ wins if present."""
|
||||
legacy = root / ".understand-anything"
|
||||
return legacy if legacy.is_dir() else root / ".ua"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Regex patterns
|
||||
# ---------------------------------------------------------------------------
|
||||
WIKILINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|([^\]]+))?\]\]")
|
||||
FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
|
||||
CODE_BLOCK_RE = re.compile(r"```(\w*)")
|
||||
HEADING_RE = re.compile(r"^(#{1,6})\s+(.+)$", re.MULTILINE)
|
||||
INDEX_SECTION_RE = re.compile(r"^##\s+(.+)$", re.MULTILINE)
|
||||
|
||||
# Files that are part of wiki infrastructure, not content articles
|
||||
INFRA_FILES = {"index.md", "log.md", "claude.md", "agents.md", "soul.md"}
|
||||
|
||||
|
||||
def find_markdown_case_insensitive(parent: Path, name: str) -> Path:
|
||||
"""Resolve a known markdown filename case-insensitively within one directory.
|
||||
|
||||
An exact match always wins (so `index.md` beats `Index.md` when both
|
||||
exist); otherwise the first case-insensitive sibling is returned. The
|
||||
unmatched candidate comes back as-is so callers can keep using
|
||||
`.is_file()` checks. Deliberately single-directory — no recursive fuzzy
|
||||
matching (see #342 non-goals).
|
||||
"""
|
||||
candidate = parent / name
|
||||
if candidate.is_file():
|
||||
return candidate
|
||||
if not parent.is_dir():
|
||||
return candidate
|
||||
wanted = name.lower()
|
||||
for child in sorted(parent.iterdir()):
|
||||
if child.is_file() and child.name.lower() == wanted:
|
||||
return child
|
||||
return candidate
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Detection: is this a Karpathy-pattern wiki?
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def detect_format(root: Path) -> dict:
|
||||
"""Detect if directory follows the Karpathy LLM wiki three-layer pattern."""
|
||||
signals = {
|
||||
"has_index": find_markdown_case_insensitive(root, "index.md").is_file()
|
||||
or find_markdown_case_insensitive(root / "wiki", "index.md").is_file(),
|
||||
"has_log": find_markdown_case_insensitive(root, "log.md").is_file()
|
||||
or find_markdown_case_insensitive(root / "wiki", "log.md").is_file(),
|
||||
"has_raw": (root / "raw").is_dir(),
|
||||
"has_schema": any(
|
||||
(root / f).is_file() or (root / "wiki" / f).is_file()
|
||||
for f in ["CLAUDE.md", "AGENTS.md"]
|
||||
),
|
||||
}
|
||||
|
||||
# Find the wiki root — could be the directory itself or a wiki/ subdirectory
|
||||
if (root / "wiki").is_dir():
|
||||
wiki_root = root / "wiki"
|
||||
else:
|
||||
wiki_root = root
|
||||
|
||||
# Count markdown files in the wiki root
|
||||
md_files = list(wiki_root.rglob("*.md"))
|
||||
signals["md_count"] = len(md_files)
|
||||
signals["wiki_root"] = str(wiki_root)
|
||||
|
||||
# Primary signal: has index.md + meaningful number of markdown files
|
||||
if signals["has_index"] and signals["md_count"] >= 3:
|
||||
signals["detected"] = True
|
||||
signals["format"] = "karpathy"
|
||||
else:
|
||||
signals["detected"] = False
|
||||
signals["format"] = "unknown"
|
||||
|
||||
return signals
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Markdown extraction helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def extract_frontmatter(text: str) -> dict:
|
||||
"""Extract YAML frontmatter as a simple key-value dict."""
|
||||
m = FRONTMATTER_RE.match(text)
|
||||
if not m:
|
||||
return {}
|
||||
fm = {}
|
||||
for line in m.group(1).split("\n"):
|
||||
if ":" in line:
|
||||
key, _, val = line.partition(":")
|
||||
fm[key.strip()] = val.strip().strip('"').strip("'")
|
||||
return fm
|
||||
|
||||
|
||||
def extract_wikilinks(text: str) -> list[dict]:
|
||||
"""Extract all [[target]] and [[target|display]] wikilinks."""
|
||||
links = []
|
||||
for m in WIKILINK_RE.finditer(text):
|
||||
links.append({
|
||||
"target": m.group(1).strip(),
|
||||
"display": m.group(2).strip() if m.group(2) else None,
|
||||
})
|
||||
return links
|
||||
|
||||
|
||||
def extract_headings(text: str) -> list[dict]:
|
||||
"""Extract all markdown headings with level and text."""
|
||||
return [
|
||||
{"level": len(m.group(1)), "text": m.group(2).strip()}
|
||||
for m in HEADING_RE.finditer(text)
|
||||
]
|
||||
|
||||
|
||||
def extract_code_blocks(text: str) -> list[str]:
|
||||
"""Extract languages from fenced code blocks."""
|
||||
return [m.group(1) for m in CODE_BLOCK_RE.finditer(text) if m.group(1)]
|
||||
|
||||
|
||||
def extract_first_paragraph(text: str) -> str:
|
||||
"""Extract the first non-empty paragraph after frontmatter and H1."""
|
||||
# Strip frontmatter
|
||||
stripped = FRONTMATTER_RE.sub("", text).strip()
|
||||
if not stripped:
|
||||
return ""
|
||||
lines = stripped.split("\n")
|
||||
|
||||
def _collect_paragraph(start_lines: list[str]) -> str:
|
||||
"""Collect the first paragraph from the given lines."""
|
||||
para: list[str] = []
|
||||
for s_raw in start_lines:
|
||||
s = s_raw.strip()
|
||||
if not s and not para:
|
||||
continue # Skip leading blank lines
|
||||
if not s and para:
|
||||
break # End of paragraph
|
||||
if s.startswith(">"):
|
||||
continue # Skip blockquotes
|
||||
if re.match(r"^[-*_]{3,}\s*$", s):
|
||||
continue # Skip horizontal rules
|
||||
if s.startswith("#"):
|
||||
if para:
|
||||
break # End paragraph at next heading
|
||||
continue # Skip headings before paragraph
|
||||
para.append(s)
|
||||
return " ".join(para)
|
||||
|
||||
# Try: find first paragraph after H1
|
||||
for i, line in enumerate(lines):
|
||||
if line.strip().startswith("# "):
|
||||
result = _collect_paragraph(lines[i + 1:])
|
||||
if result:
|
||||
if len(result) > 200:
|
||||
return result[:197] + "..."
|
||||
return result
|
||||
|
||||
# Fallback: no H1 found, take first paragraph from start
|
||||
result = _collect_paragraph(lines)
|
||||
if len(result) > 200:
|
||||
result = result[:197] + "..."
|
||||
return result or ""
|
||||
|
||||
|
||||
def extract_h1(text: str) -> str:
|
||||
"""Extract the first H1 heading."""
|
||||
for m in HEADING_RE.finditer(text):
|
||||
if len(m.group(1)) == 1:
|
||||
# Strip trailing wiki-style decorations like " — subtitle"
|
||||
return m.group(2).strip()
|
||||
return ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Index.md parsing — categories come from section headings
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def parse_index(index_path: Path) -> list[dict]:
|
||||
"""Parse index.md to extract categories from ## headings and their wikilinks."""
|
||||
if not index_path.is_file():
|
||||
return []
|
||||
text = index_path.read_text(encoding="utf-8", errors="replace")
|
||||
categories = []
|
||||
current_category = None
|
||||
|
||||
for line in text.split("\n"):
|
||||
# Detect ## section heading
|
||||
sec_match = re.match(r"^##\s+(.+)$", line)
|
||||
if sec_match:
|
||||
current_category = {
|
||||
"name": sec_match.group(1).strip(),
|
||||
"articles": [],
|
||||
}
|
||||
categories.append(current_category)
|
||||
continue
|
||||
|
||||
# Collect wikilinks under current section
|
||||
if current_category:
|
||||
for wl in WIKILINK_RE.finditer(line):
|
||||
current_category["articles"].append(wl.group(1).strip())
|
||||
|
||||
return categories
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Log.md parsing — extract operation timeline
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def parse_log(log_path: Path) -> list[dict]:
|
||||
"""Parse log.md to extract chronological entries."""
|
||||
if not log_path.is_file():
|
||||
return []
|
||||
text = log_path.read_text(encoding="utf-8", errors="replace")
|
||||
entries = []
|
||||
log_entry_re = re.compile(
|
||||
r"^##\s+\[(\d{4}-\d{2}-\d{2})\]\s+(\w+)\s*\|\s*(.+)$", re.MULTILINE
|
||||
)
|
||||
for m in log_entry_re.finditer(text):
|
||||
entries.append({
|
||||
"date": m.group(1),
|
||||
"operation": m.group(2),
|
||||
"title": m.group(3).strip(),
|
||||
})
|
||||
return entries
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main pipeline
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def build_name_to_stem_map(wiki_root: Path) -> dict[str, str]:
|
||||
"""Build a case-insensitive map from filename stem to relative stem path.
|
||||
|
||||
Full relative paths always map uniquely. Bare basenames map only when
|
||||
unambiguous — duplicate basenames are removed so they don't silently
|
||||
resolve to the wrong page.
|
||||
"""
|
||||
name_map: dict[str, str] = {}
|
||||
# Track which bare basenames appear more than once
|
||||
basename_counts: dict[str, int] = {}
|
||||
for md_file in wiki_root.rglob("*.md"):
|
||||
rel = md_file.relative_to(wiki_root)
|
||||
stem = rel.with_suffix("").as_posix() # e.g., "decisions/decision-foo"
|
||||
basename = md_file.stem # e.g., "decision-foo"
|
||||
# Full relative path always maps uniquely
|
||||
name_map[stem.lower()] = stem
|
||||
# Track basename for ambiguity detection
|
||||
key = basename.lower()
|
||||
basename_counts[key] = basename_counts.get(key, 0) + 1
|
||||
name_map[key] = stem
|
||||
|
||||
# Remove ambiguous basename entries (appear more than once)
|
||||
for key, count in basename_counts.items():
|
||||
if count > 1 and key in name_map:
|
||||
del name_map[key]
|
||||
|
||||
return name_map
|
||||
|
||||
|
||||
def resolve_wikilink(
|
||||
target: str,
|
||||
name_map: dict[str, str],
|
||||
node_ids: set[str] | None = None,
|
||||
root_prefix: str | None = None,
|
||||
) -> str | None:
|
||||
"""Resolve a wikilink target to an article node ID.
|
||||
|
||||
If node_ids is provided, only resolve to IDs that exist in the set.
|
||||
root_prefix is the article-root directory name (e.g. "wiki") — links
|
||||
written from the repository root include it ([[wiki/concepts/Index]])
|
||||
while name_map keys are relative to the article root, so such targets
|
||||
are tried both as written and with the prefix stripped.
|
||||
"""
|
||||
key = target.lower().strip()
|
||||
# Skip targets that are clearly not page names (shell flags, etc.)
|
||||
if key.startswith("-"):
|
||||
return None
|
||||
keys = [key]
|
||||
if root_prefix and key.startswith(root_prefix + "/"):
|
||||
keys.append(key[len(root_prefix) + 1:])
|
||||
for k in keys:
|
||||
stem = name_map.get(k)
|
||||
if stem:
|
||||
candidate = f"article:{stem}"
|
||||
# If we have a node set, verify the target exists
|
||||
if node_ids is not None and candidate not in node_ids:
|
||||
return None
|
||||
return candidate
|
||||
# Try without directory prefix
|
||||
for k in keys:
|
||||
for stored_key, stored_stem in name_map.items():
|
||||
if stored_key.endswith("/" + k) or stored_key == k:
|
||||
candidate = f"article:{stored_stem}"
|
||||
if node_ids is not None and candidate not in node_ids:
|
||||
return None
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def parse_wiki(root: Path) -> dict:
|
||||
"""Parse a Karpathy-pattern wiki and produce the scan manifest."""
|
||||
detection = detect_format(root)
|
||||
if not detection["detected"]:
|
||||
print(json.dumps({"error": "Not a Karpathy-pattern wiki", "detection": detection}),
|
||||
file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
wiki_root = Path(detection["wiki_root"])
|
||||
raw_root = root / "raw"
|
||||
|
||||
# Build name resolution map
|
||||
name_map = build_name_to_stem_map(wiki_root)
|
||||
|
||||
# Find index.md and log.md (case-insensitively — Index.md/Log.md are a
|
||||
# reasonable convention on case-sensitive filesystems)
|
||||
index_path = find_markdown_case_insensitive(wiki_root, "index.md")
|
||||
if not index_path.is_file():
|
||||
index_path = find_markdown_case_insensitive(root, "index.md")
|
||||
log_path = find_markdown_case_insensitive(wiki_root, "log.md")
|
||||
if not log_path.is_file():
|
||||
log_path = find_markdown_case_insensitive(root, "log.md")
|
||||
|
||||
# Parse index for categories
|
||||
categories = parse_index(index_path)
|
||||
log_entries = parse_log(log_path)
|
||||
|
||||
# Article ids are relative to wiki_root, but a root index.md commonly
|
||||
# links with the article-root prefix included ([[wiki/concepts/Index]]).
|
||||
# Register/resolve such targets both as written and prefix-stripped.
|
||||
root_prefix = wiki_root.name.lower() if wiki_root != root else None
|
||||
|
||||
# Build category lookup: wikilink target → category name
|
||||
category_lookup: dict[str, str] = {}
|
||||
for cat in categories:
|
||||
for article_target in cat["articles"]:
|
||||
t = article_target.lower()
|
||||
category_lookup[t] = cat["name"]
|
||||
if root_prefix and t.startswith(root_prefix + "/"):
|
||||
category_lookup[t[len(root_prefix) + 1:]] = cat["name"]
|
||||
|
||||
# --- Pre-compute article IDs (for edge resolution validation) ---
|
||||
# Only skip infra files at the wiki root level, not in subdirectories
|
||||
# (e.g., wiki/index.md is infra, but wiki/concepts/index.md is content)
|
||||
article_ids: set[str] = set()
|
||||
for md_file in sorted(wiki_root.rglob("*.md")):
|
||||
rel = md_file.relative_to(wiki_root)
|
||||
stem = rel.with_suffix("").as_posix()
|
||||
# Only filter infra files at root level (no parent directory)
|
||||
if rel.parent == Path(".") and rel.name.lower() in INFRA_FILES:
|
||||
continue
|
||||
article_ids.add(f"article:{stem}")
|
||||
|
||||
# --- Build article nodes ---
|
||||
nodes = []
|
||||
edges = []
|
||||
warnings = []
|
||||
stats = {"articles": 0, "sources": 0, "topics": 0, "wikilinks": 0, "unresolved": 0}
|
||||
|
||||
for md_file in sorted(wiki_root.rglob("*.md")):
|
||||
rel = md_file.relative_to(wiki_root)
|
||||
stem = rel.with_suffix("").as_posix()
|
||||
basename = md_file.stem
|
||||
|
||||
# Skip infrastructure files only at wiki root level
|
||||
if rel.parent == Path(".") and rel.name.lower() in INFRA_FILES:
|
||||
continue
|
||||
|
||||
text = md_file.read_text(encoding="utf-8", errors="replace")
|
||||
h1 = extract_h1(text)
|
||||
frontmatter = extract_frontmatter(text)
|
||||
wikilinks = extract_wikilinks(text)
|
||||
headings = extract_headings(text)
|
||||
code_langs = extract_code_blocks(text)
|
||||
summary = extract_first_paragraph(text)
|
||||
line_count = text.count("\n") + 1
|
||||
word_count = len(text.split())
|
||||
|
||||
# Derive category from index.md lookup
|
||||
category = category_lookup.get(basename.lower(), "")
|
||||
if not category:
|
||||
# Try stem match
|
||||
category = category_lookup.get(stem.lower(), "")
|
||||
|
||||
# Derive tags (deduplicated)
|
||||
tag_set: set[str] = set()
|
||||
if category:
|
||||
tag_set.add(category.lower())
|
||||
if rel.parent != Path("."):
|
||||
tag_set.add(str(rel.parent))
|
||||
fm_tags = frontmatter.get("tags", "")
|
||||
if fm_tags:
|
||||
tag_set.update(t.strip() for t in fm_tags.split(",") if t.strip())
|
||||
tags = sorted(tag_set)
|
||||
|
||||
# Complexity from wikilink density
|
||||
wl_count = len(wikilinks)
|
||||
if wl_count > 15:
|
||||
complexity = "complex"
|
||||
elif wl_count > 5:
|
||||
complexity = "moderate"
|
||||
else:
|
||||
complexity = "simple"
|
||||
|
||||
node_id = f"article:{stem}"
|
||||
nodes.append({
|
||||
"id": node_id,
|
||||
"type": "article",
|
||||
"name": h1 or basename,
|
||||
"filePath": str(rel),
|
||||
"summary": summary or f"Wiki article: {h1 or basename}",
|
||||
"tags": tags,
|
||||
"complexity": complexity,
|
||||
"knowledgeMeta": {
|
||||
"wikilinks": [wl["target"] for wl in wikilinks],
|
||||
**({"category": category} if category else {}),
|
||||
"content": text[:3000], # First 3000 chars for LLM analysis
|
||||
},
|
||||
})
|
||||
stats["articles"] += 1
|
||||
stats["wikilinks"] += wl_count
|
||||
|
||||
# Build edges from wikilinks (resolve against known article IDs)
|
||||
for wl in wikilinks:
|
||||
target_id = resolve_wikilink(wl["target"], name_map, article_ids, root_prefix)
|
||||
if target_id and target_id != node_id:
|
||||
edges.append({
|
||||
"source": node_id,
|
||||
"target": target_id,
|
||||
"type": "related",
|
||||
"direction": "forward",
|
||||
"weight": 0.7,
|
||||
})
|
||||
elif not target_id:
|
||||
warnings.append(f"Unresolved wikilink: [[{wl['target']}]] in {rel}")
|
||||
stats["unresolved"] += 1
|
||||
|
||||
# --- Build topic nodes from index.md categories ---
|
||||
for cat in categories:
|
||||
topic_id = f"topic:{cat['name'].lower().replace(' ', '-')}"
|
||||
nodes.append({
|
||||
"id": topic_id,
|
||||
"type": "topic",
|
||||
"name": cat["name"],
|
||||
"summary": f"Category from index: {cat['name']} ({len(cat['articles'])} articles)",
|
||||
"tags": ["category"],
|
||||
"complexity": "simple",
|
||||
})
|
||||
stats["topics"] += 1
|
||||
|
||||
# categorized_under edges (only resolve to known article nodes)
|
||||
for article_target in cat["articles"]:
|
||||
article_id = resolve_wikilink(article_target, name_map, article_ids, root_prefix)
|
||||
if article_id:
|
||||
edges.append({
|
||||
"source": article_id,
|
||||
"target": topic_id,
|
||||
"type": "categorized_under",
|
||||
"direction": "forward",
|
||||
"weight": 0.6,
|
||||
})
|
||||
|
||||
# --- Build source nodes from raw/ ---
|
||||
if raw_root.is_dir():
|
||||
for raw_file in sorted(raw_root.rglob("*")):
|
||||
if raw_file.is_file() and not raw_file.name.startswith("."):
|
||||
rel_raw = raw_file.relative_to(root)
|
||||
ext = raw_file.suffix.lower()
|
||||
size_kb = raw_file.stat().st_size / 1024
|
||||
source_id = f"source:{raw_file.relative_to(raw_root).with_suffix('')}"
|
||||
nodes.append({
|
||||
"id": source_id,
|
||||
"type": "source",
|
||||
"name": raw_file.name,
|
||||
"filePath": str(rel_raw),
|
||||
"summary": f"Raw source ({ext or 'unknown'}, {size_kb:.0f} KB)",
|
||||
"tags": ["raw", ext.lstrip(".") or "unknown"],
|
||||
"complexity": "simple",
|
||||
})
|
||||
stats["sources"] += 1
|
||||
|
||||
# --- Compute backlinks ---
|
||||
backlink_map: dict[str, list[str]] = {}
|
||||
for edge in edges:
|
||||
if edge["type"] == "related":
|
||||
target = edge["target"]
|
||||
source = edge["source"]
|
||||
backlink_map.setdefault(target, []).append(source)
|
||||
for node in nodes:
|
||||
if node["type"] == "article" and "knowledgeMeta" in node:
|
||||
bl = backlink_map.get(node["id"], [])
|
||||
node["knowledgeMeta"]["backlinks"] = bl
|
||||
|
||||
# --- Deduplicate edges ---
|
||||
seen_edges: set[tuple[str, str, str]] = set()
|
||||
deduped_edges = []
|
||||
for edge in edges:
|
||||
key = (edge["source"], edge["target"], edge["type"])
|
||||
if key not in seen_edges:
|
||||
seen_edges.add(key)
|
||||
deduped_edges.append(edge)
|
||||
|
||||
return {
|
||||
"format": "karpathy",
|
||||
"stats": stats,
|
||||
"categories": [{"name": c["name"], "count": len(c["articles"])} for c in categories],
|
||||
"logEntries": len(log_entries),
|
||||
"nodes": nodes,
|
||||
"edges": deduped_edges,
|
||||
"warnings": warnings[:50], # Cap warnings
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: parse-knowledge-base.py <wiki-directory>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
root = Path(sys.argv[1]).resolve()
|
||||
if not root.is_dir():
|
||||
print(f"Error: {root} is not a directory", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
manifest = parse_wiki(root)
|
||||
|
||||
# Write output
|
||||
out_dir = resolve_ua_dir(root) / "intermediate"
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
out_path = out_dir / "scan-manifest.json"
|
||||
out_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
|
||||
|
||||
# Report to stderr
|
||||
s = manifest["stats"]
|
||||
print(f"[parse] Karpathy wiki: {s['articles']} articles, {s['sources']} sources, "
|
||||
f"{s['topics']} topics, {s['wikilinks']} wikilinks "
|
||||
f"({s['unresolved']} unresolved)", file=sys.stderr)
|
||||
print(f"[parse] Output: {out_path}", file=sys.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,404 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
merge-subdomain-graphs.py — Merge subdomain knowledge-graph files into one.
|
||||
|
||||
Auto-discovers *knowledge-graph*.json files in the project's data dir
|
||||
(`.ua/`, or legacy `.understand-anything/` when that directory already exists)
|
||||
excluding knowledge-graph.json itself, loads the existing
|
||||
knowledge-graph.json as a base if present, and merges everything
|
||||
into a single knowledge-graph.json.
|
||||
|
||||
Usage:
|
||||
python merge-subdomain-graphs.py <project-root> [file1.json file2.json ...]
|
||||
|
||||
If no files are specified, auto-discovers subdomain graphs. The main
|
||||
knowledge-graph.json is loaded as a base but never as a discovery input
|
||||
(prevents self-merging on repeated runs).
|
||||
|
||||
Output:
|
||||
<ua-dir>/knowledge-graph.json
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def resolve_ua_dir(root: Path) -> Path:
|
||||
"""Mirror core resolveUaDir: legacy .understand-anything/ wins if present."""
|
||||
legacy = root / ".understand-anything"
|
||||
return legacy if legacy.is_dir() else root / ".ua"
|
||||
|
||||
# Edge types that carry the domain hierarchy. Dropping one of these changes
|
||||
# downstream graph traversal (unlike a routine `related` edge), so they are
|
||||
# warned about loudly and re-tried on later runs via merge-report.json —
|
||||
# subdomain graph files are cleaned up after assembly, so a drop would
|
||||
# otherwise be permanent even once the missing endpoint's subdomain arrives.
|
||||
STRUCTURAL_EDGE_TYPES = {"contains_flow", "flow_step", "cross_domain"}
|
||||
|
||||
|
||||
def _num(v: Any) -> float:
|
||||
"""Coerce a value to float for safe comparison (handles string weights)."""
|
||||
try:
|
||||
return float(v)
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
|
||||
def load_graph(path: Path) -> dict[str, Any] | None:
|
||||
"""Load and minimally validate a knowledge graph JSON file."""
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
print(f" Skipping {path.name}: {e}", file=sys.stderr)
|
||||
return None
|
||||
|
||||
# Must have at minimum nodes and edges arrays
|
||||
if not isinstance(data.get("nodes"), list) or not isinstance(data.get("edges"), list):
|
||||
print(f" Skipping {path.name}: missing nodes or edges array", file=sys.stderr)
|
||||
return None
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def merge_graphs(graphs: list[dict[str, Any]]) -> tuple[dict[str, Any], list[str], list[dict[str, Any]]]:
|
||||
"""Merge multiple knowledge graph dicts into one.
|
||||
|
||||
Returns (merged, report_lines, dropped_edges) — dropped_edges holds the
|
||||
full edge dicts that were removed for missing endpoints, each with an
|
||||
extra "missing" key naming which endpoint(s) were absent.
|
||||
"""
|
||||
|
||||
# ── Pattern counters for "Fixed" report ──────────────────────────
|
||||
node_dedup_by_type: Counter[str] = Counter()
|
||||
|
||||
# ── Detail lists for "Could not fix" report ──────────────────────
|
||||
unfixable: list[str] = []
|
||||
|
||||
total_input_nodes = sum(len(g.get("nodes", [])) for g in graphs)
|
||||
total_input_edges = sum(len(g.get("edges", [])) for g in graphs)
|
||||
|
||||
# ── Nodes: deduplicate by id, later occurrence wins ───────────────
|
||||
nodes_by_id: dict[str, dict] = {}
|
||||
for g in graphs:
|
||||
for node in g.get("nodes", []):
|
||||
nid = node.get("id")
|
||||
if not nid:
|
||||
unfixable.append(f"Node with no 'id' (name={node.get('name', '?')}, type={node.get('type', '?')})")
|
||||
continue
|
||||
if nid in nodes_by_id:
|
||||
node_type = node.get("type", "?")
|
||||
node_dedup_by_type[node_type] += 1
|
||||
nodes_by_id[nid] = node
|
||||
|
||||
# ── Edges: deduplicate by (source, target, type), higher weight wins
|
||||
edge_dedup_count = 0
|
||||
edges_by_key: dict[tuple[str, str, str], dict] = {}
|
||||
for g in graphs:
|
||||
for edge in g.get("edges", []):
|
||||
key = (edge.get("source", ""), edge.get("target", ""), edge.get("type", ""))
|
||||
existing = edges_by_key.get(key)
|
||||
if existing is None:
|
||||
edges_by_key[key] = edge
|
||||
else:
|
||||
edge_dedup_count += 1
|
||||
if _num(edge.get("weight", 0)) > _num(existing.get("weight", 0)):
|
||||
edges_by_key[key] = edge
|
||||
|
||||
# Drop edges referencing missing nodes
|
||||
node_ids = set(nodes_by_id.keys())
|
||||
valid_edges: list[dict] = []
|
||||
dropped_edges: list[dict[str, Any]] = []
|
||||
structural_warnings: list[str] = []
|
||||
for e in edges_by_key.values():
|
||||
src, tgt = e.get("source", ""), e.get("target", "")
|
||||
if src in node_ids and tgt in node_ids:
|
||||
valid_edges.append(e)
|
||||
else:
|
||||
missing = []
|
||||
if src not in node_ids:
|
||||
missing.append(f"source '{src}'")
|
||||
if tgt not in node_ids:
|
||||
missing.append(f"target '{tgt}'")
|
||||
etype = e.get("type", "?")
|
||||
dropped_edges.append({**e, "missing": missing})
|
||||
if etype in STRUCTURAL_EDGE_TYPES:
|
||||
structural_warnings.append(
|
||||
f"Warning: dropped structural edge {src} → {tgt} ({etype}), "
|
||||
f"missing {', '.join(missing)} — will retry on the next merge run"
|
||||
)
|
||||
else:
|
||||
unfixable.append(f"Edge {src} → {tgt} ({etype}): dropped, missing {', '.join(missing)}")
|
||||
|
||||
# ── Layers: merge by id, union nodeIds ────────────────────────────
|
||||
layers_by_id: dict[str, dict] = {}
|
||||
for g in graphs:
|
||||
for layer in g.get("layers", []):
|
||||
lid = layer.get("id", "")
|
||||
if lid in layers_by_id:
|
||||
existing_ids = set(layers_by_id[lid].get("nodeIds", []))
|
||||
existing_ids.update(layer.get("nodeIds", []))
|
||||
layers_by_id[lid]["nodeIds"] = list(existing_ids)
|
||||
else:
|
||||
layers_by_id[lid] = {**layer}
|
||||
|
||||
# Drop dangling layer nodeIds
|
||||
dropped_layer_refs = 0
|
||||
for layer in layers_by_id.values():
|
||||
before = len(layer.get("nodeIds", []))
|
||||
layer["nodeIds"] = [nid for nid in layer.get("nodeIds", []) if nid in node_ids]
|
||||
diff = before - len(layer["nodeIds"])
|
||||
if diff:
|
||||
dropped_layer_refs += diff
|
||||
|
||||
# ── Tour: concatenate, merge steps with same title ─────────────────
|
||||
all_tour_steps: list[dict] = []
|
||||
title_to_step: dict[str, dict] = {}
|
||||
for g in graphs:
|
||||
for step in g.get("tour", []):
|
||||
title = step.get("title", "")
|
||||
if title in title_to_step:
|
||||
# Merge nodeIds from duplicate-titled steps (e.g. both
|
||||
# subdomains produce a "Project Overview" step 1)
|
||||
existing = title_to_step[title]
|
||||
for nid in step.get("nodeIds", []):
|
||||
if nid not in existing.get("nodeIds", []):
|
||||
existing.setdefault("nodeIds", []).append(nid)
|
||||
# Keep the longer description
|
||||
if len(step.get("description", "")) > len(existing.get("description", "")):
|
||||
existing["description"] = step["description"]
|
||||
else:
|
||||
new_step = {**step}
|
||||
title_to_step[title] = new_step
|
||||
all_tour_steps.append(new_step)
|
||||
|
||||
# Drop dangling tour nodeIds and re-number
|
||||
dropped_tour_refs = 0
|
||||
for i, step in enumerate(all_tour_steps, start=1):
|
||||
step["order"] = i
|
||||
before = len(step.get("nodeIds", []))
|
||||
step["nodeIds"] = [nid for nid in step.get("nodeIds", []) if nid in node_ids]
|
||||
diff = before - len(step["nodeIds"])
|
||||
if diff:
|
||||
dropped_tour_refs += diff
|
||||
|
||||
# ── Project metadata: merge ───────────────────────────────────────
|
||||
languages: list[str] = []
|
||||
frameworks: list[str] = []
|
||||
descriptions: list[str] = []
|
||||
latest_at = ""
|
||||
latest_hash = ""
|
||||
project_name = ""
|
||||
|
||||
for g in graphs:
|
||||
proj = g.get("project", {})
|
||||
project_name = proj.get("name", "") or project_name
|
||||
for lang in proj.get("languages", []):
|
||||
if lang not in languages:
|
||||
languages.append(lang)
|
||||
for fw in proj.get("frameworks", []):
|
||||
if fw not in frameworks:
|
||||
frameworks.append(fw)
|
||||
desc = proj.get("description", "")
|
||||
if desc and desc not in descriptions:
|
||||
descriptions.append(desc)
|
||||
analyzed = proj.get("analyzedAt", "")
|
||||
if analyzed > latest_at:
|
||||
latest_at = analyzed
|
||||
latest_hash = proj.get("gitCommitHash", latest_hash)
|
||||
|
||||
# ── Build report ─────────────────────────────────────────────────
|
||||
report: list[str] = []
|
||||
report.append(f"Input: {total_input_nodes} nodes, {total_input_edges} edges (from {len(graphs)} graphs)")
|
||||
|
||||
# Fixed section
|
||||
fixed_lines: list[str] = []
|
||||
if node_dedup_by_type:
|
||||
for ntype, count in node_dedup_by_type.most_common():
|
||||
fixed_lines.append(f" {count:>4} × duplicate '{ntype}' nodes removed (kept later)")
|
||||
if edge_dedup_count:
|
||||
fixed_lines.append(f" {edge_dedup_count:>4} × duplicate edges removed (kept higher weight)")
|
||||
if dropped_layer_refs:
|
||||
fixed_lines.append(f" {dropped_layer_refs:>4} × dangling layer nodeId refs removed")
|
||||
if dropped_tour_refs:
|
||||
fixed_lines.append(f" {dropped_tour_refs:>4} × dangling tour nodeId refs removed")
|
||||
|
||||
if fixed_lines:
|
||||
total_fixed = sum(node_dedup_by_type.values()) + edge_dedup_count + dropped_layer_refs + dropped_tour_refs
|
||||
report.append("")
|
||||
report.append(f"Fixed ({total_fixed} corrections):")
|
||||
report.extend(fixed_lines)
|
||||
|
||||
# Structural drops surface as loud warnings, not buried counters —
|
||||
# losing hierarchy/cross-domain edges changes downstream traversal.
|
||||
if structural_warnings:
|
||||
report.append("")
|
||||
report.extend(structural_warnings)
|
||||
|
||||
# Could not fix section
|
||||
if unfixable:
|
||||
report.append("")
|
||||
report.append(f"Could not fix ({len(unfixable)} issues — needs agent review):")
|
||||
for detail in unfixable:
|
||||
report.append(f" - {detail}")
|
||||
|
||||
# Output stats
|
||||
report.append("")
|
||||
report.append(f"Output: {len(nodes_by_id)} nodes, {len(valid_edges)} edges, {len(layers_by_id)} layers, {len(all_tour_steps)} tour steps")
|
||||
|
||||
merged: dict[str, Any] = {
|
||||
"version": "1.0.0",
|
||||
"project": {
|
||||
"name": project_name,
|
||||
"languages": languages,
|
||||
"frameworks": frameworks,
|
||||
"description": " | ".join(descriptions) if len(descriptions) > 1 else (descriptions[0] if descriptions else ""),
|
||||
"analyzedAt": latest_at,
|
||||
"gitCommitHash": latest_hash,
|
||||
},
|
||||
"nodes": list(nodes_by_id.values()),
|
||||
"edges": valid_edges,
|
||||
"layers": list(layers_by_id.values()),
|
||||
"tour": all_tour_steps,
|
||||
}
|
||||
|
||||
return merged, report, dropped_edges
|
||||
|
||||
|
||||
def load_pending_structural_edges(report_path: Path) -> list[dict[str, Any]]:
|
||||
"""Read structural edges dropped by a previous run from merge-report.json.
|
||||
|
||||
Subdomain graph files are cleaned up after assembly, so these edges only
|
||||
survive in the report — re-injecting them lets a later run resolve them
|
||||
once the missing endpoint's subdomain graph has been merged.
|
||||
"""
|
||||
if not report_path.exists():
|
||||
return []
|
||||
try:
|
||||
data = json.loads(report_path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
print(f"Warning: could not read {report_path.name}: {e}", file=sys.stderr)
|
||||
return []
|
||||
pending = []
|
||||
for entry in data.get("droppedEdges", []):
|
||||
if isinstance(entry, dict) and entry.get("type") in STRUCTURAL_EDGE_TYPES:
|
||||
pending.append({k: v for k, v in entry.items() if k != "missing"})
|
||||
return pending
|
||||
|
||||
|
||||
def write_merge_report(
|
||||
report_path: Path,
|
||||
merged: dict[str, Any],
|
||||
dropped_edges: list[dict[str, Any]],
|
||||
recovered_count: int,
|
||||
) -> None:
|
||||
"""Persist the dropped-edge report so investigations don't require re-instrumenting the script."""
|
||||
report = {
|
||||
"generatedBy": "merge-subdomain-graphs.py",
|
||||
"output": {
|
||||
"nodes": len(merged.get("nodes", [])),
|
||||
"edges": len(merged.get("edges", [])),
|
||||
"layers": len(merged.get("layers", [])),
|
||||
"tourSteps": len(merged.get("tour", [])),
|
||||
},
|
||||
"recoveredStructuralEdges": recovered_count,
|
||||
"droppedEdges": dropped_edges,
|
||||
}
|
||||
report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: python merge-subdomain-graphs.py <project-root> [file1.json file2.json ...]", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
project_root = Path(sys.argv[1]).resolve()
|
||||
ua_dir = resolve_ua_dir(project_root)
|
||||
|
||||
if not ua_dir.is_dir():
|
||||
print(f"Error: {ua_dir} does not exist", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
output_path = ua_dir / "knowledge-graph.json"
|
||||
|
||||
# Determine which files to merge
|
||||
if len(sys.argv) > 2:
|
||||
# Explicit file list
|
||||
graph_files = [Path(f).resolve() for f in sys.argv[2:]]
|
||||
else:
|
||||
# Auto-discover subdomain graphs — exclude the main output file
|
||||
# to avoid self-merging on repeated runs
|
||||
graph_files = sorted(
|
||||
p for p in ua_dir.glob("*knowledge-graph*.json")
|
||||
if p.name != "knowledge-graph.json"
|
||||
)
|
||||
|
||||
if not graph_files:
|
||||
print("No subdomain graphs found to merge", file=sys.stderr)
|
||||
sys.exit(0)
|
||||
|
||||
print(f"Found {len(graph_files)} subdomain graphs:", file=sys.stderr)
|
||||
for f in graph_files:
|
||||
print(f" - {f.name}", file=sys.stderr)
|
||||
|
||||
# Load subdomain graphs
|
||||
graphs: list[dict[str, Any]] = []
|
||||
for f in graph_files:
|
||||
g = load_graph(f)
|
||||
if g is not None:
|
||||
graphs.append(g)
|
||||
node_count = len(g.get("nodes", []))
|
||||
edge_count = len(g.get("edges", []))
|
||||
print(f" Loaded {f.name}: {node_count} nodes, {edge_count} edges", file=sys.stderr)
|
||||
|
||||
if not graphs:
|
||||
print("Error: no valid subdomain graphs loaded", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Load the existing main graph as base (if it exists)
|
||||
if output_path.exists():
|
||||
base = load_graph(output_path)
|
||||
if base:
|
||||
node_count = len(base.get("nodes", []))
|
||||
edge_count = len(base.get("edges", []))
|
||||
print(f" Loaded base knowledge-graph.json: {node_count} nodes, {edge_count} edges", file=sys.stderr)
|
||||
graphs.insert(0, base) # Base first — subdomain data wins on conflict
|
||||
|
||||
# Re-inject structural edges a previous run had to drop; if their missing
|
||||
# endpoints have arrived in the meantime, this run resolves them.
|
||||
report_path = ua_dir / "merge-report.json"
|
||||
pending_edges = load_pending_structural_edges(report_path)
|
||||
if pending_edges:
|
||||
print(f" Retrying {len(pending_edges)} structural edges dropped by a previous run", file=sys.stderr)
|
||||
graphs.append({"nodes": [], "edges": pending_edges})
|
||||
|
||||
# Merge
|
||||
merged, report, dropped_edges = merge_graphs(graphs)
|
||||
|
||||
# Print report
|
||||
print("", file=sys.stderr)
|
||||
for line in report:
|
||||
print(line, file=sys.stderr)
|
||||
|
||||
# Count how many previously-dropped structural edges made it in this time
|
||||
merged_edge_keys = {(e.get("source", ""), e.get("target", ""), e.get("type", "")) for e in merged["edges"]}
|
||||
recovered = sum(
|
||||
1 for e in pending_edges
|
||||
if (e.get("source", ""), e.get("target", ""), e.get("type", "")) in merged_edge_keys
|
||||
)
|
||||
if recovered:
|
||||
print(f"Recovered {recovered} structural edges dropped by a previous run", file=sys.stderr)
|
||||
|
||||
# Write output
|
||||
output_path.write_text(json.dumps(merged, indent=2, ensure_ascii=False), encoding="utf-8")
|
||||
write_merge_report(report_path, merged, dropped_edges, recovered)
|
||||
print(f"Merge report written to {report_path}", file=sys.stderr)
|
||||
|
||||
size_kb = output_path.stat().st_size / 1024
|
||||
print(f"\nWritten to {output_path} ({size_kb:.0f} KB)", file=sys.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user