# src/codegraphcontext/tools/graph_builder.py

import asyncio

import pathspec

from pathlib import Path

from typing import Any, Coroutine, Dict, Optional, Tuple

from datetime import datetime

from ..core.database import DatabaseManager

from ..core.jobs import JobManager, JobStatus

from ..utils.debug_log import debug_log, info_logger, error_logger, warning_logger

# New imports for tree-sitter (using tree-sitter-language-pack)

from tree_sitter import Language, Parser

from ..utils.tree_sitter_manager import get_tree_sitter_manager

from ..cli.config_manager import get_config_value

from ..utils.path_ignore import file_path_has_ignore_dir_segment

import fnmatch

DEFAULT_IGNORE_PATTERNS = [

# Vendor / env dirs (gitignore-style; complements IGNORE_DIRS during indexing)

"node_modules/",

"venv/",

".venv/",

"env/",

".env/",

"dist/",

"build/",

"target/",

"out/",

".git/",

"__pycache__/",

"*.png",

"*.jpg",

"*.jpeg",

"*.gif",

"*.svg",

"*.mp4",

"*.mp3",

"*.zip",

"*.tar",

"*.gz",

]

class TreeSitterParser:

"""A generic parser wrapper for a specific language using tree-sitter."""

def __init__(self, language_name: str):

self.language_name = language_name

self.ts_manager = get_tree_sitter_manager()

# Get the language (cached) and create a new parser for this instance

self.language: Language = self.ts_manager.get_language_safe(language_name)

# In tree-sitter 0.25+, Parser takes language in constructor

self.parser = Parser(self.language)

self.language_specific_parser = None

if self.language_name == 'python':

from .languages.python import PythonTreeSitterParser

self.language_specific_parser = PythonTreeSitterParser(self)

elif self.language_name == 'javascript':

from .languages.javascript import JavascriptTreeSitterParser

self.language_specific_parser = JavascriptTreeSitterParser(self)

elif self.language_name == 'go':

from .languages.go import GoTreeSitterParser

self.language_specific_parser = GoTreeSitterParser(self)

elif self.language_name == 'typescript':

from .languages.typescript import TypescriptTreeSitterParser

self.language_specific_parser = TypescriptTreeSitterParser(self)

elif self.language_name == 'cpp':

from .languages.cpp import CppTreeSitterParser

self.language_specific_parser = CppTreeSitterParser(self)

elif self.language_name == 'rust':

from .languages.rust import RustTreeSitterParser

self.language_specific_parser = RustTreeSitterParser(self)

elif self.language_name == 'c':

from .languages.c import CTreeSitterParser

self.language_specific_parser = CTreeSitterParser(self)

elif self.language_name == 'java':

from .languages.java import JavaTreeSitterParser

self.language_specific_parser = JavaTreeSitterParser(self)

elif self.language_name == 'ruby':

from .languages.ruby import RubyTreeSitterParser

self.language_specific_parser = RubyTreeSitterParser(self)

elif self.language_name == 'c_sharp':

from .languages.csharp import CSharpTreeSitterParser

self.language_specific_parser = CSharpTreeSitterParser(self)

elif self.language_name == 'php':

from .languages.php import PhpTreeSitterParser

self.language_specific_parser = PhpTreeSitterParser(self)

elif self.language_name == 'kotlin':

from .languages.kotlin import KotlinTreeSitterParser

self.language_specific_parser = KotlinTreeSitterParser(self)

elif self.language_name == 'scala':

from .languages.scala import ScalaTreeSitterParser

self.language_specific_parser = ScalaTreeSitterParser(self)

elif self.language_name == 'swift':

from .languages.swift import SwiftTreeSitterParser

self.language_specific_parser = SwiftTreeSitterParser(self)

elif self.language_name == 'haskell':

from .languages.haskell import HaskellTreeSitterParser

self.language_specific_parser = HaskellTreeSitterParser(self)

elif self.language_name == 'dart':

from .languages.dart import DartTreeSitterParser

self.language_specific_parser = DartTreeSitterParser(self)

elif self.language_name == 'perl':

from .languages.perl import PerlTreeSitterParser

self.language_specific_parser = PerlTreeSitterParser(self)

elif self.language_name == 'elixir':

from .languages.elixir import ElixirTreeSitterParser

self.language_specific_parser = ElixirTreeSitterParser(self)

def parse(self, path: Path, is_dependency: bool = False, **kwargs) -> Dict:

"""Dispatches parsing to the language-specific parser."""

if self.language_specific_parser:

return self.language_specific_parser.parse(path, is_dependency, **kwargs)

else:

raise NotImplementedError(f"No language-specific parser implemented for {self.language_name}")

class GraphBuilder:

"""Module for building and managing the Neo4j code graph."""

def __init__(self, db_manager: DatabaseManager, job_manager: JobManager, loop: asyncio.AbstractEventLoop):

self.db_manager = db_manager

self.job_manager = job_manager

self.loop = loop

self.driver = self.db_manager.get_driver()

