From d9fad6f9f8f64c258661aede329bb301298aa079 Mon Sep 17 00:00:00 2001 From: neo <397763@gmail.com> Date: Sat, 27 Jun 2026 13:15:12 +0500 Subject: [PATCH] Add scripts/ingest.py --- scripts/ingest.py | 232 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 232 insertions(+) create mode 100644 scripts/ingest.py diff --git a/scripts/ingest.py b/scripts/ingest.py new file mode 100644 index 0000000..d626304 --- /dev/null +++ b/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))