refactor(extract): migrate pascal, objc, and julia extractors to extractors/

Now that the resolution passes and the tree-sitter engine live in their own
modules, these three bespoke extractors are self-contained and move cleanly
(verbatim), pulling only the specific engine/resolution/base helpers they need:

- extractors/pascal.py: extract_pascal + regex helpers + _PAS_* constants
- extractors/objc.py: extract_objc + Objective-C member-call resolution
- extractors/julia.py: extract_julia

Re-exported from extract.py and registered in extractors/__init__.py (27 langs).
The remaining in-file extractors (js/ts config family + vue/svelte/astro/xaml)
stay because they share _JS_CONFIG/_TS_CONFIG and the xaml dispatch caches with
extract_js/extract_csharp and the dispatcher.

extract.py 5,947 -> 4,740 LOC (17,054 at branch start, -72%). Full suite
unchanged: 3036 passed, 29 skipped; skillgen --check OK.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
tpateeq
2026-07-09 02:03:52 +05:30
co-authored by Claude Opus 4.8
parent 5015212fc1
commit 563bb599fa
5 changed files with 1249 additions and 1213 deletions
+6 -1213
View File
File diff suppressed because it is too large Load Diff
+6
View File
@@ -19,7 +19,10 @@ from graphify.extractors.elixir import extract_elixir
from graphify.extractors.fortran import extract_fortran
from graphify.extractors.go import extract_go
from graphify.extractors.json_config import extract_json
from graphify.extractors.julia import extract_julia
from graphify.extractors.markdown import extract_markdown
from graphify.extractors.objc import extract_objc
from graphify.extractors.pascal import extract_pascal
from graphify.extractors.pascal_forms import extract_delphi_form, extract_lazarus_form
from graphify.extractors.powershell import extract_powershell, extract_powershell_manifest
from graphify.extractors.razor import extract_razor
@@ -44,8 +47,11 @@ LANGUAGE_EXTRACTORS: dict[str, Callable[[Path], dict]] = {
"fortran": extract_fortran,
"go": extract_go,
"json": extract_json,
"julia": extract_julia,
"lazarus_form": extract_lazarus_form,
"markdown": extract_markdown,
"objc": extract_objc,
"pascal": extract_pascal,
"powershell": extract_powershell,
"powershell_manifest": extract_powershell_manifest,
"razor": extract_razor,
+275
View File
@@ -0,0 +1,275 @@
"""julia — moved verbatim from graphify/extract.py."""
from __future__ import annotations
from graphify.extractors.base import _file_stem, _make_id, _read_text
from graphify.extractors.engine import _semantic_reference_edge
from pathlib import Path
def extract_julia(path: Path) -> dict:
"""Extract modules, structs, functions, imports, and calls from a .jl file."""
try:
import tree_sitter_julia as tsjulia
from tree_sitter import Language, Parser
except ImportError:
return {"nodes": [], "edges": [], "error": "tree-sitter-julia not installed"}
try:
language = Language(tsjulia.language())
parser = Parser(language)
source = path.read_bytes()
tree = parser.parse(source)
root = tree.root_node
except Exception as e:
return {"nodes": [], "edges": [], "error": str(e)}
stem = _file_stem(path)
str_path = str(path)
nodes: list[dict] = []
edges: list[dict] = []
seen_ids: set[str] = set()
function_bodies: list[tuple[str, object]] = []
def add_node(nid: str, label: str, line: int) -> None:
if nid not in seen_ids:
seen_ids.add(nid)
nodes.append({
"id": nid,
"label": label,
"file_type": "code",
"source_file": str_path,
"source_location": f"L{line}",
})
def add_edge(src: str, tgt: str, relation: str, line: int,
confidence: str = "EXTRACTED", weight: float = 1.0,
context: str | None = None) -> None:
edge = {
"source": src,
"target": tgt,
"relation": relation,
"confidence": confidence,
"source_file": str_path,
"source_location": f"L{line}",
"weight": weight,
}
if context:
edge["context"] = context
edges.append(edge)
file_nid = _make_id(str(path))
add_node(file_nid, path.name, 1)
def ensure_named_node(name: str, line: int) -> str:
nid = _make_id(stem, name)
if nid in seen_ids:
return nid
nid = _make_id(name)
if nid not in seen_ids:
# The name isn't defined in this file, so this is a cross-file reference
# (e.g. a `Thing` type annotation imported from another module). Emit a
# SOURCELESS stub — like the inheritance-base path below — so the
# corpus-level rewire can collapse it onto the real definition. A sourced
# stub here makes _disambiguate_colliding_node_ids bake the referencing
# file's path (with extension) into the id and blocks the rewire, which is
# the phantom-duplicate-node bug (#1402).
seen_ids.add(nid)
nodes.append({
"id": nid,
"label": name,
"file_type": "code",
"source_file": "",
"source_location": "",
"origin_file": str_path,
})
return nid
def _func_name_from_signature(sig_node) -> str | None:
"""Extract function name from a Julia signature node (call_expression > identifier)."""
for child in sig_node.children:
if child.type == "call_expression":
callee = child.children[0] if child.children else None
if callee and callee.type == "identifier":
return _read_text(callee, source)
return None
def walk_calls(body_node, func_nid: str) -> None:
if body_node is None:
return
t = body_node.type
if t in ("function_definition", "short_function_definition"):
return
if t == "call_expression" and body_node.children:
callee = body_node.children[0]
# Direct call: foo(...)
if callee.type == "identifier":
callee_name = _read_text(callee, source)
target_nid = _make_id(stem, callee_name)
add_edge(func_nid, target_nid, "calls", body_node.start_point[0] + 1,
confidence="EXTRACTED", context="call")
# Method call: obj.method(...)
elif callee.type == "field_expression" and len(callee.children) >= 3:
method_node = callee.children[-1]
method_name = _read_text(method_node, source)
target_nid = _make_id(stem, method_name)
add_edge(func_nid, target_nid, "calls", body_node.start_point[0] + 1,
confidence="EXTRACTED", context="call")
for child in body_node.children:
walk_calls(child, func_nid)
def walk(node, scope_nid: str) -> None:
t = node.type
# Module
if t == "module_definition":
name_node = next((c for c in node.children if c.type == "identifier"), None)
if name_node:
mod_name = _read_text(name_node, source)
mod_nid = _make_id(stem, mod_name)
line = node.start_point[0] + 1
add_node(mod_nid, mod_name, line)
add_edge(file_nid, mod_nid, "defines", line)
for child in node.children:
walk(child, mod_nid)
return
# Struct (struct / mutable struct — both map to struct_definition in tree-sitter-julia)
if t == "struct_definition":
# type_head may contain: identifier (simple) or binary_expression (Foo <: Bar)
type_head = next((c for c in node.children if c.type == "type_head"), None)
if not type_head:
return
struct_name: str | None = None
super_name: str | None = None
bin_expr = next((c for c in type_head.children if c.type == "binary_expression"), None)
if bin_expr:
identifiers = [c for c in bin_expr.children if c.type == "identifier"]
if identifiers:
struct_name = _read_text(identifiers[0], source)
if len(identifiers) >= 2:
super_name = _read_text(identifiers[-1], source)
else:
name_node = next((c for c in type_head.children if c.type == "identifier"), None)
if name_node:
struct_name = _read_text(name_node, source)
if not struct_name:
return
struct_nid = _make_id(stem, struct_name)
line = node.start_point[0] + 1
add_node(struct_nid, struct_name, line)
add_edge(scope_nid, struct_nid, "defines", line)
if super_name:
add_edge(struct_nid, ensure_named_node(super_name, line),
"inherits", line, confidence="EXTRACTED")
# Field types: each `name::Type` lowers to a typed_expression child of struct_definition
for child in node.children:
if child.type == "typed_expression":
type_ids = [c for c in child.children if c.type == "identifier"]
if len(type_ids) >= 2:
field_line = child.start_point[0] + 1
type_name = _read_text(type_ids[-1], source)
type_nid = ensure_named_node(type_name, field_line)
edges.append(_semantic_reference_edge(
struct_nid, type_nid, "field", str_path, field_line))
return
# Abstract type
if t == "abstract_definition":
# type_head > identifier
type_head = next((c for c in node.children if c.type == "type_head"), None)
if type_head:
name_node = next((c for c in type_head.children if c.type == "identifier"), None)
if name_node:
abs_name = _read_text(name_node, source)
abs_nid = _make_id(stem, abs_name)
line = node.start_point[0] + 1
add_node(abs_nid, abs_name, line)
add_edge(scope_nid, abs_nid, "defines", line)
return
# Function: function foo(...) ... end
if t == "function_definition":
sig_node = next((c for c in node.children if c.type == "signature"), None)
if sig_node:
func_name = _func_name_from_signature(sig_node)
if func_name:
func_nid = _make_id(stem, func_name)
line = node.start_point[0] + 1
add_node(func_nid, f"{func_name}()", line)
add_edge(scope_nid, func_nid, "defines", line)
function_bodies.append((func_nid, node))
return
# Short function: foo(x) = expr
if t == "assignment":
lhs = node.children[0] if node.children else None
if lhs and lhs.type == "call_expression" and lhs.children:
callee = lhs.children[0]
if callee.type == "identifier":
func_name = _read_text(callee, source)
func_nid = _make_id(stem, func_name)
line = node.start_point[0] + 1
add_node(func_nid, f"{func_name}()", line)
add_edge(scope_nid, func_nid, "defines", line)
# Only walk the RHS (index 2 after lhs and operator) to avoid self-loops
rhs = node.children[-1] if len(node.children) >= 3 else None
if rhs:
function_bodies.append((func_nid, rhs))
return
# Using / Import
if t in ("using_statement", "import_statement"):
line = node.start_point[0] + 1
def _julia_mod_name(n):
# identifier (`Foo`), scoped_identifier (`Base.Threads`), or
# import_path (relative `..Sibling`) -> the module name. Only bare
# identifiers were handled, so qualified/relative imports — and the
# scoped package of a `selected_import` — were silently dropped.
if n.type == "import_path":
ids = [c for c in n.children if c.type == "identifier"]
return _read_text(ids[-1], source) if ids else None
if n.type in ("identifier", "scoped_identifier"):
return _read_text(n, source)
return None
def _emit_import(name):
if not name:
return
imp_nid = _make_id(name)
add_node(imp_nid, name, line)
add_edge(scope_nid, imp_nid, "imports", line, context="import")
for child in node.children:
if child.type in ("identifier", "scoped_identifier", "import_path"):
_emit_import(_julia_mod_name(child))
elif child.type == "selected_import":
# `import Base.Threads: nthreads` — the package (first named
# child) may itself be a scoped_identifier/import_path.
pkg = next(
(c for c in child.children
if c.type in ("identifier", "scoped_identifier", "import_path")),
None,
)
if pkg is not None:
_emit_import(_julia_mod_name(pkg))
return
for child in node.children:
walk(child, scope_nid)
walk(root, file_nid)
for func_nid, body_node in function_bodies:
# For function_definition nodes, walk children directly to avoid
# the boundary check returning early on the top-level node itself.
# Skip the "signature" child — it contains the function's own call_expression
# which would create a self-loop.
if body_node.type == "function_definition":
for child in body_node.children:
if child.type != "signature":
walk_calls(child, func_nid)
else:
walk_calls(body_node, func_nid)
return {"nodes": nodes, "edges": edges}
+430
View File
@@ -0,0 +1,430 @@
"""objc — moved verbatim from graphify/extract.py."""
from __future__ import annotations
from graphify.extractors.base import _file_stem, _make_id, _read_text
from graphify.extractors.engine import _cpp_declarator_name, _semantic_reference_edge
from graphify.extractors.resolution import _resolve_c_include_path
from pathlib import Path
from typing import Any
def _objc_local_var_types(body_node, source: bytes, table: dict[str, str]) -> None:
"""Collect ``var -> ClassName`` from ObjC local declarations (``Foo *f = ...;``)
in a method body, for receiver typing in the cross-file message-send pass
(#1556). Only a capitalized ``type_identifier`` with a single named declarator
is recorded; a built-in/lower-cased type or an un-nameable declarator is skipped
(precision over recall). Reuses the C++ declarator unwrapper (identical grammar).
"""
stack = [body_node]
while stack:
n = stack.pop()
if n.type == "method_definition" and n is not body_node:
continue
if n.type == "declaration":
type_node = n.child_by_field_name("type")
if type_node is None:
for c in n.children:
if c.type == "type_identifier":
type_node = c
break
if type_node is not None and type_node.type == "type_identifier":
type_name = _read_text(type_node, source).strip()
declarators = [
c for c in n.children
if c.type in ("identifier", "pointer_declarator", "init_declarator")
]
if type_name and type_name[:1].isupper() and len(declarators) == 1:
var = _cpp_declarator_name(declarators[0], source)
if var and var not in table:
table[var] = type_name
for c in n.children:
stack.append(c)
def extract_objc(path: Path) -> dict:
"""Extract interfaces, implementations, protocols, methods, and imports from .m/.mm/.h files."""
try:
import tree_sitter_objc as tsobjc
from tree_sitter import Language, Parser
except ImportError:
return {"nodes": [], "edges": [], "error": "tree_sitter_objc not installed"}
try:
language = Language(tsobjc.language())
parser = Parser(language)
source = path.read_bytes()
# tree-sitter-objc cannot expand these argument-less annotation macros (no
# trailing ';'), and their presence before @interface makes the parser fail to
# emit a class_interface node (#1475). Blank them to equal-length spaces so byte
# offsets / line numbers are preserved and the interface parses.
_OBJC_BLANK_MACROS = (b"NS_ASSUME_NONNULL_BEGIN", b"NS_ASSUME_NONNULL_END")
for _m in _OBJC_BLANK_MACROS:
source = source.replace(_m, b" " * len(_m))
tree = parser.parse(source)
root = tree.root_node
except Exception as e:
return {"nodes": [], "edges": [], "error": str(e)}
stem = _file_stem(path)
str_path = str(path)
nodes: list[dict] = []
edges: list[dict] = []
seen_ids: set[str] = set()
method_bodies: list[tuple[str, Any, str]] = []
# #1556: unresolved message sends saved for the cross-file ObjC resolver, plus a
# per-file `var -> ClassName` table from `Foo *f = ...;` local declarations.
raw_calls: list[dict] = []
objc_type_table: dict[str, str] = {}
def add_node(nid: str, label: str, line: int) -> None:
if nid not in seen_ids:
seen_ids.add(nid)
nodes.append({"id": nid, "label": label, "file_type": "code",
"source_file": str_path, "source_location": f"L{line}"})
def add_edge(src: str, tgt: str, relation: str, line: int,
confidence: str = "EXTRACTED", weight: float = 1.0,
context: str | None = None) -> None:
edge = {"source": src, "target": tgt, "relation": relation,
"confidence": confidence, "source_file": str_path,
"source_location": f"L{line}", "weight": weight}
if context:
edge["context"] = context
edges.append(edge)
file_nid = _make_id(str(path))
add_node(file_nid, path.name, 1)
def _read(node) -> str:
return source[node.start_byte:node.end_byte].decode("utf-8", errors="replace")
def _get_name(node, field: str) -> str | None:
n = node.child_by_field_name(field)
return _read(n) if n else None
def _type_identifiers(node):
"""Yield every type_identifier under a property's type node, descending
through generic_specifier/type_name so NSArray<Product *> yields both
NSArray and the element type Product (the generic case was invisible
because the type was wrapped in a generic_specifier, not a bare
type_identifier child) (#1475)."""
if node.type == "type_identifier":
yield node
return
for c in node.children:
yield from _type_identifiers(c)
def ensure_named_node(name: str, line: int) -> str:
nid = _make_id(stem, name)
if nid in seen_ids:
return nid
nid = _make_id(name)
if nid not in seen_ids:
# The name isn't defined in this file, so this is a cross-file reference
# (e.g. a `Thing` type annotation imported from another module). Emit a
# SOURCELESS stub — like the inheritance-base path below — so the
# corpus-level rewire can collapse it onto the real definition. A sourced
# stub here makes _disambiguate_colliding_node_ids bake the referencing
# file's path (with extension) into the id and blocks the rewire, which is
# the phantom-duplicate-node bug (#1402).
seen_ids.add(nid)
nodes.append({
"id": nid,
"label": name,
"file_type": "code",
"source_file": "",
"source_location": "",
"origin_file": str_path,
})
return nid
def walk(node, parent_nid: str | None = None) -> None:
t = node.type
line = node.start_point[0] + 1
if t == "preproc_include":
# #import <Foundation/Foundation.h> or #import "MyClass.h"
for child in node.children:
if child.type == "system_lib_string":
raw = _read(child).strip("<>")
module = raw.split("/")[-1].replace(".h", "")
if module:
tgt_nid = _make_id(module)
add_edge(file_nid, tgt_nid, "imports", line, context="import")
elif child.type == "string_literal":
# recurse into string_literal to find string_content
for sub in child.children:
if sub.type == "string_content":
raw = _read(sub)
# Resolve the quoted include to a real file so the target id
# matches the (possibly disambiguated) node id _make_id gives
# that file; the bare-stem id never survives
# _disambiguate_colliding_node_ids when a .h/.m pair exists,
# so the edge dangled and was dropped (#1475).
resolved = _resolve_c_include_path(raw, str_path)
if resolved is not None:
add_edge(file_nid, _make_id(str(resolved)), "imports", line, context="import")
else:
module = raw.split("/")[-1].replace(".h", "")
if module:
add_edge(file_nid, _make_id(module), "imports", line, context="import")
return
if t == "module_import":
# @import Foundation; / @import Foundation.NSString;
path_node = node.child_by_field_name("path")
if path_node is not None:
module = _read(path_node).split(".")[0].strip()
if module:
add_edge(file_nid, _make_id(module), "imports", line, context="import")
return
if t == "class_interface":
# @interface ClassName : SuperClass <Protocols>
# children: @interface, identifier(name), ':', identifier(super), parameterized_arguments, ...
identifiers = [c for c in node.children if c.type == "identifier"]
if not identifiers:
for child in node.children:
walk(child, parent_nid)
return
name = _read(identifiers[0])
cls_nid = _make_id(stem, name)
add_node(cls_nid, name, line)
add_edge(file_nid, cls_nid, "contains", line)
# superclass is second identifier after ':'
colon_seen = False
for child in node.children:
if child.type == ":":
colon_seen = True
elif colon_seen and child.type == "identifier":
super_nid = ensure_named_node(_read(child), line)
add_edge(cls_nid, super_nid, "inherits", line)
colon_seen = False
elif child.type == "parameterized_arguments":
# protocols adopted: @interface Foo : Bar <Proto1, Proto2>
for sub in child.children:
if sub.type == "type_name":
for s in sub.children:
if s.type == "type_identifier":
proto_nid = ensure_named_node(_read(s), line)
add_edge(cls_nid, proto_nid, "implements", line)
elif child.type == "property_declaration":
prop_line = child.start_point[0] + 1
for sub in child.children:
if sub.type == "struct_declaration":
# The type is either a direct type_identifier
# (NSString *x) or wrapped in a generic_specifier
# (NSArray<Product *> *xs). Walk every type name in the
# type portion, skipping the declarator (the *field
# name), so generic collections are no longer invisible.
seen_types: set[str] = set()
for s in sub.children:
if s.type in ("struct_declarator", ";"):
continue
for ti in _type_identifiers(s):
tname = _read(ti)
if tname in seen_types:
continue
seen_types.add(tname)
type_nid = ensure_named_node(tname, prop_line)
edges.append(_semantic_reference_edge(
cls_nid, type_nid, "field", str_path, prop_line))
elif child.type == "method_declaration":
walk(child, cls_nid)
return
if t == "class_implementation":
# @implementation ClassName
name = None
for child in node.children:
if child.type == "identifier":
name = _read(child)
break
if not name:
for child in node.children:
walk(child, parent_nid)
return
impl_nid = _make_id(stem, name)
if impl_nid not in seen_ids:
add_node(impl_nid, name, line)
add_edge(file_nid, impl_nid, "contains", line)
for child in node.children:
if child.type == "implementation_definition":
for sub in child.children:
walk(sub, impl_nid)
return
if t == "protocol_declaration":
name = None
for child in node.children:
if child.type == "identifier":
name = _read(child)
break
if name:
proto_nid = _make_id(stem, name)
add_node(proto_nid, f"<{name}>", line)
add_edge(file_nid, proto_nid, "contains", line)
# Adopted protocols: `@protocol Derived <Base, Other>`. These
# nest under a protocol_reference_list node (distinct from the
# parameterized_arguments node used by @interface adoption), so
# they were never emitted. Emit an `implements` edge for each,
# matching how @interface protocol adoption is handled.
for child in node.children:
if child.type == "protocol_reference_list":
for sub in child.children:
if sub.type == "identifier":
base_nid = ensure_named_node(_read(sub), line)
if base_nid != proto_nid:
add_edge(proto_nid, base_nid, "implements", line)
for child in node.children:
walk(child, proto_nid)
return
if t in ("method_declaration", "method_definition"):
container = parent_nid or file_nid
# Class methods start with '+', instance methods with '-' (the grammar
# emits the sigil as the first child). The selector is the concatenation
# of the direct identifier children: one for a simple selector (-go),
# several for a compound one (-tableView:numberOfRowsInSection: ->
# "tableViewnumberOfRowsInSection"); method_parameter holds the arg
# types/names, not selector keywords, so it is correctly skipped.
prefix = "-"
for child in node.children:
if child.type in ("+", "-"):
prefix = child.type
break
parts = [_read(c) for c in node.children if c.type == "identifier"]
method_name = "".join(parts) if parts else None
if method_name:
method_nid = _make_id(container, method_name)
add_node(method_nid, f"{prefix}{method_name}", line)
add_edge(container, method_nid, "method", line)
if t == "method_definition":
method_bodies.append((method_nid, node, container))
return
for child in node.children:
walk(child, parent_nid)
walk(root)
# Second pass: resolve calls inside method bodies
all_method_nids = {n["id"] for n in nodes if n["id"] != file_nid}
class_method_nids: dict[str, set[str]] = {}
for m_nid, _, container_nid in method_bodies:
class_method_nids.setdefault(container_nid, set()).add(m_nid)
seen_calls: set[tuple[str, str]] = set()
# #1556: per-file `var -> ClassName` table from local declarations in every
# method body, so the cross-file resolver can type a `[f doThing]` receiver.
for _m_nid, body_node, _container in method_bodies:
_objc_local_var_types(body_node, source, objc_type_table)
for caller_nid, body_node, container_nid in method_bodies:
sibling_nids = class_method_nids.get(container_nid, set())
def walk_calls(n) -> None:
if n.type == "message_expression":
# `[[Foo alloc] init]` is a message_expression whose method is the
# identifier `alloc` and whose receiver is the bare class identifier
# `Foo`; resolve that class name and emit a `references` edge so the
# allocating method links to the allocated type. ensure_named_node
# emits a sourceless stub for unknown names, which the corpus rewire
# collapses ONLY when exactly one real class of that name exists, so an
# unknown/ambiguous class produces no false resolved edge (#1475).
meth = n.child_by_field_name("method")
recv = n.child_by_field_name("receiver")
if (meth is not None and meth.type == "identifier" and _read(meth) == "alloc"
and recv is not None and recv.type == "identifier"):
tname = _read(recv)
ref_line = n.start_point[0] + 1
type_nid = ensure_named_node(tname, ref_line)
if type_nid != caller_nid:
edges.append(_semantic_reference_edge(
caller_nid, type_nid, "type", str_path, ref_line))
# [receiver sel] and [receiver kw1:a kw2:b] both parse to a
# message_expression whose selector parts carry the field name
# "method" (one for a simple selector, several for a compound one);
# the receiver carries field name "receiver". Reconstruct the
# selector from every "method" child so self/super/ClassName
# receivers are never mistaken for a selector, and compound sends
# resolve too (the whole second pass was previously dead code for
# ObjC because the grammar emits these as `identifier`, not
# `selector`/`keyword_argument_list`) (#1475).
sel_parts = [
_read(child)
for i, child in enumerate(n.children)
if n.field_name_for_child(i) == "method" and child.type == "identifier"
]
method_name = "".join(sel_parts)
if method_name:
needle = _make_id("", method_name).lstrip("_")
for candidate in all_method_nids:
if candidate.endswith(needle):
pair = (caller_nid, candidate)
if pair not in seen_calls and caller_nid != candidate:
seen_calls.add(pair)
add_edge(caller_nid, candidate, "calls", n.start_point[0] + 1,
confidence="EXTRACTED", weight=1.0, context="call")
# #1556: also emit a raw_call so the cross-file resolver can type
# the receiver and link to a method in ANOTHER file. A bare
# identifier receiver (`f`, `self`, `Foo`) is captured; a nested
# message send (`[[Foo alloc] init]`) has no simple receiver name
# to type, so it is left to the alloc/init `references` edge above.
if recv is not None and recv.type == "identifier":
raw_calls.append({
"caller_nid": caller_nid,
"callee": method_name,
"is_member_call": True,
"source_file": str_path,
"source_location": f"L{n.start_point[0] + 1}",
"receiver": _read(recv),
"lang": "objc",
})
elif n.type == "field_expression":
# self.name / self.product.name — dot-syntax sugar for [self name].
# Resolve to a sibling method of the SAME class, matched by EXACT
# node id (a method id is _make_id(container, name)). A suffix
# substring match would mis-resolve self.name -> -surname and would
# let a substring-colliding sibling (-surname) suppress the real
# -name edge, so it must be an exact match (#1475).
for child in n.children:
if child.type == "field_identifier":
field_name = _read(child)
target = _make_id(container_nid, field_name)
if target in sibling_nids and target != caller_nid:
pair = (caller_nid, target)
if pair not in seen_calls:
seen_calls.add(pair)
add_edge(caller_nid, target, "accesses",
n.start_point[0] + 1,
confidence="EXTRACTED", weight=1.0)
elif n.type == "selector_expression":
# @selector(doSomething:withParam:) — compile-time method ref.
# Match the selector name EXACTLY (a method id is
# _make_id(container, name)) against every class's methods, and emit
# only when exactly one method matches, to avoid ambiguous fan-out.
# Exact match (not a suffix) keeps -doThing distinct from
# -reallyDoThing (#1475).
sel_parts = [_read(c) for c in n.children if c.type == "identifier"]
sel_name = "".join(sel_parts)
if sel_name:
matches = sorted({
m for m, _, cont in method_bodies
if m == _make_id(cont, sel_name) and m != caller_nid
})
if len(matches) == 1:
pair = (caller_nid, matches[0])
if pair not in seen_calls:
seen_calls.add(pair)
add_edge(caller_nid, matches[0], "calls",
n.start_point[0] + 1,
confidence="EXTRACTED", weight=1.0,
context="call")
for child in n.children:
walk_calls(child)
walk_calls(body_node)
result = {"nodes": nodes, "edges": edges, "raw_calls": raw_calls,
"input_tokens": 0, "output_tokens": 0}
if objc_type_table:
result["objc_type_table"] = {"path": str_path, "table": objc_type_table}
return result
+532
View File
@@ -0,0 +1,532 @@
"""pascal — moved verbatim from graphify/extract.py."""
from __future__ import annotations
import re
from graphify.extractors.base import _file_stem, _make_id
from graphify.extractors.resolution import _pascal_resolve_class, _pascal_resolve_unit
from pathlib import Path
from typing import Any
_PAS_TOKEN_RE = re.compile(
r"'(?:''|[^'])*'"
r"|\{[^}]*\}"
r"|\(\*.*?\*\)"
r"|//[^\n]*",
re.DOTALL,
)
_PAS_MODULE_RE = re.compile(
r"\b(unit|program|library)\s+([A-Za-z_][\w.]*)\s*;",
re.IGNORECASE,
)
_PAS_USES_RE = re.compile(
r"\buses\b\s*([^;]+);",
re.IGNORECASE | re.DOTALL,
)
_PAS_TYPE_HEADER_RE = re.compile(
r"\b(?P<name>[A-Za-z_]\w*)(?:\s*<[^>]+>)?\s*=\s*(?:packed\s+)?"
r"(?P<kind>class|interface)\b"
r"(?:\s*\(\s*(?P<bases>[^)]*)\s*\))?",
re.IGNORECASE,
)
_PAS_END_SEMI_RE = re.compile(r"\bend\s*;", re.IGNORECASE)
_PAS_METHOD_DECL_RE = re.compile(
r"\b(?:procedure|function|constructor|destructor)\s+"
r"(?P<name>[A-Za-z_]\w*)"
r"(?:\s*\([^)]*\))?"
r"(?:\s*:\s*[\w<>,\s.]+)?"
r"\s*;",
re.IGNORECASE,
)
_PAS_IMPL_HEADER_RE = re.compile(
r"\b(?:procedure|function|constructor|destructor)\s+"
r"(?P<qual>[A-Za-z_]\w*(?:\.[A-Za-z_]\w*)?)"
r"(?:\s*<[^>]+>)?"
r"(?:\s*\([^)]*\))?"
r"(?:\s*:\s*[\w<>,\s.]+)?"
r"\s*;",
re.IGNORECASE,
)
_PAS_BEGIN_END_TOKEN_RE = re.compile(
r"\b(begin|end|case|try|asm|record)\b", re.IGNORECASE
)
_PAS_CALL_RE = re.compile(r"\b([A-Za-z_]\w*(?:\.[A-Za-z_]\w*)*)\s*[(;]")
_PAS_KEYWORDS = frozenset({
"begin", "end", "if", "then", "else", "while", "do", "for", "to",
"downto", "repeat", "until", "case", "of", "try", "finally", "except",
"with", "inherited", "result", "var", "const", "type", "nil", "true",
"false", "exit", "break", "continue", "uses", "unit", "program",
"library", "interface", "implementation", "initialization", "finalization",
"procedure", "function", "constructor", "destructor", "class", "record",
"object", "array", "string", "integer", "boolean", "real", "char",
"writeln", "write", "readln", "read", "assigned", "length", "high",
"low", "inc", "dec", "new", "dispose", "setlength", "copy", "pos",
"trim", "format", "inttostr", "strtoint", "ord", "chr", "sizeof",
"create", "free", "destroy",
})
def _pascal_strip_comments(text: str) -> str:
"""Strip Pascal comments ({}, (* *), //) while preserving newlines."""
def _sub(m: re.Match) -> str:
tok = m.group(0)
if tok.startswith("'"):
return tok
return "".join(c if c == "\n" else " " for c in tok)
return _PAS_TOKEN_RE.sub(_sub, text)
def _pascal_split_sections(text: str) -> tuple[str, int, str, int]:
"""Split into (iface_text, iface_offset, impl_text, impl_offset).
Files without interface/implementation sections (dpr/lpr/inc) return
the whole text as impl with offset 0.
"""
iface_m = re.search(r"\binterface\b", text, re.IGNORECASE)
impl_m = re.search(r"\bimplementation\b", text, re.IGNORECASE)
if iface_m and impl_m:
iface_off = iface_m.end()
impl_off = impl_m.end()
end_m = re.search(
r"\b(initialization|finalization)\b", text[impl_off:], re.IGNORECASE
)
impl_end = impl_off + end_m.start() if end_m else len(text)
return text[iface_off:impl_m.start()], iface_off, text[impl_off:impl_end], impl_off
return "", 0, text, 0
def _pascal_split_uses(s: str) -> list[str]:
"""Split a uses list string, handling 'Foo in ''bar.pas''' syntax."""
out = []
for chunk in s.split(","):
name = re.split(r"\s+in\s+", chunk.strip(), maxsplit=1, flags=re.IGNORECASE)[0]
name = name.strip().strip(";")
if name and re.match(r"[A-Za-z_][\w.]*$", name):
out.append(name)
return out
def _pascal_split_bases(s: str) -> list[str]:
"""Split inheritance list, handling generics like TList<T, U>."""
out, depth, buf = [], 0, []
for ch in s:
if ch == "<":
depth += 1
buf.append(ch)
elif ch == ">":
depth -= 1
buf.append(ch)
elif ch == "," and depth == 0:
name = re.sub(r"<.*$", "", "".join(buf).strip())
if name:
out.append(name)
buf = []
else:
buf.append(ch)
name = re.sub(r"<.*$", "", "".join(buf).strip())
if name:
out.append(name)
return [n for n in out if re.match(r"[A-Za-z_]\w*$", n)]
def _pascal_find_body(text: str, start: int) -> tuple[int, int]:
"""Find balanced begin..end after start. Returns (body_start, body_end).
Returns (0, 0) if no begin found.
"""
m = re.search(r"\bbegin\b", text[start:], re.IGNORECASE)
if not m:
return (0, 0)
body_start = start + m.end()
depth = 1
for tok in _PAS_BEGIN_END_TOKEN_RE.finditer(text, body_start):
kw = tok.group(1).lower()
if kw in ("begin", "case", "try", "asm", "record"):
depth += 1
elif kw == "end":
depth -= 1
if depth == 0:
return (body_start, tok.start())
return (body_start, len(text))
def _extract_pascal_regex(path: Path) -> dict:
"""Regex fallback for Pascal/Delphi extraction when tree-sitter-pascal
is unavailable. Produces the same node/edge schema as the tree-sitter pass.
"""
try:
raw = path.read_text(encoding="utf-8", errors="replace")
except Exception as exc:
return {"nodes": [], "edges": [], "error": str(exc)}
str_path = str(path)
stem = _file_stem(path)
nodes: list[dict] = []
edges: list[dict] = []
seen_ids: set[str] = set()
seen_call_pairs: set[tuple[str, str]] = set()
def _add_node(nid: str, label: str, line: int) -> None:
if nid not in seen_ids:
seen_ids.add(nid)
nodes.append({
"id": nid,
"label": label,
"file_type": "code",
"source_file": str_path,
"source_location": f"L{line}",
})
def _add_edge(src: str, tgt: str, relation: str, line: int, context: str | None = None) -> None:
edge: dict = {
"source": src,
"target": tgt,
"relation": relation,
"confidence": "EXTRACTED",
"source_file": str_path,
"source_location": f"L{line}",
"weight": 1.0,
}
if context:
edge["context"] = context
edges.append(edge)
def _lineno(text: str, offset: int) -> int:
return text.count("\n", 0, offset) + 1
file_nid = _make_id(str_path)
_add_node(file_nid, path.name, 1)
stripped = _pascal_strip_comments(raw)
# Module header
module_nid = file_nid
mod_m = _PAS_MODULE_RE.search(stripped)
if mod_m:
mod_name = mod_m.group(2)
module_nid = _make_id(stem, mod_name)
_add_node(module_nid, mod_name, _lineno(stripped, mod_m.start()))
_add_edge(file_nid, module_nid, "contains", _lineno(stripped, mod_m.start()))
iface_text, iface_off, impl_text, impl_off = _pascal_split_sections(stripped)
# Uses clauses
for section_text, section_off in ((iface_text, iface_off), (impl_text, impl_off)):
for um in _PAS_USES_RE.finditer(section_text):
line = _lineno(stripped, section_off + um.start())
for unit_name in _pascal_split_uses(um.group(1)):
tgt_nid = _pascal_resolve_unit(path, unit_name)
_add_edge(module_nid, tgt_nid, "imports", line, context="import")
# Type declarations (classes / interfaces) in interface section
search_text = iface_text if iface_text else stripped
search_off = iface_off if iface_text else 0
pos = 0
while pos < len(search_text):
hm = _PAS_TYPE_HEADER_RE.search(search_text, pos)
if not hm:
break
type_name = hm.group("name")
bases_raw = hm.group("bases") or ""
line = _lineno(stripped, search_off + hm.start())
cls_nid = _make_id(stem, type_name)
_add_node(cls_nid, type_name, line)
_add_edge(module_nid, cls_nid, "contains", line)
for base_name in _pascal_split_bases(bases_raw):
resolved = _pascal_resolve_class(path, base_name)
base_nid = resolved if resolved else _make_id(base_name)
if base_nid not in seen_ids:
_add_node(base_nid, base_name, line)
_add_edge(cls_nid, base_nid, "inherits", line)
# Find class body (up to next end;)
end_m = _PAS_END_SEMI_RE.search(search_text, hm.end())
body_text = search_text[hm.end():end_m.start()] if end_m else ""
body_off = search_off + hm.end()
# Forward method declarations inside the class body
for mm in _PAS_METHOD_DECL_RE.finditer(body_text):
mname = mm.group("name")
mline = _lineno(stripped, body_off + mm.start())
method_nid = _make_id(cls_nid, mname)
_add_node(method_nid, f"{mname}()", mline)
_add_edge(cls_nid, method_nid, "method", mline)
pos = end_m.end() if end_m else len(search_text)
# Implementation headers (procedure/function/constructor/destructor)
impl_records: list[tuple[str, int, str]] = []
for fm in _PAS_IMPL_HEADER_RE.finditer(impl_text):
qualified = fm.group("qual")
line = _lineno(stripped, impl_off + fm.start())
if "." in qualified:
cls_part, method_part = qualified.split(".", 1)
cls_nid = _make_id(stem, cls_part)
container = cls_nid if cls_nid in seen_ids else module_nid
relation = "method" if cls_nid in seen_ids else "contains"
label = f"{method_part}()"
else:
container, relation = module_nid, "contains"
label = f"{qualified}()"
proc_nid = _make_id(stem, qualified)
_add_node(proc_nid, label, line)
_add_edge(container, proc_nid, relation, line)
body_start, body_end = _pascal_find_body(impl_text, fm.end())
body_text = impl_text[body_start:body_end] if body_start else ""
impl_records.append((proc_nid, line, body_text))
# Intra-file call edges
all_procs: dict[str, str] = {
n["label"].removesuffix("()").lower(): n["id"]
for n in nodes
if n["id"] != file_nid and n["label"].endswith("()")
}
for caller_nid, caller_line, body_text in impl_records:
for cm in _PAS_CALL_RE.finditer(body_text):
callee_name = cm.group(1).split(".")[-1].lower()
if callee_name in _PAS_KEYWORDS:
continue
callee_nid = all_procs.get(callee_name)
if not callee_nid or callee_nid == caller_nid:
continue
pair = (caller_nid, callee_nid)
if pair in seen_call_pairs:
continue
seen_call_pairs.add(pair)
call_line = caller_line + body_text.count("\n", 0, cm.start())
_add_edge(caller_nid, callee_nid, "calls", call_line, context="call")
return {"nodes": nodes, "edges": edges, "input_tokens": 0, "output_tokens": 0}
def extract_pascal(path: Path) -> dict:
"""Extract units, classes, procedures, uses-imports, and calls from Pascal/Delphi files.
Produces nodes for:
- The file itself
- unit / program / library declarations
- class and interface type declarations
- procedure / function implementations (including qualified TClass.Method names)
Produces edges for:
- file --contains--> module
- module --imports--> other file node (via uses clause, resolved to path-based IDs)
- class --inherits--> base class
- class/module --contains--> method forward declaration
- class/module --contains--> procedure/function implementation
- procedure --calls--> other procedure (within the same file)
Uses tree-sitter-pascal when available; falls back to a regex-based extractor
(_extract_pascal_regex) when it isn't installed or fails to parse, so Pascal
extraction works out of the box without an extra pip install.
"""
try:
import tree_sitter_pascal as tspascal
from tree_sitter import Language, Parser
except ImportError:
return _extract_pascal_regex(path)
try:
language = Language(tspascal.language())
parser = Parser(language)
source = path.read_bytes()
tree = parser.parse(source)
root = tree.root_node
except Exception:
return _extract_pascal_regex(path)
stem = _file_stem(path)
str_path = str(path)
nodes: list[dict] = []
edges: list[dict] = []
seen_ids: set[str] = set()
proc_bodies: list[tuple[str, Any]] = []
def _read(node) -> str: # type: ignore[no-untyped-def]
return source[node.start_byte:node.end_byte].decode("utf-8", errors="replace")
def add_node(nid: str, label: str, line: int) -> None:
if nid not in seen_ids:
seen_ids.add(nid)
nodes.append({
"id": nid, "label": label, "file_type": "code",
"source_file": str_path, "source_location": f"L{line}",
})
def add_edge(
src: str, tgt: str, relation: str, line: int,
confidence: str = "EXTRACTED", weight: float = 1.0,
context: str | None = None,
) -> None:
edge: dict[str, Any] = {
"source": src, "target": tgt, "relation": relation,
"confidence": confidence, "source_file": str_path,
"source_location": f"L{line}", "weight": weight,
}
if context:
edge["context"] = context
edges.append(edge)
file_nid = _make_id(str(path))
add_node(file_nid, path.name, 1)
module_nid = file_nid
def _proc_name(header_node) -> str | None: # type: ignore[no-untyped-def]
name_node = header_node.child_by_field_name("name")
if name_node:
return _read(name_node)
for child in header_node.children:
if child.type in ("identifier", "genericDot", "genericTpl"):
return _read(child)
return None
def walk(node, parent_nid: str) -> None: # type: ignore[no-untyped-def]
nonlocal module_nid
t = node.type
line = node.start_point[0] + 1
if t in ("unit", "program", "library"):
name_node = next((c for c in node.children if c.type == "moduleName"), None)
mod_name = _read(name_node) if name_node else path.stem
mod_nid = _make_id(stem, mod_name)
add_node(mod_nid, mod_name, line)
add_edge(file_nid, mod_nid, "contains", line)
module_nid = mod_nid
for child in node.children:
walk(child, mod_nid)
return
if t == "declUses":
for child in node.children:
if child.type == "moduleName":
mod_name = _read(child)
tgt_nid = _pascal_resolve_unit(path, mod_name)
add_edge(parent_nid, tgt_nid, "imports", line, context="import")
return
if t == "declType":
type_name = None
kind_node = None
for child in node.children:
if child.type == "identifier" and type_name is None:
type_name = _read(child)
elif child.type in ("declClass", "declIntf", "declHelper") and kind_node is None:
kind_node = child
if type_name and kind_node:
cls_nid = _make_id(stem, type_name)
add_node(cls_nid, type_name, line)
add_edge(parent_nid, cls_nid, "contains", line)
for child in kind_node.children:
if child.type == "typeref":
base_name = _read(child)
base_nid = _make_id(stem, base_name)
if base_nid not in seen_ids:
# Try cross-file resolution (TFooBar → FooBar.pas)
resolved = _pascal_resolve_class(path, base_name)
base_nid = resolved if resolved else _make_id(base_name)
if base_nid not in seen_ids:
# Stub for RTL/external/cross-file base classes
add_node(base_nid, base_name, line)
add_edge(cls_nid, base_nid, "inherits", line)
for child in kind_node.children:
walk(child, cls_nid)
return
for child in node.children:
walk(child, parent_nid)
return
if t == "declProcFwd":
header = next((c for c in node.children if c.type == "declProc"), None)
if header:
name = _proc_name(header)
if name and "." not in name:
method_nid = _make_id(parent_nid, name)
add_node(method_nid, f"{name}()", line)
add_edge(parent_nid, method_nid, "method", line)
return
if t == "defProc":
header = next((c for c in node.children if c.type == "declProc"), None)
body_node = next((c for c in node.children if c.type == "block"), None)
if not header:
for child in node.children:
walk(child, parent_nid)
return
name = _proc_name(header)
if not name:
for child in node.children:
walk(child, parent_nid)
return
container = parent_nid
if "." in name:
parts = name.split(".", 1)
cls_nid = _make_id(stem, parts[0])
if cls_nid in seen_ids:
container = cls_nid
label = f"{parts[-1]}()"
else:
label = f"{name}()"
proc_nid = _make_id(stem, name)
add_node(proc_nid, label, line)
add_edge(
container, proc_nid,
"method" if container != parent_nid else "contains",
line,
)
if body_node:
proc_bodies.append((proc_nid, body_node))
return
for child in node.children:
walk(child, parent_nid)
walk(root, file_nid)
# Second pass: resolve calls inside procedure/function bodies
all_procs: dict[str, str] = {
n["label"].removesuffix("()").lower(): n["id"]
for n in nodes if n["id"] != file_nid
}
seen_call_pairs: set[tuple[str, str]] = set()
def walk_calls(node, caller_nid: str) -> None: # type: ignore[no-untyped-def]
if node.type == "exprCall":
callee_text = None
for child in node.children:
if child.is_named and child.type not in ("exprArgs",):
callee_text = _read(child).split(".")[-1]
break
if callee_text:
callee_nid = all_procs.get(callee_text.lower())
if callee_nid and callee_nid != caller_nid:
pair = (caller_nid, callee_nid)
if pair not in seen_call_pairs:
seen_call_pairs.add(pair)
add_edge(
caller_nid, callee_nid, "calls",
node.start_point[0] + 1, context="call",
)
elif node.type == "statement":
# Pascal bare procedure calls with no args: `Reset;`
# tree-sitter represents these as statement → identifier (no exprCall wrapper)
named = [c for c in node.children if c.is_named]
if len(named) == 1 and named[0].type == "identifier":
callee_text = _read(named[0])
callee_nid = all_procs.get(callee_text.lower())
if callee_nid and callee_nid != caller_nid:
pair = (caller_nid, callee_nid)
if pair not in seen_call_pairs:
seen_call_pairs.add(pair)
add_edge(
caller_nid, callee_nid, "calls",
node.start_point[0] + 1, context="call",
)
for child in node.children:
walk_calls(child, caller_nid)
for proc_nid, body_node in proc_bodies:
walk_calls(body_node, proc_nid)
return {"nodes": nodes, "edges": edges, "input_tokens": 0, "output_tokens": 0}