wehub-resource-sync
2293 lines
94 KiB
Python
2293 lines
94 KiB
Python
"""resolution — moved verbatim from graphify/extract.py."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Callable
|
|
from pathlib import Path
|
|
from graphify.extractors.models import LanguageConfig, _JS_CACHE_BYPASS_SUFFIXES, _NamespaceExportFact, _StarExportFact, _SymbolAliasFact, _SymbolDeclarationFact, _SymbolExportFact, _SymbolImportFact, _SymbolResolutionFacts, _SymbolUseFact, _WORKSPACE_PACKAGE_CACHE # noqa: E402,F401
|
|
from graphify.extractors.base import ( # noqa: F401
|
|
_LANGUAGE_BUILTIN_GLOBALS,
|
|
_file_stem,
|
|
_make_id,
|
|
_read_text,
|
|
)
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
|
|
_TSCONFIG_ALIAS_CACHE: dict[str, dict[str, list[str]]] = {}
|
|
|
|
_WORKSPACE_MANIFEST_NAMES = ("pnpm-workspace.yaml", "package.json")
|
|
|
|
_JS_RESOLVE_EXTS = (".ts", ".tsx", ".mts", ".cts", ".svelte", ".js", ".jsx", ".mjs")
|
|
|
|
_JS_INDEX_FILES = ("index.ts", "index.tsx", "index.svelte", "index.js", "index.jsx", "index.mjs")
|
|
|
|
def _resolve_js_import_path(candidate: Path) -> Path:
|
|
"""Resolve a JS/TS/Svelte import target to a local file when it exists."""
|
|
candidate = Path(os.path.normpath(candidate))
|
|
if candidate.is_file():
|
|
return candidate
|
|
|
|
# TS ESM convention: imports often spell .js/.jsx while source is .ts/.tsx.
|
|
if candidate.suffix == ".js":
|
|
ts_candidate = candidate.with_suffix(".ts")
|
|
if ts_candidate.is_file():
|
|
return ts_candidate
|
|
elif candidate.suffix == ".jsx":
|
|
tsx_candidate = candidate.with_suffix(".tsx")
|
|
if tsx_candidate.is_file():
|
|
return tsx_candidate
|
|
|
|
# Append extensions to the full filename, which covers extensionless imports,
|
|
# multi-dot helpers, and Svelte 5 rune files like Foo.svelte.ts.
|
|
for ext in _JS_RESOLVE_EXTS:
|
|
with_ext = candidate.parent / f"{candidate.name}{ext}"
|
|
if with_ext.is_file():
|
|
return with_ext
|
|
|
|
# Only fall back to directory indexes after file candidates lose.
|
|
if candidate.is_dir():
|
|
for index_name in _JS_INDEX_FILES:
|
|
index_candidate = candidate / index_name
|
|
if index_candidate.is_file():
|
|
return index_candidate
|
|
|
|
return candidate
|
|
|
|
def _strip_jsonc(text: str) -> str:
|
|
"""Strip // line comments, /* */ block comments, and trailing commas from JSONC.
|
|
|
|
Preserves string contents (including // and /* inside strings) by skipping over
|
|
quoted spans first. Required for tsconfig.json files generated by SvelteKit,
|
|
NestJS, Vite, T3, Astro, etc., which use JSONC by default (#700).
|
|
"""
|
|
# Remove block and line comments while leaving string literals untouched.
|
|
pattern = re.compile(
|
|
r'"(?:\\.|[^"\\])*"' # double-quoted string (with escapes)
|
|
r"|/\*.*?\*/" # /* block comment */
|
|
r"|//[^\n]*", # // line comment
|
|
re.DOTALL,
|
|
)
|
|
|
|
def _replace(match: re.Match) -> str:
|
|
token = match.group(0)
|
|
if token.startswith('"'):
|
|
return token
|
|
return ""
|
|
|
|
stripped = pattern.sub(_replace, text)
|
|
# Remove trailing commas before } or ] (allowing whitespace between).
|
|
stripped = re.sub(r",(\s*[}\]])", r"\1", stripped)
|
|
return stripped
|
|
|
|
def _read_tsconfig_aliases(tsconfig: Path, base_dir: Path, seen: set) -> dict[str, list[str]]:
|
|
"""Recursively read path aliases from a tsconfig, following extends chains.
|
|
|
|
Child config paths override parent. Circular extends are detected via seen set.
|
|
npm package configs (e.g. @tsconfig/svelte) are skipped since they're not on disk.
|
|
Handles JSONC (comments + trailing commas) which is the default tsconfig format
|
|
for SvelteKit, NestJS, Vite, T3, Astro, etc. (#700).
|
|
"""
|
|
if str(tsconfig) in seen:
|
|
return {}
|
|
seen.add(str(tsconfig))
|
|
try:
|
|
raw = tsconfig.read_text(encoding="utf-8")
|
|
except Exception as e:
|
|
print(f" warning: could not read {tsconfig} ({type(e).__name__}: {e})", file=sys.stderr, flush=True)
|
|
return {}
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
try:
|
|
data = json.loads(_strip_jsonc(raw))
|
|
except json.JSONDecodeError as e:
|
|
print(f" warning: failed to parse {tsconfig} as JSON/JSONC ({e.msg} at line {e.lineno} col {e.colno})", file=sys.stderr, flush=True)
|
|
return {}
|
|
except Exception as e:
|
|
print(f" warning: failed to parse {tsconfig} ({type(e).__name__}: {e})", file=sys.stderr, flush=True)
|
|
return {}
|
|
|
|
aliases: dict[str, list[str]] = {}
|
|
# `extends` may be a string or, since TypeScript 5.0, an array of paths.
|
|
# For an array, parents are processed in order with later entries
|
|
# overriding earlier ones; the extending config (paths below) overrides
|
|
# all parents. Without the list branch, an array `extends` raised
|
|
# `AttributeError: 'list' object has no attribute 'startswith'`, which
|
|
# _safe_extract turned into a skip of the whole file.
|
|
extends = data.get("extends")
|
|
if isinstance(extends, str):
|
|
extends_list = [extends]
|
|
elif isinstance(extends, list):
|
|
extends_list = [e for e in extends if isinstance(e, str)]
|
|
else:
|
|
extends_list = []
|
|
for ext in extends_list:
|
|
# Skip scoped npm package configs (e.g. @tsconfig/svelte) — not on disk.
|
|
if not ext or ext.startswith("@"):
|
|
continue
|
|
extended_path = (base_dir / ext).resolve()
|
|
if not extended_path.suffix:
|
|
extended_path = extended_path.with_suffix(".json")
|
|
if extended_path.exists():
|
|
aliases.update(_read_tsconfig_aliases(extended_path, extended_path.parent, seen))
|
|
|
|
# tsconfig `paths` are resolved relative to `baseUrl` (itself relative to
|
|
# the tsconfig's directory), not the tsconfig directory directly. Honoring
|
|
# baseUrl is required for the common monorepo / NestJS layout where
|
|
# baseUrl points at a subdirectory, e.g. baseUrl "./src" with
|
|
# "@services/*": ["services/*"] must resolve to <dir>/src/services rather
|
|
# than <dir>/services. Defaults to "." so configs without baseUrl (paths
|
|
# relative to the tsconfig dir, the TS 4.1+ behavior) keep working.
|
|
compiler_options = data.get("compilerOptions", {})
|
|
base_url = compiler_options.get("baseUrl") or "."
|
|
paths_base = base_dir / base_url
|
|
paths = compiler_options.get("paths", {})
|
|
for alias, targets in paths.items():
|
|
if not targets:
|
|
continue
|
|
# Keep ALL targets in declared order — tsc tries each until one resolves
|
|
# on disk. Discarding the fallbacks (#1531) misresolved/dropped imports
|
|
# whose file lived at a non-first target. Preserve wildcard tokens in
|
|
# both sides until the resolver substitutes the captured segment, then
|
|
# normalizes the concrete path (#927). Empty/non-string entries are skipped.
|
|
target_patterns = [
|
|
str(paths_base / t)
|
|
for t in targets
|
|
if isinstance(t, str) and t
|
|
]
|
|
if target_patterns:
|
|
aliases[alias] = target_patterns
|
|
|
|
return aliases
|
|
|
|
def _load_tsconfig_aliases(start_dir: Path) -> dict[str, list[str]]:
|
|
"""Walk up from start_dir to find tsconfig.json and return compilerOptions.paths aliases.
|
|
|
|
Follows extends chains so SvelteKit/Nuxt/NestJS inherited aliases are included.
|
|
Returns a dict mapping alias patterns to ordered resolved target patterns;
|
|
wildcard tokens remain intact for substitution during resolution (#927).
|
|
Result is cached by tsconfig path string.
|
|
"""
|
|
current = start_dir.resolve()
|
|
for candidate in [current, *current.parents]:
|
|
tsconfig = candidate / "tsconfig.json"
|
|
if tsconfig.exists():
|
|
key = str(tsconfig)
|
|
if key not in _TSCONFIG_ALIAS_CACHE:
|
|
_TSCONFIG_ALIAS_CACHE[key] = _read_tsconfig_aliases(tsconfig, candidate, seen=set())
|
|
return _TSCONFIG_ALIAS_CACHE[key]
|
|
return {}
|
|
|
|
def _match_tsconfig_alias(raw: str, pattern: str) -> "tuple[tuple[int, int], str, bool] | None":
|
|
"""Return (specificity, captured text, is_wildcard) when pattern matches raw.
|
|
|
|
Exact aliases win first. Wildcard aliases follow TypeScript's longest-prefix
|
|
rule. The final branch preserves Graphify's existing support for treating a
|
|
non-wildcard alias as a directory prefix, but only after real wildcard matches.
|
|
"""
|
|
if "*" in pattern:
|
|
if pattern.count("*") != 1:
|
|
return None
|
|
prefix, suffix = pattern.split("*", 1)
|
|
if not raw.startswith(prefix) or not raw.endswith(suffix):
|
|
return None
|
|
end = len(raw) - len(suffix) if suffix else len(raw)
|
|
if end < len(prefix):
|
|
return None
|
|
return (1, -len(prefix)), raw[len(prefix):end], True
|
|
|
|
if raw == pattern:
|
|
return (0, -len(pattern)), "", False
|
|
|
|
prefix = pattern.rstrip("/")
|
|
if prefix and raw.startswith(prefix + "/"):
|
|
return (2, -len(prefix)), raw[len(prefix):].lstrip("/"), False
|
|
return None
|
|
|
|
def _resolve_tsconfig_alias(raw: str, aliases: dict[str, list[str]]) -> "Path | None":
|
|
"""Resolve `raw` against the most specific matching tsconfig alias pattern.
|
|
|
|
Within that pattern, try targets in declared order and return the first whose
|
|
candidate resolves to a real file. If none exist, return the first candidate
|
|
so existing phantom/external-edge behavior stays unchanged.
|
|
"""
|
|
best: "tuple[tuple[int, int], str, bool, list[str]] | None" = None
|
|
for pattern, targets in aliases.items():
|
|
match = _match_tsconfig_alias(raw, pattern)
|
|
if match is None:
|
|
continue
|
|
specificity, captured, is_wildcard = match
|
|
if best is None or specificity < best[0]:
|
|
best = specificity, captured, is_wildcard, targets
|
|
|
|
if best is None:
|
|
return None
|
|
|
|
_, captured, is_wildcard, targets = best
|
|
first = None
|
|
for target in targets:
|
|
if is_wildcard:
|
|
# TypeScript substitutes only when the matched star is non-empty.
|
|
substituted = target.replace("*", captured, 1) if captured else target
|
|
cand = Path(os.path.normpath(substituted))
|
|
else:
|
|
cand = Path(target)
|
|
if captured:
|
|
cand = Path(os.path.normpath(cand / captured))
|
|
resolved = _resolve_js_import_path(cand)
|
|
if resolved.is_file():
|
|
return resolved
|
|
if first is None:
|
|
first = cand
|
|
return first
|
|
|
|
def _find_workspace_root(start_dir: Path) -> Path | None:
|
|
current = start_dir.resolve()
|
|
for candidate in [current, *current.parents]:
|
|
if (candidate / "pnpm-workspace.yaml").exists():
|
|
return candidate
|
|
package_json = candidate / "package.json"
|
|
if package_json.is_file():
|
|
try:
|
|
data = json.loads(package_json.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
continue
|
|
if "workspaces" in data:
|
|
return candidate
|
|
return None
|
|
|
|
def _pnpm_workspace_globs(workspace_file: Path) -> list[str]:
|
|
globs: list[str] = []
|
|
in_packages = False
|
|
for raw_line in workspace_file.read_text(encoding="utf-8", errors="replace").splitlines():
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if line.startswith("packages:"):
|
|
in_packages = True
|
|
continue
|
|
if in_packages and line.startswith("-"):
|
|
value = line[1:].strip().strip("'\"")
|
|
if value and not value.startswith("!"):
|
|
globs.append(value)
|
|
continue
|
|
if in_packages and not raw_line.startswith((" ", "\t")):
|
|
break
|
|
return globs
|
|
|
|
def _workspace_globs(root: Path) -> list[str]:
|
|
pnpm_workspace = root / "pnpm-workspace.yaml"
|
|
if pnpm_workspace.exists():
|
|
return _pnpm_workspace_globs(pnpm_workspace)
|
|
|
|
package_json = root / "package.json"
|
|
try:
|
|
data = json.loads(package_json.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return []
|
|
|
|
workspaces = data.get("workspaces")
|
|
if isinstance(workspaces, list):
|
|
return [item for item in workspaces if isinstance(item, str) and not item.startswith("!")]
|
|
if isinstance(workspaces, dict):
|
|
packages = workspaces.get("packages")
|
|
if isinstance(packages, list):
|
|
return [item for item in packages if isinstance(item, str) and not item.startswith("!")]
|
|
return []
|
|
|
|
def _load_workspace_packages(start_dir: Path) -> dict[str, Path]:
|
|
root = _find_workspace_root(start_dir)
|
|
if root is None:
|
|
return {}
|
|
manifest_mtimes = tuple(
|
|
(name, (root / name).stat().st_mtime_ns)
|
|
for name in _WORKSPACE_MANIFEST_NAMES
|
|
if (root / name).is_file()
|
|
)
|
|
key = str((root, manifest_mtimes))
|
|
if key in _WORKSPACE_PACKAGE_CACHE:
|
|
return _WORKSPACE_PACKAGE_CACHE[key]
|
|
|
|
packages: dict[str, Path] = {}
|
|
for pattern in _workspace_globs(root):
|
|
package_dirs: list[Path] = [root] if pattern in (".", "./") else list(root.glob(pattern))
|
|
for package_dir in package_dirs:
|
|
manifest = package_dir / "package.json"
|
|
if not manifest.is_file():
|
|
continue
|
|
try:
|
|
data = json.loads(manifest.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
continue
|
|
name = data.get("name")
|
|
if isinstance(name, str) and name:
|
|
packages[name] = package_dir
|
|
_WORKSPACE_PACKAGE_CACHE[key] = packages
|
|
return packages
|
|
|
|
_EXPORT_CONDITION_PRIORITY = (
|
|
"source", "import", "module", "svelte", "types", "require", "default",
|
|
)
|
|
|
|
def _resolve_export_target(value: Any) -> str | None:
|
|
"""Resolve an `exports` map value (string or condition object) to a
|
|
relative target string, honouring _EXPORT_CONDITION_PRIORITY for objects
|
|
and recursing into nested condition objects."""
|
|
if isinstance(value, str):
|
|
return value
|
|
if isinstance(value, dict):
|
|
for cond in _EXPORT_CONDITION_PRIORITY:
|
|
v = value.get(cond)
|
|
if isinstance(v, str):
|
|
return v
|
|
if isinstance(v, dict):
|
|
nested = _resolve_export_target(v)
|
|
if nested:
|
|
return nested
|
|
return None
|
|
|
|
def _contained_in_package(resolved: Path, package_dir: Path) -> bool:
|
|
"""Guard against `exports` targets that escape the package directory
|
|
(e.g. "./evil": "../../../etc/passwd"). Only accept paths that stay
|
|
within package_dir after resolution."""
|
|
try:
|
|
return resolved.resolve().is_relative_to(package_dir.resolve())
|
|
except ValueError:
|
|
return False
|
|
|
|
def _package_entry_candidates(package_dir: Path, subpath: str) -> list[Path]:
|
|
manifest = package_dir / "package.json"
|
|
manifest_data: dict[str, Any] = {}
|
|
try:
|
|
manifest_data = json.loads(manifest.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
pass
|
|
|
|
if subpath:
|
|
# Consult the package's `exports` subpath map before the bare-path
|
|
# fallback (#1308): "./browser" -> conditions -> file, plus single
|
|
# wildcard "./*" patterns. Targets that escape the package dir are
|
|
# rejected; resolution then falls through to the bare path.
|
|
exports = manifest_data.get("exports")
|
|
if isinstance(exports, dict):
|
|
subpath_key = "./" + subpath
|
|
target = _resolve_export_target(exports.get(subpath_key))
|
|
if target:
|
|
candidate = package_dir / target
|
|
if _contained_in_package(candidate, package_dir):
|
|
return [candidate]
|
|
else:
|
|
for pattern, pattern_value in exports.items():
|
|
if "*" in pattern and pattern.count("*") == 1:
|
|
prefix, suffix = pattern.split("*", 1)
|
|
if (subpath_key.startswith(prefix)
|
|
and (not suffix or subpath_key.endswith(suffix))):
|
|
matched = subpath_key[len(prefix):len(subpath_key) - len(suffix) if suffix else None]
|
|
resolved = _resolve_export_target(pattern_value)
|
|
if resolved and "*" in resolved:
|
|
candidate = package_dir / resolved.replace("*", matched)
|
|
if _contained_in_package(candidate, package_dir):
|
|
return [candidate]
|
|
return [package_dir / subpath]
|
|
|
|
exports = manifest_data.get("exports")
|
|
if isinstance(exports, str):
|
|
return [package_dir / exports]
|
|
if isinstance(exports, dict):
|
|
dot_target = _resolve_export_target(exports.get("."))
|
|
if dot_target:
|
|
return [package_dir / dot_target]
|
|
|
|
candidates: list[Path] = []
|
|
for key in ("svelte", "module", "main", "types"):
|
|
value = manifest_data.get(key)
|
|
if isinstance(value, str):
|
|
candidates.append(package_dir / value)
|
|
candidates.append(package_dir / "src/index")
|
|
candidates.append(package_dir / "index")
|
|
return candidates
|
|
|
|
def _resolve_workspace_import(raw: str, start_dir: Path) -> Path | None:
|
|
packages = _load_workspace_packages(start_dir)
|
|
for package_name, package_dir in packages.items():
|
|
if raw == package_name:
|
|
subpath = ""
|
|
elif raw.startswith(package_name + "/"):
|
|
subpath = raw[len(package_name) + 1:]
|
|
else:
|
|
continue
|
|
for candidate in _package_entry_candidates(package_dir, subpath):
|
|
resolved = _resolve_js_import_path(candidate)
|
|
if resolved.is_file():
|
|
return resolved
|
|
return None
|
|
|
|
def _resolve_js_module_path(raw: str | Path, start_dir: Path | None = None) -> Path | None:
|
|
"""Resolve a JS/TS module path or specifier to a local source file.
|
|
|
|
With a Path argument this preserves the path-based helper API used by
|
|
import-extension tests. With a string plus start_dir it resolves JS/TS
|
|
module specifiers including relative paths, tsconfig aliases, and workspace
|
|
packages.
|
|
"""
|
|
if isinstance(raw, Path):
|
|
return _resolve_js_import_path(raw)
|
|
if start_dir is None:
|
|
return _resolve_js_import_path(Path(raw))
|
|
if raw.startswith("."):
|
|
return _resolve_js_import_path(start_dir / raw)
|
|
|
|
aliases = _load_tsconfig_aliases(start_dir)
|
|
hit = _resolve_tsconfig_alias(raw, aliases)
|
|
if hit is not None:
|
|
return _resolve_js_import_path(hit)
|
|
|
|
return _resolve_workspace_import(raw, start_dir)
|
|
|
|
def _resolve_js_import_target(raw: str, str_path: str) -> "tuple[str, Path | None] | None":
|
|
"""Resolve a JS/TS import path string to (target_nid, resolved_path).
|
|
|
|
Handles relative paths, tsconfig path aliases, workspace packages, and
|
|
bare/scoped imports.
|
|
Returns None if `raw` is empty.
|
|
"""
|
|
if not raw:
|
|
return None
|
|
resolved_path = _resolve_js_module_path(raw, Path(str_path).parent)
|
|
if resolved_path is not None:
|
|
return _make_id(str(resolved_path)), resolved_path
|
|
module_name = raw.split("/")[-1]
|
|
if not module_name:
|
|
return None
|
|
# Unresolved: relative/absolute, tsconfig-alias and workspace resolution have
|
|
# all run and failed, so this is an external package (or a dangling local
|
|
# path). Namespace the id with the "ref" prefix — the J-4 convention already
|
|
# used for tsconfig `extends`/`$ref` externals — so it can NEVER collapse to
|
|
# the same _make_id as a local file/symbol node. Without it, the bare
|
|
# last-segment id (e.g. "tailwindcss/colors" -> "colors") collides with any
|
|
# unrelated local file of that stem via build.py's pre-migration alias index,
|
|
# producing a confident (EXTRACTED) cross-language phantom imports_from edge
|
|
# (#1638). The ref-namespaced target has no node, so build drops it as an
|
|
# external reference — the correct outcome for a third-party import.
|
|
return _make_id("ref", raw), None
|
|
|
|
def _resolve_c_include_path(raw: str, str_path: str) -> "Path | None":
|
|
"""Resolve a quoted #include path to a real file on disk.
|
|
|
|
Searches relative to the including file's directory. Returns None for
|
|
system headers (<...>) or paths that don't exist on disk.
|
|
"""
|
|
if not raw:
|
|
return None
|
|
candidate = (Path(str_path).parent / raw).resolve()
|
|
if candidate.is_file():
|
|
return candidate
|
|
return None
|
|
|
|
def _resolve_lua_import_target(raw_module: str, str_path: str) -> str:
|
|
"""Resolve a Lua require() module name to a node id.
|
|
|
|
Lua module names use dots as path separators: `require("pkg.b")` looks for
|
|
`pkg/b.lua` (or `pkg/b/init.lua`) relative to a package root. We probe the
|
|
importing file's directory and walk upward looking for a matching file on
|
|
disk; if found, the returned id matches the file node id `_extract_generic`
|
|
assigns to that file (`_make_id(str(path))`), so the edge lands on a real
|
|
node. When nothing matches, fall back to `_make_id` of the full dotted
|
|
module name so cross-file resolution can still complete via the symbol
|
|
resolution pass instead of dropping the edge entirely (#1075).
|
|
"""
|
|
if not raw_module:
|
|
return ""
|
|
rel = raw_module.replace(".", "/")
|
|
try:
|
|
start_dir = Path(str_path).parent
|
|
except Exception:
|
|
start_dir = None
|
|
if start_dir is not None:
|
|
probe = start_dir
|
|
# Walk up a few levels so requires from nested files still resolve when
|
|
# the package root is above the importing file.
|
|
for _ in range(6):
|
|
for suffix in (".lua", ".luau"):
|
|
cand = probe / f"{rel}{suffix}"
|
|
if cand.is_file():
|
|
return _make_id(str(cand))
|
|
for suffix in (".lua", ".luau"):
|
|
cand = probe / rel / f"init{suffix}"
|
|
if cand.is_file():
|
|
return _make_id(str(cand))
|
|
if probe.parent == probe:
|
|
break
|
|
probe = probe.parent
|
|
return _make_id(raw_module)
|
|
|
|
_VUE_SCRIPT_RE = re.compile(
|
|
r"""(<script\b(?:"[^"]*"|'[^']*'|[^>"'])*>)([\s\S]*?)(</script\s*>)""",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
_VUE_SCRIPT_LANG_RE = re.compile(
|
|
r"""\blang\s*=\s*['"]?([A-Za-z]+)['"]?""", re.IGNORECASE
|
|
)
|
|
|
|
def _vue_mask_non_script(src: str) -> tuple[str, str | None]:
|
|
"""Blank everything outside ``<script>`` bodies, keeping ``\\r``/``\\n``.
|
|
|
|
Replaces template/style/tags with spaces so a JS/TS grammar sees only the
|
|
script, while preserved newlines keep line numbers accurate. Returns
|
|
``(masked_source, lang)``; ``lang`` is the first block's declared ``lang``.
|
|
"""
|
|
def _blank(s: str) -> str:
|
|
return re.sub(r"[^\r\n]", " ", s)
|
|
|
|
out: list[str] = []
|
|
pos = 0
|
|
lang: str | None = None
|
|
for m in _VUE_SCRIPT_RE.finditer(src):
|
|
out.append(_blank(src[pos:m.start()])) # markup/style before this block
|
|
out.append(_blank(m.group(1))) # <script …> open tag
|
|
out.append(m.group(2)) # script body, verbatim
|
|
out.append(_blank(m.group(3))) # </script> close tag
|
|
pos = m.end()
|
|
if lang is None:
|
|
lang_m = _VUE_SCRIPT_LANG_RE.search(m.group(1))
|
|
if lang_m:
|
|
lang = lang_m.group(1).lower()
|
|
out.append(_blank(src[pos:]))
|
|
return "".join(out), lang
|
|
|
|
def _source_key(source_file: str, root: Path) -> str:
|
|
if not source_file:
|
|
return ""
|
|
source_path = Path(source_file)
|
|
try:
|
|
return str(source_path.resolve().relative_to(root))
|
|
except Exception:
|
|
return str(source_path)
|
|
|
|
def _node_disambiguation_source_key(node: dict, root: Path) -> str:
|
|
source_file = str(node.get("source_file", ""))
|
|
if source_file:
|
|
return _source_key(source_file, root)
|
|
return _source_key(str(node.get("origin_file", "")), root)
|
|
|
|
def _disambiguate_colliding_node_ids(
|
|
nodes: list[dict],
|
|
edges: list[dict],
|
|
raw_calls: list[dict],
|
|
root: Path,
|
|
) -> None:
|
|
"""Rewrite only colliding node IDs, using source path as the disambiguator.
|
|
|
|
Module anchor nodes (#1327) are exempt: ``import CoreKit`` from three files
|
|
yields three ``type=module`` nodes with the same id but different
|
|
source_files. Those are the *same* module, not distinct same-named symbols,
|
|
so they must collapse to one shared node — disambiguating them by path would
|
|
scatter a single module across N file-qualified duplicates.
|
|
"""
|
|
by_id: dict[str, list[dict]] = {}
|
|
for node in nodes:
|
|
if node.get("type") in ("module", "namespace"):
|
|
continue
|
|
nid = node.get("id")
|
|
if isinstance(nid, str) and nid:
|
|
by_id.setdefault(nid, []).append(node)
|
|
|
|
remap: dict[tuple[str, str], str] = {}
|
|
ambiguous_ids: set[str] = set()
|
|
for old_id, group in by_id.items():
|
|
source_keys = {_node_disambiguation_source_key(node, root) for node in group}
|
|
if len(group) < 2 or len(source_keys) < 2:
|
|
continue
|
|
ambiguous_ids.add(old_id)
|
|
# Salt the colliding id with the *path* it came from. The naive salt is
|
|
# ``_make_id(source_key, old_id)`` — source_key is the raw repo-relative
|
|
# path. But _make_id collapses every separator, so two DISTINCT paths
|
|
# whose only difference is a separator-vs-inner-punctuation swap
|
|
# (``a/b/c.md`` vs ``a.b/c.md``, ``foo/bar_baz.md`` vs ``foo_bar/baz.md``)
|
|
# normalize to the SAME salted id and still collide (#1522 — the residual
|
|
# of #1504 the 0.9.0 full-path stem didn't reach). When that happens,
|
|
# append a short stable hash of the *raw* source_key, which IS injective
|
|
# over distinct paths, so the colliders separate. Computed in code from
|
|
# source_file (never trusted from the LLM), so AST↔semantic parity holds.
|
|
naive: dict[str, str] = {} # source_key -> _make_id(source_key, old_id)
|
|
for source_key in source_keys:
|
|
if source_key:
|
|
naive[source_key] = _make_id(source_key, old_id)
|
|
# source_keys that, after normalization, are not unique among themselves.
|
|
seen: dict[str, int] = {}
|
|
for nid in naive.values():
|
|
seen[nid] = seen.get(nid, 0) + 1
|
|
needs_hash = {sk for sk, nid in naive.items() if seen.get(nid, 0) > 1}
|
|
for node in group:
|
|
source_key = _node_disambiguation_source_key(node, root)
|
|
if not source_key:
|
|
continue
|
|
if source_key in needs_hash:
|
|
salt = hashlib.sha1(source_key.encode("utf-8")).hexdigest()[:6]
|
|
new_id = _make_id(source_key, old_id, salt)
|
|
else:
|
|
new_id = naive.get(source_key) or _make_id(source_key, old_id)
|
|
remap[(old_id, source_key)] = new_id
|
|
if new_id != old_id:
|
|
node["id"] = new_id
|
|
|
|
if not remap:
|
|
return
|
|
|
|
unambiguous_remaps: dict[str, str] = {}
|
|
for old_id, group in by_id.items():
|
|
if old_id in ambiguous_ids:
|
|
continue
|
|
candidates = {
|
|
node["id"] for node in group
|
|
if isinstance(node.get("id"), str) and node["id"] != old_id
|
|
}
|
|
if len(candidates) == 1:
|
|
unambiguous_remaps[old_id] = next(iter(candidates))
|
|
|
|
# A C/ObjC/C++ `#include "foo.h"` / `#import "foo.h"` resolves to the header's
|
|
# file node, but `foo.h` and its sibling `foo.c`/`foo.m`/`foo.cpp` collapse to
|
|
# the same `foo` file id, so disambiguation salts them apart by path. A
|
|
# cross-file import edge from a THIRD file carries neither salt's source_key, so
|
|
# the (target, edge_source_key) lookup misses and the edge dangles on the now
|
|
# dead `foo` id. Repoint those import edges to the HEADER variant (the include
|
|
# always targeted the header), keyed by the original colliding id (#1475).
|
|
_HEADER_SUFFIXES = (".h", ".hpp", ".hh", ".hxx")
|
|
header_remaps: dict[str, str] = {}
|
|
for old_id in ambiguous_ids:
|
|
for node in by_id.get(old_id, []):
|
|
sk = _node_disambiguation_source_key(node, root)
|
|
if sk and Path(sk).suffix.lower() in _HEADER_SUFFIXES:
|
|
new_id = remap.get((old_id, sk))
|
|
if new_id:
|
|
header_remaps[old_id] = new_id
|
|
break
|
|
|
|
for edge in edges:
|
|
edge_source_key = _source_key(str(edge.get("source_file", "")), root)
|
|
source_key = (edge.get("source", ""), edge_source_key)
|
|
target_key = (edge.get("target", ""), edge_source_key)
|
|
if source_key in remap:
|
|
edge["source"] = remap[source_key]
|
|
elif edge.get("source") in unambiguous_remaps:
|
|
edge["source"] = unambiguous_remaps[str(edge["source"])]
|
|
# imports/imports_from always target a header file, so they must resolve to
|
|
# the header variant BEFORE the same-source-file salt is considered. Keying
|
|
# the import target by the importer's own source file mis-points a `.m`
|
|
# importing its own `.h` back at itself (self-loop), and is wrong for any
|
|
# cross-file import whose importer shares the colliding id (#1475).
|
|
if (edge.get("relation") in ("imports", "imports_from")
|
|
and edge.get("target") in header_remaps):
|
|
edge["target"] = header_remaps[str(edge["target"])]
|
|
elif target_key in remap:
|
|
edge["target"] = remap[target_key]
|
|
elif edge.get("target") in unambiguous_remaps:
|
|
edge["target"] = unambiguous_remaps[str(edge["target"])]
|
|
|
|
for raw_call in raw_calls:
|
|
call_source_key = _source_key(str(raw_call.get("source_file", "")), root)
|
|
caller_key = (raw_call.get("caller_nid", ""), call_source_key)
|
|
if caller_key in remap:
|
|
raw_call["caller_nid"] = remap[caller_key]
|
|
elif raw_call.get("caller_nid") in unambiguous_remaps:
|
|
raw_call["caller_nid"] = unambiguous_remaps[str(raw_call["caller_nid"])]
|
|
|
|
def _is_type_like_definition(node: dict) -> bool:
|
|
if node.get("type") == "namespace":
|
|
return False
|
|
label = str(node.get("label", "")).strip()
|
|
if not label:
|
|
return False
|
|
if label.endswith(")") or label.startswith("."):
|
|
return False
|
|
if "." in label:
|
|
return False
|
|
return node.get("file_type") == "code"
|
|
|
|
def _js_source_path(source_file: str, root: Path) -> Path | None:
|
|
if not source_file:
|
|
return None
|
|
path = Path(source_file)
|
|
if not path.is_absolute():
|
|
path = root / path
|
|
try:
|
|
return path.resolve()
|
|
except Exception:
|
|
return path
|
|
|
|
def _apply_symbol_resolution_facts(
|
|
paths: list[Path],
|
|
nodes: list[dict],
|
|
edges: list[dict],
|
|
root: Path,
|
|
facts: _SymbolResolutionFacts,
|
|
) -> None:
|
|
"""Apply language-provided import/export/use facts to graph edges."""
|
|
if not (
|
|
facts.declarations
|
|
or facts.imports
|
|
or facts.aliases
|
|
or facts.exports
|
|
or facts.star_exports
|
|
or facts.namespace_exports
|
|
or facts.uses
|
|
or facts.module_imports
|
|
):
|
|
return
|
|
|
|
path_by_resolved = {path.resolve(): path for path in paths}
|
|
source_file_id = {path.resolve(): _make_id(str(path)) for path in paths}
|
|
symbol_nodes: dict[tuple[Path, str], str] = {}
|
|
for node in nodes:
|
|
source_path = _js_source_path(str(node.get("source_file", "")), root)
|
|
if source_path is None:
|
|
continue
|
|
label = str(node.get("label", "")).strip().strip("()").lstrip(".")
|
|
if label and node.get("id"):
|
|
symbol_nodes[(source_path, label)] = str(node["id"])
|
|
|
|
def ensure_symbol_node(path: Path, name: str, line: int) -> str:
|
|
resolved_path = path.resolve()
|
|
existing = symbol_nodes.get((resolved_path, name))
|
|
if existing is not None:
|
|
return existing
|
|
node_id = _make_id(_file_stem(path), name)
|
|
symbol_nodes[(resolved_path, name)] = node_id
|
|
nodes.append({
|
|
"id": node_id,
|
|
"label": name,
|
|
"file_type": "code",
|
|
"source_file": str(path),
|
|
"source_location": f"L{line}",
|
|
})
|
|
return node_id
|
|
|
|
existing_edges = {
|
|
(
|
|
str(edge.get("source")),
|
|
str(edge.get("target")),
|
|
str(edge.get("relation")),
|
|
str(edge.get("context") or ""),
|
|
)
|
|
for edge in edges
|
|
}
|
|
|
|
def add_edge(source: str, target: str, relation: str, context: str, line: int, source_path: Path) -> None:
|
|
key = (source, target, relation, context or "")
|
|
if key in existing_edges:
|
|
return
|
|
existing_edges.add(key)
|
|
edges.append({
|
|
"source": source,
|
|
"target": target,
|
|
"relation": relation,
|
|
"context": context,
|
|
"confidence": "EXTRACTED",
|
|
"source_file": str(source_path),
|
|
"source_location": f"L{line}",
|
|
"weight": 1.0,
|
|
})
|
|
|
|
for declaration in facts.declarations:
|
|
ensure_symbol_node(declaration.file_path, declaration.name, declaration.line)
|
|
|
|
local_aliases_by_file: dict[Path, dict[str, tuple[Path, str]]] = {}
|
|
for import_fact in facts.imports:
|
|
file_path = import_fact.file_path.resolve()
|
|
local_aliases_by_file.setdefault(file_path, {})[import_fact.local_name] = (
|
|
import_fact.target_path.resolve(),
|
|
import_fact.imported_name,
|
|
)
|
|
|
|
pending_aliases_by_file: dict[Path, list[_SymbolAliasFact]] = {}
|
|
for alias_fact in facts.aliases:
|
|
pending_aliases_by_file.setdefault(alias_fact.file_path.resolve(), []).append(alias_fact)
|
|
|
|
for file_path, aliases in pending_aliases_by_file.items():
|
|
local_aliases = local_aliases_by_file.setdefault(file_path, {})
|
|
changed = True
|
|
while changed:
|
|
changed = False
|
|
for alias_fact in aliases:
|
|
if alias_fact.alias in local_aliases:
|
|
continue
|
|
origin = local_aliases.get(alias_fact.target_name)
|
|
if origin is not None:
|
|
local_aliases[alias_fact.alias] = origin
|
|
changed = True
|
|
|
|
named_exports_by_file: dict[Path, dict[str, tuple[Path, str]]] = {}
|
|
star_exports_by_file: dict[Path, list[Path]] = {}
|
|
|
|
for star_fact in facts.star_exports:
|
|
source_path = star_fact.file_path.resolve()
|
|
target_path = star_fact.target_path.resolve()
|
|
star_exports_by_file.setdefault(source_path, []).append(target_path)
|
|
source_id = source_file_id.get(source_path)
|
|
if source_id is not None:
|
|
add_edge(
|
|
source_id,
|
|
_make_id(str(path_by_resolved.get(target_path, target_path))),
|
|
"re_exports",
|
|
"export",
|
|
star_fact.line,
|
|
star_fact.file_path,
|
|
)
|
|
|
|
for namespace_fact in facts.namespace_exports:
|
|
source_path = namespace_fact.file_path.resolve()
|
|
target_path = namespace_fact.target_path.resolve()
|
|
namespace_id = ensure_symbol_node(
|
|
namespace_fact.file_path,
|
|
namespace_fact.exported_name,
|
|
namespace_fact.line,
|
|
)
|
|
named_exports_by_file.setdefault(source_path, {})[
|
|
namespace_fact.exported_name
|
|
] = (source_path, namespace_fact.exported_name)
|
|
source_id = source_file_id.get(source_path)
|
|
if source_id is not None:
|
|
add_edge(
|
|
source_id,
|
|
namespace_id,
|
|
"contains",
|
|
"namespace_export",
|
|
namespace_fact.line,
|
|
namespace_fact.file_path,
|
|
)
|
|
add_edge(
|
|
source_id,
|
|
_make_id(str(path_by_resolved.get(target_path, target_path))),
|
|
"re_exports",
|
|
"export",
|
|
namespace_fact.line,
|
|
namespace_fact.file_path,
|
|
)
|
|
|
|
for export_fact in facts.exports:
|
|
file_path = export_fact.file_path.resolve()
|
|
origin: tuple[Path, str] | None = None
|
|
if export_fact.target_path is not None and export_fact.target_name is not None:
|
|
origin = (export_fact.target_path.resolve(), export_fact.target_name)
|
|
elif export_fact.local_name is not None:
|
|
origin = local_aliases_by_file.get(file_path, {}).get(export_fact.local_name)
|
|
if origin is None and (file_path, export_fact.local_name) in symbol_nodes:
|
|
origin = (file_path, export_fact.local_name)
|
|
if origin is None:
|
|
continue
|
|
named_exports_by_file.setdefault(file_path, {})[export_fact.exported_name] = origin
|
|
if origin[0] != file_path:
|
|
source_id = source_file_id.get(file_path)
|
|
if source_id is not None:
|
|
add_edge(
|
|
source_id,
|
|
_make_id(str(path_by_resolved.get(origin[0], origin[0]))),
|
|
"re_exports",
|
|
"export",
|
|
export_fact.line,
|
|
export_fact.file_path,
|
|
)
|
|
|
|
def resolve_exported_origin(target_path: Path, imported_name: str, seen: set[tuple[Path, str]] | None = None) -> tuple[Path, str]:
|
|
target_path = target_path.resolve()
|
|
key = (target_path, imported_name)
|
|
if seen is None:
|
|
seen = set()
|
|
if key in seen:
|
|
return key
|
|
seen.add(key)
|
|
origin = named_exports_by_file.get(target_path, {}).get(imported_name)
|
|
if origin is not None:
|
|
return resolve_exported_origin(origin[0], origin[1], seen)
|
|
for star_target in star_exports_by_file.get(target_path, []):
|
|
star_key = (star_target, imported_name)
|
|
if star_key in symbol_nodes:
|
|
return star_key
|
|
resolved = resolve_exported_origin(star_target, imported_name, seen)
|
|
if resolved in symbol_nodes:
|
|
return resolved
|
|
return key
|
|
|
|
for import_fact in facts.imports:
|
|
source_id = source_file_id.get(import_fact.file_path.resolve())
|
|
if source_id is None:
|
|
continue
|
|
origin_path, origin_symbol = resolve_exported_origin(
|
|
import_fact.target_path,
|
|
import_fact.imported_name,
|
|
)
|
|
target_id = symbol_nodes.get((origin_path, origin_symbol))
|
|
if target_id is None:
|
|
continue
|
|
add_edge(
|
|
source_id,
|
|
target_id,
|
|
"imports",
|
|
"import",
|
|
import_fact.line,
|
|
import_fact.file_path,
|
|
)
|
|
|
|
# #1146: emit file-to-file imports_from edges for package-form submodule imports.
|
|
for from_path, to_path, line in facts.module_imports:
|
|
try:
|
|
from_rel = from_path.relative_to(root)
|
|
to_rel = to_path.relative_to(root)
|
|
except ValueError:
|
|
continue
|
|
source_id = _make_id(_file_stem(from_rel))
|
|
target_id = _make_id(_file_stem(to_rel))
|
|
add_edge(source_id, target_id, "imports_from", "submodule_import", line, from_path)
|
|
|
|
for use_fact in facts.uses:
|
|
file_path = use_fact.file_path.resolve()
|
|
target_id = None
|
|
unresolved_origin = local_aliases_by_file.get(file_path, {}).get(use_fact.local_name)
|
|
if unresolved_origin is not None:
|
|
origin_path, origin_symbol = resolve_exported_origin(*unresolved_origin)
|
|
target_id = symbol_nodes.get((origin_path, origin_symbol))
|
|
if target_id is None and use_fact.relation in ("inherits", "implements"):
|
|
# Same-file fallback for HERITAGE only: a base declared in the same
|
|
# file (`class X extends Y`, `interface A extends B`) has no import
|
|
# alias, so resolve it directly against the file's own symbol nodes.
|
|
# Scoped to heritage because same-file calls/uses already resolve via
|
|
# the dedicated call-graph pass; widening this would duplicate those
|
|
# edges. Import resolution still takes precedence (#1095).
|
|
target_id = symbol_nodes.get((file_path, use_fact.local_name))
|
|
if target_id is None:
|
|
continue
|
|
add_edge(
|
|
use_fact.source_id,
|
|
target_id,
|
|
use_fact.relation,
|
|
use_fact.context,
|
|
use_fact.line,
|
|
use_fact.file_path,
|
|
)
|
|
|
|
def _parse_js_tree(path: Path):
|
|
try:
|
|
from tree_sitter import Language, Parser
|
|
# .vue embeds the script in non-JS markup; mask it out and parse the
|
|
# <script> with TS.
|
|
vue_lang: str | None = None
|
|
if path.suffix == ".vue":
|
|
masked, vue_lang = _vue_mask_non_script(
|
|
path.read_text(encoding="utf-8", errors="replace")
|
|
)
|
|
source = masked.encode("utf-8")
|
|
else:
|
|
source = path.read_bytes()
|
|
use_ts = path.suffix in (".ts", ".tsx", ".mts", ".cts") or (
|
|
path.suffix == ".vue" and vue_lang not in ("js", "jsx")
|
|
)
|
|
if use_ts:
|
|
import tree_sitter_typescript as tstypescript
|
|
language = Language(tstypescript.language_typescript())
|
|
else:
|
|
import tree_sitter_javascript as tsjavascript
|
|
language = Language(tsjavascript.language())
|
|
parser = Parser(language)
|
|
return source, parser.parse(source).root_node
|
|
except Exception:
|
|
return None
|
|
|
|
def _walk_js_tree(node):
|
|
# Iterative DFS avoids Python's O(depth) generator-chain overhead.
|
|
# Recursive yield-from creates one generator frame per level — at 26+
|
|
# levels deep each leaf's value had to propagate through 26 frames.
|
|
stack = [node]
|
|
while stack:
|
|
n = stack.pop()
|
|
yield n
|
|
stack.extend(reversed(n.children))
|
|
|
|
def _js_module_specifier(node, source: bytes) -> str | None:
|
|
source_node = node.child_by_field_name("source")
|
|
if source_node is None:
|
|
for child in node.children:
|
|
if child.type == "string":
|
|
source_node = child
|
|
break
|
|
if source_node is None:
|
|
return None
|
|
raw = _read_text(source_node, source).strip()
|
|
return raw.strip("'\"`") or None
|
|
|
|
def _js_named_specifiers(node, source: bytes, specifier_type: str) -> list[tuple[str, str]]:
|
|
pairs: list[tuple[str, str]] = []
|
|
for child in _walk_js_tree(node):
|
|
if child.type != specifier_type:
|
|
continue
|
|
name_node = child.child_by_field_name("name")
|
|
if name_node is None:
|
|
continue
|
|
alias_node = child.child_by_field_name("alias")
|
|
name = _read_text(name_node, source)
|
|
exposed = _read_text(alias_node, source) if alias_node is not None else name
|
|
if name and exposed:
|
|
pairs.append((name, exposed))
|
|
return pairs
|
|
|
|
def _js_export_clause(node):
|
|
for child in node.children:
|
|
if child.type == "export_clause":
|
|
return child
|
|
return None
|
|
|
|
def _js_export_statement_is_star(node) -> bool:
|
|
return any(child.type == "*" for child in node.children)
|
|
|
|
def _js_namespace_export_name(node, source: bytes) -> str | None:
|
|
for child in node.children:
|
|
if child.type != "namespace_export":
|
|
continue
|
|
for sub in child.children:
|
|
if sub.type == "identifier":
|
|
return _read_text(sub, source) or None
|
|
return None
|
|
|
|
def _js_lexical_aliases(node, source: bytes) -> list[tuple[str, str]]:
|
|
aliases: list[tuple[str, str]] = []
|
|
if node.type != "lexical_declaration":
|
|
return aliases
|
|
for child in node.children:
|
|
if child.type != "variable_declarator":
|
|
continue
|
|
name_node = child.child_by_field_name("name")
|
|
value_node = child.child_by_field_name("value")
|
|
if (
|
|
name_node is not None
|
|
and value_node is not None
|
|
and value_node.type in ("identifier", "type_identifier")
|
|
):
|
|
aliases.append((_read_text(name_node, source), _read_text(value_node, source)))
|
|
return aliases
|
|
|
|
def _js_exported_declaration_names(node, source: bytes) -> list[str]:
|
|
names: list[str] = []
|
|
declaration = node.child_by_field_name("declaration")
|
|
if declaration is None:
|
|
return names
|
|
|
|
if declaration.type == "lexical_declaration":
|
|
names.extend(alias for alias, _target in _js_lexical_aliases(declaration, source))
|
|
return names
|
|
|
|
if declaration.type in (
|
|
"class_declaration",
|
|
"abstract_class_declaration",
|
|
"interface_declaration",
|
|
"type_alias_declaration",
|
|
"function_declaration",
|
|
):
|
|
name_node = declaration.child_by_field_name("name")
|
|
if name_node is not None:
|
|
names.append(_read_text(name_node, source))
|
|
return names
|
|
|
|
def _js_default_import_name(node, source: bytes) -> str | None:
|
|
"""Local binding of a default import: the `Foo` in `import Foo from './x'`.
|
|
|
|
The default binding is a bare identifier child of the import_clause (named
|
|
imports live in a `named_imports` node, namespace imports in a
|
|
`namespace_import` node), so it is also picked up from the mixed form
|
|
`import Foo, { Bar } from './x'`.
|
|
"""
|
|
for child in node.children:
|
|
if child.type == "import_clause":
|
|
for sub in child.children:
|
|
if sub.type == "identifier":
|
|
return _read_text(sub, source)
|
|
return None
|
|
|
|
def _js_default_export_name(node, source: bytes) -> str | None:
|
|
"""Local name of a default export, or None for anonymous defaults.
|
|
|
|
Handles `export default class Foo {}`, `export default function foo() {}`,
|
|
`export default abstract class Foo {}` (name on the `declaration` field) and
|
|
`export default Foo` (an identifier on the `value` field). Anonymous defaults
|
|
(`export default class {}`, `export default {...}`) have no resolvable symbol
|
|
and return None.
|
|
"""
|
|
if not any(child.type == "default" for child in node.children):
|
|
return None
|
|
declaration = node.child_by_field_name("declaration")
|
|
if declaration is not None:
|
|
name_node = declaration.child_by_field_name("name")
|
|
return _read_text(name_node, source) if name_node is not None else None
|
|
value = node.child_by_field_name("value")
|
|
if value is not None and value.type == "identifier":
|
|
return _read_text(value, source)
|
|
return None
|
|
|
|
def _js_top_level_function_bodies(path: Path, root_node, source: bytes) -> list[tuple[str, object]]:
|
|
bodies: list[tuple[str, object]] = []
|
|
stem = _file_stem(path)
|
|
for node in root_node.children:
|
|
if node.type == "function_declaration":
|
|
name_node = node.child_by_field_name("name")
|
|
body = node.child_by_field_name("body")
|
|
if name_node is not None and body is not None:
|
|
bodies.append((_make_id(stem, _read_text(name_node, source)), body))
|
|
continue
|
|
if node.type != "lexical_declaration":
|
|
continue
|
|
for child in node.children:
|
|
if child.type != "variable_declarator":
|
|
continue
|
|
name_node = child.child_by_field_name("name")
|
|
value_node = child.child_by_field_name("value")
|
|
if (
|
|
name_node is not None
|
|
and value_node is not None
|
|
and value_node.type == "arrow_function"
|
|
):
|
|
bodies.append((_make_id(stem, _read_text(name_node, source)), value_node))
|
|
return bodies
|
|
|
|
def _js_call_identifier(node, source: bytes) -> str | None:
|
|
if node.type != "call_expression":
|
|
return None
|
|
function_node = node.child_by_field_name("function")
|
|
if function_node is None:
|
|
for child in node.children:
|
|
if child.is_named:
|
|
function_node = child
|
|
break
|
|
if function_node is not None and function_node.type in ("identifier", "type_identifier"):
|
|
return _read_text(function_node, source)
|
|
return None
|
|
|
|
_JS_PRIMITIVE_TYPES = frozenset({
|
|
"string", "number", "boolean", "any", "unknown", "void", "never",
|
|
"object", "null", "undefined", "bigint", "symbol", "this",
|
|
})
|
|
|
|
def _ts_heritage_clause_entries(clause_node, source: bytes) -> list[str]:
|
|
"""Return base/interface type names from an extends_clause or implements_clause."""
|
|
out: list[str] = []
|
|
for child in clause_node.children:
|
|
if not child.is_named:
|
|
continue
|
|
if child.type in ("identifier", "type_identifier"):
|
|
name = _read_text(child, source)
|
|
if name:
|
|
out.append(name)
|
|
elif child.type == "generic_type":
|
|
name_node = child.child_by_field_name("name")
|
|
if name_node is None:
|
|
for sub in child.children:
|
|
if sub.type in ("type_identifier", "nested_type_identifier", "identifier"):
|
|
name_node = sub
|
|
break
|
|
if name_node is not None:
|
|
text = _read_text(name_node, source).rsplit(".", 1)[-1]
|
|
if text:
|
|
out.append(text)
|
|
elif child.type == "nested_type_identifier":
|
|
text = _read_text(child, source).rsplit(".", 1)[-1]
|
|
if text:
|
|
out.append(text)
|
|
return out
|
|
|
|
def _ts_collect_type_refs(node, source: bytes, generic: bool, out: list[tuple[str, str]]) -> None:
|
|
"""Walk a TS type annotation tree; append (name, role) tuples.
|
|
|
|
role is 'type' for the outermost type position and 'generic_arg' for entries
|
|
that appear inside `type_arguments`.
|
|
"""
|
|
if node is None:
|
|
return
|
|
t = node.type
|
|
if t == "type_annotation":
|
|
for c in node.children:
|
|
if c.is_named:
|
|
_ts_collect_type_refs(c, source, generic, out)
|
|
return
|
|
if t in ("type_identifier", "identifier"):
|
|
name = _read_text(node, source)
|
|
if name and name not in _JS_PRIMITIVE_TYPES:
|
|
out.append((name, "generic_arg" if generic else "type"))
|
|
return
|
|
if t == "nested_type_identifier":
|
|
tail = _read_text(node, source).rsplit(".", 1)[-1]
|
|
if tail and tail not in _JS_PRIMITIVE_TYPES:
|
|
out.append((tail, "generic_arg" if generic else "type"))
|
|
return
|
|
if t == "generic_type":
|
|
name_node = node.child_by_field_name("name")
|
|
if name_node is not None:
|
|
text = _read_text(name_node, source).rsplit(".", 1)[-1]
|
|
if text and text not in _JS_PRIMITIVE_TYPES:
|
|
out.append((text, "generic_arg" if generic else "type"))
|
|
else:
|
|
for c in node.children:
|
|
if c.type in ("type_identifier", "nested_type_identifier"):
|
|
text = _read_text(c, source).rsplit(".", 1)[-1]
|
|
if text and text not in _JS_PRIMITIVE_TYPES:
|
|
out.append((text, "generic_arg" if generic else "type"))
|
|
break
|
|
for c in node.children:
|
|
if c.type == "type_arguments":
|
|
for sub in c.children:
|
|
if sub.is_named:
|
|
_ts_collect_type_refs(sub, source, True, out)
|
|
return
|
|
if node.is_named:
|
|
for c in node.children:
|
|
if c.is_named:
|
|
_ts_collect_type_refs(c, source, generic, out)
|
|
|
|
def _ts_walk_class_members(class_node, source: bytes, path: Path, class_nid: str,
|
|
facts: _SymbolResolutionFacts) -> None:
|
|
"""Emit type-relation and type-reference use facts for a class declaration node."""
|
|
line = class_node.start_point[0] + 1
|
|
for child in class_node.children:
|
|
if child.type == "class_heritage":
|
|
for clause in child.children:
|
|
if clause.type == "extends_clause":
|
|
for name in _ts_heritage_clause_entries(clause, source):
|
|
facts.uses.append(
|
|
_SymbolUseFact(path, class_nid, name, "inherits", "type",
|
|
clause.start_point[0] + 1)
|
|
)
|
|
elif clause.type == "implements_clause":
|
|
for name in _ts_heritage_clause_entries(clause, source):
|
|
facts.uses.append(
|
|
_SymbolUseFact(path, class_nid, name, "implements", "type",
|
|
clause.start_point[0] + 1)
|
|
)
|
|
elif child.type == "extends_type_clause":
|
|
# Interface heritage (`interface A extends B, C`) is an
|
|
# extends_type_clause node, NOT a class_heritage. Its base entries
|
|
# are the same node types extends_clause holds, so the helper is
|
|
# reusable. Without this branch interface inheritance is dropped (#1095).
|
|
for name in _ts_heritage_clause_entries(child, source):
|
|
facts.uses.append(
|
|
_SymbolUseFact(path, class_nid, name, "inherits", "type",
|
|
child.start_point[0] + 1)
|
|
)
|
|
|
|
body = class_node.child_by_field_name("body")
|
|
if body is None:
|
|
return
|
|
|
|
for member in body.children:
|
|
m_line = member.start_point[0] + 1
|
|
if member.type in ("method_definition", "method_signature", "abstract_method_signature"):
|
|
name_node = member.child_by_field_name("name")
|
|
if name_node is None:
|
|
continue
|
|
method_name = _read_text(name_node, source)
|
|
method_nid = _make_id(class_nid, method_name)
|
|
params = member.child_by_field_name("parameters")
|
|
if params is not None:
|
|
for p in params.children:
|
|
if p.type not in ("required_parameter", "optional_parameter"):
|
|
continue
|
|
type_anno = p.child_by_field_name("type")
|
|
if type_anno is None:
|
|
continue
|
|
refs: list[tuple[str, str]] = []
|
|
_ts_collect_type_refs(type_anno, source, False, refs)
|
|
for name, role in refs:
|
|
ctx = "generic_arg" if role == "generic_arg" else "parameter_type"
|
|
facts.uses.append(
|
|
_SymbolUseFact(path, method_nid, name, "references", ctx, m_line)
|
|
)
|
|
return_type = member.child_by_field_name("return_type")
|
|
if return_type is not None:
|
|
refs = []
|
|
_ts_collect_type_refs(return_type, source, False, refs)
|
|
for name, role in refs:
|
|
ctx = "generic_arg" if role == "generic_arg" else "return_type"
|
|
facts.uses.append(
|
|
_SymbolUseFact(path, method_nid, name, "references", ctx, m_line)
|
|
)
|
|
elif member.type in ("public_field_definition", "property_signature"):
|
|
type_anno = None
|
|
for c in member.children:
|
|
if c.type == "type_annotation":
|
|
type_anno = c
|
|
break
|
|
if type_anno is None:
|
|
continue
|
|
refs = []
|
|
_ts_collect_type_refs(type_anno, source, False, refs)
|
|
for name, role in refs:
|
|
ctx = "generic_arg" if role == "generic_arg" else "field"
|
|
facts.uses.append(
|
|
_SymbolUseFact(path, class_nid, name, "references", ctx, m_line)
|
|
)
|
|
|
|
def _collect_js_symbol_resolution_facts(paths: list[Path], facts: _SymbolResolutionFacts) -> None:
|
|
js_paths = [
|
|
path for path in paths
|
|
if path.suffix in _JS_CACHE_BYPASS_SUFFIXES
|
|
]
|
|
if not js_paths:
|
|
return
|
|
|
|
trees: dict[Path, tuple[bytes, object]] = {}
|
|
|
|
for path in js_paths:
|
|
resolved_path = path.resolve()
|
|
parsed = _parse_js_tree(path)
|
|
if parsed is None:
|
|
continue
|
|
source, root_node = parsed
|
|
trees[resolved_path] = parsed
|
|
|
|
for node in _walk_js_tree(root_node):
|
|
if node.type == "export_statement":
|
|
for name in _js_exported_declaration_names(node, source):
|
|
facts.declarations.append(
|
|
_SymbolDeclarationFact(path, name, node.start_point[0] + 1)
|
|
)
|
|
|
|
if node.type != "import_statement":
|
|
continue
|
|
raw_module = _js_module_specifier(node, source)
|
|
if raw_module is None:
|
|
continue
|
|
target_path = _resolve_js_module_path(raw_module, path.parent)
|
|
if target_path is None:
|
|
continue
|
|
target_path = target_path.resolve()
|
|
for imported_name, local_name in _js_named_specifiers(node, source, "import_specifier"):
|
|
facts.imports.append(
|
|
_SymbolImportFact(
|
|
path,
|
|
local_name,
|
|
target_path,
|
|
imported_name,
|
|
node.start_point[0] + 1,
|
|
)
|
|
)
|
|
default_local = _js_default_import_name(node, source)
|
|
if default_local is not None:
|
|
facts.imports.append(
|
|
_SymbolImportFact(
|
|
path,
|
|
default_local,
|
|
target_path,
|
|
"default",
|
|
node.start_point[0] + 1,
|
|
)
|
|
)
|
|
|
|
for node in _walk_js_tree(root_node):
|
|
for alias, target in _js_lexical_aliases(node, source):
|
|
facts.aliases.append(
|
|
_SymbolAliasFact(path, alias, target, node.start_point[0] + 1)
|
|
)
|
|
|
|
for path in js_paths:
|
|
resolved_path = path.resolve()
|
|
parsed = trees.get(resolved_path)
|
|
if parsed is None:
|
|
continue
|
|
source, root_node = parsed
|
|
|
|
for node in _walk_js_tree(root_node):
|
|
if node.type != "export_statement":
|
|
continue
|
|
|
|
raw_module = _js_module_specifier(node, source)
|
|
export_clause = _js_export_clause(node)
|
|
if raw_module is not None:
|
|
target_path = _resolve_js_module_path(raw_module, path.parent)
|
|
if target_path is None:
|
|
continue
|
|
target_path = target_path.resolve()
|
|
namespace_name = _js_namespace_export_name(node, source)
|
|
if namespace_name is not None:
|
|
facts.namespace_exports.append(
|
|
_NamespaceExportFact(
|
|
path,
|
|
namespace_name,
|
|
target_path,
|
|
node.start_point[0] + 1,
|
|
)
|
|
)
|
|
elif _js_export_statement_is_star(node):
|
|
facts.star_exports.append(
|
|
_StarExportFact(path, target_path, node.start_point[0] + 1)
|
|
)
|
|
if export_clause is not None:
|
|
for original_name, exported_name in _js_named_specifiers(
|
|
export_clause, source, "export_specifier"
|
|
):
|
|
facts.exports.append(
|
|
_SymbolExportFact(
|
|
path,
|
|
exported_name,
|
|
node.start_point[0] + 1,
|
|
target_path=target_path,
|
|
target_name=original_name,
|
|
)
|
|
)
|
|
continue
|
|
|
|
if export_clause is not None:
|
|
for local_name, exported_name in _js_named_specifiers(
|
|
export_clause, source, "export_specifier"
|
|
):
|
|
facts.exports.append(
|
|
_SymbolExportFact(
|
|
path,
|
|
exported_name,
|
|
node.start_point[0] + 1,
|
|
local_name=local_name,
|
|
)
|
|
)
|
|
continue
|
|
|
|
for exported_name in _js_exported_declaration_names(node, source):
|
|
facts.exports.append(
|
|
_SymbolExportFact(
|
|
path,
|
|
exported_name,
|
|
node.start_point[0] + 1,
|
|
local_name=exported_name,
|
|
)
|
|
)
|
|
|
|
# `export default class Foo {}` / `export default foo` exposes the
|
|
# symbol under the name "default"; record that so a default import
|
|
# (imported_name="default") resolves to it. `export { X as default }`
|
|
# is already handled via the export_clause path above.
|
|
default_name = _js_default_export_name(node, source)
|
|
if default_name is not None:
|
|
facts.exports.append(
|
|
_SymbolExportFact(
|
|
path,
|
|
"default",
|
|
node.start_point[0] + 1,
|
|
local_name=default_name,
|
|
)
|
|
)
|
|
|
|
for path in js_paths:
|
|
resolved_path = path.resolve()
|
|
parsed = trees.get(resolved_path)
|
|
if parsed is None:
|
|
continue
|
|
source, root_node = parsed
|
|
for source_id, body in _js_top_level_function_bodies(path, root_node, source):
|
|
for node in _walk_js_tree(body):
|
|
imported_name = _js_call_identifier(node, source)
|
|
if imported_name is None:
|
|
continue
|
|
facts.uses.append(
|
|
_SymbolUseFact(
|
|
path,
|
|
source_id,
|
|
imported_name,
|
|
"calls",
|
|
"call",
|
|
node.start_point[0] + 1,
|
|
)
|
|
)
|
|
|
|
for path in js_paths:
|
|
resolved_path = path.resolve()
|
|
parsed = trees.get(resolved_path)
|
|
if parsed is None:
|
|
continue
|
|
source, root_node = parsed
|
|
stem = _file_stem(path)
|
|
for node in _walk_js_tree(root_node):
|
|
if node.type not in (
|
|
"class_declaration",
|
|
"abstract_class_declaration",
|
|
"interface_declaration",
|
|
):
|
|
continue
|
|
name_node = node.child_by_field_name("name")
|
|
if name_node is None:
|
|
continue
|
|
class_name = _read_text(name_node, source)
|
|
if not class_name:
|
|
continue
|
|
class_nid = _make_id(stem, class_name)
|
|
_ts_walk_class_members(node, source, path, class_nid, facts)
|
|
|
|
def _parse_python_tree(path: Path):
|
|
try:
|
|
import tree_sitter_python as tspython
|
|
from tree_sitter import Language, Parser
|
|
source = path.read_bytes()
|
|
parser = Parser(Language(tspython.language()))
|
|
return source, parser.parse(source).root_node
|
|
except Exception:
|
|
return None
|
|
|
|
def _walk_python_tree(node):
|
|
yield node
|
|
for child in node.children:
|
|
yield from _walk_python_tree(child)
|
|
|
|
def _python_import_from_module(node, source: bytes) -> tuple[int, str] | None:
|
|
level = 0
|
|
module_name = ""
|
|
for child in node.children:
|
|
if child.type == "import":
|
|
break
|
|
if child.type == "relative_import":
|
|
raw = _read_text(child, source)
|
|
level = len(raw) - len(raw.lstrip("."))
|
|
remainder = raw.lstrip(".")
|
|
if remainder:
|
|
module_name = remainder
|
|
for sub in child.children:
|
|
if sub.type == "dotted_name":
|
|
module_name = _read_text(sub, source)
|
|
elif child.type == "dotted_name":
|
|
module_name = _read_text(child, source)
|
|
if level == 0 and not module_name:
|
|
return None
|
|
return level, module_name
|
|
|
|
def _python_imported_names(node, source: bytes) -> list[tuple[str, str]]:
|
|
names: list[tuple[str, str]] = []
|
|
past_import = False
|
|
for child in node.children:
|
|
if child.type == "import":
|
|
past_import = True
|
|
continue
|
|
if not past_import:
|
|
continue
|
|
if child.type == "dotted_name":
|
|
name = _read_text(child, source)
|
|
names.append((name, name.split(".")[-1]))
|
|
elif child.type == "aliased_import":
|
|
name_node = child.child_by_field_name("name")
|
|
alias_node = child.child_by_field_name("alias")
|
|
if name_node is None:
|
|
continue
|
|
name = _read_text(name_node, source)
|
|
local = _read_text(alias_node, source) if alias_node is not None else name.split(".")[-1]
|
|
names.append((name, local))
|
|
return names
|
|
|
|
def _resolve_python_module_path(module_name: str, current_path: Path, root: Path, level: int) -> Path | None:
|
|
if level > 0:
|
|
base = current_path.parent
|
|
for _ in range(level - 1):
|
|
base = base.parent
|
|
candidate = base / module_name.replace(".", "/") if module_name else base
|
|
else:
|
|
candidate = root / module_name.replace(".", "/")
|
|
|
|
if candidate.is_dir():
|
|
init_path = candidate / "__init__.py"
|
|
if init_path.is_file():
|
|
return init_path
|
|
if candidate.is_file():
|
|
return candidate
|
|
py_candidate = candidate.with_suffix(".py")
|
|
if py_candidate.is_file():
|
|
return py_candidate
|
|
return None
|
|
|
|
def _python_top_level_function_bodies(path: Path, root_node, source: bytes) -> list[tuple[str, object]]:
|
|
bodies: list[tuple[str, object]] = []
|
|
stem = _file_stem(path)
|
|
for node in root_node.children:
|
|
if node.type != "function_definition":
|
|
continue
|
|
name_node = node.child_by_field_name("name")
|
|
body = node.child_by_field_name("body")
|
|
if name_node is not None and body is not None:
|
|
bodies.append((_make_id(stem, _read_text(name_node, source)), body))
|
|
return bodies
|
|
|
|
def _python_call_identifier(node, source: bytes) -> str | None:
|
|
if node.type != "call":
|
|
return None
|
|
function_node = node.child_by_field_name("function")
|
|
if function_node is not None and function_node.type == "identifier":
|
|
return _read_text(function_node, source)
|
|
return None
|
|
|
|
def _collect_python_symbol_resolution_facts(
|
|
paths: list[Path],
|
|
root: Path,
|
|
facts: _SymbolResolutionFacts,
|
|
) -> None:
|
|
py_paths = [path for path in paths if path.suffix == ".py"]
|
|
if not py_paths:
|
|
return
|
|
|
|
trees: dict[Path, tuple[bytes, object]] = {}
|
|
for path in py_paths:
|
|
parsed = _parse_python_tree(path)
|
|
if parsed is None:
|
|
continue
|
|
source, root_node = parsed
|
|
trees[path.resolve()] = parsed
|
|
|
|
for node in _walk_python_tree(root_node):
|
|
if node.type != "import_from_statement":
|
|
continue
|
|
module = _python_import_from_module(node, source)
|
|
if module is None:
|
|
continue
|
|
level, module_name = module
|
|
target_path = _resolve_python_module_path(module_name, path, root, level)
|
|
if target_path is None:
|
|
continue
|
|
# #1146: `from pkg import submod` — if the target is a package
|
|
# (__init__.py) and an imported name matches a submodule file on
|
|
# disk, emit a file-level import edge to that submodule rather
|
|
# than only to the package.
|
|
pkg_dir = target_path.parent if target_path.name == "__init__.py" else None
|
|
for imported_name, local_name in _python_imported_names(node, source):
|
|
line = node.start_point[0] + 1
|
|
if pkg_dir is not None:
|
|
sub_py = pkg_dir / f"{imported_name}.py"
|
|
sub_pkg = pkg_dir / imported_name / "__init__.py"
|
|
submodule = sub_py if sub_py.is_file() else (sub_pkg if sub_pkg.is_file() else None)
|
|
if submodule is not None:
|
|
facts.module_imports.append((path, submodule, line))
|
|
continue
|
|
facts.imports.append(
|
|
_SymbolImportFact(path, local_name, target_path, imported_name, line)
|
|
)
|
|
if path.name == "__init__.py":
|
|
facts.exports.append(
|
|
_SymbolExportFact(
|
|
path,
|
|
local_name,
|
|
line,
|
|
target_path=target_path,
|
|
target_name=imported_name,
|
|
)
|
|
)
|
|
|
|
for path in py_paths:
|
|
parsed = trees.get(path.resolve())
|
|
if parsed is None:
|
|
continue
|
|
source, root_node = parsed
|
|
for source_id, body in _python_top_level_function_bodies(path, root_node, source):
|
|
for node in _walk_python_tree(body):
|
|
imported_name = _python_call_identifier(node, source)
|
|
if imported_name is None:
|
|
continue
|
|
facts.uses.append(
|
|
_SymbolUseFact(
|
|
path,
|
|
source_id,
|
|
imported_name,
|
|
"calls",
|
|
"call",
|
|
node.start_point[0] + 1,
|
|
)
|
|
)
|
|
|
|
def _augment_symbol_resolution_edges(
|
|
paths: list[Path],
|
|
nodes: list[dict],
|
|
edges: list[dict],
|
|
root: Path,
|
|
) -> None:
|
|
facts = _SymbolResolutionFacts()
|
|
_collect_js_symbol_resolution_facts(paths, facts)
|
|
_collect_python_symbol_resolution_facts(paths, root, facts)
|
|
_apply_symbol_resolution_facts(paths, nodes, edges, root, facts)
|
|
|
|
def _resolve_cross_file_imports(
|
|
per_file: list[dict],
|
|
paths: list[Path],
|
|
) -> list[dict]:
|
|
"""
|
|
Two-pass import resolution: turn file-level imports into class-level edges.
|
|
|
|
Pass 1 - build a global map: class/function name → node_id, per stem.
|
|
Pass 2 - for each `from .module import Name`, look up Name in the global
|
|
map and add a direct INFERRED edge from each class in the
|
|
importing file to the imported entity.
|
|
|
|
This turns:
|
|
auth.py --imports_from--> models.py (obvious, filtered out)
|
|
Into:
|
|
DigestAuth --uses--> Response [INFERRED] (cross-file, interesting!)
|
|
BasicAuth --uses--> Request [INFERRED]
|
|
"""
|
|
try:
|
|
import tree_sitter_python as tspython
|
|
from tree_sitter import Language, Parser
|
|
except ImportError:
|
|
return []
|
|
|
|
language = Language(tspython.language())
|
|
parser = Parser(language)
|
|
|
|
# Pass 1: _file_stem(path) → {ClassName: node_id}
|
|
# Keyed by directory-qualified stem (e.g. "auth_models") to avoid collisions
|
|
# when multiple files share the same filename in different directories.
|
|
# A secondary bare-stem index handles absolute imports where only the module
|
|
# name is known — first writer wins when names collide (inherently ambiguous).
|
|
stem_to_entities: dict[str, dict[str, str]] = {}
|
|
bare_to_qualified: dict[str, str] = {}
|
|
for file_result in per_file:
|
|
for node in file_result.get("nodes", []):
|
|
src = node.get("source_file", "")
|
|
if not src:
|
|
continue
|
|
src_path = Path(src)
|
|
fq_stem = _file_stem(src_path)
|
|
label = node.get("label", "")
|
|
nid = node.get("id", "")
|
|
# Index class-level entities only. Function/method labels end in "()"
|
|
# so are excluded by the `endswith(")")` filter; file nodes end in ".py";
|
|
# private/internal labels start with "_"; rationale nodes carry
|
|
# file_type=="rationale" and must never participate in cross-file
|
|
# import resolution (#563).
|
|
if (
|
|
label
|
|
and not label.endswith((")", ".py"))
|
|
and "_" not in label[:1]
|
|
and node.get("file_type") != "rationale"
|
|
):
|
|
stem_to_entities.setdefault(fq_stem, {})[label] = nid
|
|
if src_path.stem not in bare_to_qualified:
|
|
bare_to_qualified[src_path.stem] = fq_stem
|
|
|
|
# Pass 2: for each file, find `from .X import A, B, C` and resolve
|
|
new_edges: list[dict] = []
|
|
stem_to_path: dict[str, Path] = {_file_stem(p): p for p in paths}
|
|
|
|
for file_result, path in zip(per_file, paths):
|
|
stem = _file_stem(path)
|
|
str_path = str(path)
|
|
|
|
# Find all classes defined in this file (the importers).
|
|
# Excludes rationale nodes whose labels happen not to end in ")" or ".py"
|
|
# but which must never be treated as importing entities (#563).
|
|
local_classes = [
|
|
n["id"] for n in file_result.get("nodes", [])
|
|
if n.get("source_file") == str_path
|
|
and not n["label"].endswith((")", ".py"))
|
|
and n["id"] != _make_id(stem) # exclude file-level node
|
|
and n.get("file_type") != "rationale"
|
|
]
|
|
if not local_classes:
|
|
continue
|
|
|
|
# Parse imports from this file
|
|
try:
|
|
source = path.read_bytes()
|
|
tree = parser.parse(source)
|
|
except Exception:
|
|
continue
|
|
|
|
def walk_imports(node) -> None:
|
|
if node.type == "import_from_statement":
|
|
# Find the module name - handles both absolute and relative imports.
|
|
# Relative: `from .models import X` → relative_import → dotted_name
|
|
# Absolute: `from models import X` → module_name field
|
|
# target_fq is the directory-qualified stem used as the key in
|
|
# stem_to_entities. Relative imports are resolved exactly via the
|
|
# importing file's directory; absolute imports fall back to the
|
|
# bare-stem secondary index (first-writer-wins when names collide).
|
|
target_fq: str | None = None
|
|
for child in node.children:
|
|
if child.type == "relative_import":
|
|
for sub in child.children:
|
|
if sub.type == "dotted_name":
|
|
raw = source[sub.start_byte:sub.end_byte].decode("utf-8", errors="replace")
|
|
bare = raw.split(".")[-1]
|
|
# Resolve relative import to exact qualified stem.
|
|
candidate = path.parent / f"{bare}.py"
|
|
target_fq = _file_stem(candidate)
|
|
break
|
|
break
|
|
if child.type == "dotted_name" and target_fq is None:
|
|
raw = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
|
bare = raw.split(".")[-1]
|
|
target_fq = bare_to_qualified.get(bare)
|
|
|
|
if not target_fq or target_fq not in stem_to_entities:
|
|
return
|
|
|
|
# Collect imported names: dotted_name children of import_from_statement
|
|
# that come AFTER the 'import' keyword token.
|
|
imported_names: list[str] = []
|
|
past_import_kw = False
|
|
for child in node.children:
|
|
if child.type == "import":
|
|
past_import_kw = True
|
|
continue
|
|
if not past_import_kw:
|
|
continue
|
|
if child.type == "dotted_name":
|
|
imported_names.append(
|
|
source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
|
)
|
|
elif child.type == "aliased_import":
|
|
# `import X as Y` - take the original name
|
|
name_node = child.child_by_field_name("name")
|
|
if name_node:
|
|
imported_names.append(
|
|
source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
|
)
|
|
|
|
line = node.start_point[0] + 1
|
|
for name in imported_names:
|
|
tgt_nid = stem_to_entities[target_fq].get(name)
|
|
if tgt_nid:
|
|
for src_class_nid in local_classes:
|
|
new_edges.append({
|
|
"source": src_class_nid,
|
|
"target": tgt_nid,
|
|
"relation": "uses",
|
|
"confidence": "INFERRED",
|
|
"source_file": str_path,
|
|
"source_location": f"L{line}",
|
|
"weight": 0.8,
|
|
})
|
|
for child in node.children:
|
|
walk_imports(child)
|
|
|
|
walk_imports(tree.root_node)
|
|
|
|
return new_edges
|
|
|
|
_DECLDEF_HEADER_SUFFIXES = frozenset({".h", ".hpp", ".hh", ".hxx"})
|
|
|
|
_DECLDEF_IMPL_SUFFIXES = frozenset({".m", ".mm", ".cpp", ".cc", ".cxx", ".c"})
|
|
|
|
def _decldef_class_stem(source_file: str) -> tuple[str, str] | None:
|
|
"""Return ``(dir, base_stem)`` for a header/impl source file, else None.
|
|
|
|
The base stem strips an ObjC category suffix (``Foo+Cat.m`` -> ``Foo``) so a
|
|
category implementation pairs with its ``Foo.h`` declaration. Files with an
|
|
extension that is neither a header nor an impl extension return None and are
|
|
never considered for the merge.
|
|
"""
|
|
if not source_file:
|
|
return None
|
|
p = Path(source_file)
|
|
suffix = p.suffix.lower()
|
|
if suffix not in _DECLDEF_HEADER_SUFFIXES and suffix not in _DECLDEF_IMPL_SUFFIXES:
|
|
return None
|
|
stem = p.stem.split("+", 1)[0] # ObjC category: Foo+Cat -> Foo
|
|
if not stem:
|
|
return None
|
|
return (str(p.parent), stem)
|
|
|
|
def _merge_decl_def_classes(
|
|
all_nodes: list[dict],
|
|
all_edges: list[dict],
|
|
) -> None:
|
|
"""Merge a class (and its methods) declared in a header with its definition in
|
|
a sibling impl file into ONE node, for C/C++/ObjC (#1547, #1556).
|
|
|
|
A class declared in ``Foo.h`` (``class Foo`` / ``@interface Foo``) and defined
|
|
in the sibling ``Foo.cpp`` / ``Foo.m`` (``@implementation Foo``, plus — after
|
|
the C++ qualified-name fix — out-of-class method definitions ``Foo::bar``)
|
|
produces TWO nodes per symbol. Both are keyed off the file *stem*, and
|
|
``_file_stem`` drops the extension, so the header symbol and its impl
|
|
counterpart get the IDENTICAL id and differ only in ``source_file`` and label
|
|
(the C++ def label is ``Foo::bar()`` vs the decl's ``bar``; the ObjC impl class
|
|
label equals the interface's). Left alone, ``_disambiguate_colliding_node_ids``
|
|
SPLITS those id-collisions apart by path, fragmenting one class into two def
|
|
nodes — which then trips every resolver's single-definition god-node guard
|
|
(``len(defs) != 1`` -> bail), cascading into lost .h<->.m/.cpp linkage and dead
|
|
cross-file calls.
|
|
|
|
This pass runs BEFORE disambiguation and collapses each such id-collision to
|
|
ONE node — the header (declaration) variant, consistent with the #1475
|
|
header_remaps direction — so disambiguation sees a single source_file per id
|
|
and leaves it alone, and the downstream resolvers see ONE definition. Because
|
|
the colliding nodes already share an id, no edge re-pointing is needed: every
|
|
edge that referenced the impl symbol already points at the surviving id. We
|
|
only drop the redundant duplicate node and prefer the header's label.
|
|
|
|
GOD-NODE GUARDS (false merges are the main risk):
|
|
|
|
* Collapse fires ONLY when every node in an id-collision group comes from a
|
|
SIBLING header/impl set — same directory, same base stem (ObjC categories
|
|
``Foo+Cat.m`` compare by the stem before ``+``), header extension paired
|
|
with impl extension — AND the group contains exactly ONE header file.
|
|
* Two unrelated ``class Logger`` in DIFFERENT directories never collide on id
|
|
(the id embeds the full file stem / directory path), so they are never
|
|
grouped and never merge. Two same-named classes in the SAME directory but
|
|
different base stems likewise key to different ids. Any id-collision that
|
|
is NOT a clean single-header sibling set is left untouched for
|
|
disambiguation to split (the conservative default).
|
|
|
|
The class and its method/field members fold in together: members are keyed
|
|
``_make_id(class_id, name)`` (ObjC) or, for an out-of-class C++ definition,
|
|
``_make_id(stem, "Foo::bar")`` which normalizes to the same id as the in-class
|
|
member ``_make_id(class_id, "bar")``. So every decl/def member pair is itself an
|
|
id-collision across the same sibling file set and collapses by the same rule.
|
|
"""
|
|
# Group every code node by id, recording the distinct source files involved.
|
|
by_id: dict[str, list[dict]] = {}
|
|
for n in all_nodes:
|
|
if n.get("file_type") != "code":
|
|
continue
|
|
nid = n.get("id")
|
|
sf = str(n.get("source_file", ""))
|
|
if not isinstance(nid, str) or not nid or not sf:
|
|
continue
|
|
by_id.setdefault(nid, []).append(n)
|
|
|
|
# Identify, per surviving id, which node to keep (header preferred). We can't
|
|
# mutate all_nodes mid-scan, so collect a set of node object ids to drop.
|
|
drop_objs: set[int] = set()
|
|
for nid, group in by_id.items():
|
|
if len(group) < 2:
|
|
continue
|
|
# The distinct source files of this collision must form a clean sibling
|
|
# header/impl set with exactly one header. Each file must parse as a
|
|
# header/impl file (others -> bail), share one directory + base stem.
|
|
sibling_keys: set[tuple[str, str]] = set()
|
|
headers: list[dict] = []
|
|
ok = True
|
|
for node in group:
|
|
sf = str(node.get("source_file", ""))
|
|
ds = _decldef_class_stem(sf)
|
|
if ds is None:
|
|
ok = False
|
|
break
|
|
sibling_keys.add(ds)
|
|
if Path(sf).suffix.lower() in _DECLDEF_HEADER_SUFFIXES:
|
|
headers.append(node)
|
|
if not ok:
|
|
continue
|
|
# All from one (dir, base_stem) sibling family, with a UNIQUE header.
|
|
if len(sibling_keys) != 1 or len(headers) != 1:
|
|
continue
|
|
keeper = headers[0]
|
|
for node in group:
|
|
if node is not keeper:
|
|
drop_objs.add(id(node))
|
|
|
|
if not drop_objs:
|
|
return
|
|
|
|
# Drop the redundant duplicate nodes. The surviving (header) node keeps its
|
|
# own label/source_file; edges are unchanged because the id is identical. Then
|
|
# de-dup any now-identical edges (e.g. the impl file's `contains`/`method`
|
|
# edge that duplicates the header's after the collapse).
|
|
all_nodes[:] = [n for n in all_nodes if id(n) not in drop_objs]
|
|
|
|
seen_keys: set[tuple] = set()
|
|
rewritten: list[dict] = []
|
|
for e in all_edges:
|
|
src = e.get("source")
|
|
tgt = e.get("target")
|
|
if src == tgt:
|
|
continue
|
|
k = (src, tgt, e.get("relation"), e.get("context"))
|
|
if k in seen_keys:
|
|
continue
|
|
seen_keys.add(k)
|
|
rewritten.append(e)
|
|
all_edges[:] = rewritten
|
|
|
|
def _resolve_cross_file_java_imports(
|
|
per_file: list[dict],
|
|
paths: list[Path],
|
|
) -> list[dict]:
|
|
"""Two-pass Java import resolution.
|
|
|
|
Pass 1: build a global index {ClassName: [node_id, ...]} across all Java nodes.
|
|
Pass 2: re-parse each Java file; for every `import a.b.C;`, resolve C against
|
|
the index. Wildcard and stdlib imports produce no edge.
|
|
"""
|
|
try:
|
|
import tree_sitter_java as tsjava
|
|
from tree_sitter import Language, Parser
|
|
except ImportError:
|
|
return []
|
|
|
|
language = Language(tsjava.language())
|
|
parser = Parser(language)
|
|
|
|
# Pass 1: class-name → node_id index (only internal, uppercase-starting names)
|
|
name_to_ids: dict[str, list[str]] = {}
|
|
for file_result in per_file:
|
|
for node in file_result.get("nodes", []):
|
|
label = node.get("label", "")
|
|
nid = node.get("id", "")
|
|
src = node.get("source_file", "")
|
|
if not label or not nid or not src:
|
|
continue
|
|
if label.endswith(")") or label.endswith(".java"):
|
|
continue
|
|
if not label[0].isalpha() or not label[0].isupper():
|
|
continue
|
|
name_to_ids.setdefault(label, []).append(nid)
|
|
|
|
# Pass 2: resolve imports to real node IDs
|
|
new_edges: list[dict] = []
|
|
seen_pairs: set[tuple[str, str]] = set()
|
|
for path in paths:
|
|
file_nid = _make_id(str(path))
|
|
try:
|
|
source = path.read_bytes()
|
|
tree = parser.parse(source)
|
|
except Exception:
|
|
continue
|
|
|
|
def walk(n) -> None:
|
|
if n.type == "import_declaration":
|
|
raw = _read_text(n, source).strip()
|
|
body = raw[len("import"):].strip().rstrip(";").strip()
|
|
if body.startswith("static "):
|
|
body = body[len("static "):].strip()
|
|
if body.endswith(".*"):
|
|
return
|
|
parts = body.split(".")
|
|
if not parts:
|
|
return
|
|
last = parts[-1]
|
|
if last and last[0].islower() and len(parts) >= 2:
|
|
last = parts[-2]
|
|
at_line = n.start_point[0] + 1
|
|
for tgt_nid in name_to_ids.get(last, []):
|
|
if tgt_nid == file_nid:
|
|
continue
|
|
key = (file_nid, tgt_nid)
|
|
if key in seen_pairs:
|
|
continue
|
|
seen_pairs.add(key)
|
|
new_edges.append({
|
|
"source": file_nid,
|
|
"target": tgt_nid,
|
|
"relation": "imports",
|
|
"confidence": "EXTRACTED",
|
|
"confidence_score": 1.0,
|
|
"source_file": str(path),
|
|
"source_location": f"L{at_line}",
|
|
"weight": 1.0,
|
|
})
|
|
for child in n.children:
|
|
walk(child)
|
|
|
|
walk(tree.root_node)
|
|
|
|
return new_edges
|
|
|
|
def _resolve_java_type_references(
|
|
per_file: list[dict],
|
|
paths: list[Path],
|
|
all_nodes: list[dict],
|
|
all_edges: list[dict],
|
|
) -> None:
|
|
"""Re-point dangling Java ``implements``/``inherits`` edges to the real
|
|
definition, using the referencing file's ``import`` statements (+ package)
|
|
for exact disambiguation.
|
|
|
|
Cross-file type references resolve by bare name and fall back to a no-source
|
|
"shadow" stub. ``_rewire_unique_stub_nodes`` repairs that only when the name
|
|
is globally unique; when two packages define a same-named type it bails, so
|
|
the ``implements`` edge stays stuck on the shadow node and the real interface
|
|
is wrongly isolated (#1318). An ``import com.a.handler.AIResponseHandler``
|
|
names the exact package, so it disambiguates where bare-name matching cannot.
|
|
|
|
Mutates ``all_nodes``/``all_edges`` in place. Runs after id-disambiguation so
|
|
target ids are final, and after ``_rewire_unique_stub_nodes`` so it only has
|
|
to handle the ambiguous remainder.
|
|
"""
|
|
try:
|
|
import tree_sitter_java as tsjava
|
|
from tree_sitter import Language, Parser
|
|
except ImportError:
|
|
return
|
|
|
|
language = Language(tsjava.language())
|
|
parser = Parser(language)
|
|
|
|
# package + simple-name->FQN imports, keyed by the source_file string the
|
|
# file's own nodes use (so it matches edge/node source_file exactly).
|
|
pkg_by_file: dict[str, str] = {}
|
|
imports_by_file: dict[str, dict[str, str]] = {}
|
|
for path, result in zip(paths, per_file):
|
|
srcs = {n.get("source_file") for n in result.get("nodes", []) if n.get("source_file")}
|
|
if not srcs:
|
|
continue
|
|
try:
|
|
source = path.read_bytes()
|
|
tree = parser.parse(source)
|
|
except Exception:
|
|
continue
|
|
pkg = ""
|
|
imps: dict[str, str] = {}
|
|
|
|
def walk(n) -> None:
|
|
nonlocal pkg
|
|
if n.type == "package_declaration":
|
|
pkg = _read_text(n, source).strip()[len("package"):].strip().rstrip(";").strip()
|
|
elif n.type == "import_declaration":
|
|
body = _read_text(n, source).strip()[len("import"):].strip().rstrip(";").strip()
|
|
if body.startswith("static "):
|
|
body = body[len("static "):].strip()
|
|
if body.endswith(".*") or "." not in body:
|
|
return
|
|
simple = body.split(".")[-1]
|
|
if simple and simple[0].isupper():
|
|
imps[simple] = body
|
|
for child in n.children:
|
|
walk(child)
|
|
|
|
walk(tree.root_node)
|
|
for s in srcs:
|
|
pkg_by_file[s] = pkg
|
|
imports_by_file[s] = imps
|
|
|
|
# FQN (package.Class) -> definition node id, for type-like defs with a source.
|
|
fqn_to_id: dict[str, str] = {}
|
|
for node in all_nodes:
|
|
label = node.get("label", "")
|
|
src = node.get("source_file", "")
|
|
nid = node.get("id", "")
|
|
if not (label and src and nid) or src not in pkg_by_file:
|
|
continue
|
|
if not label[:1].isupper() or label.endswith(")") or label.endswith(".java"):
|
|
continue
|
|
pkg = pkg_by_file[src]
|
|
fqn_to_id.setdefault(f"{pkg}.{label}" if pkg else label, nid)
|
|
|
|
# Bare shadow stubs: no source_file, type-like label.
|
|
stub_label: dict[str, str] = {
|
|
node["id"]: node.get("label", "")
|
|
for node in all_nodes
|
|
if node.get("id") and not node.get("source_file") and node.get("label", "")[:1].isupper()
|
|
}
|
|
if not stub_label:
|
|
return
|
|
|
|
# `imports` is included so the file-level import edge that also lands on the
|
|
# shadow stub gets re-pointed too, leaving the stub unreferenced (and dropped).
|
|
# External/stdlib imports never resolve (no internal def / same-package match),
|
|
# so their edges correctly stay on their stub. `references` (field/parameter/
|
|
# return-type uses) is included so a cross-module reference to a same-named
|
|
# class doesn't dangle on a sourceless phantom node when two packages define
|
|
# the same simple name — the node itself survives with a path-scoped id, but
|
|
# the reference must point at the RIGHT one (#1744). Mirrors the C# resolver,
|
|
# whose REPOINT set already covers `references`.
|
|
REPOINT_RELATIONS = {"implements", "inherits", "extends", "imports", "references"}
|
|
repointed_from: set[str] = set()
|
|
for edge in all_edges:
|
|
if edge.get("relation") not in REPOINT_RELATIONS:
|
|
continue
|
|
tgt = edge.get("target")
|
|
label = stub_label.get(tgt)
|
|
if not label:
|
|
continue
|
|
ref_file = edge.get("source_file", "")
|
|
resolved = None
|
|
fqn = imports_by_file.get(ref_file, {}).get(label)
|
|
if fqn:
|
|
resolved = fqn_to_id.get(fqn)
|
|
if resolved is None: # same-package reference (no explicit import)
|
|
pkg = pkg_by_file.get(ref_file, "")
|
|
resolved = fqn_to_id.get(f"{pkg}.{label}" if pkg else label)
|
|
if resolved and resolved != tgt:
|
|
edge["target"] = resolved
|
|
repointed_from.add(tgt)
|
|
|
|
if not repointed_from:
|
|
return
|
|
|
|
# Drop shadow stubs that no edge references anymore.
|
|
still_referenced: set[str] = set()
|
|
for edge in all_edges:
|
|
still_referenced.add(edge.get("source"))
|
|
still_referenced.add(edge.get("target"))
|
|
all_nodes[:] = [
|
|
node for node in all_nodes
|
|
if node.get("id") not in repointed_from or node.get("id") in still_referenced
|
|
]
|
|
|
|
_pascal_unit_cache: dict[str, dict[str, str]] = {}
|
|
|
|
_pascal_class_stem_cache: dict[str, dict[str, str]] = {} # root_key → {stem_lower: _file_stem}
|
|
|
|
def _pascal_project_root(from_path: Path) -> Path:
|
|
"""Return the highest ancestor directory that looks like a Pascal project root.
|
|
|
|
Walks up the directory tree and tracks the topmost directory that:
|
|
- is NOT a filesystem root (e.g. D:/, C:/, /)
|
|
- has at least 2 .pas files OR at least 1 .dpr file as direct children
|
|
|
|
The minimum-2 threshold avoids treating a level as the root just because a
|
|
single stray .pas file was copied there. The filesystem-root exclusion
|
|
prevents overshoot on drives that have a stray file directly at D:/.
|
|
|
|
Falls back to from_path.parent if nothing better is found.
|
|
"""
|
|
best = from_path.parent
|
|
current = from_path.parent
|
|
for _ in range(12):
|
|
if len(current.parts) <= 1:
|
|
break # never use a filesystem root (D:/, C:/, /)
|
|
pas_count = sum(1 for _ in current.glob("*.pas"))
|
|
dpr_count = sum(1 for _ in current.glob("*.dpr"))
|
|
if pas_count >= 2 or dpr_count >= 1:
|
|
best = current
|
|
parent = current.parent
|
|
if parent == current:
|
|
break
|
|
current = parent
|
|
return best
|
|
|
|
def _pascal_resolve_unit(from_path: Path, unit_name: str) -> str:
|
|
"""Resolve a Pascal unit name to the graphify node ID of its source file.
|
|
|
|
Scans all Pascal files under the project root (the highest ancestor that
|
|
directly contains .pas/.dpr files) and returns _make_id(str(matched_path)).
|
|
Result is cached per project root so the rglob runs at most once per
|
|
project. Falls back to _make_id(unit_name) for units not found on disk
|
|
(e.g. standard RTL units like SysUtils, Windows).
|
|
"""
|
|
root = _pascal_project_root(from_path)
|
|
root_key = str(root)
|
|
if root_key not in _pascal_unit_cache:
|
|
unit_map: dict[str, str] = {}
|
|
for ext in (".pas", ".pp", ".dpr", ".dpk", ".inc"):
|
|
for f in root.rglob("*" + ext):
|
|
unit_map[f.stem.lower()] = _make_id(str(f))
|
|
_pascal_unit_cache[root_key] = unit_map
|
|
return _pascal_unit_cache[root_key].get(unit_name.lower(), _make_id(unit_name))
|
|
|
|
def _pascal_resolve_class(from_path: Path, class_name: str) -> str | None:
|
|
"""Resolve a Pascal class/interface name to the node ID of its defining file's class node.
|
|
|
|
Pascal convention: TFooBar is defined in FooBar.pas, IFooBar in FooBar.pas.
|
|
Strips the leading T/I prefix, finds the file, and returns
|
|
_make_id(_file_stem(found_file), class_name).
|
|
|
|
Returns None when no matching file is found on disk (RTL, stdlib, or
|
|
unconventionally-named class — caller should create a stub node).
|
|
"""
|
|
prefix = class_name[:1]
|
|
unit_name = class_name[1:] if prefix in ("T", "I") else class_name
|
|
|
|
root = _pascal_project_root(from_path)
|
|
root_key = str(root)
|
|
if root_key not in _pascal_class_stem_cache:
|
|
stem_map: dict[str, str] = {}
|
|
for ext in (".pas", ".pp", ".dpr", ".dpk"):
|
|
for f in root.rglob("*" + ext):
|
|
stem_map[f.stem.lower()] = _file_stem(f)
|
|
_pascal_class_stem_cache[root_key] = stem_map
|
|
|
|
file_stem = _pascal_class_stem_cache[root_key].get(unit_name.lower())
|
|
if file_stem:
|
|
return _make_id(file_stem, class_name)
|
|
return None
|