File size: 2,417 Bytes
12496fc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | """Bounded source indexing. AST symbols/import edges for Python; lexical for others."""
import ast
from collections import Counter
from pathlib import Path
import re
from .tools import Executor
EXTENSIONS = {".py", ".js", ".ts", ".tsx", ".jsx", ".c", ".h", ".cpp", ".hpp", ".cs", ".java", ".go", ".rs", ".sql", ".sh", ".ps1", ".html", ".css", ".kt", ".swift", ".md", ".json"}
EXCLUDED = {".git", "node_modules", ".venv", ".cache", "private", "__pycache__", "artifacts", "reports"}
def index_repository(executor: Executor, max_files=2000, max_bytes=10_000_000):
if "READ" not in executor.policy.permissions:
raise PermissionError("Repository indexing requires READ")
result, total = [], 0
for path in sorted(executor.root.rglob("*")):
if len(result) >= max_files:
break
rel = path.relative_to(executor.root)
if any(p in EXCLUDED for p in rel.parts) or path.suffix not in EXTENSIONS or not path.is_file():
continue
try:
safe = executor.path(str(rel))
size = safe.stat().st_size
if size > executor.policy.max_file_bytes or total + size > max_bytes:
continue
text = safe.read_text(encoding="utf-8")
except (OSError, UnicodeError, PermissionError):
continue
total += size
symbols, imports = [], []
if path.suffix == ".py":
try:
tree = ast.parse(text)
symbols = [{"name": n.name, "line": n.lineno, "kind": type(n).__name__} for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))]
imports = [n.module for n in ast.walk(tree) if isinstance(n, ast.ImportFrom)] + [a.name for n in ast.walk(tree) if isinstance(n, ast.Import) for a in n.names]
except SyntaxError:
pass
result.append({"path": rel.as_posix(), "symbols": symbols, "imports": imports,
"terms": dict(Counter(re.findall(r"[A-Za-z_][A-Za-z_0-9]*", text.lower())))})
return result
def retrieve(index, query, limit=8):
terms = set(re.findall(r"[A-Za-z_][A-Za-z_0-9]*", query.lower()))
scored = [(sum(min(r["terms"].get(t, 0), 5) for t in terms), r) for r in index]
return [{k: v for k, v in r.items() if k != "terms"} for score, r in sorted(scored, key=lambda x: x[0], reverse=True) if score > 0][:limit]
|