self.parsers = {

'.py': 'python',

'.ipynb': 'python',

'.js': 'javascript',

'.jsx': 'javascript',

'.mjs': 'javascript',

'.cjs': 'javascript',

'.go': 'go',

'.ts': 'typescript',

'.tsx': 'typescript',

'.cpp': 'cpp',

'.h': 'cpp',

'.hpp': 'cpp',

'.hh': 'cpp',

'.rs': 'rust',

'.c': 'c',

# '.h': 'c', # Need to write an algo for distinguishing C vs C++ headers

'.java': 'java',

'.rb': 'ruby',

'.cs': 'c_sharp',

'.php': 'php',

'.kt': 'kotlin',

'.scala': 'scala',

'.sc': 'scala',

'.swift': 'swift',

'.hs': 'haskell',

'.dart': 'dart',

'.pl': 'perl',

'.pm': 'perl',

'.ex': 'elixir',

'.exs': 'elixir',

}

self._parsed_cache = {}

self.create_schema()

def get_parser(self, extension: str) -> Optional[TreeSitterParser]:

"""Gets or creates a TreeSitterParser for the given extension."""

lang_name = self.parsers.get(extension)

if not lang_name:

return None

if lang_name not in self._parsed_cache:

try:

self._parsed_cache[lang_name] = TreeSitterParser(lang_name)

except Exception as e:

warning_logger(f"Failed to initialize parser for {lang_name}: {e}")

return None

return self._parsed_cache[lang_name]

# A general schema creation based on common features across languages

def create_schema(self):

"""Create constraints and indexes in Neo4j."""

# When adding a new node type with a unique key, add its constraint here.

with self.driver.session() as session:

try:

session.run("CREATE CONSTRAINT repository_path IF NOT EXISTS FOR (r:Repository) REQUIRE r.path IS UNIQUE")

session.run("CREATE CONSTRAINT path IF NOT EXISTS FOR (f:File) REQUIRE f.path IS UNIQUE")

session.run("CREATE CONSTRAINT directory_path IF NOT EXISTS FOR (d:Directory) REQUIRE d.path IS UNIQUE")

session.run("CREATE CONSTRAINT function_unique IF NOT EXISTS FOR (f:Function) REQUIRE (f.name, f.path, f.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT class_unique IF NOT EXISTS FOR (c:Class) REQUIRE (c.name, c.path, c.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT trait_unique IF NOT EXISTS FOR (t:Trait) REQUIRE (t.name, t.path, t.line_number) IS UNIQUE") # Added trait constraint

session.run("CREATE CONSTRAINT interface_unique IF NOT EXISTS FOR (i:Interface) REQUIRE (i.name, i.path, i.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT macro_unique IF NOT EXISTS FOR (m:Macro) REQUIRE (m.name, m.path, m.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT variable_unique IF NOT EXISTS FOR (v:Variable) REQUIRE (v.name, v.path, v.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT module_name IF NOT EXISTS FOR (m:Module) REQUIRE m.name IS UNIQUE")

session.run("CREATE CONSTRAINT struct_cpp IF NOT EXISTS FOR (cstruct: Struct) REQUIRE (cstruct.name, cstruct.path, cstruct.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT enum_cpp IF NOT EXISTS FOR (cenum: Enum) REQUIRE (cenum.name, cenum.path, cenum.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT union_cpp IF NOT EXISTS FOR (cunion: Union) REQUIRE (cunion.name, cunion.path, cunion.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT annotation_unique IF NOT EXISTS FOR (a:Annotation) REQUIRE (a.name, a.path, a.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT record_unique IF NOT EXISTS FOR (r:Record) REQUIRE (r.name, r.path, r.line_number) IS UNIQUE")

session.run("CREATE CONSTRAINT property_unique IF NOT EXISTS FOR (p:Property) REQUIRE (p.name, p.path, p.line_number) IS UNIQUE")

# Indexes for language attribute

session.run("CREATE INDEX function_lang IF NOT EXISTS FOR (f:Function) ON (f.lang)")

session.run("CREATE INDEX class_lang IF NOT EXISTS FOR (c:Class) ON (c.lang)")

session.run("CREATE INDEX annotation_lang IF NOT EXISTS FOR (a:Annotation) ON (a.lang)")

is_falkordb = getattr(self.db_manager, 'get_backend_type', lambda: 'neo4j')() != 'neo4j'

if is_falkordb:

# FalkorDB uses db.idx.fulltext.createNodeIndex per label

for label in ['Function', 'Class']:

try:

session.run(f"CALL db.idx.fulltext.createNodeIndex('{label}', 'name', 'source', 'docstring')")

except Exception:

pass # Index may already exist

else:

session.run("""

CREATE FULLTEXT INDEX code_search_index IF NOT EXISTS

FOR (n:Function|Class|Variable)

ON EACH [n.name, n.source, n.docstring]

""")

info_logger("Database schema verified/created successfully")

except Exception as e:

warning_logger(f"Schema creation warning: {e}")

# Neo4j RANGE indexes have an ~8 kB key-size limit. Long C++ template

# function names (e.g. from llama.cpp) can exceed this, causing

# "Property value is too large to index" errors. We cap string properties

# at 4096 chars, which is comfortably under the 8 kB boundary.

_MAX_STR_LEN = 4096

@staticmethod

def _sanitize_props(props: Dict) -> Dict:

