"""Tree-sitter based multi-language code parser. Extracts structural nodes (classes, functions, imports, types) and edges (calls, inheritance, contains) from source files. """ from __future__ import annotations import ast import hashlib import html import importlib import json import logging import math import os import re import subprocess import sys import threading from dataclasses import dataclass, field from functools import lru_cache from pathlib import Path, PurePath from typing import Any, NamedTuple, Optional try: import tomllib except ModuleNotFoundError: # pragma: no cover - Python 3.10 import tomli as tomllib # type: ignore[import-not-found,no-redef] from .config_keys import is_spring_config_path, normalize_spring_config_key from .custom_languages import CustomLanguage, load_custom_languages try: import yaml as _yaml # type: ignore[import-untyped] from yaml import MappingNode as _YamlMapping from yaml import ScalarNode as _YamlScalar from yaml import SequenceNode as _YamlSequence except ImportError: _yaml = None # type: ignore[assignment] _YamlMapping = _YamlSequence = _YamlScalar = None # type: ignore[assignment,misc] from .tsconfig_resolver import TsconfigResolver class CellInfo(NamedTuple): """Represents a single cell in a notebook with its language.""" cell_index: int language: str source: str _SQL_TABLE_RE = re.compile( r"(?:FROM|JOIN|INTO|CREATE\s+(?:OR\s+REPLACE\s+)?(?:TABLE|VIEW)|INSERT\s+OVERWRITE)" r"\s+((?:`[^`]+`|\w+)(?:\.(?:`[^`]+`|\w+))*)", re.IGNORECASE, ) # dbt model dependencies: {{ ref('model') }}, {{ ref('package', 'model') }} # and {{ source('source_name', 'table') }}. String-literal arguments only — # dynamic ref() calls cannot be resolved statically. _DBT_REF_RE = re.compile( r"\{\{-?\s*(ref|source)\s*\(\s*" r"['\"]([^'\"]+)['\"]" r"(?:\s*,\s*['\"]([^'\"]+)['\"])?" r"\s*\)", ) _PYTHON_STAR_CACHE_MAX = 15_000 _PYTHON_STAR_EXPORT_CACHE: dict[tuple[str, int, int], dict[str, str]] = {} _PYTHON_STAR_EXPORT_CACHE_LOCK = threading.RLock() @lru_cache(maxsize=512) def _read_cargo_manifest( manifest_path: str, _mtime_ns: int, _size: int, ) -> dict[str, Any]: """Read one Cargo manifest, keyed by immutable file identity metadata.""" try: parsed = tomllib.loads( Path(manifest_path).read_text(encoding="utf-8", errors="replace"), ) except (OSError, tomllib.TOMLDecodeError): return {} return parsed if isinstance(parsed, dict) else {} def _load_cargo_manifest(path: Path) -> dict[str, Any]: try: stat = path.stat() resolved = path.resolve() except (OSError, RuntimeError, ValueError): return {} return _read_cargo_manifest(str(resolved), stat.st_mtime_ns, stat.st_size) class _PythonScopeBindingVisitor(ast.NodeVisitor): """Collect names bound in one Python lexical scope.""" def __init__(self) -> None: self.names: set[str] = set() def visit_Name(self, node: ast.Name) -> None: # noqa: N802 if isinstance(node.ctx, (ast.Store, ast.Del)): self.names.add(node.id) def visit_FunctionDef(self, node: ast.FunctionDef) -> None: # noqa: N802 self.names.add(node.name) def visit_AsyncFunctionDef( # noqa: N802 self, node: ast.AsyncFunctionDef, ) -> None: self.names.add(node.name) def visit_ClassDef(self, node: ast.ClassDef) -> None: # noqa: N802 self.names.add(node.name) def visit_Lambda(self, node: ast.Lambda) -> None: # noqa: N802 return def visit_Import(self, node: ast.Import) -> None: # noqa: N802 for alias in node.names: self.names.add(alias.asname or alias.name.split(".", 1)[0]) def visit_ImportFrom(self, node: ast.ImportFrom) -> None: # noqa: N802 for alias in node.names: if alias.name != "*": self.names.add(alias.asname or alias.name) def _python_type_checking_aliases( tree: ast.Module, ) -> tuple[frozenset[str], frozenset[str]]: """Return unshadowed aliases for ``typing.TYPE_CHECKING``.""" names: set[str] = set() modules: set[str] = set() shadowed: set[str] = set() for statement in tree.body: if isinstance(statement, ast.Import): for alias in statement.names: if alias.name == "typing": modules.add(alias.asname or "typing") else: shadowed.add( alias.asname or alias.name.split(".", 1)[0], ) elif isinstance(statement, ast.ImportFrom) and statement.module == "typing": for alias in statement.names: if alias.name == "TYPE_CHECKING": names.add(alias.asname or alias.name) elif alias.name != "*": shadowed.add(alias.asname or alias.name) else: bindings = _PythonScopeBindingVisitor() bindings.visit(statement) shadowed.update(bindings.names) return frozenset(names - shadowed), frozenset(modules - shadowed) def _python_static_truth( node: ast.expr, type_checking_names: frozenset[str] = frozenset(), typing_modules: frozenset[str] = frozenset(), ) -> Optional[bool]: """Return a truth value for the small constant subset we can prove.""" if isinstance(node, ast.Constant) and isinstance(node.value, (bool, int)): return bool(node.value) if isinstance(node, ast.Name) and node.id in type_checking_names: return False if ( isinstance(node, ast.Attribute) and node.attr == "TYPE_CHECKING" and isinstance(node.value, ast.Name) and node.value.id in typing_modules ): return False if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not): value = _python_static_truth( node.operand, type_checking_names, typing_modules, ) return None if value is None else not value if isinstance(node, ast.BoolOp): values = [ _python_static_truth( value, type_checking_names, typing_modules, ) for value in node.values ] if isinstance(node.op, ast.And): if False in values: return False return True if all(value is True for value in values) else None if isinstance(node.op, ast.Or): if True in values: return True return False if all(value is False for value in values) else None return None class _PythonUnreachableCallVisitor(ast.NodeVisitor): """Collect calls inside branches whose condition is statically false.""" def __init__( self, type_checking_names: frozenset[str], typing_modules: frozenset[str], ) -> None: self._dead = False self.positions: set[tuple[int, int]] = set() self._type_checking_names = type_checking_names self._typing_modules = typing_modules def visit_Call(self, node: ast.Call) -> None: # noqa: N802 if self._dead: self.positions.add((node.lineno, node.col_offset)) self.generic_visit(node) def visit_If(self, node: ast.If) -> None: # noqa: N802 self.visit(node.test) truth = _python_static_truth( node.test, self._type_checking_names, self._typing_modules, ) self._visit_statements(node.body, dead=truth is False) self._visit_statements(node.orelse, dead=truth is True) def visit_FunctionDef(self, node: ast.FunctionDef) -> None: # noqa: N802 self._visit_function(node) def visit_AsyncFunctionDef( # noqa: N802 self, node: ast.AsyncFunctionDef, ) -> None: self._visit_function(node) def visit_ClassDef(self, node: ast.ClassDef) -> None: # noqa: N802 for decorator in node.decorator_list: self.visit(decorator) for base in node.bases: self.visit(base) for keyword in node.keywords: self.visit(keyword.value) bindings = _PythonScopeBindingVisitor() for statement in node.body: bindings.visit(statement) outer_names = self._type_checking_names outer_modules = self._typing_modules self._type_checking_names = outer_names - bindings.names self._typing_modules = outer_modules - bindings.names self._visit_statements(node.body, dead=False) self._type_checking_names = outer_names self._typing_modules = outer_modules def _visit_function( self, node: ast.FunctionDef | ast.AsyncFunctionDef, ) -> None: for decorator in node.decorator_list: self.visit(decorator) for default in (*node.args.defaults, *node.args.kw_defaults): if default is not None: self.visit(default) all_args = ( *node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs, ) for argument in all_args: if argument.annotation is not None: self.visit(argument.annotation) for optional_argument in (node.args.vararg, node.args.kwarg): if ( optional_argument is not None and optional_argument.annotation is not None ): self.visit(optional_argument.annotation) if node.returns is not None: self.visit(node.returns) bindings = _PythonScopeBindingVisitor() for statement in node.body: bindings.visit(statement) bindings.names.update(argument.arg for argument in all_args) bindings.names.update( argument.arg for argument in (node.args.vararg, node.args.kwarg) if argument is not None ) outer_names = self._type_checking_names outer_modules = self._typing_modules self._type_checking_names = outer_names - bindings.names self._typing_modules = outer_modules - bindings.names self._visit_statements(node.body, dead=False) self._type_checking_names = outer_names self._typing_modules = outer_modules def _visit_statements( self, statements: list[ast.stmt], *, dead: bool, ) -> None: outer_dead = self._dead self._dead = outer_dead or dead for statement in statements: self.visit(statement) self._dead = outer_dead @lru_cache(maxsize=128) def _python_unreachable_call_positions( source: bytes, ) -> frozenset[tuple[int, int]]: """Return one-based line/byte-column positions of proven-dead calls.""" try: tree = ast.parse(source.decode("utf-8")) except (SyntaxError, UnicodeDecodeError, ValueError): return frozenset() type_checking_names, typing_modules = _python_type_checking_aliases(tree) visitor = _PythonUnreachableCallVisitor( type_checking_names, typing_modules, ) visitor.visit(tree) return frozenset(visitor.positions) # --------------------------------------------------------------------------- # Non-Python static dead-guard detection (tree-sitter ancestor walk). # # ``_python_unreachable_call_positions`` above uses the ``ast`` module and so # only covers Python. The helpers below cover languages ``ast`` cannot parse, # walking the tree-sitter ancestor chain of a call node: # # * Go / TypeScript / JavaScript: ``if false { ... }`` / ``if (0) { ... }`` # * C / C++: ``#if 0`` / ``#elif 0`` preprocessor blocks # # Only the consequence (true branch) is dead; ``else`` / ``#else`` / ``#elif`` # branches stay live. # --------------------------------------------------------------------------- def _node_is_in_child(node, child_node) -> bool: """Return True if *node* is *child_node* or one of its descendants. Compares byte ranges because the tree-sitter Python bindings create a fresh ``Node`` object on every ``child_by_field_name`` call, so ``is`` identity fails even when both sides refer to the same tree node. """ start_byte, end_byte = child_node.start_byte, child_node.end_byte cursor = node while cursor is not None: if cursor.start_byte == start_byte and cursor.end_byte == end_byte: return True cursor = cursor.parent return False def _is_statically_false_condition(cond) -> bool: """Return True if *cond* is a statically-false literal (non-Python). * ``parenthesized_expression`` (TS/JS wrap ``if (expr)``) is unwrapped. * ``false`` -- Go and TS/JS boolean literal. * ``number`` equal to ``0`` -- TS/JS ``if (0)``. """ if cond.type == "parenthesized_expression": inner = cond.named_children return _is_statically_false_condition(inner[0]) if inner else False if cond.type == "false": return True # Literal ``0`` only. ``0x0`` / ``0.0`` are equally falsy but are left # undetected on purpose: missing one is a dropped suppression, never a # wrongly-suppressed live call, and evaluating arbitrary numeric literals # invites its own bugs. Python's ``if 0:`` is handled by the ast path. if cond.type == "number" and cond.text == b"0": return True return False def _is_in_static_dead_guard(node) -> bool: """Return True if *node* sits in a statically-dead branch (non-Python). Two independent ancestor walks: * A walk for Go / TS / JS ``if false`` / ``if (0)``. * A walk for C/C++ ``#if 0`` / ``#elif 0``. Neither walk stops at a function or class boundary. A declaration nested inside a dead branch is never evaluated, so calls in its body are dead too -- matching what the Python ``ast`` path above already does for a ``def`` or ``class`` under ``if False:``. Unlike Python, JS/TS class declarations are not hoisted, so there is no reachable symbol to preserve either. """ # Go / TS / JS: ``if`` with a statically-false condition. cursor = node.parent while cursor is not None: node_type = cursor.type if node_type == "if_statement": condition = cursor.child_by_field_name("condition") consequence = cursor.child_by_field_name("consequence") if ( condition is not None and consequence is not None and _is_statically_false_condition(condition) and _node_is_in_child(node, consequence) ): return True cursor = cursor.parent # C / C++: ``#if 0`` / ``#elif 0`` preprocessor block. preproc = node.parent while preproc is not None: if preproc.type in ("preproc_if", "preproc_elif"): condition = preproc.child_by_field_name("condition") if ( condition is not None and condition.type == "number_literal" and condition.text == b"0" ): alternative = preproc.child_by_field_name("alternative") if not ( alternative is not None and _node_is_in_child(node, alternative) ): return True preproc = preproc.parent return False # SQL keywords that can appear after FROM/JOIN but are NOT table names. _SQL_KEYWORDS: frozenset[str] = frozenset({ "SELECT", "WHERE", "GROUP", "ORDER", "HAVING", "LIMIT", "OFFSET", "UNION", "INTERSECT", "EXCEPT", "AS", "ON", "USING", "SET", "VALUES", "DEFAULT", "NULL", "TRUE", "FALSE", "INNER", "OUTER", "LEFT", "RIGHT", "FULL", "CROSS", "NATURAL", "LATERAL", "RECURSIVE", "ONLY", "WITH", }) logger = logging.getLogger(__name__) _DEFAULT_PARSER_LOAD_TIMEOUT_SECONDS = 5.0 _PARSER_PROBE_RESULTS: dict[str, bool] = {} _PARSER_PROBE_FAILURE_DETAILS: dict[str, str] = {} _PARSER_PROBE_LOCK = threading.Lock() _EXPECTED_PARSER_LOAD_ERRORS = (ImportError, LookupError, OSError, ValueError) def _parser_load_timeout_seconds() -> float: """Return a safe positive timeout for native grammar probes.""" raw = os.environ.get( "CRG_PARSER_LOAD_TIMEOUT_SECONDS", str(_DEFAULT_PARSER_LOAD_TIMEOUT_SECONDS), ) try: timeout = float(raw) except ValueError: timeout = _DEFAULT_PARSER_LOAD_TIMEOUT_SECONDS if not math.isfinite(timeout) or timeout <= 0: timeout = _DEFAULT_PARSER_LOAD_TIMEOUT_SECONDS return timeout def _run_parser_load_probe(grammar: str, timeout_seconds: float) -> bool: """Probe one native grammar in a disposable interpreter process.""" code = ( "from tree_sitter_language_pack import get_parser\n" "import sys\n" "get_parser(sys.argv[1])\n" ) try: completed = subprocess.run( [sys.executable, "-c", code, grammar], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=timeout_seconds, check=False, ) except (OSError, subprocess.TimeoutExpired) as exc: _PARSER_PROBE_FAILURE_DETAILS[grammar] = str(exc) logger.debug("tree-sitter parser probe failed for %s: %s", grammar, exc) return False if completed.returncode == 0: _PARSER_PROBE_FAILURE_DETAILS.pop(grammar, None) return True raw_stderr = getattr(completed, "stderr", b"") if isinstance(raw_stderr, bytes): stderr = raw_stderr.decode("utf-8", errors="replace") else: stderr = str(raw_stderr) detail = next( (line.strip() for line in reversed(stderr.splitlines()) if line.strip()), f"probe exited with status {completed.returncode}", ) _PARSER_PROBE_FAILURE_DETAILS[grammar] = detail[:500] return False def _parser_load_probe_succeeds( grammar: str, timeout_seconds: float | None = None, ) -> bool: """Return a process-cached result for one bounded grammar probe. The lock deliberately covers the subprocess call: parallel parser users must not start duplicate probes for the same grammar while the first one is still running. """ with _PARSER_PROBE_LOCK: cached = _PARSER_PROBE_RESULTS.get(grammar) if cached is not None: return cached timeout = ( _parser_load_timeout_seconds() if timeout_seconds is None else timeout_seconds ) result = _run_parser_load_probe(grammar, timeout) _PARSER_PROBE_RESULTS[grammar] = result if not result: detail = _PARSER_PROBE_FAILURE_DETAILS.get(grammar) if detail: logger.warning( "Skipping unavailable tree-sitter parser for %s: %s", grammar, detail, ) else: logger.warning( "Skipping unavailable tree-sitter parser for %s", grammar, ) return result def _mark_parser_unavailable(grammar: str) -> None: """Prevent repeated parent-process loads after an expected failure.""" with _PARSER_PROBE_LOCK: _PARSER_PROBE_RESULTS[grammar] = False def _clear_parser_probe_cache() -> None: """Clear process-level probe state (used by focused tests).""" with _PARSER_PROBE_LOCK: _PARSER_PROBE_RESULTS.clear() _PARSER_PROBE_FAILURE_DETAILS.clear() def _load_tree_sitter_parser(grammar: str): """Load a probed grammar, suppressing only known availability errors.""" if not _parser_load_probe_succeeds(grammar): return None try: language_pack = importlib.import_module("tree_sitter_language_pack") return language_pack.get_parser(grammar) # type: ignore[attr-defined] except _EXPECTED_PARSER_LOAD_ERRORS as exc: _mark_parser_unavailable(grammar) logger.debug("tree-sitter parser unavailable for %s: %s", grammar, exc) return None _PhpPsr4Mappings = tuple[tuple[str, tuple[str, ...]], ...] def _path_is_within(path: Path, root: Path) -> bool: """Return whether *path* is inside *root* after both are resolved.""" try: path.relative_to(root) except ValueError: return False return True @lru_cache(maxsize=128) def _read_php_composer_psr4( composer_path: str, repo_root: str, _mtime_ns: int, _size: int, ) -> _PhpPsr4Mappings: """Read immutable, shape-safe PSR-4 mappings from one composer.json.""" composer = Path(composer_path) root = Path(repo_root) try: data = json.loads( composer.read_text(encoding="utf-8", errors="replace"), ) except (OSError, json.JSONDecodeError): return () if not isinstance(data, dict): return () combined: dict[str, list[str]] = {} for section_name in ("autoload", "autoload-dev"): section = data.get(section_name) if not isinstance(section, dict): continue psr4 = section.get("psr-4") if not isinstance(psr4, dict): continue for raw_prefix, raw_paths in psr4.items(): if not isinstance(raw_prefix, str): continue prefix = raw_prefix.lstrip("\\").rstrip("\\") candidate_paths = ( [raw_paths] if isinstance(raw_paths, str) else raw_paths if isinstance(raw_paths, list) else [] ) destinations = combined.setdefault(prefix, []) for raw_path in candidate_paths: if not isinstance(raw_path, str): continue try: destination = (composer.parent / raw_path).resolve() except (OSError, RuntimeError, ValueError): continue if not _path_is_within(destination, root): continue destination_str = str(destination) if destination_str not in destinations: destinations.append(destination_str) return tuple( (prefix, tuple(destinations)) for prefix, destinations in sorted( combined.items(), key=lambda item: (-len(item[0]), item[0]), ) if destinations ) # --------------------------------------------------------------------------- # Data models for extracted entities # --------------------------------------------------------------------------- def normalize_file_path(path: "str | PurePath") -> str: """Return *path* as a forward-slash (POSIX) string for graph identity. ``file_path`` values and the path component of qualified names are graph identity: they must be separator-stable across operating systems so a graph built on Windows produces the same identifiers as one built on Linux/macOS, and so consumers that reconstruct identifiers from ``Path`` objects always agree with the parser. See issue #774. Only apply this to file *paths* — never to symbol names: PHP namespace identifiers (``App\\Domain\\Job``) legitimately contain backslashes. """ if isinstance(path, PurePath): return path.as_posix() return str(path).replace("\\", "/") @dataclass class NodeInfo: kind: str # File, Class, Function, Type, Test name: str file_path: str line_start: int line_end: int language: str = "" parent_name: Optional[str] = None # enclosing class/module params: Optional[str] = None return_type: Optional[str] = None modifiers: Optional[str] = None is_test: bool = False extra: dict = field(default_factory=dict) identity_name: Optional[str] = None def __post_init__(self) -> None: # Identity invariant (#774): file paths always use POSIX separators. # File nodes carry their path in ``name`` as well. self.file_path = normalize_file_path(self.file_path) if self.kind == "File": self.name = normalize_file_path(self.name) @dataclass class EdgeInfo: # CALLS, IMPORTS_FROM, INHERITS, IMPLEMENTS, CONTAINS, # TESTED_BY, DEPENDS_ON, REFERENCES kind: str source: str # qualified name or path target: str # qualified name or path file_path: str line: int = 0 extra: dict = field(default_factory=dict) def __post_init__(self) -> None: # Identity invariant (#774): file paths always use POSIX separators. # ``source``/``target`` are left alone — they may contain qualified # names whose symbol part legitimately embeds backslashes (PHP FQNs). self.file_path = normalize_file_path(self.file_path) # --------------------------------------------------------------------------- # Language extension mapping # --------------------------------------------------------------------------- EXTENSION_TO_LANGUAGE: dict[str, str] = { ".py": "python", ".js": "javascript", ".jsx": "javascript", ".ts": "typescript", ".tsx": "tsx", ".go": "go", ".rs": "rust", ".java": "java", ".cs": "csharp", ".rb": "ruby", ".cpp": "cpp", ".cc": "cpp", ".cxx": "cpp", ".c": "c", ".h": "c", ".hpp": "cpp", ".hh": "cpp", ".kt": "kotlin", ".swift": "swift", ".php": "php", ".scala": "scala", ".sol": "solidity", ".vue": "vue", ".dart": "dart", ".r": "r", # .lower() in detect_language handles .R → .r ".mjs": "javascript", ".astro": "typescript", ".pl": "perl", ".pm": "perl", ".t": "perl", ".xs": "c", # Perl XS: parsed as C to capture functions/structs/includes ".lua": "lua", ".luau": "luau", ".m": "objc", # Objective-C (.h still maps to C; .mm defers to C++ for simplicity) ".sh": "bash", ".bash": "bash", ".zsh": "bash", ".ksh": "bash", # Korn shell — close enough to bash for tree-sitter-bash (#235) ".ex": "elixir", ".exs": "elixir", ".ipynb": "notebook", ".zig": "zig", ".ps1": "powershell", ".psm1": "powershell", ".psd1": "powershell", ".svelte": "svelte", ".jl": "julia", # ReScript: .res is implementation, .resi is interface. Both share one # language label; the parser flags interface files via extra metadata. # No tree-sitter grammar is bundled in tree_sitter_language_pack, so # extraction is regex-based (see _parse_rescript). ".res": "rescript", ".resi": "rescript", ".gd": "gdscript", ".nix": "nix", # SystemVerilog/Verilog ".sv": "verilog", ".svh": "verilog", ".v": "verilog", ".vh": "verilog", # tree-sitter-language-pack does not currently bundle Visual Basic. # Keep the fallback deliberately structural and repository-local. ".vb": "vbnet", ".sql": "sql", ".tf": "hcl", ".hcl": "hcl", ".properties": "properties", ".yml": "yaml", ".yaml": "yaml", } # ``.h`` is shared by C and C++. Keep C as the extension default, then promote # a header only when the C++ grammar finds syntax that C cannot express. Weak # compatibility markers such as ``__cplusplus`` and ``extern "C"`` are # deliberately excluded because they are common in otherwise-C headers. _CPP_HEADER_EVIDENCE_TYPES = frozenset({ "access_specifier", "alias_declaration", "base_class_clause", "class_specifier", "concept_definition", "lambda_expression", "namespace_definition", "noexcept", "template_declaration", "trailing_return_type", "using_declaration", }) _CPP_HEADER_EVIDENCE_QUALIFIERS = frozenset({b"consteval", b"constinit"}) _CPP_QT_STRUCTURAL_MACRO_REPLACEMENTS = { b"QT_BEGIN_NAMESPACE": b" " * len(b"QT_BEGIN_NAMESPACE"), b"QT_END_NAMESPACE": b" " * len(b"QT_END_NAMESPACE"), b"Q_OBJECT": b" " * len(b"Q_OBJECT"), b"Q_SIGNALS": b"public" + b" " * (len(b"Q_SIGNALS") - len(b"public")), b"Q_SLOTS": b" " * len(b"Q_SLOTS"), b"Q_EMIT": b" " * len(b"Q_EMIT"), } # Shebang interpreter → language mapping for extension-less Unix scripts. # Each key is the **basename** of the interpreter path as it appears after # ``#!`` (or after ``#!/usr/bin/env``). Only languages already registered # above are listed — this file strictly routes extension-less scripts, it # does NOT introduce new languages on its own. See issue #237. SHEBANG_INTERPRETER_TO_LANGUAGE: dict[str, str] = { # POSIX / bash-compatible shells — all routed through tree-sitter-bash "bash": "bash", "sh": "bash", "zsh": "bash", "ksh": "bash", "dash": "bash", "ash": "bash", # Python (every common variant) "python": "python", "python2": "python", "python3": "python", "pypy": "python", "pypy3": "python", # JavaScript via Node "node": "javascript", "nodejs": "javascript", # Ruby / Perl / Lua / R / PHP "ruby": "ruby", "perl": "perl", "lua": "lua", "Rscript": "r", "php": "php", } # Maximum bytes to read from the head of a file when probing for a shebang. # 256 is enough for any reasonable shebang line (``#!/usr/bin/env python3 -u\n`` # is ~30 chars) while keeping the worst-case read tiny even on fat binaries. _SHEBANG_PROBE_BYTES = 256 # --------------------------------------------------------------------------- # Ansible YAML constants # --------------------------------------------------------------------------- # Path components that strongly suggest an Ansible project layout _ANSIBLE_PATH_COMPONENTS: frozenset[str] = frozenset({ "playbooks", "roles", "tasks", "handlers", "group_vars", "host_vars", }) # Common top-level playbook filenames (still require content confirmation) _ANSIBLE_PLAYBOOK_NAMES: frozenset[str] = frozenset({ "site.yml", "site.yaml", "main.yml", "main.yaml", "install.yml", "install.yaml", "deploy.yml", "deploy.yaml", }) # Play-level keys that are ONLY valid in Ansible plays. # `hosts:` alone is not sufficient to identify a play — require at least one of these. _ANSIBLE_PLAY_KEYS: frozenset[str] = frozenset({ "tasks", "handlers", "pre_tasks", "post_tasks", "roles", "gather_facts", "become", "become_user", "become_method", "serial", "strategy", "vars_files", "vars_prompt", "any_errors_fatal", "max_fail_percentage", "ignore_errors", }) # Bare module names for content sniffing; also used after FQCN prefix strip _ANSIBLE_MODULE_KEYS: frozenset[str] = frozenset({ "apt", "yum", "dnf", "package", "pip", "copy", "template", "file", "service", "systemd", "command", "shell", "raw", "git", "user", "stat", "include_tasks", "import_tasks", "include_role", "import_role", "set_fact", "debug", "fail", "assert", "wait_for", "pause", "lineinfile", "blockinfile", "get_url", "uri", "unarchive", "add_host", "group_by", "include_vars", }) # Task mapping keys that are metadata, NOT module invocations _TASK_META_KEYS: frozenset[str] = frozenset({ "name", "when", "loop", "loop_control", "with_items", "with_first_found", "with_fileglob", "with_dict", "with_subelements", "with_nested", "with_sequence", "with_indexed_items", "register", "notify", "tags", "become", "become_user", "become_method", "ignore_errors", "vars", "no_log", "check_mode", "environment", "any_errors_fatal", "run_once", "delegate_to", "delegate_facts", "block", "rescue", "always", "changed_when", "failed_when", "retries", "delay", "until", "listen", "connection", "timeout", }) # Tree-sitter node type mappings per language # Maps (language) -> dict of semantic role -> list of TS node types _CLASS_TYPES: dict[str, list[str]] = { "python": ["class_definition"], "javascript": ["class_declaration", "class"], # TS types are declarations, not just runtime classes: an interface or type # alias is the thing callers depend on, so it needs a node of its own the way # Java/C#/PHP interfaces do. Without them a types-only module (types.ts, # *.d.ts) contributes zero symbol nodes and its blast radius collapses to # whole-file IMPORTS_FROM fan-out. See: #737 "typescript": [ "class_declaration", "class", "interface_declaration", "type_alias_declaration", "enum_declaration", ], "tsx": [ "class_declaration", "class", "interface_declaration", "type_alias_declaration", "enum_declaration", ], "go": ["type_declaration"], # impl_item is a scope for methods, not a second type definition. It is # dispatched separately so repeated impl blocks cannot overwrite structs. "rust": ["struct_item", "enum_item", "trait_item"], "java": ["class_declaration", "interface_declaration", "enum_declaration"], "c": ["struct_specifier", "type_definition"], "cpp": ["class_specifier", "struct_specifier"], "csharp": [ "class_declaration", "interface_declaration", "enum_declaration", "struct_declaration", "record_declaration", "record_struct_declaration", ], "ruby": ["class", "module"], "r": [], # Classes detected via call pattern-matching, not AST node types "perl": ["package_statement", "class_statement", "role_statement"], "kotlin": ["class_declaration", "object_declaration"], "swift": ["class_declaration", "struct_declaration", "protocol_declaration"], "php": [ "class_declaration", "interface_declaration", "trait_declaration", "enum_declaration", ], "scala": [ "class_definition", "trait_definition", "object_definition", "enum_definition", ], "solidity": [ "contract_declaration", "interface_declaration", "library_declaration", "struct_declaration", "enum_declaration", "error_declaration", "user_defined_type_definition", ], "dart": ["class_definition", "mixin_declaration", "enum_declaration"], "lua": [], # Lua has no class keyword; table-based OOP handled via constructs handler "luau": ["type_definition"], # Luau type aliases; table-based OOP via constructs handler "objc": [ "class_interface", "class_implementation", "category_interface", "protocol_declaration", ], "bash": [], # Shell has no classes # Elixir: `defmodule Name do ... end` is a ``call`` node whose first # identifier is literally "defmodule". Dispatched via # _extract_elixir_constructs to avoid matching every ``call`` here. "elixir": [], # Nix: attrset bindings aren't "classes"; dispatched via # _extract_nix_constructs. "nix": [], # Zig has no single class node; struct/union/enum/opaque are VarDecl # whose RHS is a SuffixExpr > ContainerDecl. Dispatched via # _extract_zig_constructs. "zig": [], "powershell": ["class_statement"], "julia": [ "struct_definition", "abstract_definition", "module_definition", ], "verilog": [ "module_declaration", "interface_declaration", "class_declaration", "package_declaration", ], # GDScript: inner classes use ``class Name:`` (class_definition); the # file-level ``class_name Name`` gives the script itself an identity. "gdscript": ["class_definition", "class_name_statement"], # SQL: CREATE TABLE / CREATE VIEW are handled via _parse_sql dispatch. "sql": [], # HCL/Terraform: all constructs are blocks; dispatched via # _extract_hcl_constructs. "hcl": [], } # TS/TSX heritage clauses. Classes wrap theirs in class_heritage; interfaces use # extends_type_clause. A type_identifier inside one is already covered by an # INHERITS edge, so it must not also emit REFERENCES. _TS_HERITAGE_CLAUSES = frozenset({ "extends_clause", "implements_clause", "extends_type_clause", }) # TS/TSX declarations whose ``name`` field is a type_identifier. That occurrence # is the definition site, not a use of the type. _TS_TYPE_DECLARATIONS = frozenset({ "class_declaration", "abstract_class_declaration", "interface_declaration", "type_alias_declaration", "enum_declaration", }) _FUNCTION_TYPES: dict[str, list[str]] = { "python": ["function_definition"], "javascript": ["function_declaration", "method_definition", "arrow_function"], "typescript": ["function_declaration", "method_definition", "arrow_function"], "tsx": ["function_declaration", "method_definition", "arrow_function"], "go": ["function_declaration", "method_declaration"], "rust": ["function_item", "function_signature_item"], "java": ["method_declaration", "constructor_declaration"], "c": ["function_definition"], "cpp": ["function_definition", "declaration", "field_declaration"], "csharp": ["method_declaration", "constructor_declaration"], "ruby": ["method", "singleton_method"], "r": ["function_definition"], "perl": ["subroutine_declaration_statement", "method_declaration_statement"], "kotlin": ["function_declaration"], # Swift: initializers, deinitializers and subscripts are separate node # types, not `function_declaration`s, so they need listing alongside it — # the same way java/csharp list `constructor_declaration`. Their names come # from the `_get_name` Swift branch (the grammar has no usable name field). "swift": [ "function_declaration", "init_declaration", "deinit_declaration", "subscript_declaration", ], "php": ["function_definition", "method_declaration"], "scala": ["function_definition", "function_declaration"], # Solidity: events and modifiers use kind="Function" because the graph # schema has no dedicated kind for them. State variables are also modeled # as Function nodes (public ones auto-generate getters) and distinguished # via extra["solidity_kind"]. "solidity": [ "function_definition", "constructor_definition", "modifier_definition", "event_definition", "fallback_receive_definition", ], # Dart: function_signature covers both top-level functions and class methods # (class methods appear as method_signature > function_signature pairs; # the parser recurses into method_signature generically and then matches # function_signature inside it). "dart": ["function_signature"], "lua": ["function_declaration"], "luau": ["function_declaration"], # Objective-C: method_definition lives inside implementation_definition # inside class_implementation. C-style function_definition is also present # for main() and helper functions. "objc": ["method_definition", "function_definition"], # Bash: only function_definition; everything else is a command. "bash": ["function_definition"], # Elixir: def/defp/defmacro are all ``call`` nodes whose first # identifier matches. Dispatched via _extract_elixir_constructs. "elixir": [], # Nix: `attrpath = expr;` bindings become Function nodes — # handled in _extract_nix_constructs. "nix": [], # Zig: FnProto+Block pairs sit inside a Decl node; the standard generic # walker can't bridge the FnProto signature to its sibling Block body, # so the whole thing is dispatched via _extract_zig_constructs. "zig": [], "powershell": ["function_statement"], # Julia: short-form functions `f(x) = expr` parse as `assignment` nodes # (not a dedicated definition node) and are handled in # _extract_julia_constructs. "julia": [ "function_definition", "macro_definition", ], "verilog": ["task_declaration", "function_declaration", "always_construct"], # GDScript: ``func name(args) -> ReturnType:`` — includes ``static func``. "gdscript": ["function_definition"], # SQL: CREATE FUNCTION / CREATE PROCEDURE handled via _parse_sql dispatch. "sql": [], # HCL/Terraform: dispatched via _extract_hcl_constructs. "hcl": [], } _IMPORT_TYPES: dict[str, list[str]] = { "python": ["import_statement", "import_from_statement"], "javascript": ["import_statement"], "typescript": ["import_statement"], "tsx": ["import_statement"], "go": ["import_declaration"], "rust": ["use_declaration"], "java": ["import_declaration"], "c": ["preproc_include"], "cpp": ["preproc_include"], "csharp": ["using_directive"], "ruby": ["call"], # require/require_relative "r": ["call"], # library(), require(), source() — filtered downstream "perl": ["use_statement", "require_expression"], "kotlin": ["import_header"], "swift": ["import_declaration"], "php": ["namespace_use_declaration"], "scala": ["import_declaration"], "solidity": ["import_directive"], # Dart: import_or_export wraps library_import > import_specification > configurable_uri "dart": ["import_or_export"], # Lua/Luau: require() is a function_call, handled via _extract_lua_constructs "lua": [], "luau": [], # Objective-C: #import "..." and #include "..." both arrive as preproc_include # (tree-sitter-objc doesn't distinguish via a separate preproc_import node). "objc": ["preproc_include"], # Bash: source / . is a command — handled in _extract_bash_source below. "bash": [], # Elixir: alias/import/require/use are all ``call`` nodes — # handled in _extract_elixir_constructs. "elixir": [], # Nix: `import ./x.nix`, `callPackage ./y.nix {}`, and flake # `inputs.*.url` strings become IMPORTS_FROM edges — # handled in _extract_nix_constructs. "nix": [], # Zig: @import("path") is a SuffixExpr containing a BUILTINIDENTIFIER # "@import" + FnCallArguments holding a STRINGLITERALSINGLE. Handled in # _extract_zig_constructs as part of VarDecl processing. "zig": [], "powershell": [], # Julia: import/using are import_statement nodes. "julia": ["import_statement", "using_statement"], "verilog": ["package_import_declaration"], # GDScript has no ``import`` keyword. The closest analogue is # ``extends OtherClass`` / ``extends "res://path.gd"``, which establishes # a hard dependency on the parent script. preload()/load() calls remain # as ordinary CALLS edges. "gdscript": ["extends_statement"], # SQL: table references extracted as IMPORTS_FROM via _parse_sql dispatch. "sql": [], # HCL/Terraform: module source attributes become IMPORTS_FROM via # _extract_hcl_constructs. "hcl": [], } _CALL_TYPES: dict[str, list[str]] = { "python": ["call"], "javascript": ["call_expression", "new_expression"], "typescript": ["call_expression", "new_expression"], "tsx": ["call_expression", "new_expression"], "go": ["call_expression"], "rust": ["call_expression", "macro_invocation"], "java": ["method_invocation", "object_creation_expression", "method_reference"], "c": ["call_expression"], "cpp": ["call_expression"], "csharp": ["invocation_expression", "object_creation_expression"], "ruby": ["call", "method_call"], "r": ["call"], "perl": [ "function_call_expression", "method_call_expression", "ambiguous_function_call_expression", ], "kotlin": ["call_expression"], "swift": ["call_expression"], "php": [ "function_call_expression", "member_call_expression", "scoped_call_expression", "nullsafe_member_call_expression", "object_creation_expression", ], "scala": ["call_expression", "instance_expression", "generic_function"], "solidity": ["call_expression"], "lua": ["function_call"], "luau": ["function_call"], # Objective-C: [receiver message:args] produces message_expression; # C-style foo(x) produces call_expression. "objc": ["message_expression", "call_expression"], # Bash: every command invocation is a "command" node. "bash": ["command"], # Elixir: everything is a ``call`` node — dispatched via # _extract_elixir_constructs which filters out def/defmodule/alias/etc. # before treating what's left as a real call. "elixir": [], # Nix: function application is ubiquitous; only import/callPackage # produce edges, in _extract_nix_constructs. "nix": [], # Zig calls are SuffixExpr/FieldOrFnCall nodes containing FnCallArguments. # Mapping SuffixExpr here would over-match (every expression is a # SuffixExpr); calls are walked explicitly in # _extract_zig_calls_in_subtree from inside function bodies. "zig": [], "powershell": ["command_expression"], "julia": [ "call_expression", "broadcast_call_expression", "macrocall_expression", ], "verilog": [ "module_instantiation", "interface_instantiation", "function_subroutine_call", "subroutine_call", "system_tf_call", ], # GDScript: bare calls produce ``call``; ``obj.method()`` is an # ``attribute`` node whose right-hand side is an ``attribute_call``. "gdscript": ["call", "attribute_call"], # SQL: no call edges extracted (grammar too unreliable for procedure calls). "sql": [], # HCL/Terraform: resource references dispatched via _extract_hcl_constructs. "hcl": [], } def _builtin_language_names() -> frozenset[str]: """All built-in language identifiers. Used to stop config-driven custom languages (languages.toml) from shadowing a built-in language name — built-ins always win. """ return ( frozenset(EXTENSION_TO_LANGUAGE.values()) | frozenset(_CLASS_TYPES) | frozenset(_FUNCTION_TYPES) | frozenset(_IMPORT_TYPES) | frozenset(_CALL_TYPES) ) # Patterns that indicate a test function _TEST_PATTERNS = [ re.compile(r"^test_"), re.compile(r"^Test"), re.compile(r"_test$"), re.compile(r"\.test\."), re.compile(r"\.spec\."), re.compile(r"_spec$"), ] _TEST_FILE_PATTERNS = [ re.compile(r"test_.*\.py$"), re.compile(r".*_test\.py$"), re.compile(r".*\.test\.[jt]sx?$"), re.compile(r".*\.spec\.[jt]sx?$"), re.compile(r".*_test\.go$"), re.compile(r"tests?/"), re.compile(r"[\\/]__tests__[\\/]"), re.compile(r".*_test\.dart$"), re.compile(r"test[_-].*\.[rR]$"), re.compile(r"tests/testthat/"), re.compile(r".*Test\.kt$"), re.compile(r".*Test\.java$"), re.compile(r".*_test\.resi?$"), re.compile(r".*\.test\.resi?$"), re.compile(r"test/runtests\.jl$"), re.compile(r"test/.*\.jl$"), ] _TEST_RUNNER_NAMES = frozenset({ "describe", "it", "test", "beforeEach", "afterEach", "beforeAll", "afterAll", # Mocha TDD interface: `suite` is the describe-equivalent. # `test`, the it-equivalent, is already covered above. "suite", }) # Annotations/decorators that mark test methods (JUnit, TestNG, etc.) _TEST_ANNOTATIONS = frozenset({ "Test", "ParameterizedTest", "RepeatedTest", "TestFactory", "org.junit.Test", "org.junit.jupiter.api.Test", # Rust: built-in `#[test]` plus common async-runtime + framework # variants. Stripped of the `#[ ]` wrapper before lookup. "test", "tokio::test", "async_std::test", "rstest", "rstest::rstest", "proptest", }) # Spring stereotype annotations that mark classes as managed beans _SPRING_STEREOTYPE_ANNOTATIONS = frozenset({ "Component", "Service", "Repository", "Controller", "RestController", "Configuration", "Indexed", "ControllerAdvice", "RestControllerAdvice", "EventListener", }) # Spring DI injection annotations (field/setter/constructor-level) _SPRING_INJECT_ANNOTATIONS = frozenset({ "Autowired", "Inject", "Resource", }) _SPRING_PLACEHOLDER_RE = re.compile(r"\$\{([^{}]+)\}") # Temporal workflow/activity interface markers _TEMPORAL_INTERFACE_ANNOTATIONS = frozenset({ "WorkflowInterface", "ActivityInterface", }) # Temporal method-level markers _TEMPORAL_METHOD_ANNOTATIONS = frozenset({ "WorkflowMethod", "ActivityMethod", "SignalMethod", "QueryMethod", }) # Kafka consumer annotations (annotation-based pattern) _KAFKA_LISTENER_ANNOTATIONS = frozenset({"KafkaListener", "KafkaHandler"}) # Kafka consumer field types (reactive / imperative) _KAFKA_CONSUMER_TYPES = frozenset({ "KafkaReceiver", "ReactiveKafkaConsumerTemplate", "MessageListenerContainer", "ConcurrentMessageListenerContainer", }) # Kafka producer field types _KAFKA_PRODUCER_TYPES = frozenset({ "KafkaTemplate", "KafkaOperations", "ReactiveKafkaProducerTemplate", "KafkaSender", }) # Spring scheduling annotations. ``Scheduled`` is repeatable; ``Schedules`` # is its explicit Java container form. _SPRING_SCHEDULED_ANNOTATIONS = frozenset({"Scheduled", "Schedules"}) _SPRING_EVENT_LISTENER_ANNOTATIONS = frozenset({"EventListener"}) _SPRING_EVENT_PUBLISH_METHODS = frozenset({"publishEvent"}) _JAVA_PACKAGE_KEY = "__crg_java_package__" _SPRING_REQUEST_PREFIX_KEY = "__crg_spring_request_prefix__:" _JS_IMPORT_ORIGINAL_PREFIX_KEY = "__crg_js_import_original__:" _SPRING_REQUEST_MAPPINGS = { "DeleteMapping": ("DELETE",), "GetMapping": ("GET",), "PatchMapping": ("PATCH",), "PostMapping": ("POST",), "PutMapping": ("PUT",), "RequestMapping": (), } _SPRING_WEBFLUX_HTTP_VERBS = frozenset({"DELETE", "GET", "PATCH", "POST", "PUT"}) _HTTP_REQUEST_METHODS = frozenset({ "CONNECT", "DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT", "TRACE", }) # --------------------------------------------------------------------------- # VB.NET regex patterns and helpers (no tree-sitter grammar bundled) # --------------------------------------------------------------------------- _VBNET_IDENT = r"(?:[A-Za-z_][A-Za-z0-9_]*|\[[^\]\r\n]+\])" _VBNET_DOTTED_IDENT = rf"{_VBNET_IDENT}(?:\.{_VBNET_IDENT})*" _VBNET_MODIFIER_WORDS = ( "Public", "Private", "Protected", "Friend", "Partial", "Shared", "Static", "MustInherit", "NotInheritable", "Overridable", "Overrides", "MustOverride", "Overloads", "Default", "ReadOnly", "WriteOnly", "Shadows", "Async", "Iterator", "Declare", "Narrowing", "Widening", ) _VBNET_MODIFIER_RE = rf"(?:(?:{'|'.join(_VBNET_MODIFIER_WORDS)})\s+)*" _VBNET_IMPORT_RE = re.compile(r"^\s*Imports\s+(.+?)\s*$", re.IGNORECASE) _VBNET_NAMESPACE_RE = re.compile( rf"^\s*Namespace\s+(?P{_VBNET_DOTTED_IDENT})\s*$", re.IGNORECASE, ) _VBNET_END_NAMESPACE_RE = re.compile( r"^\s*End\s+Namespace\b", re.IGNORECASE, ) _VBNET_TYPE_RE = re.compile( rf"^\s*{_VBNET_MODIFIER_RE}" rf"(?PClass|Interface|Structure|Module|Enum)\s+" rf"(?P{_VBNET_IDENT})\b(?P.*)$", re.IGNORECASE, ) _VBNET_END_TYPE_RE = re.compile( r"^\s*End\s+(?PClass|Interface|Structure|Module|Enum)\b", re.IGNORECASE, ) _VBNET_MEMBER_RE = re.compile( rf"^\s*(?P{_VBNET_MODIFIER_RE})" rf"(?PFunction|Sub|Property)\s+" rf"(?P{_VBNET_IDENT})\b(?P.*)$", re.IGNORECASE, ) _VBNET_OPERATOR_RE = re.compile( rf"^\s*(?P{_VBNET_MODIFIER_RE})" r"(?POperator)\s+(?P\S+)\s*(?P.*)$", re.IGNORECASE, ) _VBNET_END_MEMBER_RE = re.compile( r"^\s*End\s+(Function|Sub|Property|Operator)\b", re.IGNORECASE, ) _VBNET_INHERITS_RE = re.compile(r"\bInherits\s+(.+?)\s*$", re.IGNORECASE) _VBNET_IMPLEMENTS_RE = re.compile(r"\bImplements\s+(.+?)\s*$", re.IGNORECASE) _VBNET_NEW_RE = re.compile( rf"\bNew\s+(?P{_VBNET_DOTTED_IDENT})\s*(?:\(Of\b[^)]*\))?\s*\(", re.IGNORECASE, ) _VBNET_CALL_RE = re.compile( rf"(?{_VBNET_DOTTED_IDENT})\s*\(", re.IGNORECASE, ) _VBNET_CALL_KEYWORDS = frozenset({ "addhandler", "and", "andalso", "as", "call", "case", "catch", "class", "cobj", "continue", "ctype", "directcast", "do", "each", "else", "elseif", "end", "enum", "erase", "error", "event", "exit", "finally", "for", "function", "get", "gettype", "getxmlnamespace", "global", "gosub", "goto", "if", "implements", "imports", "inherits", "interface", "loop", "module", "mustinherit", "new", "next", "not", "nothing", "operator", "option", "or", "orelse", "property", "raiseevent", "redim", "rem", "removehandler", "resume", "return", "select", "set", "step", "stop", "structure", "sub", "synclock", "then", "throw", "to", "try", "typeof", "until", "using", "when", "while", "with", "withevents", "xor", }) def _vbnet_normalize_name(value: str) -> str: """Remove VB escaping while preserving dotted identity.""" return ".".join( part[1:-1] if part.startswith("[") and part.endswith("]") else part for part in value.strip().split(".") ) def _strip_vbnet_noise(text: str) -> str: """Blank VB comments and string contents while preserving line numbers.""" cleaned_lines: list[str] = [] for raw_line in text.splitlines(keepends=True): line = raw_line.rstrip("\r\n") ending = raw_line[len(line):] if re.match(r"^\s*Rem\b", line, re.IGNORECASE): cleaned_lines.append(" " * len(line) + ending) continue out: list[str] = [] in_string = False i = 0 while i < len(line): char = line[i] if char == '"': out.append(char) if in_string and i + 1 < len(line) and line[i + 1] == '"': out.append(" ") i += 2 continue in_string = not in_string i += 1 continue if not in_string and char == "'": out.append(" " * (len(line) - i)) break out.append(" " if in_string else char) i += 1 cleaned_lines.append("".join(out) + ending) return "".join(cleaned_lines) def _vbnet_logical_lines(cleaned: str) -> list[tuple[int, int, str]]: """Join explicit and parenthesized VB continuations with source ranges.""" logical: list[tuple[int, int, str]] = [] parts: list[str] = [] start_line = 1 depth = 0 for line_no, raw_line in enumerate(cleaned.splitlines(), start=1): stripped = raw_line.strip() if not stripped and not parts: continue if not parts: start_line = line_no explicit = stripped.endswith("_") if explicit: stripped = stripped[:-1].rstrip() parts.append(stripped) depth += stripped.count("(") - stripped.count(")") if explicit or depth > 0: continue logical.append((start_line, line_no, " ".join(parts))) parts = [] depth = 0 if parts: logical.append((start_line, len(cleaned.splitlines()) or 1, " ".join(parts))) return logical def _vbnet_parenthesized(text: str, start: int) -> tuple[str, int] | None: """Return one balanced parenthesized group and its exclusive end.""" if start >= len(text) or text[start] != "(": return None depth = 0 for index in range(start, len(text)): if text[index] == "(": depth += 1 elif text[index] == ")": depth -= 1 if depth == 0: return text[start + 1:index], index + 1 return None def _vbnet_split_top_level(value: str) -> list[str]: """Split a comma list without splitting generic/array groups.""" parts: list[str] = [] start = 0 depth = 0 for index, char in enumerate(value): if char == "(": depth += 1 elif char == ")": depth = max(0, depth - 1) elif char == "," and depth == 0: parts.append(value[start:index]) start = index + 1 parts.append(value[start:]) return parts def _vbnet_type_parameters(rest: str) -> tuple[list[str], str]: tail = rest.lstrip() if not tail.lower().startswith("(of "): return [], rest group = _vbnet_parenthesized(tail, 0) if group is None: return [], rest content, end = group params = [ part.strip().split()[0] for part in _vbnet_split_top_level(content[3:]) if part.strip() ] return params, tail[end:] def _vbnet_signature_parts( rest: str, ) -> tuple[Optional[str], Optional[str], list[str]]: """Extract parameters, return type, and method type parameters.""" type_params, tail = _vbnet_type_parameters(rest) tail = tail.lstrip() params: Optional[str] = None if tail.startswith("("): group = _vbnet_parenthesized(tail, 0) if group is not None: raw_params, end = group params = re.sub(r"\s+", " ", raw_params).strip() tail = tail[end:] return_type = None return_match = re.search( r"\bAs\s+(.+?)(?=\s+(?:Implements|Handles)\b|$)", tail, re.IGNORECASE, ) if return_match: return_type = re.sub(r"\s+", " ", return_match.group(1)).strip() return params, return_type, type_params def _vbnet_relationship_targets(value: str) -> list[str]: targets: list[str] = [] for raw_part in _vbnet_split_top_level(value): part = raw_part.strip() if not part: continue if "=" in part: part = part.split("=", 1)[1].strip() if part.lower().startswith("global."): part = part[7:] part = re.sub(r"\s*\(Of\b.*\)\s*$", "", part, flags=re.IGNORECASE) if re.fullmatch(_VBNET_DOTTED_IDENT, part): targets.append(_vbnet_normalize_name(part)) return targets # --------------------------------------------------------------------------- # ReScript regex patterns and helpers (no tree-sitter grammar bundled) # --------------------------------------------------------------------------- _RESCRIPT_IDENT = r"[A-Za-z_][A-Za-z0-9_']*" # `module Name =`, `module type Name =`, `module Name: {`, `module Name: (Sig) => {` _RESCRIPT_MODULE_RE = re.compile( r"^\s*module\s+(?:type\s+)?([A-Z][A-Za-z0-9_']*)\s*[:=]", re.MULTILINE, ) # Optional leading decorator block on the same line, e.g. `@deriving(foo)`. _RESCRIPT_DECORATOR_PREFIX = r"(?:@[A-Za-z_][A-Za-z0-9_']*(?:\([^)]*\))?\s+)*" # `let [rec] name` / `and name` — captures binding name. Multi-line decorators # on prior lines don't interfere (they end with a newline and the anchor # restarts on the next line); same-line decorators are tolerated. _RESCRIPT_LET_RE = re.compile( rf"^\s*{_RESCRIPT_DECORATOR_PREFIX}" rf"(?:let\s+(?:rec\s+)?|and\s+)({_RESCRIPT_IDENT})\b", re.MULTILINE, ) # `external name: sig = "..."` _RESCRIPT_EXTERNAL_RE = re.compile( rf"^\s*{_RESCRIPT_DECORATOR_PREFIX}external\s+({_RESCRIPT_IDENT})\s*:", re.MULTILINE, ) # `type name` / `type rec name` / `type name<'a>` _RESCRIPT_TYPE_RE = re.compile( rf"^\s*{_RESCRIPT_DECORATOR_PREFIX}type\s+(?:rec\s+)?({_RESCRIPT_IDENT})\b", re.MULTILINE, ) # `open Foo` / `include Foo.Bar` _RESCRIPT_OPEN_RE = re.compile( r"^\s*(open|include)\s+([A-Z][A-Za-z0-9_'.]*)", re.MULTILINE, ) # `module X = Foo.Bar` with no `{` body — a module alias/re-export. Distinct # from `module X = { ... }` (handled by _RESCRIPT_MODULE_RE + brace scan). _RESCRIPT_MODULE_ALIAS_RE = re.compile( r"^\s*module\s+([A-Z][A-Za-z0-9_']*)\s*=\s*" r"([A-Z][A-Za-z0-9_']*(?:\.[A-Za-z_][A-Za-z0-9_']*)*)\s*$", re.MULTILINE, ) # JSX opening tag: ``, `<=`, `<-`, or a generic-type # parameter (we approximate by requiring the char before `<` to be space, # newline, `{`, `(`, `,`, `>`, `}`, or BOF). _RESCRIPT_JSX_RE = re.compile( r"(?:^|(?<=[\s{(,>}]))" r"<([A-Z][A-Za-z0-9_']*(?:\.[A-Z][A-Za-z0-9_']*)*)\b", re.MULTILINE, ) # `@module("path")` — source module for an external binding _RESCRIPT_MODULE_ATTR_RE = re.compile( r'@module\(\s*"([^"]+)"\s*\)', ) # `Ident(`, `Mod.fn(` — anything that looks like a call site. Preceded by a # non-identifier char to avoid matching suffixes of identifiers. _RESCRIPT_CALL_RE = re.compile( rf"(? str: """Replace ReScript comments and string/backtick content with spaces. Newlines are preserved so absolute offsets still map back to accurate line numbers. ReScript block comments may nest, so we track depth. """ out: list[str] = [] i = 0 n = len(text) while i < n: c = text[i] nxt = text[i + 1] if i + 1 < n else "" # Line comment if c == "/" and nxt == "/": while i < n and text[i] != "\n": out.append(" ") i += 1 continue # Nestable block comment if c == "/" and nxt == "*": depth = 1 out.append(" ") i += 2 while i < n and depth > 0: if i + 1 < n and text[i] == "/" and text[i + 1] == "*": depth += 1 out.append(" ") i += 2 elif i + 1 < n and text[i] == "*" and text[i + 1] == "/": depth -= 1 out.append(" ") i += 2 else: out.append("\n" if text[i] == "\n" else " ") i += 1 continue # Double-quoted string — blank content, keep quotes + newlines. if c == '"': out.append('"') i += 1 while i < n and text[i] != '"': if text[i] == "\\" and i + 1 < n: out.append(" ") i += 2 continue out.append("\n" if text[i] == "\n" else " ") i += 1 if i < n: out.append('"') i += 1 continue # Backtick template string — blank content, preserve newlines. if c == "`": out.append("`") i += 1 while i < n and text[i] != "`": out.append("\n" if text[i] == "\n" else " ") i += 1 if i < n: out.append("`") i += 1 continue out.append(c) i += 1 return "".join(out) def _rescript_brace_depth_array(cleaned: str) -> list[int]: """Compute brace depth at every offset in `cleaned` (comment/string-stripped). Returned array has length len(cleaned); `depth[i]` is the depth immediately before the character at position i. """ depth = [0] * (len(cleaned) + 1) d = 0 for i, c in enumerate(cleaned): depth[i] = d if c == "{": d += 1 elif c == "}": d = max(0, d - 1) depth[len(cleaned)] = d return depth def _scan_rescript_modules(cleaned: str, offset_to_line) -> list[dict]: """Find `module Name = { ... }` blocks and their offset/line ranges. Returns dicts with name, start/end offsets, start/end lines, and parent module name (or None for top-level). """ modules: list[dict] = [] n = len(cleaned) # Module aliases (`module X = Foo.Bar`) also match _RESCRIPT_MODULE_RE but # have no brace body — skip them here to avoid the greedy `{`-scanner # swallowing the next unrelated block (e.g. a `let` body). alias_starts = { m.start() for m in _RESCRIPT_MODULE_ALIAS_RE.finditer(cleaned) } for match in _RESCRIPT_MODULE_RE.finditer(cleaned): if match.start() in alias_starts: continue name = match.group(1) header_start = match.start() # Find the first `{` after the header's `:` or `=`. To avoid grabbing # a `{` from an unrelated following statement, require that the chars # between `match.end()` and `brace_open` contain no definition-starting # keywords (`let`, `type`, `module`, `external`). brace_open = cleaned.find("{", match.end()) if brace_open == -1: continue between = cleaned[match.end():brace_open] if re.search( r"(?:^|\s)(?:let|type|module|external|and)\s", between, ): continue # Walk braces to find the matching close. depth = 1 j = brace_open + 1 while j < n and depth > 0: c = cleaned[j] if c == "{": depth += 1 elif c == "}": depth -= 1 j += 1 brace_close = j - 1 if depth == 0 else n - 1 modules.append({ "name": name, "start_off": header_start, "end_off": brace_close, "body_start_off": brace_open + 1, "start_line": offset_to_line(header_start), "end_line": offset_to_line(brace_close), "parent": None, }) # Parent = innermost strictly-containing module. for i, m in enumerate(modules): parent_name = None parent_start = -1 for j, other in enumerate(modules): if i == j: continue if ( other["start_off"] < m["start_off"] and other["end_off"] > m["end_off"] and other["start_off"] > parent_start ): parent_name = other["name"] parent_start = other["start_off"] m["parent"] = parent_name return modules def _is_test_file(path: str) -> bool: return any(p.search(path) for p in _TEST_FILE_PATTERNS) def _is_test_function( name: str, file_path: str, decorators: tuple[str, ...] = (), ) -> bool: """A function is a test if its name matches test patterns, it lives in a test file and has a test-runner name, or it has a @Test annotation. """ if any(p.search(name) for p in _TEST_PATTERNS): return True if _is_test_file(file_path) and name in _TEST_RUNNER_NAMES: return True if decorators and any(d in _TEST_ANNOTATIONS for d in decorators): return True return False # Documentation summaries are stored in ``NodeInfo.extra`` so the graph # schema remains backward compatible. A hard cap keeps parser metadata and # semantic-search input bounded even when a source file contains a very long # API reference as its docstring. _MAX_DOCSTRING_CHARS = 400 _DOC_COMMENT_NODE_TYPES = frozenset({ "comment", "block_comment", "line_comment", "doc_comment", }) _DOC_COMMENT_SKIP_TYPES = frozenset({"attribute_item", "decorator"}) _DOC_COMMENT_WRAPPER_TYPES = frozenset({ "export_statement", "template_declaration", }) _CSHARP_SUMMARY_RE = re.compile( r"]*)?>(.*?)", re.IGNORECASE | re.DOTALL, ) _XML_TAG_RE = re.compile(r"<[^>]+>") _DOC_PARAGRAPH_TAG_RE = re.compile( r"]*)?>", re.IGNORECASE, ) _DOC_INLINE_TAG_RE = re.compile( r"\{@(?:code|literal|link)\s+([^}]+)\}", re.IGNORECASE, ) _DOC_BRIEF_RE = re.compile(r"^\s*[@\\]brief\s+", re.IGNORECASE) def _clean_docstring_summary(raw: str, language: str) -> str: """Return a whitespace-stable, first-paragraph documentation summary.""" text = raw.replace("\r\n", "\n").replace("\r", "\n") if language == "csharp": summary = _CSHARP_SUMMARY_RE.search(text) if summary: text = summary.group(1) if language != "python": text = _DOC_PARAGRAPH_TAG_RE.sub("\n\n", text) text = _DOC_INLINE_TAG_RE.sub(r"\1", text) text = _XML_TAG_RE.sub(" ", text) text = html.unescape(text) text = _DOC_BRIEF_RE.sub("", text) lines = [line.strip() for line in text.splitlines()] while lines and not lines[0]: lines.pop(0) paragraph: list[str] = [] for line in lines: if not line: break if paragraph and line.startswith(("@param", "@return", "\\param", "\\return")): break paragraph.append(line) return " ".join(" ".join(paragraph).split())[:_MAX_DOCSTRING_CHARS] def _strip_block_doc_comment(text: str) -> str: """Remove a Doxygen/Javadoc-style wrapper and per-line ``*`` prefix.""" if text.startswith(("/**", "/*!")): text = text[3:] elif text.startswith("/*"): text = text[2:] if text.endswith("*/"): text = text[:-2] lines: list[str] = [] for line in text.splitlines(): stripped = line.lstrip() if stripped.startswith("*"): stripped = stripped[1:] if stripped.startswith(" "): stripped = stripped[1:] lines.append(stripped) return "\n".join(lines) def _modifier_annotation_names(node) -> list[str]: """Return annotation names from a ``modifiers`` child of *node*. Covers Java/Kotlin/C# where annotations live inside a ``modifiers`` node as ``annotation`` / ``marker_annotation`` children. The leading ``@`` is stripped. See: #295 """ names: list[str] = [] for sub in node.children: if sub.type == "modifiers": for mod in sub.children: if mod.type in ("annotation", "marker_annotation"): text = mod.text.decode("utf-8", errors="replace") names.append(text.lstrip("@").strip()) return names def _python_decorator_names(node) -> list[str]: """Return decorators wrapping a Python definition in source order.""" parent = node.parent if parent is None or parent.type != "decorated_definition": return [] names: list[str] = [] for sibling in parent.children: if sibling.type != "decorator": continue text = sibling.text.decode("utf-8", errors="replace") names.append(text.lstrip("@").strip()) return names def _csharp_attribute_names(node) -> list[str]: """Return C# attribute names from ``attribute_list`` children of *node*. C# attributes (``[HttpGet]``, ``[Authorize]``, ``[ApiController]``) are ``attribute_list`` nodes, each wrapping one or more ``attribute`` nodes whose first ``identifier`` is the attribute name. The bracket wrapper and any argument list are dropped. See: #295 """ names: list[str] = [] for sub in node.children: if sub.type != "attribute_list": continue for attr in sub.children: if attr.type != "attribute": continue for ident in attr.children: if ident.type in ("identifier", "qualified_name"): names.append(ident.text.decode("utf-8", errors="replace").strip()) break return names _PHPUNIT_TEST_ATTRIBUTE = "phpunit\\framework\\attributes\\test" def _php_attribute_aliases(node) -> dict[str, str]: """Return aliases that resolve specifically to PHPUnit's Test attribute.""" root = node while root.parent is not None: root = root.parent aliases: dict[str, str] = {} stack = [root] while stack: current = stack.pop() if current.type == "namespace_use_clause": alias = current.child_by_field_name("alias") if alias is not None: target = next( ( child for child in current.children if child.type in ("name", "qualified_name") and child != alias ), None, ) if target is not None: alias_text = alias.text.decode( "utf-8", errors="replace", ).strip() target_text = target.text.decode( "utf-8", errors="replace", ).strip() if current.parent.type == "namespace_use_group": declaration = current.parent.parent prefix = next( ( child for child in declaration.children if child.type == "namespace_name" ), None, ) if prefix is not None: prefix_text = prefix.text.decode( "utf-8", errors="replace", ).strip() target_text = f"{prefix_text}\\{target_text}" normalized_target = target_text.lstrip("\\").casefold() if normalized_target == _PHPUNIT_TEST_ATTRIBUTE: aliases[alias_text.casefold()] = "Test" stack.extend(reversed(current.children)) return aliases def _php_attribute_names(node) -> list[str]: """Return PHP attribute names from ``attribute_list`` children of *node*. PHP 8 attributes (``#[Test]``, ``#[DataProvider('x')]``) are ``attribute_list`` nodes wrapping one or more ``attribute_group`` nodes, each wrapping one or more ``attribute`` nodes whose ``name`` child is the attribute name -- one level deeper than C#'s ``attribute_list > attribute``. See: #693 """ names: list[str] = [] for sub in node.children: if sub.type != "attribute_list": continue for group in sub.children: if group.type != "attribute_group": continue for attr in group.children: if attr.type != "attribute": continue for ident in attr.children: if ident.type in ("name", "qualified_name"): raw_name = ident.text.decode( "utf-8", errors="replace", ).strip() normalized = raw_name.lstrip("\\") if normalized.casefold() == _PHPUNIT_TEST_ATTRIBUTE: names.append("Test") else: names.append(normalized) break if not names: return [] aliases = _php_attribute_aliases(node) return [aliases.get(name.casefold(), name) for name in names] _PHP_TEST_DOC_TAG_RE = re.compile(r"(? bool: """Return True if a PHPUnit ``/** @test */`` docblock precedes *node*. PHPUnit's legacy convention marks a test method with an ``@test`` tag in the doc comment immediately above it, instead of a ``test_`` name prefix or a ``#[Test]`` attribute. Tree-sitter represents that comment as a preceding sibling of the ``method_declaration``, not a child, so it needs its own check rather than reuse of the attribute-based extraction above. See: #693 """ sib = node.prev_sibling return bool( sib is not None and sib.type == "comment" and _PHP_TEST_DOC_TAG_RE.search(sib.text.decode("utf-8", errors="replace")) ) def _csharp_namespaces(root_node) -> list[str]: """Return all namespaces declared in a C# compilation unit. Handles both the block form (``namespace_declaration``) and the C# 10+ file-scoped form (``file_scoped_namespace_declaration``). A single file may declare multiple namespaces; all are returned in source order. See: #310 """ namespaces: list[str] = [] stack = [(root_node, None)] while stack: node, parent_namespace = stack.pop() current_namespace = parent_namespace if node.type in ( "namespace_declaration", "file_scoped_namespace_declaration", ): for c in node.children: if c.type in ("qualified_name", "identifier"): text = c.text.decode("utf-8", errors="replace").strip() if text: current_namespace = ( f"{parent_namespace}.{text}" if parent_namespace else text ) namespaces.append(current_namespace) break stack.extend( (child, current_namespace) for child in reversed(node.children) ) return namespaces def file_hash(path: Path) -> str: """SHA-256 hash of file contents.""" return hashlib.sha256(path.read_bytes()).hexdigest() # --------------------------------------------------------------------------- # HCL / Terraform helpers (module-level; no self needed) # --------------------------------------------------------------------------- def _hcl_text(node) -> str: """Decode tree-sitter node bytes to a string.""" return node.text.decode("utf-8", errors="replace") def _hcl_child(node, *types: str): """Return the first direct child whose type is in *types*, or None.""" return next((c for c in node.children if c.type in types), None) def _hcl_block_name(prefix: str, labels: list[str], n_labels: int) -> Optional[str]: """Build *prefix.label0[.label1]* from the first *n_labels* labels, or None.""" if len(labels) < n_labels: return None return ".".join([prefix] + labels[:n_labels]) # Terraform reference-namespace prefixes (special roots, not resource types). # These roots are *not* resource type names, so they must never be mapped to # ``resource..*``. The block-local iterators (each, count, self) and # built-in namespace objects (path, terraform) are also included so that # expressions like ``each.value.id`` or ``terraform.workspace`` do not # generate spurious REFERENCES edges. _HCL_REF_PREFIXES: frozenset[str] = frozenset({ "var", "module", "local", "data", # block-local meta-argument iterators "each", "count", "self", # built-in namespace objects "path", "terraform", }) def _hcl_ref_target(root: str, attrs: list[str]) -> Optional[str]: """Map a ``variable_expr.get_attr*`` chain to its canonical graph name.""" if root == "var" and attrs: return f"var.{attrs[0]}" if root == "module" and attrs: return f"module.{attrs[0]}" if root == "local" and attrs: return f"local.{attrs[0]}" if root == "data" and len(attrs) >= 2: return f"data.{attrs[0]}.{attrs[1]}" if root not in _HCL_REF_PREFIXES and attrs: return f"resource.{root}.{attrs[0]}" return None def _hcl_variable_refs(expr_node): """Yield (root, attrs, line) for each ``variable_expr get_attr*`` chain that is a direct child sequence inside *expr_node*.""" children = expr_node.children i = 0 while i < len(children): child = children[i] if child.type != "variable_expr": i += 1 continue ident = _hcl_child(child, "identifier") if ident is None: i += 1 continue j, attrs = i + 1, [] while j < len(children) and children[j].type == "get_attr": id_node = _hcl_child(children[j], "identifier") if id_node: attrs.append(_hcl_text(id_node)) j += 1 yield _hcl_text(ident), attrs, child.start_point[0] + 1 i = j # Node types to recurse into when scanning for HCL variable references. # ``function_call`` / ``function_arguments`` ensure that variable references # inside calls like ``length(var.x)`` are extracted. # ``quoted_template`` / ``template_interpolation`` ensure that variable # references inside ``"${var.x}"`` template strings are extracted. _HCL_RECURSE_TYPES: frozenset[str] = frozenset({ "expression", "body", "block", "attribute", "tuple", "object", "object_elem", "collection_value", "template_expr", "for_expr", "for_tuple_expr", "for_object_expr", "for_intro", "for_cond", "conditional", # variable refs inside function call arguments, e.g. length(var.x) "function_call", "function_arguments", # variable refs inside template string interpolations, e.g. "${var.x}" "quoted_template", "template_interpolation", }) def _hcl_for_iterator_names(for_expr_node) -> frozenset[str]: """Return loop-local symbols declared by a Terraform for-expression.""" stack = [for_expr_node] while stack: current = stack.pop() if current.type == "for_intro": return frozenset( _hcl_text(child) for child in current.children if child.type == "identifier" ) stack.extend(reversed(current.children)) return frozenset() def _hcl_dynamic_iterator_name(block_node) -> Optional[str]: """Return the iterator symbol for a ``dynamic`` block, or ``None``. Defaults to the dynamic block's string label (e.g. ``"setting"`` becomes the symbol ``setting``). Can be overridden by an ``iterator = `` attribute inside the block body. Returns ``None`` when *block_node* is not a ``dynamic`` block. This is used by ``_walk_hcl_expressions`` to build a *local_names* scope so that iterator references such as ``setting.value[...]`` or ``origin_group.key`` do not produce spurious ``resource.*`` REFERENCES edges. """ id_node = _hcl_child(block_node, "identifier") if id_node is None or _hcl_text(id_node) != "dynamic": return None # Default iterator name = the string label of the dynamic block. # Block children: identifier("dynamic"), string_lit(label), body default_name: Optional[str] = None for child in block_node.children: if child.type == "string_lit": tmpl = _hcl_child(child, "template_literal") default_name = _hcl_text(tmpl) if tmpl is not None else _hcl_text(child).strip('"') break if default_name is None: return None # Check for optional ``iterator = `` override inside the block body. # The value is an unquoted identifier expression, e.g. ``iterator = srv``. body_node = _hcl_child(block_node, "body") if body_node is not None: for attr in body_node.children: if attr.type != "attribute": continue key = _hcl_child(attr, "identifier") if key is None or _hcl_text(key) != "iterator": continue expr = _hcl_child(attr, "expression") if expr is None: continue # Handles both bare identifier (``srv``) and quoted string (``"srv"``) raw = _hcl_text(expr).strip().strip('"') if raw and raw.isidentifier(): return raw return default_name # Dispatch table: block_type → (graph_kind, name_prefix, n_labels, emit_refs) # "terraform" and unknown types are absent so they are silently skipped. _HCL_BLOCK_CFG: dict[str, tuple[str, str, int, bool]] = { "resource": ("Class", "resource", 2, True), "data": ("Class", "data", 2, True), "module": ("Class", "module", 1, True), "variable": ("Function", "var", 1, False), "output": ("Function", "output", 1, True), "provider": ("Function", "provider", 1, True), } # --------------------------------------------------------------------------- # Ansible YAML helpers (module-level so tests can import them directly) # --------------------------------------------------------------------------- def _is_ansible_path(path: Path) -> bool: """Return True if the path suggests an Ansible YAML file by directory convention.""" parts = {p.lower() for p in path.parts} return bool(parts & _ANSIBLE_PATH_COMPONENTS) or path.name.lower() in _ANSIBLE_PLAYBOOK_NAMES def _is_ansible_content(source: bytes) -> bool: """Lightweight byte-scan: does this YAML look like an Ansible file? Checks that the file is a top-level list and contains at least one Ansible-specific structural marker (hosts, tasks, handlers, import_playbook, or a known module key / FQCN pattern). """ try: text = source.decode("utf-8", errors="replace") except Exception: return False for line in text.splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#") or stripped == "---": continue # skip blank lines, comments, and YAML document markers if not (line.startswith("- ") or line == "-"): return False break else: return False has_hosts = bool(re.search(r"^\s+hosts\s*:", text, re.MULTILINE)) has_tasks = bool(re.search(r"^\s+tasks\s*:", text, re.MULTILINE)) has_handlers = bool(re.search(r"^\s+handlers\s*:", text, re.MULTILINE)) has_import_pb = bool(re.search(r"^\s+import_playbook\s*:", text, re.MULTILINE)) has_name = bool(re.search(r"^\s+-?\s*name\s*:", text, re.MULTILINE)) has_module = any( re.search(rf"^\s+{re.escape(k)}\s*:", text, re.MULTILINE) for k in _ANSIBLE_MODULE_KEYS ) or bool(re.search(r"^\s+ansible\.\w+\.\w+\s*:", text, re.MULTILINE)) return has_hosts or has_import_pb or has_tasks or has_handlers or (has_name and has_module) def _ansible_file_type(path: Path) -> str: """Classify an Ansible file by path convention. Returns one of: 'playbook', 'tasks', 'handlers', 'meta', 'vars', 'unknown'. """ parts_lower = [p.lower() for p in path.parts] name_lower = path.name.lower() if "meta" in parts_lower and name_lower in ("main.yml", "main.yaml"): return "meta" if "handlers" in parts_lower: return "handlers" if "tasks" in parts_lower: return "tasks" if any(p in parts_lower for p in ("group_vars", "host_vars", "vars", "defaults")): return "vars" if "playbooks" in parts_lower or name_lower in _ANSIBLE_PLAYBOOK_NAMES: return "playbook" return "unknown" def _ansible_fqcn_short(key: str) -> str: """Strip FQCN prefix: 'ansible.builtin.include_tasks' → 'include_tasks'.""" return key.rsplit(".", 1)[-1] def _yaml_line(node: object) -> int: return node.start_mark.line + 1 # type: ignore[attr-defined] def _yaml_end_line(node: object) -> int: return node.end_mark.line + 1 # type: ignore[attr-defined] def _yaml_get_key(mapping_node: object, key: str) -> Optional[object]: for k_node, v_node in mapping_node.value: # type: ignore[attr-defined] if isinstance(k_node, _YamlScalar) and k_node.value == key: return v_node return None def _yaml_scalar(node: object) -> Optional[str]: if isinstance(node, _YamlScalar): return node.value # type: ignore[attr-defined] return None def _ansible_is_play_item(item: object) -> bool: """True if this top-level sequence item is a definitive Ansible play or playbook import. Requires EITHER: - ``import_playbook:`` key (unambiguous), OR - ``hosts:`` key AND at least one key from ``_ANSIBLE_PLAY_KEYS``. A bare ``hosts: all`` without any other play key is too generic and is rejected. """ if not isinstance(item, _YamlMapping): return False keys: set[Optional[str]] = { _yaml_scalar(k) for k, _ in item.value # type: ignore[attr-defined] if isinstance(k, _YamlScalar) } if "import_playbook" in keys: return True return "hosts" in keys and bool(keys & _ANSIBLE_PLAY_KEYS) # --------------------------------------------------------------------------- # Parser # --------------------------------------------------------------------------- class CodeParser: """Parses source files using Tree-sitter and extracts structural information.""" _MODULE_CACHE_MAX = 15_000 # Evict cache to cap memory on huge monorepos _BLADE_COMMENT_RE = re.compile(r"\{\{--.*?(?:--\}\}|$)", re.DOTALL) _BLADE_DIRECTIVE_RE = re.compile( r"""(? None: self._repo_root = Path(repo_root).resolve() if repo_root is not None else None self._dbt_model_paths_cache: dict[Path, tuple[Path, ...]] = {} self._parsers: dict[str, object] = {} self._module_file_cache: dict[str, Optional[str]] = {} # Absolute file paths to treat as absent during import resolution. # ``forget`` sets this (via :meth:`exclude_files`) so a re-parsed # referrer resolves exactly as it would in a build where the forgotten # files never existed on disk. See ``forget.forget_files``. self._excluded_files: set[str] = set() self._export_symbol_cache: dict[str, Optional[str]] = {} self._tsconfig_resolver = TsconfigResolver() # Per-parse cache of Dart pubspec root lookups; see #87 self._dart_pubspec_cache: dict[tuple[str, str], Optional[Path]] = {} # Cargo discovery is shared by every Rust import/call in a source file. self._rust_project_cache: dict[ str, tuple[Path, Path, Path, dict[str, Path], Path] ] = {} # Config-driven custom languages (.code-review-graph/languages.toml). # The built-in tables stay shared module-level constants; only when a # repo defines custom languages does this parser switch to merged # copies, so other CodeParser instances (multi-repo registry, worker # processes for other repos) are never affected. See #320. self._extension_map: dict[str, str] = EXTENSION_TO_LANGUAGE self._class_types: dict[str, list[str]] = _CLASS_TYPES self._function_types: dict[str, list[str]] = _FUNCTION_TYPES self._import_types: dict[str, list[str]] = _IMPORT_TYPES self._call_types: dict[str, list[str]] = _CALL_TYPES self._custom_languages: dict[str, CustomLanguage] = {} if repo_root is not None: self._custom_languages = load_custom_languages( Path(repo_root), builtin_extensions=EXTENSION_TO_LANGUAGE, builtin_languages=_builtin_language_names(), ) if self._custom_languages: self._extension_map = dict(EXTENSION_TO_LANGUAGE) self._class_types = dict(_CLASS_TYPES) self._function_types = dict(_FUNCTION_TYPES) self._import_types = dict(_IMPORT_TYPES) self._call_types = dict(_CALL_TYPES) for custom in self._custom_languages.values(): for ext in custom.extensions: self._extension_map[ext] = custom.name self._class_types[custom.name] = list(custom.class_node_types) self._function_types[custom.name] = list(custom.function_node_types) self._import_types[custom.name] = list(custom.import_node_types) self._call_types[custom.name] = list(custom.call_node_types) def _get_parser(self, language: str): # type: ignore[arg-type] if language not in self._parsers: # Custom languages map their name onto a packaged grammar. custom = self._custom_languages.get(language) grammar = custom.grammar if custom is not None else language parser = _load_tree_sitter_parser(grammar) if parser is None: return None self._parsers[language] = parser return self._parsers[language] def detect_language(self, path: Path, source: Optional[bytes] = None) -> Optional[str]: """Map a file path to its language name. Extension-based lookup is tried first. For extension-less files (typical for Unix scripts like ``bin/myapp`` or ``.git/hooks/pre-commit``) we fall back to reading the first line for a shebang. Files that already have a known extension are never re-read — shebang probing only runs when the extension lookup returns ``None`` **and** the path has no suffix at all. See issue #237. When *source* is provided, the shebang is sniffed from those bytes instead of re-reading the file. Callers that hash-and-parse one byte snapshot MUST pass it: a separate disk read can race a concurrent save, mis-detect the language, and store a wrong parse under the snapshot's hash (issue #746). """ if path.name.lower().endswith(".blade.php"): return "blade" suffix = path.suffix.lower() lang = self._extension_map.get(suffix) if lang == "yaml" and _is_ansible_path(path): return "ansible" if lang in ("properties", "yaml") and is_spring_config_path(path): return "spring_config" if lang == "properties": return None if lang is not None: return lang # Only probe shebang for files without any extension — "README", "LICENSE", # and other extension-less text files also fall here, but the probe is a # cheap 256-byte read that returns None when no shebang is found. if suffix == "": head = source[:_SHEBANG_PROBE_BYTES] if source is not None else None return self._detect_language_from_shebang(path, head) return None @staticmethod def _detect_language_from_shebang( path: Path, head: Optional[bytes] = None, ) -> Optional[str]: """Inspect the first line of ``path`` for a shebang interpreter. When *head* is given it is used as the first bytes of the file and the file is not read from disk (TOCTOU-safe for callers that already hold the byte snapshot being parsed, see issue #746). Returns the mapped language name or ``None`` if the file has no shebang, is unreadable, or names an interpreter we don't map. Accepted shapes:: #!/bin/bash #!/usr/bin/env python3 #!/usr/bin/env -S node --experimental-vm-modules #!/usr/bin/bash -e Only the basename of the interpreter is consulted. Trailing flags after the interpreter are ignored. Windows-style ``\r\n`` line endings are handled. Binary files read as garbage bytes simply fail the ``#!`` prefix check and return ``None``. """ if head is None: try: with path.open("rb") as fh: head = fh.read(_SHEBANG_PROBE_BYTES) except (OSError, PermissionError): return None if not head.startswith(b"#!"): return None # Take just the first line, stripped of leading "#!" and any # surrounding whitespace. Split on NUL to defend against accidental # binary content following a ``#!`` prefix. first_line = head.split(b"\n", 1)[0].split(b"\0", 1)[0] try: line = first_line[2:].decode("utf-8", errors="strict").strip() except UnicodeDecodeError: return None if not line: return None tokens = line.split() if not tokens: return None first = tokens[0] # `/usr/bin/env` indirection: the interpreter is the next token. # `/usr/bin/env -S node --flag` is also valid — skip any leading # ``-`` options after env. if first.endswith("/env") or first == "env": interpreter_token: Optional[str] = None for tok in tokens[1:]: if tok.startswith("-"): # ``-S`` takes no argument in most envs; skip and continue. continue interpreter_token = tok break if interpreter_token is None: return None interpreter = interpreter_token.rsplit("/", 1)[-1] else: # Direct form: ``#!/bin/bash`` or ``#!/usr/local/bin/python3``. interpreter = first.rsplit("/", 1)[-1] return SHEBANG_INTERPRETER_TO_LANGUAGE.get(interpreter) def parse_file(self, path: Path) -> tuple[list[NodeInfo], list[EdgeInfo]]: """Parse a single file and return extracted nodes and edges.""" try: source = path.read_bytes() except (OSError, PermissionError): return [], [] return self.parse_bytes(path, source) def parse_bytes(self, path: Path, source: bytes) -> tuple[list[NodeInfo], list[EdgeInfo]]: """Parse pre-read bytes and return extracted nodes and edges. This avoids re-reading the file from disk, eliminating TOCTOU gaps when the caller has already read the bytes (e.g. for hashing): every parse decision, including shebang language detection, derives from *source* so the stored file hash always describes the bytes that were actually parsed (issue #746). """ language = self.detect_language(path, source) if not language: return [], [] parser = None tree = None parse_source = source if language == "c" and path.suffix.lower() == ".h": cpp_parser = self._get_parser("cpp") if cpp_parser is not None: cpp_source = self._mask_cpp_qt_macros(source) cpp_tree = cpp_parser.parse(cpp_source) if self._has_cpp_header_evidence(cpp_tree.root_node): language = "cpp" parser = cpp_parser tree = cpp_tree parse_source = cpp_source elif language == "cpp": parse_source = self._mask_cpp_qt_macros(source) if language == "blade": return self._parse_blade(path, source) # Vue SFCs: parse with vue parser, then delegate script blocks to JS/TS if language == "vue": return self._parse_vue(path, source) # Svelte SFCs: same approach as Vue — extract