"""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 hashlib import json import logging import re from dataclasses import dataclass, field from pathlib import Path from typing import NamedTuple, Optional import tree_sitter_language_pack as tslp from .custom_languages import CustomLanguage, load_custom_languages 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, ) # 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__) # --------------------------------------------------------------------------- # Data models for extracted entities # --------------------------------------------------------------------------- @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) @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) # --------------------------------------------------------------------------- # 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", ".sql": "sql", } # 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 # 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"], "typescript": ["class_declaration", "class"], "tsx": ["class_declaration", "class"], "go": ["type_declaration"], "rust": ["struct_item", "enum_item", "impl_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", ], "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"], "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": ["container_declaration"], "powershell": ["class_statement"], "julia": [ "struct_definition", "abstract_definition", "module_definition", ], "verilog": ["module_declaration", "interface_declaration", "class_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": [], } _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"], "java": ["method_declaration", "constructor_declaration"], "c": ["function_definition"], "cpp": ["function_definition"], "csharp": ["method_declaration", "constructor_declaration"], "ruby": ["method", "singleton_method"], "r": ["function_definition"], "perl": ["subroutine_declaration_statement", "method_declaration_statement"], "kotlin": ["function_declaration"], "swift": ["function_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": ["fn_proto", "fn_decl"], "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": [], } _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("...") is a builtin_call_expr — handled # generically via call types below. "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": [], } _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"], "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", ], "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": ["call_expression", "builtin_call_expr"], "powershell": ["command_expression"], "julia": [ "call_expression", "broadcast_call_expression", "macrocall_expression", ], "verilog": [ "module_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": [], } 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", }) # Lombok annotations that trigger constructor injection of final fields _LOMBOK_CONSTRUCTOR_ANNOTATIONS = frozenset({ "RequiredArgsConstructor", "AllArgsConstructor", }) # 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", }) # --------------------------------------------------------------------------- # 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 def file_hash(path: Path) -> str: """SHA-256 hash of file contents.""" return hashlib.sha256(path.read_bytes()).hexdigest() # --------------------------------------------------------------------------- # 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 def __init__(self, repo_root: Optional[Path] = None) -> None: self._parsers: dict[str, object] = {} self._module_file_cache: dict[str, Optional[str]] = {} 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]] = {} # 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 try: self._parsers[language] = tslp.get_parser(grammar) # type: ignore[arg-type] except (LookupError, ValueError, ImportError) as exc: # language not packaged, or grammar load failed logger.debug("tree-sitter parser unavailable for %s: %s", language, exc) return None return self._parsers[language] def detect_language(self, path: Path) -> 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. """ suffix = path.suffix.lower() lang = self._extension_map.get(suffix) 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 == "": return self._detect_language_from_shebang(path) return None @staticmethod def _detect_language_from_shebang(path: Path) -> Optional[str]: """Inspect the first line of ``path`` for a shebang interpreter. 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``. """ 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). """ language = self.detect_language(path) if not language: return [], [] # 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