"""Return a copy of *props* with all values coerced to database-safe types.

FalkorDB and KùzuDB only accept node properties that are primitives

(str, int, float, bool, None) or flat lists of primitives. Complex

values such as tuples, dicts, or lists-of-dicts that come from language

parsers (e.g. C's ``detailed_args`` or Scala's tuple ``class_context``)

are serialized to a JSON string so the data is preserved rather than

being silently dropped.

Additionally, string values are truncated to _MAX_STR_LEN characters to

avoid Neo4j's RANGE-index 8 kB property-size limit (triggered by very

long C++ template-mangled function names).

"""

import json

MAX = GraphBuilder._MAX_STR_LEN

def _is_primitive(v):

return isinstance(v, (str, int, float, bool)) or v is None

def _is_flat_list(v):

return isinstance(v, list) and all(_is_primitive(item) for item in v)

def _coerce(v):

if isinstance(v, str):

# Truncate long strings to stay within Neo4j RANGE index limits

return v[:MAX] if len(v) > MAX else v

if _is_primitive(v):

return v

if _is_flat_list(v):

# Truncate any long strings in lists too

return [s[:MAX] if isinstance(s, str) and len(s) > MAX else s for s in v]

# Tuples, dicts, lists-of-dicts, nested structures → JSON string

try:

serialized = json.dumps(v, default=str)

return serialized[:MAX] if len(serialized) > MAX else serialized

except Exception:

s = str(v)

return s[:MAX] if len(s) > MAX else s

return {k: _coerce(v) for k, v in props.items()}

def _pre_scan_for_imports(self, files: list[Path]) -> dict:

"""Dispatches pre-scan to the correct language-specific implementation."""

imports_map = {}

# Group files by language/extension

files_by_lang = {}

for file in files:

if file.suffix in self.parsers:

lang_ext = file.suffix

if lang_ext not in files_by_lang:

files_by_lang[lang_ext] = []

files_by_lang[lang_ext].append(file)

if '.py' in files_by_lang:

from .languages import python as python_lang_module

imports_map.update(python_lang_module.pre_scan_python(files_by_lang['.py'], self.get_parser('.py')))

if '.ipynb' in files_by_lang:

from .languages import python as python_lang_module

imports_map.update(python_lang_module.pre_scan_python(files_by_lang['.ipynb'], self.get_parser('.ipynb')))

if '.js' in files_by_lang:

from .languages import javascript as js_lang_module

imports_map.update(js_lang_module.pre_scan_javascript(files_by_lang['.js'], self.get_parser('.js')))

if '.jsx' in files_by_lang:

from .languages import javascript as js_lang_module

imports_map.update(js_lang_module.pre_scan_javascript(files_by_lang['.jsx'], self.get_parser('.jsx')))

if '.mjs' in files_by_lang:

from .languages import javascript as js_lang_module

imports_map.update(js_lang_module.pre_scan_javascript(files_by_lang['.mjs'], self.get_parser('.mjs')))

if '.cjs' in files_by_lang:

from .languages import javascript as js_lang_module

imports_map.update(js_lang_module.pre_scan_javascript(files_by_lang['.cjs'], self.get_parser('.cjs')))

if '.go' in files_by_lang:

from .languages import go as go_lang_module

imports_map.update(go_lang_module.pre_scan_go(files_by_lang['.go'], self.get_parser('.go')))

if '.ts' in files_by_lang:

from .languages import typescript as ts_lang_module

imports_map.update(ts_lang_module.pre_scan_typescript(files_by_lang['.ts'], self.get_parser('.ts')))

if '.tsx' in files_by_lang:

from .languages import typescriptjsx as tsx_lang_module

imports_map.update(tsx_lang_module.pre_scan_typescript(files_by_lang['.tsx'], self.get_parser('.tsx')))

if '.cpp' in files_by_lang:

from .languages import cpp as cpp_lang_module

imports_map.update(cpp_lang_module.pre_scan_cpp(files_by_lang['.cpp'], self.get_parser('.cpp')))

if '.h' in files_by_lang:

from .languages import cpp as cpp_lang_module

imports_map.update(cpp_lang_module.pre_scan_cpp(files_by_lang['.h'], self.get_parser('.h')))

if '.hpp' in files_by_lang:

from .languages import cpp as cpp_lang_module

imports_map.update(cpp_lang_module.pre_scan_cpp(files_by_lang['.hpp'], self.get_parser('.hpp')))

if '.hh' in files_by_lang:

from .languages import cpp as cpp_lang_module

imports_map.update(cpp_lang_module.pre_scan_cpp(files_by_lang['.hh'], self.get_parser('.hh')))

if '.rs' in files_by_lang:

from .languages import rust as rust_lang_module

imports_map.update(rust_lang_module.pre_scan_rust(files_by_lang['.rs'], self.get_parser('.rs')))

if '.c' in files_by_lang:

from .languages import c as c_lang_module

imports_map.update(c_lang_module.pre_scan_c(files_by_lang['.c'], self.get_parser('.c')))

elif '.java' in files_by_lang:

from .languages import java as java_lang_module

imports_map.update(java_lang_module.pre_scan_java(files_by_lang['.java'], self.get_parser('.java')))

elif '.rb' in files_by_lang:

from .languages import ruby as ruby_lang_module

imports_map.update(ruby_lang_module.pre_scan_ruby(files_by_lang['.rb'], self.get_parser('.rb')))

