#!/usr/bin/env python3 """ Source-free Python structural extraction for Neurcode Repository Graph V2. Reads JSON from stdin: {"filePath": "...", "sourceText": "...", "contentHash": "...", "timeoutMs": 2000} Writes JSON to stdout with symbols, imports, exports — never source bodies. Uses stdlib ast only; no network; bounded by caller timeout. """ from __future__ import annotations import ast import hashlib import json import sys from typing import Any def sha256(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def stable_id(prefix: str, key: str) -> str: return f"{prefix}:{sha256(key)[:24]}" def param_arity(args: ast.arguments) -> int: positional = len(getattr(args, 'posonlyargs', [])) + len(args.args) if args.vararg: positional += 1 positional += len(args.kwonlyargs) if args.kwarg: positional += 1 return positional def structural_fingerprint(kind: str, name: str, arity: int | None) -> str: return sha256(f"{kind}:{name}:{arity if arity is not None else ''}") def extract_symbols(tree: ast.AST, file_path: str) -> list[dict[str, Any]]: symbols: list[dict[str, Any]] = [] ordinals: dict[str, int] = {} def add_symbol(name: str, kind: str, line: int, exported: bool, arity: int | None) -> None: ordinal_key = f"{kind}:{name}" ordinal = ordinals.get(ordinal_key, 0) ordinals[ordinal_key] = ordinal + 1 sym_id = stable_id("symbol", f"{file_path}:{kind}:{name}:{ordinal}") symbols.append({ "id": sym_id, "name": name, "kind": kind, "language": "python", "filePath": file_path, "line": line, "exported": exported, "local": name.startswith("_"), "arity": arity, "signatureHash": sha256(f"{kind}:{arity if arity is not None else ''}"), "structuralFingerprint": structural_fingerprint(kind, name, arity), "parserDepth": "syntax_tree", }) for node in ast.iter_child_nodes(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): add_symbol( node.name, "function", node.lineno, not node.name.startswith("_"), param_arity(node.args), ) elif isinstance(node, ast.ClassDef): add_symbol(node.name, "class", node.lineno, not node.name.startswith("_"), None) for child in ast.iter_child_nodes(node): if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)): add_symbol( child.name, "method", child.lineno, not child.name.startswith("_"), param_arity(child.args), ) return symbols def imported_names_from_importfrom(node: ast.ImportFrom) -> list[str]: names: list[str] = [] for alias in node.names: if alias.name == "*": names.append("*") else: names.append(alias.asname or alias.name) return sorted(names) def extract_imports(tree: ast.AST, file_path: str) -> list[dict[str, Any]]: imports: list[dict[str, Any]] = [] for index, node in enumerate(ast.walk(tree)): if isinstance(node, ast.ImportFrom): module = node.module or "" level = node.level or 0 if level > 0: target = "." * level + module else: target = module imports.append({ "id": stable_id("import", f"{file_path}:{target}:{node.lineno}:{index}"), "fromFile": file_path, "target": target, "resolvedFile": None, "importedNames": imported_names_from_importfrom(node), "kind": "python_import", "line": node.lineno, "parserDepth": "syntax_tree", }) elif isinstance(node, ast.Import): for alias in node.names: # Resolve the real dotted module (alias.name); the `as` binding name is # irrelevant to module-path resolution. target = alias.name imports.append({ "id": stable_id("import", f"{file_path}:{target}:{node.lineno}:{index}"), "fromFile": file_path, "target": target, "resolvedFile": None, "importedNames": [], "kind": "python_import", "line": node.lineno, "parserDepth": "syntax_tree", }) elif isinstance(node, ast.Call): # Dynamic imports (`importlib.import_module(...)`, `__import__(...)`) are NEVER # statically resolvable. Emit them with kind="dynamic" so the resolver records # them as not_evaluated rather than silently dropping the dependency. func = node.func is_dynamic = ( (isinstance(func, ast.Name) and func.id == "__import__") or (isinstance(func, ast.Attribute) and func.attr in ("import_module", "__import__")) ) if is_dynamic: literal = "" if node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str): literal = node.args[0].value imports.append({ "id": stable_id("import", f"{file_path}:dynamic:{node.lineno}:{index}"), "fromFile": file_path, "target": literal, "resolvedFile": None, "importedNames": [], "kind": "dynamic", "line": node.lineno, "parserDepth": "syntax_tree", }) return imports def extract_exports(symbols: list[dict[str, Any]], file_path: str) -> list[dict[str, Any]]: exports: list[dict[str, Any]] = [] for sym in symbols: if not sym.get("exported"): continue exports.append({ "id": stable_id("export", f"{file_path}:{sym['name']}:{sym['line']}"), "filePath": file_path, "symbolName": sym["name"], "target": None, "kind": "python_public", "line": sym["line"], "parserDepth": "syntax_tree", }) return exports def analyze(payload: dict[str, Any]) -> dict[str, Any]: file_path = str(payload.get("filePath") or "") source = str(payload.get("sourceText") or "") if not file_path: return {"ok": False, "error": "filePath_required", "symbols": [], "imports": [], "exports": [], "errors": []} try: tree = ast.parse(source, filename=file_path) except SyntaxError as exc: return { "ok": False, "error": "syntax_error", "symbols": [], "imports": [], "exports": [], "errors": [f"syntax_error:{exc.lineno or 0}"], "limitations": ["Python source could not be parsed; falling back may be required."], } symbols = extract_symbols(tree, file_path) imports = extract_imports(tree, file_path) exports = extract_exports(symbols, file_path) return { "ok": True, "parserId": "python-structural-ast", "parserDepth": "syntax_tree", "symbols": symbols, "imports": imports, "exports": exports, "errors": [], "limitations": [ "Python AST extraction is structural only; call and reference edges are not emitted.", ], } def main() -> None: try: raw = sys.stdin.read() payload = json.loads(raw) if raw.strip() else {} except json.JSONDecodeError: print(json.dumps({"ok": False, "error": "invalid_json", "symbols": [], "imports": [], "exports": [], "errors": []})) return result = analyze(payload) print(json.dumps(result, separators=(",", ":"))) if __name__ == "__main__": main()