125 lines
4.3 KiB
Python
125 lines
4.3 KiB
Python
"""Adaptive RAG mixin module."""
|
|
from __future__ import annotations
|
|
|
|
from .shared import *
|
|
|
|
|
|
class AdaptiveRAGSymbolSnapshotMixin:
|
|
def _symbol_cache_key(self) -> str:
|
|
return f"{self.tenant_id}:backend_symbol_snapshot"
|
|
|
|
async def _get_symbol_snapshot_cached(self, *, ttl_seconds: int = 300) -> dict[str, Any]:
|
|
now = time.time()
|
|
key = self._symbol_cache_key()
|
|
cached = self._symbol_snapshot_cache.get(key)
|
|
if cached and cached[0] > now:
|
|
return dict(cached[1])
|
|
try:
|
|
service = SymbolIndexService(self.db, tenant_id=self.tenant_id)
|
|
built = await service.get_overview(top_n_files=10, auto_sync_if_empty=True)
|
|
except Exception:
|
|
built = self._build_symbol_index_snapshot()
|
|
self._symbol_snapshot_cache[key] = (now + max(60, int(ttl_seconds)), dict(built))
|
|
return built
|
|
|
|
def _build_symbol_index_snapshot(
|
|
self,
|
|
*,
|
|
max_files: int = 240,
|
|
max_symbols: int = 4000,
|
|
) -> dict[str, Any]:
|
|
backend_root = Path(__file__).resolve().parents[2]
|
|
app_root = backend_root / "app"
|
|
if not app_root.exists():
|
|
return {
|
|
"root": str(app_root),
|
|
"indexed_files": 0,
|
|
"symbol_total": 0,
|
|
"python_symbols": 0,
|
|
"typescript_symbols": 0,
|
|
"top_modules": [],
|
|
"generated_at": now_cn_aware().isoformat(),
|
|
}
|
|
|
|
candidates = [
|
|
p
|
|
for p in app_root.rglob("*")
|
|
if p.is_file() and p.suffix.lower() in {".py", ".ts", ".tsx", ".js", ".jsx"}
|
|
]
|
|
candidates.sort(key=lambda p: str(p))
|
|
file_paths = candidates[: max(1, int(max_files))]
|
|
|
|
python_symbols = 0
|
|
ts_symbols = 0
|
|
total_symbols = 0
|
|
module_counter: dict[str, int] = {}
|
|
|
|
for path in file_paths:
|
|
rel = str(path.relative_to(backend_root))
|
|
module = rel.split("/", 2)[:2]
|
|
module_name = "/".join(module) if module else rel
|
|
try:
|
|
text = path.read_text(encoding="utf-8", errors="ignore")
|
|
except Exception:
|
|
continue
|
|
|
|
symbols_in_file = 0
|
|
if path.suffix.lower() == ".py":
|
|
extracted = self._extract_python_symbols(text)
|
|
symbols_in_file = len(extracted)
|
|
python_symbols += symbols_in_file
|
|
else:
|
|
extracted = self._extract_text_symbols(text)
|
|
symbols_in_file = len(extracted)
|
|
ts_symbols += symbols_in_file
|
|
if symbols_in_file <= 0:
|
|
continue
|
|
module_counter[module_name] = int(module_counter.get(module_name, 0)) + symbols_in_file
|
|
total_symbols += symbols_in_file
|
|
if total_symbols >= max_symbols:
|
|
break
|
|
|
|
top_modules = [
|
|
{"module": name, "symbols": int(count)}
|
|
for name, count in sorted(module_counter.items(), key=lambda item: item[1], reverse=True)[:10]
|
|
]
|
|
return {
|
|
"root": str(app_root),
|
|
"indexed_files": len(file_paths),
|
|
"symbol_total": int(total_symbols),
|
|
"python_symbols": int(python_symbols),
|
|
"typescript_symbols": int(ts_symbols),
|
|
"top_modules": top_modules,
|
|
"generated_at": now_cn_aware().isoformat(),
|
|
}
|
|
|
|
@staticmethod
|
|
def _extract_python_symbols(text: str) -> list[str]:
|
|
raw = str(text or "")
|
|
if not raw.strip():
|
|
return []
|
|
try:
|
|
tree = ast.parse(raw)
|
|
except Exception:
|
|
return []
|
|
out: list[str] = []
|
|
for node in tree.body:
|
|
if isinstance(node, ast.ClassDef):
|
|
out.append(f"class:{node.name}")
|
|
elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
out.append(f"function:{node.name}")
|
|
return out
|
|
|
|
@staticmethod
|
|
def _extract_text_symbols(text: str) -> list[str]:
|
|
raw = str(text or "")
|
|
if not raw.strip():
|
|
return []
|
|
out: list[str] = []
|
|
for match in TS_SYMBOL_RE.finditer(raw):
|
|
name = str(match.group(1) or "").strip()
|
|
if name:
|
|
out.append(name)
|
|
return out
|
|
|