elif '.cs' in files_by_lang:

from .languages import csharp as csharp_lang_module

imports_map.update(csharp_lang_module.pre_scan_csharp(files_by_lang['.cs'], self.get_parser('.cs')))

if '.kt' in files_by_lang:

from .languages import kotlin as kotlin_lang_module

imports_map.update(kotlin_lang_module.pre_scan_kotlin(files_by_lang['.kt'], self.get_parser('.kt')))

if '.scala' in files_by_lang:

from .languages import scala as scala_lang_module

imports_map.update(scala_lang_module.pre_scan_scala(files_by_lang['.scala'], self.get_parser('.scala')))

if '.sc' in files_by_lang:

from .languages import scala as scala_lang_module

imports_map.update(scala_lang_module.pre_scan_scala(files_by_lang['.sc'], self.get_parser('.sc')))

if '.swift' in files_by_lang:

from .languages import swift as swift_lang_module

imports_map.update(swift_lang_module.pre_scan_swift(files_by_lang['.swift'], self.get_parser('.swift')))

if '.dart' in files_by_lang:

from .languages import dart as dart_lang_module

imports_map.update(dart_lang_module.pre_scan_dart(files_by_lang['.dart'], self.get_parser('.dart')))

if '.pl' in files_by_lang:

from .languages import perl as perl_lang_module

imports_map.update(perl_lang_module.pre_scan_perl(files_by_lang['.pl'], self.get_parser('.pl')))

if '.pm' in files_by_lang:

from .languages import perl as perl_lang_module

imports_map.update(perl_lang_module.pre_scan_perl(files_by_lang['.pm'], self.get_parser('.pm')))

if '.ex' in files_by_lang:

from .languages import elixir as elixir_lang_module

imports_map.update(elixir_lang_module.pre_scan_elixir(files_by_lang['.ex'], self.get_parser('.ex')))

if '.exs' in files_by_lang:

from .languages import elixir as elixir_lang_module

imports_map.update(elixir_lang_module.pre_scan_elixir(files_by_lang['.exs'], self.get_parser('.exs')))

return imports_map

# Language-agnostic method

def add_repository_to_graph(self, repo_path: Path, is_dependency: bool = False):

"""Adds a repository node using its absolute path as the unique key."""

repo_name = repo_path.name

repo_path_str = str(repo_path.resolve())

with self.driver.session() as session:

session.run(

"""

MERGE (r:Repository {path: $path})

SET r.name = $name, r.is_dependency = $is_dependency

""",

path=repo_path_str,

name=repo_name,

is_dependency=is_dependency,

)

# First pass to add file and its contents

def add_file_to_graph(self, file_data: Dict, repo_name: str, imports_map: dict):

calls_count = len(file_data.get('function_calls', []))

debug_log(f"Executing add_file_to_graph for {file_data.get('path', 'unknown')} - Calls found: {calls_count}")

"""Adds a file and its contents within a single, unified session."""

file_path_str = str(Path(file_data['path']).resolve())

file_name = Path(file_path_str).name

is_dependency = file_data.get('is_dependency', False)

with self.driver.session() as session:

try:

# Match repository by path, not name, to avoid conflicts with same-named folders at different locations

repo_result = session.run("MATCH (r:Repository {path: $repo_path}) RETURN r.path as path", repo_path=str(Path(file_data['repo_path']).resolve())).single()

relative_path = str(Path(file_path_str).relative_to(Path(repo_result['path']))) if repo_result else file_name

except ValueError:

relative_path = file_name

session.run("""

MERGE (f:File {path: $path})

SET f.name = $name, f.relative_path = $relative_path, f.is_dependency = $is_dependency

""", path=file_path_str, name=file_name, relative_path=relative_path, is_dependency=is_dependency)

file_path_obj = Path(file_path_str)

if repo_result:

repo_path_obj = Path(repo_result['path'])

else:

# Fallback to the path we queried for

warning_logger(f"Repository node not found for {file_data['repo_path']} during indexing of {file_name}. Using original path.")

repo_path_obj = Path(file_data['repo_path']).resolve()

relative_path_to_file = file_path_obj.relative_to(repo_path_obj)

parent_path = str(repo_path_obj)

parent_label = 'Repository'

for part in relative_path_to_file.parts[:-1]:

current_path = Path(parent_path) / part

current_path_str = str(current_path)

session.run(f"""

MATCH (p:{parent_label} {{path: $parent_path}})

MERGE (d:Directory {{path: $current_path}})

SET d.name = $part

MERGE (p)-[:CONTAINS]->(d)

""", parent_path=parent_path, current_path=current_path_str, part=part)

parent_path = current_path_str

parent_label = 'Directory'

session.run(f"""

MATCH (p:{parent_label} {{path: $parent_path}})

MATCH (f:File {{path: $path}})

MERGE (p)-[:CONTAINS]->(f)

""", parent_path=parent_path, path=file_path_str)

# CONTAINS relationships for functions, classes, and variables

# To add a new language-specific node type (e.g., 'Trait' for Rust):

# 1. Ensure your language-specific parser returns a list under a unique key (e.g., 'traits': [...] ).

# 2. Add a new constraint for the new label in the `create_schema` method.

# 3. Add a new entry to the `item_mappings` list below (e.g., (file_data.get('traits', []), 'Trait') ).

