"""Build a stable node ID from one or more name parts."""

combined = "_".join(p.strip("_.") for p in parts if p)

cleaned = re.sub(r"[^a-zA-Z0-9]+", "_", combined)

"""Extract classes, functions, and imports from a .py file via tree-sitter AST."""

try:

import tree_sitter_python as tspython

from tree_sitter import Language, Parser

except ImportError:

return {"nodes": [], "edges": [], "error": "tree-sitter-python not installed"}

try:

language = Language(tspython.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 = path.stem

str_path = str(path)

nodes: list[dict] = []

edges: list[dict] = []

seen_ids: set[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) -> None:

# Only add edge if both endpoints exist or src is the file node

edges.append({

"source": src,

"target": tgt,

"relation": relation,

"confidence": "EXTRACTED",

"source_file": str_path,

"source_location": f"L{line}",

"weight": 1.0,

})

# File-level node - stable ID based on stem only

file_nid = _make_id(stem)

add_node(file_nid, path.name, 1)

def walk(node, parent_class_nid: str | None = None) -> None:

t = node.type

if t == "import_statement":

for child in node.children:

if child.type in ("dotted_name", "aliased_import"):

raw = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")

module_name = raw.split(" as ")[0].strip().lstrip(".")

tgt_nid = _make_id(module_name)

add_edge(file_nid, tgt_nid, "imports", node.start_point[0] + 1)

return

if t == "import_from_statement":

module_node = node.child_by_field_name("module_name")

if module_node:

raw = source[module_node.start_byte:module_node.end_byte].decode("utf-8", errors="replace").lstrip(".")

tgt_nid = _make_id(raw)

add_edge(file_nid, tgt_nid, "imports_from", node.start_point[0] + 1)

return

if t == "class_definition":

name_node = node.child_by_field_name("name")

if not name_node:

return

class_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

class_nid = _make_id(stem, class_name)

line = node.start_point[0] + 1

add_node(class_nid, class_name, line)

add_edge(file_nid, class_nid, "contains", line)

# Inheritance - create stub node for external bases so the edge is never dropped

args = node.child_by_field_name("superclasses")

if args:

for arg in args.children:

if arg.type == "identifier":

base = source[arg.start_byte:arg.end_byte].decode("utf-8", errors="replace")

# Try same-file base first; fall back to a bare stub

base_nid = _make_id(stem, base)

if base_nid not in seen_ids:

# External or forward-declared base - add a stub so edge survives

base_nid = _make_id(base)

if base_nid not in seen_ids:

nodes.append({

"id": base_nid,

"label": base,

"file_type": "code",

"source_file": "",

"source_location": "",

})

seen_ids.add(base_nid)

add_edge(class_nid, base_nid, "inherits", line)

# Walk class body for methods

body = node.child_by_field_name("body")

if body:

for child in body.children:

walk(child, parent_class_nid=class_nid)

return

if t == "function_definition":

name_node = node.child_by_field_name("name")

if not name_node:

return

func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = node.start_point[0] + 1

if parent_class_nid:

func_nid = _make_id(parent_class_nid, func_name)

add_node(func_nid, f".{func_name}()", line)

add_edge(parent_class_nid, func_nid, "method", line)

else:

func_nid = _make_id(stem, func_name)

add_node(func_nid, f"{func_name}()", line)

add_edge(file_nid, func_nid, "contains", line)

# Collect body for the call-graph pass below

body = node.child_by_field_name("body")

if body:

function_bodies.append((func_nid, body))

return

for child in node.children:

walk(child, parent_class_nid=None)

function_bodies: list[tuple[str, object]] = []

walk(root)

# ── Call-graph pass ───────────────────────────────────────────────────────

# Build label→nid lookup from all nodes collected above.

# Normalise: strip "()" suffix and leading "." so "cohesion_score()" and

# ".cohesion_score()" both map to the same entry.

label_to_nid: dict[str, str] = {}

for n in nodes:

raw = n["label"]

normalised = raw.strip("()").lstrip(".")

label_to_nid[normalised.lower()] = n["id"]

seen_call_pairs: set[tuple[str, str]] = set()

def walk_calls(node, caller_nid: str) -> None:

# Don't recurse into nested function definitions - they have their own context.

if node.type == "function_definition":

return

if node.type == "call":

func_node = node.child_by_field_name("function")

callee_name: str | None = None

if func_node:

if func_node.type == "identifier":

callee_name = source[func_node.start_byte:func_node.end_byte].decode("utf-8", errors="replace")

elif func_node.type == "attribute":

attr = func_node.child_by_field_name("attribute")

if attr:

callee_name = source[attr.start_byte:attr.end_byte].decode("utf-8", errors="replace")

if callee_name:

tgt_nid = label_to_nid.get(callee_name.lower())

if tgt_nid and tgt_nid != caller_nid:

pair = (caller_nid, tgt_nid)

if pair not in seen_call_pairs:

seen_call_pairs.add(pair)

line = node.start_point[0] + 1

edges.append({

"source": caller_nid,

"target": tgt_nid,

"relation": "calls",

"confidence": "INFERRED",

"source_file": str_path,

"source_location": f"L{line}",

"weight": 0.8,

})

for child in node.children:

walk_calls(child, caller_nid)

for caller_nid, body_node in function_bodies:

walk_calls(body_node, caller_nid)

# ─────────────────────────────────────────────────────────────────────────

# Post-process: remove edges whose source or target was never added as a node

# (dangling import edges pointing to external libraries are fine to keep,

# but edges between internal entities must be valid)

valid_ids = seen_ids

clean_edges = []

for edge in edges:

src, tgt = edge["source"], edge["target"]

# Keep if both endpoints are known, OR if it's an import edge (tgt may be external)

if src in valid_ids and (tgt in valid_ids or edge["relation"] in ("imports", "imports_from")):

clean_edges.append(edge)

"""Extract classes, functions, arrow functions, and imports from a .js/.ts/.tsx file."""

try:

if path.suffix in (".ts", ".tsx"):

import tree_sitter_typescript as tslang

from tree_sitter import Language, Parser

language = Language(tslang.language_typescript())

else:

import tree_sitter_javascript as tslang

from tree_sitter import Language, Parser

language = Language(tslang.language())

except ImportError:

return {"nodes": [], "edges": [], "error": "tree-sitter-javascript/typescript not installed"}

try:

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 = path.stem

str_path = str(path)

nodes: list[dict] = []

edges: list[dict] = []

seen_ids: set[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, confidence: str = "EXTRACTED", weight: float = 1.0) -> None:

edges.append({

"source": src,

"target": tgt,

"relation": relation,

"confidence": confidence,

"source_file": str_path,

"source_location": f"L{line}",

"weight": weight,

})

file_nid = _make_id(stem)

add_node(file_nid, path.name, 1)

function_bodies: list[tuple[str, object]] = []

def walk(node, parent_class_nid: str | None = None) -> None:

t = node.type

if t == "import_statement":

for child in node.children:

if child.type == "string":

raw = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace").strip("'\"` ")

module_name = raw.lstrip("./").split("/")[-1]

if module_name:

tgt_nid = _make_id(module_name)

add_edge(file_nid, tgt_nid, "imports_from", node.start_point[0] + 1)

return

if t == "class_declaration":

name_node = node.child_by_field_name("name")

if not name_node:

return

class_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

class_nid = _make_id(stem, class_name)

line = node.start_point[0] + 1

add_node(class_nid, class_name, line)

add_edge(file_nid, class_nid, "contains", line)

body = node.child_by_field_name("body")

if body:

for child in body.children:

walk(child, parent_class_nid=class_nid)

return

if t == "function_declaration":

name_node = node.child_by_field_name("name")

if not name_node:

return

func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = node.start_point[0] + 1

func_nid = _make_id(stem, func_name)

add_node(func_nid, f"{func_name}()", line)

add_edge(file_nid, func_nid, "contains", line)

body = node.child_by_field_name("body")

if body:

function_bodies.append((func_nid, body))

return

if t == "method_definition" and parent_class_nid:

name_node = node.child_by_field_name("name")

if not name_node:

return

method_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = node.start_point[0] + 1

method_nid = _make_id(parent_class_nid, method_name)

add_node(method_nid, f".{method_name}()", line)

add_edge(parent_class_nid, method_nid, "method", line)

body = node.child_by_field_name("body")

if body:

function_bodies.append((method_nid, body))

return

if t == "lexical_declaration":

# Arrow functions: const foo = (...) => { ... }

for child in node.children:

if child.type == "variable_declarator":

value = child.child_by_field_name("value")

if value and value.type == "arrow_function":

name_node = child.child_by_field_name("name")

if name_node:

func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = child.start_point[0] + 1

func_nid = _make_id(stem, func_name)

add_node(func_nid, f"{func_name}()", line)

add_edge(file_nid, func_nid, "contains", line)

body = value.child_by_field_name("body")

if body:

function_bodies.append((func_nid, body))

return

for child in node.children:

walk(child, parent_class_nid=None)

walk(root)

label_to_nid: dict[str, str] = {}

for n in nodes:

raw = n["label"]

normalised = raw.strip("()").lstrip(".")

label_to_nid[normalised.lower()] = n["id"]

seen_call_pairs: set[tuple[str, str]] = set()

def walk_calls(node, caller_nid: str) -> None:

if node.type in ("function_declaration", "arrow_function", "method_definition"):

return

if node.type == "call_expression":

func_node = node.child_by_field_name("function")

callee_name: str | None = None

if func_node:

if func_node.type == "identifier":

callee_name = source[func_node.start_byte:func_node.end_byte].decode("utf-8", errors="replace")

elif func_node.type == "member_expression":

prop = func_node.child_by_field_name("property")

if prop:

callee_name = source[prop.start_byte:prop.end_byte].decode("utf-8", errors="replace")

if callee_name:

tgt_nid = label_to_nid.get(callee_name.lower())

if tgt_nid and tgt_nid != caller_nid:

pair = (caller_nid, tgt_nid)

if pair not in seen_call_pairs:

seen_call_pairs.add(pair)

line = node.start_point[0] + 1

edges.append({

"source": caller_nid,

"target": tgt_nid,

"relation": "calls",

"confidence": "INFERRED",

"source_file": str_path,

"source_location": f"L{line}",

"weight": 0.8,

})

for child in node.children:

walk_calls(child, caller_nid)

for caller_nid, body_node in function_bodies:

walk_calls(body_node, caller_nid)

valid_ids = seen_ids

clean_edges = []

for edge in edges:

src, tgt = edge["source"], edge["target"]

if src in valid_ids and (tgt in valid_ids or edge["relation"] in ("imports", "imports_from")):

clean_edges.append(edge)

"""Extract functions, methods, type declarations, and imports from a .go file."""

try:

import tree_sitter_go as tsgo

from tree_sitter import Language, Parser

except ImportError:

return {"nodes": [], "edges": [], "error": "tree-sitter-go not installed"}

try:

language = Language(tsgo.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 = path.stem

str_path = str(path)

nodes: list[dict] = []

edges: list[dict] = []

seen_ids: set[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_raw(src: str, tgt: str, relation: str, line: int, confidence: str = "EXTRACTED", weight: float = 1.0) -> None:

edges.append({

"source": src,

"target": tgt,

"relation": relation,

"confidence": confidence,

"source_file": str_path,

"source_location": f"L{line}",

"weight": weight,

})

file_nid = _make_id(stem)

add_node(file_nid, path.name, 1)

function_bodies: list[tuple[str, object]] = []

def walk(node) -> None:

t = node.type

if t == "function_declaration":

name_node = node.child_by_field_name("name")

if name_node:

func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = node.start_point[0] + 1

func_nid = _make_id(stem, func_name)

add_node(func_nid, f"{func_name}()", line)

add_edge_raw(file_nid, func_nid, "contains", line)

body = node.child_by_field_name("body")

if body:

function_bodies.append((func_nid, body))

return

if t == "method_declaration":

receiver = node.child_by_field_name("receiver")

receiver_type: str | None = None

if receiver:

for param in receiver.children:

if param.type == "parameter_declaration":

type_node = param.child_by_field_name("type")

if type_node:

raw = source[type_node.start_byte:type_node.end_byte].decode("utf-8", errors="replace").lstrip("*").strip()

receiver_type = raw

break

name_node = node.child_by_field_name("name")

if name_node:

method_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = node.start_point[0] + 1

if receiver_type:

parent_nid = _make_id(stem, receiver_type)

add_node(parent_nid, receiver_type, line)

method_nid = _make_id(parent_nid, method_name)

add_node(method_nid, f".{method_name}()", line)

add_edge_raw(parent_nid, method_nid, "method", line)

else:

method_nid = _make_id(stem, method_name)

add_node(method_nid, f"{method_name}()", line)

add_edge_raw(file_nid, method_nid, "contains", line)

body = node.child_by_field_name("body")

if body:

function_bodies.append((method_nid, body))

return

if t == "type_declaration":

for child in node.children:

if child.type == "type_spec":

name_node = child.child_by_field_name("name")

if name_node:

type_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = child.start_point[0] + 1

type_nid = _make_id(stem, type_name)

add_node(type_nid, type_name, line)

add_edge_raw(file_nid, type_nid, "contains", line)

return

if t == "import_declaration":

for child in node.children:

if child.type == "import_spec_list":

for spec in child.children:

if spec.type == "import_spec":

path_node = spec.child_by_field_name("path")

if path_node:

raw = source[path_node.start_byte:path_node.end_byte].decode("utf-8", errors="replace").strip('"')

module_name = raw.split("/")[-1]

tgt_nid = _make_id(module_name)

add_edge_raw(file_nid, tgt_nid, "imports_from", spec.start_point[0] + 1)

elif child.type == "import_spec":

path_node = child.child_by_field_name("path")

if path_node:

raw = source[path_node.start_byte:path_node.end_byte].decode("utf-8", errors="replace").strip('"')

module_name = raw.split("/")[-1]

tgt_nid = _make_id(module_name)

add_edge_raw(file_nid, tgt_nid, "imports_from", child.start_point[0] + 1)

return

for child in node.children:

walk(child)

walk(root)

label_to_nid: dict[str, str] = {}

for n in nodes:

raw = n["label"]

normalised = raw.strip("()").lstrip(".")

label_to_nid[normalised.lower()] = n["id"]

seen_call_pairs: set[tuple[str, str]] = set()

def walk_calls(node, caller_nid: str) -> None:

if node.type in ("function_declaration", "method_declaration"):

return

if node.type == "call_expression":

func_node = node.child_by_field_name("function")

callee_name: str | None = None

if func_node:

if func_node.type == "identifier":

callee_name = source[func_node.start_byte:func_node.end_byte].decode("utf-8", errors="replace")

elif func_node.type == "selector_expression":

field = func_node.child_by_field_name("field")

if field:

callee_name = source[field.start_byte:field.end_byte].decode("utf-8", errors="replace")

if callee_name:

tgt_nid = label_to_nid.get(callee_name.lower())

if tgt_nid and tgt_nid != caller_nid:

pair = (caller_nid, tgt_nid)

if pair not in seen_call_pairs:

seen_call_pairs.add(pair)

line = node.start_point[0] + 1

edges.append({

"source": caller_nid,

"target": tgt_nid,

"relation": "calls",

"confidence": "INFERRED",

"source_file": str_path,

"source_location": f"L{line}",

"weight": 0.8,

})

for child in node.children:

walk_calls(child, caller_nid)

for caller_nid, body_node in function_bodies:

walk_calls(body_node, caller_nid)

valid_ids = seen_ids

clean_edges = []

for edge in edges:

src, tgt = edge["source"], edge["target"]

if src in valid_ids and (tgt in valid_ids or edge["relation"] in ("imports", "imports_from")):

clean_edges.append(edge)

"""Extract functions, structs, enums, traits, impl methods, and use declarations from a .rs file."""

try:

import tree_sitter_rust as tsrust

from tree_sitter import Language, Parser

except ImportError:

return {"nodes": [], "edges": [], "error": "tree-sitter-rust not installed"}

try:

language = Language(tsrust.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 = path.stem

str_path = str(path)

nodes: list[dict] = []

edges: list[dict] = []

seen_ids: set[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, confidence: str = "EXTRACTED", weight: float = 1.0) -> None:

edges.append({

"source": src,

"target": tgt,

"relation": relation,

"confidence": confidence,

"source_file": str_path,

"source_location": f"L{line}",

"weight": weight,

})

file_nid = _make_id(stem)

add_node(file_nid, path.name, 1)

function_bodies: list[tuple[str, object]] = []

def walk(node, parent_impl_nid: str | None = None) -> None:

t = node.type

if t == "function_item":

name_node = node.child_by_field_name("name")

if name_node:

func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = node.start_point[0] + 1

if parent_impl_nid:

func_nid = _make_id(parent_impl_nid, func_name)

add_node(func_nid, f".{func_name}()", line)

add_edge(parent_impl_nid, func_nid, "method", line)

else:

func_nid = _make_id(stem, func_name)

add_node(func_nid, f"{func_name}()", line)

add_edge(file_nid, func_nid, "contains", line)

body = node.child_by_field_name("body")

if body:

function_bodies.append((func_nid, body))

return

if t in ("struct_item", "enum_item", "trait_item"):

name_node = node.child_by_field_name("name")

if name_node:

item_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = node.start_point[0] + 1

item_nid = _make_id(stem, item_name)

add_node(item_nid, item_name, line)

add_edge(file_nid, item_nid, "contains", line)

return

if t == "impl_item":

type_node = node.child_by_field_name("type")

impl_nid: str | None = None

if type_node:

type_name = source[type_node.start_byte:type_node.end_byte].decode("utf-8", errors="replace").strip()

impl_nid = _make_id(stem, type_name)

add_node(impl_nid, type_name, node.start_point[0] + 1)

body = node.child_by_field_name("body")

if body:

for child in body.children:

walk(child, parent_impl_nid=impl_nid)

return

if t == "use_declaration":

arg = node.child_by_field_name("argument")

if arg:

raw = source[arg.start_byte:arg.end_byte].decode("utf-8", errors="replace")

clean = raw.split("{")[0].rstrip(":").rstrip("*").rstrip(":")

module_name = clean.split("::")[-1].strip()

if module_name:

tgt_nid = _make_id(module_name)

add_edge(file_nid, tgt_nid, "imports_from", node.start_point[0] + 1)

return

for child in node.children:

walk(child, parent_impl_nid=None)

walk(root)

label_to_nid: dict[str, str] = {}

for n in nodes:

raw = n["label"]

normalised = raw.strip("()").lstrip(".")

label_to_nid[normalised.lower()] = n["id"]

seen_call_pairs: set[tuple[str, str]] = set()

def walk_calls(node, caller_nid: str) -> None:

if node.type == "function_item":

return

if node.type == "call_expression":

func_node = node.child_by_field_name("function")

callee_name: str | None = None

if func_node:

if func_node.type == "identifier":

callee_name = source[func_node.start_byte:func_node.end_byte].decode("utf-8", errors="replace")

elif func_node.type == "field_expression":

field = func_node.child_by_field_name("field")

if field:

callee_name = source[field.start_byte:field.end_byte].decode("utf-8", errors="replace")

elif func_node.type == "scoped_identifier":

name = func_node.child_by_field_name("name")

if name:

callee_name = source[name.start_byte:name.end_byte].decode("utf-8", errors="replace")

if callee_name:

tgt_nid = label_to_nid.get(callee_name.lower())

if tgt_nid and tgt_nid != caller_nid:

pair = (caller_nid, tgt_nid)

if pair not in seen_call_pairs:

seen_call_pairs.add(pair)

line = node.start_point[0] + 1

edges.append({

"source": caller_nid,

"target": tgt_nid,

"relation": "calls",

"confidence": "INFERRED",

"source_file": str_path,

"source_location": f"L{line}",

"weight": 0.8,

})

for child in node.children:

walk_calls(child, caller_nid)

for caller_nid, body_node in function_bodies:

walk_calls(body_node, caller_nid)

valid_ids = seen_ids

clean_edges = []

for edge in edges:

src, tgt = edge["source"], edge["target"]

if src in valid_ids and (tgt in valid_ids or edge["relation"] in ("imports", "imports_from")):

clean_edges.append(edge)

"""Extract classes, interfaces, methods, constructors, and imports from a .java file."""

try:

import tree_sitter_java as tsjava

from tree_sitter import Language, Parser

except ImportError:

return {"nodes": [], "edges": [], "error": "tree-sitter-java not installed"}

try:

language = Language(tsjava.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 = path.stem

str_path = str(path)

nodes: list[dict] = []

edges: list[dict] = []

seen_ids: set[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_raw(src: str, tgt: str, relation: str, line: int, confidence: str = "EXTRACTED", weight: float = 1.0) -> None:

edges.append({

"source": src,

"target": tgt,

"relation": relation,

"confidence": confidence,

"source_file": str_path,

"source_location": f"L{line}",

"weight": weight,

})

file_nid = _make_id(stem)

add_node(file_nid, path.name, 1)

function_bodies: list[tuple[str, object]] = []

def _walk_scoped_identifier(node) -> str:

"""Reconstruct a dotted import path from nested scoped_identifier nodes."""

parts: list[str] = []

cur = node

while cur:

if cur.type == "scoped_identifier":

name_node = cur.child_by_field_name("name")

if name_node:

parts.append(source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace"))

cur = cur.child_by_field_name("scope")

elif cur.type == "identifier":

parts.append(source[cur.start_byte:cur.end_byte].decode("utf-8", errors="replace"))

break

else:

break

parts.reverse()

return ".".join(parts)

def walk(node, parent_class_nid: str | None = None) -> None:

t = node.type

if t == "import_declaration":

# Find scoped_identifier or identifier child

for child in node.children:

if child.type in ("scoped_identifier", "identifier"):

path_str = _walk_scoped_identifier(child)

module_name = path_str.split(".")[-1].strip("*").strip(".") or path_str.split(".")[-2]

if module_name:

tgt_nid = _make_id(module_name)

add_edge_raw(file_nid, tgt_nid, "imports", node.start_point[0] + 1)

break

return

if t in ("class_declaration", "interface_declaration"):

name_node = node.child_by_field_name("name")

if not name_node:

return

class_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

class_nid = _make_id(stem, class_name)

line = node.start_point[0] + 1

add_node(class_nid, class_name, line)

add_edge_raw(file_nid, class_nid, "contains", line)

body = node.child_by_field_name("body")

if body:

for child in body.children:

walk(child, parent_class_nid=class_nid)

return

if t in ("method_declaration", "constructor_declaration"):

name_node = node.child_by_field_name("name")

if not name_node:

return

method_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

line = node.start_point[0] + 1

if parent_class_nid:

method_nid = _make_id(parent_class_nid, method_name)

add_node(method_nid, f".{method_name}()", line)

add_edge_raw(parent_class_nid, method_nid, "method", line)

else:

method_nid = _make_id(stem, method_name)

add_node(method_nid, f"{method_name}()", line)

add_edge_raw(file_nid, method_nid, "contains", line)

body = node.child_by_field_name("body")

if body:

function_bodies.append((method_nid, body))

return

for child in node.children:

walk(child, parent_class_nid=None)

walk(root)

label_to_nid: dict[str, str] = {}

for n in nodes:

raw = n["label"]

normalised = raw.strip("()").lstrip(".")

label_to_nid[normalised.lower()] = n["id"]

seen_call_pairs: set[tuple[str, str]] = set()

def walk_calls(node, caller_nid: str) -> None:

if node.type in ("method_declaration", "constructor_declaration"):

return

if node.type == "method_invocation":

name_node = node.child_by_field_name("name")

callee_name: str | None = None

if name_node:

callee_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")

if callee_name:

tgt_nid = label_to_nid.get(callee_name.lower())

if tgt_nid and tgt_nid != caller_nid:

pair = (caller_nid, tgt_nid)

if pair not in seen_call_pairs:

seen_call_pairs.add(pair)

line = node.start_point[0] + 1

edges.append({

"source": caller_nid,

"target": tgt_nid,

"relation": "calls",

"confidence": "INFERRED",

"source_file": str_path,

"source_location": f"L{line}",

"weight": 0.8,

})

for child in node.children:

walk_calls(child, caller_nid)

for caller_nid, body_node in function_bodies:

walk_calls(body_node, caller_nid)

valid_ids = seen_ids

clean_edges = []

for edge in edges:

src, tgt = edge["source"], edge["target"]

if src in valid_ids and (tgt in valid_ids or edge["relation"] in ("imports", "imports_from")):

clean_edges.append(edge)

"""Extract functions and includes from a .c/.h file."""

try:

import tree_sitter_c as tsc

from tree_sitter import Language, Parser

except ImportError:

return {"nodes": [], "edges": [], "error": "tree-sitter-c not installed"}

try:

language = Language(tsc.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 = path.stem

str_path = str(path)

nodes: list[dict] = []

edges: list[dict] = []

seen_ids: set[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}",