"""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