item_mappings = [

(file_data.get('functions', []), 'Function'),

(file_data.get('classes', []), 'Class'),

(file_data.get('traits', []), 'Trait'), # <-- Added trait mapping

(file_data.get('variables', []), 'Variable'),

(file_data.get('interfaces', []), 'Interface'),

(file_data.get('macros', []), 'Macro'),

(file_data.get('structs',[]), 'Struct'),

(file_data.get('enums',[]), 'Enum'),

(file_data.get('unions',[]), 'Union'),

(file_data.get('records',[]), 'Record'),

(file_data.get('properties',[]), 'Property'),

]

for item_data, label in item_mappings:

for item in item_data:

# Ensure cyclomatic_complexity is set for functions

if label == 'Function' and 'cyclomatic_complexity' not in item:

item['cyclomatic_complexity'] = 1 # Default value

query = f"""

MATCH (f:File {{path: $path}})

MERGE (n:{label} {{name: $name, path: $path, line_number: $line_number}})

SET n += $props

MERGE (f)-[:CONTAINS]->(n)

"""

# Strip non-primitive fields (dicts, tuples, lists-of-dicts)

# before writing to the database to avoid runtime errors such as

# "Property values can only be of primitive types or arrays of

# primitive types" raised by FalkorDB / KùzuDB.

# _sanitize_props also truncates long strings to avoid Neo4j's

# 8 kB RANGE-index limit (seen with long C++ template names).

safe_props = self._sanitize_props(item)

try:

session.run(query, path=file_path_str, name=item['name'], line_number=item['line_number'], props=safe_props)

except Exception as node_err:

err_str = str(node_err)

if "too large to index" in err_str or "property size" in err_str.lower():

warning_logger(

f"Skipping {label} '{item['name']}' in {file_path_str}: "

f"property value too large for index (name length={len(item['name'])})"

)

else:

raise # Re-raise unexpected errors

if label == 'Function':

for arg_name in item.get('args', []):

session.run("""

MATCH (fn:Function {name: $func_name, path: $path, line_number: $line_number})

MERGE (p:Parameter {name: $arg_name, path: $path, function_line_number: $line_number})

MERGE (fn)-[:HAS_PARAMETER]->(p)

""", func_name=item['name'], path=file_path_str, line_number=item['line_number'], arg_name=arg_name)

# --- NEW: persist Ruby Modules ---

for m in file_data.get('modules', []):

session.run("""

MERGE (mod:Module {name: $name})

ON CREATE SET mod.lang = $lang

ON MATCH SET mod.lang = coalesce(mod.lang, $lang)

""", name=m["name"], lang=file_data.get("lang"))

# Create CONTAINS relationships for nested functions

for item in file_data.get('functions', []):

if item.get("context_type") == "function_definition":

session.run("""

MATCH (outer:Function {name: $context, path: $path})

MATCH (inner:Function {name: $name, path: $path, line_number: $line_number})

MERGE (outer)-[:CONTAINS]->(inner)

""", context=item["context"], path=file_path_str, name=item["name"], line_number=item["line_number"])

# Handle imports and create IMPORTS relationships

for imp in file_data.get('imports', []):

info_logger(f"Processing import: {imp}")

lang = file_data.get('lang')

if lang == 'javascript':

# New, correct logic for JS

module_name = imp.get('source')

if not module_name: continue

# Use a map for relationship properties to handle optional alias and line_number

rel_props = {'imported_name': imp.get('name', '*')}

if imp.get('alias'):

rel_props['alias'] = imp.get('alias')

if imp.get('line_number'):

rel_props['line_number'] = imp.get('line_number')

session.run("""

MATCH (f:File {path: $path})

MERGE (m:Module {name: $module_name})

MERGE (f)-[r:IMPORTS]->(m)

SET r += $props

""", path=file_path_str, module_name=module_name, props=rel_props)

else:

# Existing logic for Python (and other languages)

# For KùzuDB, Module schema only has: name, lang, full_import_name.

# 'alias' belongs on the relationship.

set_clauses = []

if 'full_import_name' in imp:

set_clauses.append("m.full_import_name = $full_import_name")

set_clause_str = ("SET " + ", ".join(set_clauses)) if set_clauses else ""

# Build relationship properties

rel_props = {}

if imp.get('line_number'):

rel_props['line_number'] = imp.get('line_number')

if imp.get('alias'):

rel_props['alias'] = imp.get('alias')

# Ensure full_import_name is available in params for SET clause

params = imp.copy()

params['path'] = file_path_str

params['module_name'] = imp.get('name') # Use 'name' from imp as module name

# Sanitize scalar params but keep rel_props as a dict

# (FalkorDB SET r += $rel_props requires a map, not a string)

sanitized = self._sanitize_props(params)

sanitized['rel_props'] = rel_props

session.run(f"""

MATCH (f:File {{path: $path}})

MERGE (m:Module {{name: $module_name}})

{set_clause_str}

MERGE (f)-[r:IMPORTS]->(m)

SET r += $rel_props

""", **sanitized)

# Handle CONTAINS relationship between class to their children like variables

for func in file_data.get('functions', []):

if func.get('class_context'):

# Try same-file match first (Python, JS, etc.)

