#!/usr/bin/env python3 """ Cyborg Architecture — Knowledge Ingestion Script Phase 1: File-based knowledge extraction and graph building """ import subprocess import hashlib import json import sys import os from pathlib import Path from typing import Optional # --- Configuration --- MEMGRAPH_URI = os.getenv("MEMGRAPH_URI", "bolt://localhost:7687") MEMGRAPH_USER = os.getenv("MEMGRAPH_USER", "memgraph") MEMGRAPH_PASS = os.getenv("MEMGRAPH_PASS", "memgraph") QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333") OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434") OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen2.5:7b") # --- Helpers --- def run_cmd(cmd: list[str]) -> tuple[int, str]: """Run a shell command, return (exit_code, stdout).""" result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) return result.returncode, result.stdout.strip() def file_hash(path: str) -> str: """Simple SHA-256 hash of file content.""" h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): h.update(chunk) return h.hexdigest()[:16] def get_file_metadata(path: str) -> dict: """Extract metadata from a file.""" p = Path(path) lang = p.suffix.lstrip(".") or "text" lang = classify_language(lang) # Count lines lines = 0 try: with open(path, "r", errors="ignore") as f: for _ in f: lines += 1 except Exception: pass return { "path": path, "name": p.name, "lang": lang, "lines": lines, "size": p.stat().st_size, "hash": file_hash(path), "modified": p.stat().st_mtime, } def classify_language(ext: str) -> str: """Map file extension to language category.""" lang_map = { "py": "python", "js": "javascript", "ts": "typescript", "go": "go", "rs": "rust", "java": "java", "cpp": "cpp", "c": "c", "h": "c-header", "hpp": "cpp-header", "html": "html", "css": "css", "scss": "scss", "sass": "sass", "md": "markdown", "yaml": "yaml", "yml": "yaml", "json": "json", "xml": "xml", "toml": "toml", "ini": "ini", "cfg": "ini", "sh": "shell", "bash": "shell", "zsh": "shell", "dockerfile": "docker", "docker-compose": "docker", "sql": "sql", "graphql": "graphql", "proto": "protobuf", "tf": "hcl", "hcl": "hcl", "vue": "vue", "svelte": "svelte", } return lang_map.get(ext, ext) # --- Graph Operations (via Cypher) --- def create_file_node(cypher_conn, meta: dict): """Create or update a file node in Memgraph.""" # This would use the mgclient Python library in production # For now, return the Cypher statement cypher = f""" MERGE (f:File {{path: '{meta['path']}'}}) SET f.name = '{meta['name']}', f.lang = '{meta['lang']}', f.lines = {meta['lines']}, f.size = {meta['size']}, f.hash = '{meta['hash']}', f.modified = {meta['modified']}, f.indexed_at = timestamp() RETURN f """ return cypher def create_import_dependency(cypher_conn, from_file: str, to_file: str): """Create an IMPORTS relationship between files.""" cypher = f""" MATCH (a:File {{path: '{from_file}'}}), (b:File {{path: '{to_file}'}}) CREATE (a)-[:IMPORTS]->(b) RETURN a, b """ return cypher def create_project_structure(cypher_conn, path: str, level: int = 0): """Build a directory tree as a graph.""" p = Path(path) if not p.is_dir(): return [] statements = [] current = f"MATCH (d:Directory {{path: '{p}'}}) RETURN d" for item in sorted(p.iterdir()): if item.is_dir(): child_path = str(item) statements.append(f""" MERGE (parent:Directory {{path: '{p}'}}) MERGE (child:Directory {{path: '{child_path}'}}) MERGE (parent)-[:CONTAINS]->(child) """) statements.extend(create_project_structure(str(item), level + 1)) elif item.is_file(): meta = get_file_metadata(str(item)) statements.append(create_file_node(None, meta)) statements.append(f""" MERGE (d:Directory {{path: '{p}'}}) MERGE (f:File {{path: '{meta['path']}'}}) MERGE (d)-[:CONTAINS]->(f) """) return statements # --- Dependency Extraction --- def extract_imports_python(file_path: str) -> list[str]: """Parse Python imports from a file.""" imports = [] try: with open(file_path, "r", errors="ignore") as f: for line in f: line = line.strip() if line.startswith("import ") or line.startswith("from "): module = line.split()[1].split(".")[0] imports.append(module) except Exception: pass return imports def extract_imports_js(file_path: str) -> list[str]: """Parse JS/TS imports from a file.""" imports = [] try: with open(file_path, "r", errors="ignore") as f: content = f.read() # Simple regex for require/import import re for m in re.finditer(r'''(?:import\s+.*?from\s+['"]([^'"]+)['"]|require\s*\(\s*['"]([^'"]+)['"]\s*\))''', content): mod = m.group(1) or m.group(2) if not mod.startswith(".") and not mod.startswith("/"): imports.append(mod.split("/")[0]) except Exception: pass return imports def extract_imports_generic(file_path: str, lang: str) -> list[str]: """Generic import extraction.""" if lang == "python": return extract_imports_python(file_path) elif lang in ("javascript", "typescript"): return extract_imports_js(file_path) return [] # --- Main --- def ingest_directory(directory: str, dry_run: bool = False) -> dict: """Main ingestion function.""" results = { "files": [], "dependencies": [], "errors": [], } for root, dirs, files in os.walk(directory): # Skip hidden dirs and venvs dirs[:] = [d for d in dirs if not d.startswith(".") and d not in ("node_modules", "__pycache__", ".git", "venv")] for fname in files: fpath = os.path.join(root, fname) rel_path = os.path.relpath(fpath, directory) meta = get_file_metadata(fpath) results["files"].append(meta) # Extract imports imports = extract_imports_generic(fpath, meta["lang"]) if imports: results["dependencies"].append({ "file": fpath, "imports": imports, }) if dry_run: print(f"Would ingest {len(results['files'])} files") print(f"Would create {len(results['dependencies'])} dependency records") print(json.dumps(results, indent=2)[:2000]) return results if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Cyborg Architecture - Knowledge Ingest") parser.add_argument("directory", nargs="?", default=".", help="Directory to ingest") parser.add_argument("--dry-run", action="store_true", help="Show what would happen") args = parser.parse_args() results = ingest_directory(args.directory, dry_run=args.dry_run) print(json.dumps(results, indent=2, default=str))