777 lines
31 KiB
Python
777 lines
31 KiB
Python
"""Symbol 索引服务。
|
||
|
||
目标：
|
||
1. 对代码库执行增量符号扫描（def/ref）
|
||
2. 提供符号查询、详情、概览
|
||
3. 为 GraphRAG / Agent 检索提供结构化工具层
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import hashlib
|
||
import re
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
|
||
from sqlalchemy import and_, delete, func, or_, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..models import SymbolDefinition, SymbolReference
|
||
|
||
|
||
TS_SYMBOL_RE = re.compile(
|
||
r"^\s*(?:export\s+)?(?:default\s+)?(?:async\s+)?(?:function|class|interface|type|const)\s+([A-Za-z_][A-Za-z0-9_]*)",
|
||
re.MULTILINE,
|
||
)
|
||
TS_IMPORT_RE = re.compile(r"^\s*import\s+.*?\s+from\s+['\"]([^'\"]+)['\"]", re.MULTILINE)
|
||
TS_CALL_RE = re.compile(r"\b([A-Za-z_][A-Za-z0-9_]*)\s*\(")
|
||
TS_KEYWORDS = {"if", "for", "while", "switch", "catch", "return", "new", "typeof", "await"}
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class ParsedSymbol:
|
||
symbol_type: str
|
||
symbol_name: str
|
||
qualified_name: str
|
||
start_line: int
|
||
end_line: int
|
||
signature: str
|
||
docstring: str
|
||
payload: dict[str, Any]
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class ParsedReference:
|
||
relation_type: str
|
||
target_symbol_name: str
|
||
target_qualified_name: str
|
||
source_qualified_name: str
|
||
start_line: int
|
||
end_line: int
|
||
payload: dict[str, Any]
|
||
|
||
|
||
class SymbolIndexService:
|
||
def __init__(self, db: AsyncSession, *, tenant_id: int) -> None:
|
||
self.db = db
|
||
self.tenant_id = int(tenant_id)
|
||
|
||
@staticmethod
|
||
def _default_scan_roots(*, backend_root: Path, workspace_root: Path) -> list[Path]:
|
||
roots: list[Path] = []
|
||
for candidate in [
|
||
(backend_root / "app").resolve(),
|
||
(workspace_root / "frontend" / "src").resolve(),
|
||
]:
|
||
if candidate.exists() and candidate.is_dir():
|
||
roots.append(candidate)
|
||
return roots
|
||
|
||
@staticmethod
|
||
def _to_rel_path(*, file_path: Path, workspace_root: Path, backend_root: Path) -> str:
|
||
try:
|
||
return str(file_path.resolve().relative_to(workspace_root))
|
||
except Exception:
|
||
try:
|
||
return str(file_path.resolve().relative_to(backend_root))
|
||
except Exception:
|
||
return str(file_path.resolve())
|
||
|
||
async def sync_codebase_symbols(
|
||
self,
|
||
*,
|
||
base_dir: Optional[str] = None,
|
||
max_files: int = 500,
|
||
force_reindex: bool = False,
|
||
) -> dict[str, Any]:
|
||
backend_root = Path(__file__).resolve().parents[2]
|
||
workspace_root = backend_root.parent
|
||
if base_dir:
|
||
root = Path(base_dir).resolve()
|
||
if not root.exists() or not root.is_dir():
|
||
raise ValueError(f"symbol scan path not found: {root}")
|
||
scan_roots = [root]
|
||
else:
|
||
scan_roots = self._default_scan_roots(backend_root=backend_root, workspace_root=workspace_root)
|
||
if not scan_roots:
|
||
raise ValueError("symbol scan path not found: backend/app and frontend/src")
|
||
|
||
allowed = {".py", ".ts", ".tsx", ".js", ".jsx"}
|
||
files: list[Path] = []
|
||
for root in scan_roots:
|
||
files.extend([p for p in root.rglob("*") if p.is_file() and p.suffix.lower() in allowed])
|
||
files.sort(key=lambda x: str(x))
|
||
selected = files[: max(1, int(max_files))]
|
||
|
||
stats = {
|
||
"root": str(scan_roots[0]),
|
||
"roots": [str(x) for x in scan_roots],
|
||
"candidate_files": len(files),
|
||
"selected_files": len(selected),
|
||
"changed_files": 0,
|
||
"skipped_unchanged_files": 0,
|
||
"indexed_symbols": 0,
|
||
"indexed_references": 0,
|
||
"failed_files": 0,
|
||
"errors": [],
|
||
}
|
||
|
||
for file_path in selected:
|
||
language = self._detect_language(file_path)
|
||
if not language:
|
||
continue
|
||
try:
|
||
text = file_path.read_text(encoding="utf-8", errors="ignore")
|
||
rel_path = self._to_rel_path(file_path=file_path, workspace_root=workspace_root, backend_root=backend_root)
|
||
file_hash = hashlib.sha1(text.encode("utf-8")).hexdigest()
|
||
mtime = datetime.fromtimestamp(file_path.stat().st_mtime, tz=timezone.utc).replace(tzinfo=None)
|
||
if not force_reindex:
|
||
latest_hash = await self._get_latest_file_hash(rel_path=rel_path)
|
||
if latest_hash and latest_hash == file_hash:
|
||
stats["skipped_unchanged_files"] += 1
|
||
continue
|
||
|
||
symbols, refs = self._parse_file(
|
||
file_path=rel_path,
|
||
language=language,
|
||
content=text,
|
||
)
|
||
s_count, r_count = await self._replace_file_index(
|
||
file_path=rel_path,
|
||
language=language,
|
||
file_hash=file_hash,
|
||
source_mtime=mtime,
|
||
symbols=symbols,
|
||
refs=refs,
|
||
)
|
||
stats["changed_files"] += 1
|
||
stats["indexed_symbols"] += s_count
|
||
stats["indexed_references"] += r_count
|
||
except Exception as exc:
|
||
stats["failed_files"] += 1
|
||
stats["errors"].append(f"{str(file_path)}: {str(exc)[:180]}")
|
||
|
||
await self.db.commit()
|
||
stats["errors"] = stats["errors"][:20]
|
||
stats["finished_at"] = datetime.now(timezone.utc).isoformat()
|
||
return stats
|
||
|
||
async def search_symbols(
|
||
self,
|
||
*,
|
||
query: str,
|
||
language: Optional[str] = None,
|
||
symbol_type: Optional[str] = None,
|
||
limit: int = 30,
|
||
) -> dict[str, Any]:
|
||
text = str(query or "").strip()
|
||
if not text:
|
||
return {"count": 0, "items": []}
|
||
|
||
like = f"%{text}%"
|
||
stmt = select(SymbolDefinition).where(
|
||
and_(
|
||
SymbolDefinition.tenant_id == self.tenant_id,
|
||
SymbolDefinition.status == "active",
|
||
or_(
|
||
SymbolDefinition.symbol_name.ilike(like),
|
||
SymbolDefinition.qualified_name.ilike(like),
|
||
SymbolDefinition.file_path.ilike(like),
|
||
),
|
||
)
|
||
)
|
||
if language:
|
||
stmt = stmt.where(SymbolDefinition.language == str(language).strip().lower()[:20])
|
||
if symbol_type:
|
||
stmt = stmt.where(SymbolDefinition.symbol_type == str(symbol_type).strip().lower()[:32])
|
||
rows = (
|
||
await self.db.execute(
|
||
stmt.order_by(SymbolDefinition.updated_at.desc(), SymbolDefinition.id.desc()).limit(max(1, min(int(limit), 200)))
|
||
)
|
||
).scalars().all()
|
||
return {
|
||
"count": len(rows),
|
||
"items": [self._serialize_symbol(row) for row in rows],
|
||
}
|
||
|
||
async def get_symbol_detail(self, *, symbol_id: int) -> dict[str, Any]:
|
||
symbol = (
|
||
await self.db.execute(
|
||
select(SymbolDefinition).where(
|
||
and_(
|
||
SymbolDefinition.tenant_id == self.tenant_id,
|
||
SymbolDefinition.id == int(symbol_id),
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if symbol is None:
|
||
raise ValueError("symbol_not_found")
|
||
|
||
refs = (
|
||
await self.db.execute(
|
||
select(SymbolReference).where(
|
||
and_(
|
||
SymbolReference.tenant_id == self.tenant_id,
|
||
SymbolReference.source_symbol_id == int(symbol.id),
|
||
)
|
||
)
|
||
.order_by(SymbolReference.id.desc())
|
||
.limit(120)
|
||
)
|
||
).scalars().all()
|
||
|
||
inbound_refs = (
|
||
await self.db.execute(
|
||
select(SymbolReference)
|
||
.where(
|
||
and_(
|
||
SymbolReference.tenant_id == self.tenant_id,
|
||
or_(
|
||
SymbolReference.target_symbol_name == str(symbol.symbol_name),
|
||
SymbolReference.target_qualified_name == str(symbol.qualified_name or ""),
|
||
),
|
||
)
|
||
)
|
||
.order_by(SymbolReference.id.desc())
|
||
.limit(120)
|
||
)
|
||
).scalars().all()
|
||
|
||
return {
|
||
"symbol": self._serialize_symbol(symbol),
|
||
"outbound_refs": [self._serialize_ref(row) for row in refs],
|
||
"inbound_refs": [self._serialize_ref(row) for row in inbound_refs],
|
||
}
|
||
|
||
async def get_overview(
|
||
self,
|
||
*,
|
||
top_n_files: int = 10,
|
||
auto_sync_if_empty: bool = False,
|
||
base_dir: Optional[str] = None,
|
||
) -> dict[str, Any]:
|
||
total = int(
|
||
(
|
||
await self.db.execute(
|
||
select(func.count(SymbolDefinition.id)).where(SymbolDefinition.tenant_id == self.tenant_id)
|
||
)
|
||
).scalar_one()
|
||
or 0
|
||
)
|
||
if total == 0 and auto_sync_if_empty:
|
||
await self.sync_codebase_symbols(base_dir=base_dir, max_files=300, force_reindex=False)
|
||
total = int(
|
||
(
|
||
await self.db.execute(
|
||
select(func.count(SymbolDefinition.id)).where(SymbolDefinition.tenant_id == self.tenant_id)
|
||
)
|
||
).scalar_one()
|
||
or 0
|
||
)
|
||
|
||
language_groups = (
|
||
await self.db.execute(
|
||
select(SymbolDefinition.language, func.count(SymbolDefinition.id))
|
||
.where(SymbolDefinition.tenant_id == self.tenant_id)
|
||
.group_by(SymbolDefinition.language)
|
||
)
|
||
).all()
|
||
type_groups = (
|
||
await self.db.execute(
|
||
select(SymbolDefinition.symbol_type, func.count(SymbolDefinition.id))
|
||
.where(SymbolDefinition.tenant_id == self.tenant_id)
|
||
.group_by(SymbolDefinition.symbol_type)
|
||
)
|
||
).all()
|
||
file_groups = (
|
||
await self.db.execute(
|
||
select(SymbolDefinition.file_path, func.count(SymbolDefinition.id))
|
||
.where(SymbolDefinition.tenant_id == self.tenant_id)
|
||
.group_by(SymbolDefinition.file_path)
|
||
.order_by(func.count(SymbolDefinition.id).desc())
|
||
.limit(max(1, min(int(top_n_files), 50)))
|
||
)
|
||
).all()
|
||
ref_total = int(
|
||
(
|
||
await self.db.execute(
|
||
select(func.count(SymbolReference.id)).where(SymbolReference.tenant_id == self.tenant_id)
|
||
)
|
||
).scalar_one()
|
||
or 0
|
||
)
|
||
indexed_files_total = int(
|
||
(
|
||
await self.db.execute(
|
||
select(func.count(func.distinct(SymbolDefinition.file_path))).where(
|
||
SymbolDefinition.tenant_id == self.tenant_id
|
||
)
|
||
)
|
||
).scalar_one()
|
||
or 0
|
||
)
|
||
return {
|
||
"tenant_id": self.tenant_id,
|
||
"indexed_files": indexed_files_total,
|
||
"symbol_total": total,
|
||
"reference_total": ref_total,
|
||
"python_symbols": int(dict(language_groups).get("python", 0) or 0),
|
||
"typescript_symbols": int(dict(language_groups).get("typescript", 0) or 0),
|
||
"symbols_by_language": {str(k): int(v or 0) for k, v in language_groups},
|
||
"symbols_by_type": {str(k): int(v or 0) for k, v in type_groups},
|
||
"top_modules": [
|
||
{"module": str(path), "symbols": int(cnt or 0)}
|
||
for path, cnt in file_groups
|
||
],
|
||
"snapshot_at": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
|
||
async def _get_latest_file_hash(self, *, rel_path: str) -> Optional[str]:
|
||
row = (
|
||
await self.db.execute(
|
||
select(SymbolDefinition.file_hash).where(
|
||
and_(
|
||
SymbolDefinition.tenant_id == self.tenant_id,
|
||
SymbolDefinition.file_path == rel_path,
|
||
)
|
||
)
|
||
.order_by(SymbolDefinition.updated_at.desc(), SymbolDefinition.id.desc())
|
||
.limit(1)
|
||
)
|
||
).scalar_one_or_none()
|
||
return str(row).strip() if row else None
|
||
|
||
async def _replace_file_index(
|
||
self,
|
||
*,
|
||
file_path: str,
|
||
language: str,
|
||
file_hash: str,
|
||
source_mtime: datetime,
|
||
symbols: list[ParsedSymbol],
|
||
refs: list[ParsedReference],
|
||
) -> tuple[int, int]:
|
||
await self.db.execute(
|
||
delete(SymbolReference).where(
|
||
and_(
|
||
SymbolReference.tenant_id == self.tenant_id,
|
||
SymbolReference.file_path == file_path,
|
||
)
|
||
)
|
||
)
|
||
await self.db.execute(
|
||
delete(SymbolDefinition).where(
|
||
and_(
|
||
SymbolDefinition.tenant_id == self.tenant_id,
|
||
SymbolDefinition.file_path == file_path,
|
||
)
|
||
)
|
||
)
|
||
|
||
created_symbols: list[SymbolDefinition] = []
|
||
qn_to_id: dict[str, int] = {}
|
||
name_to_id: dict[str, int] = {}
|
||
for item in symbols:
|
||
row = SymbolDefinition(
|
||
tenant_id=self.tenant_id,
|
||
file_path=file_path,
|
||
language=language,
|
||
symbol_type=item.symbol_type[:32],
|
||
symbol_name=item.symbol_name[:255],
|
||
qualified_name=item.qualified_name[:700] if item.qualified_name else None,
|
||
start_line=int(item.start_line) if item.start_line > 0 else None,
|
||
end_line=int(item.end_line) if item.end_line > 0 else None,
|
||
signature=item.signature[:1000] if item.signature else None,
|
||
docstring=item.docstring[:2000] if item.docstring else None,
|
||
file_hash=file_hash,
|
||
source_mtime=source_mtime,
|
||
status="active",
|
||
payload=item.payload,
|
||
last_indexed_at=datetime.now(timezone.utc).replace(tzinfo=None),
|
||
)
|
||
self.db.add(row)
|
||
created_symbols.append(row)
|
||
await self.db.flush()
|
||
for row in created_symbols:
|
||
if row.qualified_name:
|
||
qn_to_id[str(row.qualified_name)] = int(row.id)
|
||
name_to_id[str(row.symbol_name)] = int(row.id)
|
||
|
||
ref_count = 0
|
||
# DB 层有 (tenant_id,file_path,relation_type,target_symbol_name,start_line,source_symbol_id) 唯一约束，
|
||
# 这里先去重，避免同一行重复调用（例如链式调用）触发批量写入冲突。
|
||
ref_seen_keys: set[tuple[str, str, Optional[int], Optional[int]]] = set()
|
||
for ref in refs:
|
||
source_id = None
|
||
source_qn = str(ref.source_qualified_name or "").strip()
|
||
if source_qn:
|
||
source_id = qn_to_id.get(source_qn)
|
||
if source_id is None:
|
||
source_tail = source_qn.split(".")[-1]
|
||
source_id = name_to_id.get(source_tail)
|
||
relation_type = ref.relation_type[:32]
|
||
target_symbol_name = ref.target_symbol_name[:255]
|
||
start_line = int(ref.start_line) if ref.start_line > 0 else None
|
||
dedupe_key = (
|
||
relation_type,
|
||
target_symbol_name,
|
||
start_line,
|
||
int(source_id) if source_id is not None else None,
|
||
)
|
||
if dedupe_key in ref_seen_keys:
|
||
continue
|
||
ref_seen_keys.add(dedupe_key)
|
||
row = SymbolReference(
|
||
tenant_id=self.tenant_id,
|
||
source_symbol_id=int(source_id) if source_id is not None else None,
|
||
file_path=file_path,
|
||
relation_type=relation_type,
|
||
target_symbol_name=target_symbol_name,
|
||
target_qualified_name=ref.target_qualified_name[:700] if ref.target_qualified_name else None,
|
||
start_line=start_line,
|
||
end_line=int(ref.end_line) if ref.end_line > 0 else None,
|
||
payload=ref.payload,
|
||
)
|
||
self.db.add(row)
|
||
ref_count += 1
|
||
await self.db.flush()
|
||
return (len(created_symbols), ref_count)
|
||
|
||
def _parse_file(
|
||
self,
|
||
*,
|
||
file_path: str,
|
||
language: str,
|
||
content: str,
|
||
) -> tuple[list[ParsedSymbol], list[ParsedReference]]:
|
||
if language == "python":
|
||
return self._parse_python_file(file_path=file_path, content=content)
|
||
return self._parse_typescript_file(file_path=file_path, content=content)
|
||
|
||
def _parse_python_file(self, *, file_path: str, content: str) -> tuple[list[ParsedSymbol], list[ParsedReference]]:
|
||
text = str(content or "")
|
||
if not text.strip():
|
||
return ([], [])
|
||
try:
|
||
tree = ast.parse(text)
|
||
except Exception:
|
||
return ([], [])
|
||
symbols: list[ParsedSymbol] = []
|
||
refs: list[ParsedReference] = []
|
||
|
||
class Visitor(ast.NodeVisitor):
|
||
def __init__(self):
|
||
self.class_stack: list[str] = []
|
||
self.func_stack: list[str] = []
|
||
|
||
def _current_source(self) -> str:
|
||
if self.func_stack:
|
||
return self.func_stack[-1]
|
||
if self.class_stack:
|
||
return self.class_stack[-1]
|
||
return ""
|
||
|
||
def _qualify(self, name: str) -> str:
|
||
chain = [*self.class_stack]
|
||
if self.func_stack:
|
||
chain.extend(self.func_stack[-1].split(".")[len(self.class_stack):])
|
||
chain.append(name)
|
||
uniq: list[str] = []
|
||
for item in chain:
|
||
if item and (not uniq or uniq[-1] != item):
|
||
uniq.append(item)
|
||
return ".".join(uniq) if uniq else name
|
||
|
||
def visit_ClassDef(self, node: ast.ClassDef) -> Any:
|
||
qualified = ".".join([*self.class_stack, node.name]) if self.class_stack else node.name
|
||
symbols.append(
|
||
ParsedSymbol(
|
||
symbol_type="class",
|
||
symbol_name=node.name,
|
||
qualified_name=qualified,
|
||
start_line=int(getattr(node, "lineno", 0) or 0),
|
||
end_line=int(getattr(node, "end_lineno", getattr(node, "lineno", 0)) or 0),
|
||
signature=f"class {node.name}",
|
||
docstring=str(ast.get_docstring(node) or ""),
|
||
payload={"file_path": file_path},
|
||
)
|
||
)
|
||
self.class_stack.append(node.name)
|
||
self.generic_visit(node)
|
||
self.class_stack.pop()
|
||
|
||
def visit_FunctionDef(self, node: ast.FunctionDef) -> Any:
|
||
self._visit_function_node(node=node, is_async=False)
|
||
|
||
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> Any:
|
||
self._visit_function_node(node=node, is_async=True)
|
||
|
||
def _visit_function_node(self, *, node: Any, is_async: bool) -> None:
|
||
if self.class_stack:
|
||
qualified = ".".join([*self.class_stack, node.name])
|
||
symbol_type = "method"
|
||
else:
|
||
qualified = node.name
|
||
symbol_type = "function"
|
||
arg_names = []
|
||
for arg in list(node.args.posonlyargs) + list(node.args.args):
|
||
arg_names.append(str(arg.arg))
|
||
if node.args.vararg is not None:
|
||
arg_names.append(f"*{node.args.vararg.arg}")
|
||
for arg in node.args.kwonlyargs:
|
||
arg_names.append(str(arg.arg))
|
||
if node.args.kwarg is not None:
|
||
arg_names.append(f"**{node.args.kwarg.arg}")
|
||
prefix = "async def" if is_async else "def"
|
||
signature = f"{prefix} {node.name}({', '.join(arg_names)})"
|
||
symbols.append(
|
||
ParsedSymbol(
|
||
symbol_type=symbol_type,
|
||
symbol_name=node.name,
|
||
qualified_name=qualified,
|
||
start_line=int(getattr(node, "lineno", 0) or 0),
|
||
end_line=int(getattr(node, "end_lineno", getattr(node, "lineno", 0)) or 0),
|
||
signature=signature[:1000],
|
||
docstring=str(ast.get_docstring(node) or ""),
|
||
payload={"file_path": file_path},
|
||
)
|
||
)
|
||
self.func_stack.append(qualified)
|
||
self.generic_visit(node)
|
||
self.func_stack.pop()
|
||
|
||
def visit_Import(self, node: ast.Import) -> Any:
|
||
source = self._current_source()
|
||
for alias in node.names:
|
||
target = str(alias.name or "").strip()
|
||
if not target:
|
||
continue
|
||
refs.append(
|
||
ParsedReference(
|
||
relation_type="imports",
|
||
target_symbol_name=target.split(".")[-1][:255],
|
||
target_qualified_name=target[:700],
|
||
source_qualified_name=source[:700],
|
||
start_line=int(getattr(node, "lineno", 0) or 0),
|
||
end_line=int(getattr(node, "end_lineno", getattr(node, "lineno", 0)) or 0),
|
||
payload={"alias": str(alias.asname or "")[:255] if alias.asname else None},
|
||
)
|
||
)
|
||
self.generic_visit(node)
|
||
|
||
def visit_ImportFrom(self, node: ast.ImportFrom) -> Any:
|
||
source = self._current_source()
|
||
module = str(node.module or "").strip()
|
||
for alias in node.names:
|
||
name = str(alias.name or "").strip()
|
||
if not name:
|
||
continue
|
||
target_qn = f"{module}.{name}" if module else name
|
||
refs.append(
|
||
ParsedReference(
|
||
relation_type="imports",
|
||
target_symbol_name=name[:255],
|
||
target_qualified_name=target_qn[:700],
|
||
source_qualified_name=source[:700],
|
||
start_line=int(getattr(node, "lineno", 0) or 0),
|
||
end_line=int(getattr(node, "end_lineno", getattr(node, "lineno", 0)) or 0),
|
||
payload={"module": module[:255] if module else None},
|
||
)
|
||
)
|
||
self.generic_visit(node)
|
||
|
||
def visit_Call(self, node: ast.Call) -> Any:
|
||
source = self._current_source()
|
||
target = ""
|
||
if isinstance(node.func, ast.Name):
|
||
target = str(node.func.id or "").strip()
|
||
elif isinstance(node.func, ast.Attribute):
|
||
target = str(node.func.attr or "").strip()
|
||
if target:
|
||
refs.append(
|
||
ParsedReference(
|
||
relation_type="calls",
|
||
target_symbol_name=target[:255],
|
||
target_qualified_name=target[:700],
|
||
source_qualified_name=source[:700],
|
||
start_line=int(getattr(node, "lineno", 0) or 0),
|
||
end_line=int(getattr(node, "end_lineno", getattr(node, "lineno", 0)) or 0),
|
||
payload={},
|
||
)
|
||
)
|
||
self.generic_visit(node)
|
||
|
||
Visitor().visit(tree)
|
||
return (symbols[:5000], refs[:12000])
|
||
|
||
def _parse_typescript_file(self, *, file_path: str, content: str) -> tuple[list[ParsedSymbol], list[ParsedReference]]:
|
||
text = str(content or "")
|
||
if not text.strip():
|
||
return ([], [])
|
||
symbols: list[ParsedSymbol] = []
|
||
refs: list[ParsedReference] = []
|
||
lines = text.splitlines()
|
||
line_symbol_map: dict[int, str] = {}
|
||
|
||
for idx, line in enumerate(lines, start=1):
|
||
match = re.match(
|
||
r"^\s*(?:export\s+)?(?:default\s+)?(?:async\s+)?(function|class|interface|type|const)\s+([A-Za-z_][A-Za-z0-9_]*)",
|
||
line,
|
||
)
|
||
if not match:
|
||
continue
|
||
symbol_type = str(match.group(1) or "").strip().lower()
|
||
name = str(match.group(2) or "").strip()
|
||
if not name:
|
||
continue
|
||
signature = line.strip()
|
||
symbols.append(
|
||
ParsedSymbol(
|
||
symbol_type=symbol_type,
|
||
symbol_name=name,
|
||
qualified_name=name,
|
||
start_line=idx,
|
||
end_line=idx,
|
||
signature=signature[:1000],
|
||
docstring="",
|
||
payload={"file_path": file_path},
|
||
)
|
||
)
|
||
line_symbol_map[idx] = name
|
||
|
||
if "frontend/src/router/" in str(file_path).replace("\\", "/"):
|
||
for idx, line in enumerate(lines, start=1):
|
||
m = re.search(
|
||
r'path:\s*"([^"]+)"\s*,\s*element:\s*<([A-Za-z_][A-Za-z0-9_]*)',
|
||
line,
|
||
)
|
||
if not m:
|
||
continue
|
||
route_path = str(m.group(1) or "").strip()
|
||
component = str(m.group(2) or "").strip()
|
||
if not route_path:
|
||
continue
|
||
clean_route = route_path if route_path.startswith("/") else f"/{route_path}"
|
||
symbol_name = f"route:{clean_route}"
|
||
symbols.append(
|
||
ParsedSymbol(
|
||
symbol_type="route",
|
||
symbol_name=symbol_name[:255],
|
||
qualified_name=symbol_name[:700],
|
||
start_line=idx,
|
||
end_line=idx,
|
||
signature=line.strip()[:1000],
|
||
docstring=f"component:{component}",
|
||
payload={"file_path": file_path, "route_path": clean_route, "component": component},
|
||
)
|
||
)
|
||
|
||
if str(file_path).replace("\\", "/").endswith("components/layout/Sidebar.tsx"):
|
||
for idx, line in enumerate(lines, start=1):
|
||
m = re.search(r'to:\s*"([^"]+)"\s*,\s*label:\s*"([^"]+)"', line)
|
||
if not m:
|
||
continue
|
||
route_path = str(m.group(1) or "").strip()
|
||
label = str(m.group(2) or "").strip()
|
||
if not route_path or not label:
|
||
continue
|
||
clean_route = route_path if route_path.startswith("/") else f"/{route_path}"
|
||
symbol_name = f"feature:{label}"
|
||
symbols.append(
|
||
ParsedSymbol(
|
||
symbol_type="feature",
|
||
symbol_name=symbol_name[:255],
|
||
qualified_name=f"{label}@{clean_route}"[:700],
|
||
start_line=idx,
|
||
end_line=idx,
|
||
signature=line.strip()[:1000],
|
||
docstring=f"menu:{clean_route}",
|
||
payload={"file_path": file_path, "feature_label": label, "route_path": clean_route},
|
||
)
|
||
)
|
||
|
||
imports = TS_IMPORT_RE.finditer(text)
|
||
for match in imports:
|
||
module = str(match.group(1) or "").strip()
|
||
if not module:
|
||
continue
|
||
start = text[: match.start()].count("\n") + 1
|
||
refs.append(
|
||
ParsedReference(
|
||
relation_type="imports",
|
||
target_symbol_name=module.split("/")[-1][:255],
|
||
target_qualified_name=module[:700],
|
||
source_qualified_name="",
|
||
start_line=start,
|
||
end_line=start,
|
||
payload={},
|
||
)
|
||
)
|
||
|
||
for idx, line in enumerate(lines, start=1):
|
||
source_symbol = ""
|
||
candidate_lines = [x for x in line_symbol_map.keys() if x <= idx]
|
||
if candidate_lines:
|
||
source_symbol = line_symbol_map[max(candidate_lines)]
|
||
for call in TS_CALL_RE.finditer(line):
|
||
name = str(call.group(1) or "").strip()
|
||
if not name or name.lower() in TS_KEYWORDS:
|
||
continue
|
||
refs.append(
|
||
ParsedReference(
|
||
relation_type="calls",
|
||
target_symbol_name=name[:255],
|
||
target_qualified_name=name[:700],
|
||
source_qualified_name=source_symbol[:700],
|
||
start_line=idx,
|
||
end_line=idx,
|
||
payload={},
|
||
)
|
||
)
|
||
return (symbols[:5000], refs[:12000])
|
||
|
||
@staticmethod
|
||
def _detect_language(path: Path) -> str:
|
||
ext = path.suffix.lower()
|
||
if ext == ".py":
|
||
return "python"
|
||
if ext in {".ts", ".tsx", ".js", ".jsx"}:
|
||
return "typescript"
|
||
return ""
|
||
|
||
@staticmethod
|
||
def _serialize_symbol(row: SymbolDefinition) -> dict[str, Any]:
|
||
return {
|
||
"id": int(row.id),
|
||
"file_path": str(row.file_path or ""),
|
||
"language": str(row.language or ""),
|
||
"symbol_type": str(row.symbol_type or ""),
|
||
"symbol_name": str(row.symbol_name or ""),
|
||
"qualified_name": str(row.qualified_name or ""),
|
||
"start_line": int(row.start_line) if row.start_line is not None else None,
|
||
"end_line": int(row.end_line) if row.end_line is not None else None,
|
||
"signature": str(row.signature or ""),
|
||
"docstring": str(row.docstring or ""),
|
||
"status": str(row.status or ""),
|
||
"updated_at": row.updated_at.isoformat() if row.updated_at else None,
|
||
"last_indexed_at": row.last_indexed_at.isoformat() if row.last_indexed_at else None,
|
||
}
|
||
|
||
@staticmethod
|
||
def _serialize_ref(row: SymbolReference) -> dict[str, Any]:
|
||
return {
|
||
"id": int(row.id),
|
||
"source_symbol_id": int(row.source_symbol_id) if row.source_symbol_id is not None else None,
|
||
"file_path": str(row.file_path or ""),
|
||
"relation_type": str(row.relation_type or ""),
|
||
"target_symbol_name": str(row.target_symbol_name or ""),
|
||
"target_qualified_name": str(row.target_qualified_name or ""),
|
||
"start_line": int(row.start_line) if row.start_line is not None else None,
|
||
"end_line": int(row.end_line) if row.end_line is not None else None,
|
||
"updated_at": row.updated_at.isoformat() if row.updated_at else None,
|
||
}
|