if not self._safe_run_create(session, """

MATCH (c:Class {name: $class_name, path: $path})

MATCH (fn:Function {name: $func_name, path: $path, line_number: $func_line})

MERGE (c)-[:CONTAINS]->(fn)

RETURN count(*) as created

""", {

'class_name': func['class_context'],

'path': file_path_str,

'func_name': func['name'],

'func_line': func['line_number']

}):

# Cross-file match for C/C++ where class is in .h and method in .cpp.

# Note: matches by class name only (no path constraint), so classes

# with identical names in different files could get false links.

self._safe_run_create(session, """

MATCH (c:Class {name: $class_name})

MATCH (fn:Function {name: $func_name, path: $path, line_number: $func_line})

MERGE (c)-[:CONTAINS]->(fn)

RETURN count(*) as created

""", {

'class_name': func['class_context'],

'path': file_path_str,

'func_name': func['name'],

'func_line': func['line_number']

})

# --- NEW: Class INCLUDES Module (Ruby mixins) ---

for inc in file_data.get('module_inclusions', []):

session.run("""

MATCH (c:Class {name: $class_name, path: $path})

MERGE (m:Module {name: $module_name})

MERGE (c)-[:INCLUDES]->(m)

""",

class_name=inc["class"],

path=file_path_str,

module_name=inc["module"])

# Class inheritance is handled in a separate pass after all files are processed.

# Function calls are also handled in a separate pass after all files are processed.

# Second pass to create relationships that depend on all files being present like call functions and class inheritance

def _safe_run_create(self, session, query, params) -> bool:

"""Helper to run a creation query safely, catching exceptions and checking result."""

try:

result = session.run(query, **params)

row = result.single()

return row is not None and row.get('created', 0) > 0

except Exception as e:

# Optionally log, but suppress to allow fallback

return False

def _create_function_calls(self, session, file_data: Dict, imports_map: dict):

"""Create CALLS relationships with a unified, prioritized logic flow for all call types."""

caller_file_path = str(Path(file_data['path']).resolve())

num_calls = len(file_data.get('function_calls', []))

if num_calls > 0:

debug_log(f"Creating function calls for {caller_file_path} (Count: {num_calls})")

local_names = {f['name'] for f in file_data.get('functions', [])} | \

{c['name'] for c in file_data.get('classes', [])}

local_imports = {imp.get('alias') or imp['name'].split('.')[-1]: imp['name']

for imp in file_data.get('imports', [])}

# Check if we should skip external resolution attempts -

skip_external = (get_config_value("SKIP_EXTERNAL_RESOLUTION") or "false").lower() == "true"

for call in file_data.get('function_calls', []):

called_name = call['name']

# debug_log(f"Processing call: {called_name}")

if called_name in __builtins__: continue

resolved_path = None

full_call = call.get('full_name', called_name)

base_obj = full_call.split('.')[0] if '.' in full_call else None

# For chained calls like self.graph_builder.method(), we need to look up 'method'

# For direct calls like self.method(), we can use the caller's file

is_chained_call = full_call.count('.') > 1 if '.' in full_call else False

# Determine the lookup name:

# - For chained calls (self.attr.method), use the actual method name

# - For direct calls (self.method or module.function), use the base object

if is_chained_call and base_obj in ('self', 'this', 'super', 'super()', 'cls', '@'):

lookup_name = called_name # Use the actual method name for lookup

else:

lookup_name = base_obj if base_obj else called_name

# 1. Check for local context keywords/direct local names

# Only resolve to caller_file_path for DIRECT self/this calls, not chained ones

if base_obj in ('self', 'this', 'super', 'super()', 'cls', '@') and not is_chained_call:

resolved_path = caller_file_path

elif lookup_name in local_names:

resolved_path = caller_file_path

# 2. Check inferred type if available

elif call.get('inferred_obj_type'):

obj_type = call['inferred_obj_type']

possible_paths = imports_map.get(obj_type, [])

if len(possible_paths) > 0:

resolved_path = possible_paths[0]

# 3. Check imports map with validation against local imports

if not resolved_path:

possible_paths = imports_map.get(lookup_name, [])

if len(possible_paths) == 1:

resolved_path = possible_paths[0]

elif len(possible_paths) > 1:

if lookup_name in local_imports:

full_import_name = local_imports[lookup_name]

# Optimization: Check if the FQN is directly in imports_map (from pre-scan)

if full_import_name in imports_map:

direct_paths = imports_map[full_import_name]

if direct_paths and len(direct_paths) == 1:

resolved_path = direct_paths[0]

if not resolved_path:

for path in possible_paths:

if full_import_name.replace('.', '/') in path:

resolved_path = path

break

if not resolved_path:

# Only log warning if we're not skipping external resolution

if not skip_external:

warning_logger(f"Could not resolve call {called_name} (lookup: {lookup_name}) in {caller_file_path}")

# Track that this was an unresolved external call

is_unresolved_external = True

else:

is_unresolved_external = False

# else:

# info_logger(f"Resolved call {called_name} -> {resolved_path}")

# Legacy fallback block (was mis-indented)

if not resolved_path:

possible_paths = imports_map.get(lookup_name, [])

if len(possible_paths) > 0:

# Final fallback: global candidate

# Check if it was imported explicitly, otherwise risky

