Add scripts/ingest.py

scripts/ingest.py

@@ -0,0 +1,232 @@
+#!/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))