if lookup_name in local_imports:

# We already tried specific matching above, but if we are here

# it means we had ambiguity without matching path?

pass

else:

# Fallback to first available if not imported? Or skip?

# Original logic: resolved_path = possible_paths[0]

# But wait, original code logic was:

pass

if not resolved_path:

if called_name in local_names:

resolved_path = caller_file_path

is_unresolved_external = False # This is a local call, not external

elif called_name in imports_map and imports_map[called_name]:

# Check if any path in imports_map for called_name matches current file's imports

candidates = imports_map[called_name]

for path in candidates:

for imp_name in local_imports.values():

if imp_name.replace('.', '/') in path:

resolved_path = path

is_unresolved_external = False # Found a match

break

if resolved_path: break

if not resolved_path:

resolved_path = candidates[0]

else:

resolved_path = caller_file_path

# Skip creating CALLS relationship for unresolved external calls when skip_external is enabled

if skip_external and is_unresolved_external:

continue

caller_context = call.get('context')

if caller_context and len(caller_context) == 3 and caller_context[0] is not None:

caller_name, _, caller_line_number = caller_context

# KùzuDB workaround: Try Function->Function first, then other combinations

# This avoids polymorphic MERGE which KùzuDB doesn't support

call_params = self._sanitize_props({

'caller_name': caller_name,

'caller_file_path': caller_file_path,

'caller_line_number': caller_line_number,

'called_name': called_name,

'called_file_path': resolved_path,

'line_number': call['line_number'],

'args': call.get('args', []),

'full_call_name': call.get('full_name', called_name)

})

# Try Function caller -> Function callee

if not self._safe_run_create(session, """

OPTIONAL MATCH (caller:Function {name: $caller_name, path: $caller_file_path})

OPTIONAL MATCH (called:Function {name: $called_name, path: $called_file_path})

WITH caller, called

WHERE caller IS NOT NULL AND called IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(called)

RETURN count(*) as created

""", call_params):

# Try Function caller -> Class.__init__ / constructor

if not self._safe_run_create(session, """

OPTIONAL MATCH (caller:Function {name: $caller_name, path: $caller_file_path})

OPTIONAL MATCH (called:Class {name: $called_name, path: $called_file_path})

OPTIONAL MATCH (called)-[:CONTAINS]->(init:Function)

WHERE init.name IN ["__init__", "constructor"]

WITH caller, init

WHERE caller IS NOT NULL AND init IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(init)

RETURN count(*) as created

""", call_params):

# No __init__ found - link directly to the Class node

self._safe_run_create(session, """

OPTIONAL MATCH (caller:Function {name: $caller_name, path: $caller_file_path})

OPTIONAL MATCH (called:Class {name: $called_name, path: $called_file_path})

WITH caller, called

WHERE caller IS NOT NULL AND called IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(called)

RETURN count(*) as created

""", call_params)

# Try Class caller -> Function callee

if not self._safe_run_create(session, """

OPTIONAL MATCH (caller:Class {name: $caller_name, path: $caller_file_path})

OPTIONAL MATCH (called:Function {name: $called_name, path: $called_file_path})

WITH caller, called

WHERE caller IS NOT NULL AND called IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(called)

RETURN count(*) as created

""", call_params):

# Try Class caller -> Class.__init__ / constructor

if not self._safe_run_create(session, """

OPTIONAL MATCH (caller:Class {name: $caller_name, path: $caller_file_path})

OPTIONAL MATCH (called:Class {name: $called_name, path: $called_file_path})

OPTIONAL MATCH (called)-[:CONTAINS]->(init:Function)

WHERE init.name IN ["__init__", "constructor"]

WITH caller, init

WHERE caller IS NOT NULL AND init IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(init)

RETURN count(*) as created

""", call_params):

# No __init__ - link directly to the Class node

if not self._safe_run_create(session, """

OPTIONAL MATCH (caller:Class {name: $caller_name, path: $caller_file_path})

OPTIONAL MATCH (called:Class {name: $called_name, path: $called_file_path})

WITH caller, called

WHERE caller IS NOT NULL AND called IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(called)

RETURN count(*) as created

""", call_params):

# Fallback: Relaxed Global Search (Function caller -> any Function callee)

if not self._safe_run_create(session, """

OPTIONAL MATCH (caller:Function {name: $caller_name, path: $caller_file_path})

OPTIONAL MATCH (called:Function {name: $called_name})

WITH caller, called

WHERE caller IS NOT NULL AND called IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(called)

""", call_params):

# Fallback: Class caller -> any Function callee

self._safe_run_create(session, """

OPTIONAL MATCH (caller:Class {name: $caller_name, path: $caller_file_path})

OPTIONAL MATCH (called:Function {name: $called_name})

WITH caller, called

WHERE caller IS NOT NULL AND called IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(called)

""", call_params)

else:

# File-level calls: Try Function first, then Class

call_params = self._sanitize_props({

'caller_file_path': caller_file_path,

'called_name': called_name,

'called_file_path': resolved_path,

'line_number': call['line_number'],

'args': call.get('args', []),

'full_call_name': call.get('full_name', called_name)

})

if not self._safe_run_create(session, """

OPTIONAL MATCH (caller:File {path: $caller_file_path})

OPTIONAL MATCH (called:Function {name: $called_name, path: $called_file_path})

WITH caller, called

WHERE caller IS NOT NULL AND called IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(called)

RETURN count(*) as created

""", call_params):

# Try File caller -> Class.__init__ / constructor

if not self._safe_run_create(session, """

OPTIONAL MATCH (caller:File {path: $caller_file_path})

OPTIONAL MATCH (called:Class {name: $called_name, path: $called_file_path})

OPTIONAL MATCH (called)-[:CONTAINS]->(init:Function)

WHERE init.name IN ["__init__", "constructor"]

WITH caller, init

WHERE caller IS NOT NULL AND init IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(init)

RETURN count(*) as created

""", call_params):

# No __init__ - link directly to the Class node

if not self._safe_run_create(session, """

OPTIONAL MATCH (caller:File {path: $caller_file_path})

OPTIONAL MATCH (called:Class {name: $called_name, path: $called_file_path})

WITH caller, called

WHERE caller IS NOT NULL AND called IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(called)

RETURN count(*) as created

""", call_params):

# Fallback: Relaxed Global Search (File -> any Function)

self._safe_run_create(session, """

OPTIONAL MATCH (caller:File {path: $caller_file_path})

OPTIONAL MATCH (called:Function {name: $called_name})

WITH caller, called

WHERE caller IS NOT NULL AND called IS NOT NULL

MERGE (caller)-[:CALLS {line_number: $line_number, args: $args, full_call_name: $full_call_name}]->(called)

""", call_params)

def _create_all_function_calls(self, all_file_data: list[Dict], imports_map: dict):

"""Create CALLS relationships for all functions after all files have been processed."""

debug_log(f"_create_all_function_calls called with {len(all_file_data)} files")

with self.driver.session() as session:

for idx, file_data in enumerate(all_file_data):

debug_log(f"Processing file {idx+1}/{len(all_file_data)}: {file_data.get('path', 'unknown')}")

self._create_function_calls(session, file_data, imports_map)

def _create_inheritance_links(self, session, file_data: Dict, imports_map: dict):

"""Create INHERITS relationships with a more robust resolution logic."""

caller_file_path = str(Path(file_data['path']).resolve())

local_class_names = {c['name'] for c in file_data.get('classes', [])}

# Create a map of local import aliases/names to full import names

local_imports = {imp.get('alias') or imp['name'].split('.')[-1]: imp['name']

for imp in file_data.get('imports', [])}

for class_item in file_data.get('classes', []):

if not class_item.get('bases'):

continue

for base_class_str in class_item['bases']:

if base_class_str == 'object':

continue

resolved_path = None

target_class_name = base_class_str.split('.')[-1]

# Handle qualified names like module.Class or alias.Class

if '.' in base_class_str:

lookup_name = base_class_str.split('.')[0]

# Case 1: The prefix is a known import

if lookup_name in local_imports:

full_import_name = local_imports[lookup_name]

possible_paths = imports_map.get(target_class_name, [])

# Find the path that corresponds to the imported module

for path in possible_paths:

if full_import_name.replace('.', '/') in path:

resolved_path = path

break

# Handle simple names

else:

lookup_name = base_class_str

# Case 2: The base class is in the same file

if lookup_name in local_class_names:

resolved_path = caller_file_path

# Case 3: The base class was imported directly (e.g., from module import Parent)

elif lookup_name in local_imports:

full_import_name = local_imports[lookup_name]

possible_paths = imports_map.get(target_class_name, [])

for path in possible_paths:

if full_import_name.replace('.', '/') in path:

resolved_path = path

break

# Case 4: Fallback to global map (less reliable)

elif lookup_name in imports_map:

possible_paths = imports_map[lookup_name]

if len(possible_paths) == 1:

resolved_path = possible_paths[0]

# If a path was found, create the relationship

if resolved_path:

session.run("""

MATCH (child:Class {name: $child_name, path: $path})

MATCH (parent:Class {name: $parent_name, path: $resolved_parent_file_path})

MERGE (child)-[:INHERITS]->(parent)

""",

child_name=class_item['name'],

path=caller_file_path,

parent_name=target_class_name,

resolved_parent_file_path=resolved_path)

def _create_csharp_inheritance_and_interfaces(self, session, file_data: Dict, imports_map: dict):

"""Create INHERITS and IMPLEMENTS relationships for C# types."""

if file_data.get('lang') != 'c_sharp':

return

caller_file_path = str(Path(file_data['path']).resolve())

# Collect all local type names

local_type_names = set()

for type_list in ['classes', 'interfaces', 'structs', 'records']:

local_type_names.update(t['name'] for t in file_data.get(type_list, []))

# Process all type declarations that can have bases

for type_list_name, type_label in [('classes', 'Class'), ('structs', 'Struct'), ('records', 'Record'), ('interfaces', 'Interface')]:

for type_item in file_data.get(type_list_name, []):

if not type_item.get('bases'):

continue

for base_str in type_item['bases']:

# Clean up the base name (remove generic parameters, etc.)

base_name = base_str.split('<')[0].strip()

# Determine if this is an interface

is_interface = False

resolved_path = caller_file_path

# Check if base is a local interface

for iface in file_data.get('interfaces', []):

if iface['name'] == base_name:

is_interface = True