from importlib.metadata import files

from pathlib import Path

from typing import Any, Dict, Optional, Tuple

import re

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

from codegraphcontext.utils.tree_sitter_manager import execute_query

HASKELL_QUERIES = {

"functions":"""

[

(function_declaration

name: (simple_identifier) @name

parameters: (parameters)* @params

) @function_node

(init_declaration

parameters: (parameter)* @params

) @init_node

]

""",

"classes":"""

[

(class_declaration

name: (type_identifier) @name

) @class

(

struct_declaration

name: (type_identifier) @name

) @struct

(

enum_declaration

name: (type_identifier) @name

) @enum

(

protocol_declaration

name: (type_identifier) @name

) @protocol

]

""",

"imports":"""

(import_declaration) @import

""",

"calls":"""

(call_expression) @call_node

""",

"variables":"""

(property_declaration

(variable_declaration

(simple_identifier) @name

)

) @variable

""",

}

class HaskellTreeSitterParser:

def __init__(self,generic_parser_wrapper: Any):

self.generic_parser_wrapper= generic_parser_wrapper

self.language_name="haskell"

self.language= generic_parser_wrapper.language

self.parser= generic_parser_wrapper.parser

def parse(self,path: Path, is_dependency: bool= False, index_source: bool= False)->Dict[str,Any]:

try:

self.index_source = index_source

with open(path, 'r', encoding='utf-8', errors='ignore') as f:

source_code= f.read()

if not source_code.strip():

warning_logger(f"Empty or whitespace-only file: {path}")

return {

"path":str(path),

"functions":[],

"classes":[],

"variables":[],

"imports": [],

"function_calls": [],

"is_dependency": is_dependency,

"lang": self.language_name,

}

tree = self.parser.parse(bytes(source_code, "utf8"))

parsed_functions = []

parsed_classes= []

parsed_variables= []

parsed_imports = []

parsed_calls = []

# parse variables first to populate for inference

if 'variables' in HASKELL_QUERIES:

results = execute_query(self.language, HASKELL_QUERIES['variables'], tree.root_node)

parsed_variables = self._parse_variables(results, source_code, path)

for capture_name, query in HASKELL_QUERIES.items():

if capture_name == 'variables' : continue

results = execute_query(self.language, query, tree.root_node)

if capture_name == "functions":

parsed_functions.extend(self._parse_functions(results, source_code, path))

elif capture_name == "classes":

parsed_classes.extend(self._parse_classes(results, source_code, path))

elif capture_name == "imports":

parsed_classes.extend(self._parse_imports(results, source_code))

elif capture_name == "calls":

parsed_classes.extend(self._parse_calls(results, source_code, path, parsed_variables))

return {

"path": str(path),

"functions": parsed_functions,

"classes": parsed_classes,

"variables": parsed_variables,

"imports": parsed_imports,

"function_calls": parsed_calls,

"is_dependency": is_dependency,

"lang": self.language_name,

}

except Exception as e:

error_logger(f"Error parsing Haskell file {path}: {e}")

return {

"path": str(path),

"functions": [],

"classes": [],

"variables": [],

"imports": [],

"function_calls": [],

"is_dependency": is_dependency,

"lang": self.language_name,

}

def _get_parent_context(self, node: Any) -> Tuple[Optional[str], Optional[str], Optional[int]]:

curr = node.parent

while curr:

if curr.type in ('class_declaration',):

name_node = None

for child in curr.children:

if child.type == "simple_identifier":

name_node = child

break

return (

self._get_node_text(name_node) if name_node else None,

curr.type,

curr.start_point[0] + 1

)

if curr.type in ("class_declaration", "object_declaration"):

for child in curr.children:

if child.type in ("simple_identifier", "type_identifier"):

return (

self._get_node_text(child),

curr.type,

curr.start_point[0] + 1

)

# check for secondary constructors

if curr.type == "secondary_constructor":

return (

"constructor",

curr.type,

curr.start_point[0] + 1

)

if curr.type == "companion_object":

name = "Companion"

for child in curr.children:

if child.type in ("simple_identifier", "type_identifier"):

name = self._get_node_text(child)

break

return (

name,

curr.type,

curr.start_point[0] + 1

)

# Handle anonymous objects (object_literal)

if curr.type == "object_literal":

# checking if it is assigned to a variable to get a name?

# or simply "AnonymousObject"

# it is usually hard to name them without variable context.

name = "AnonymousObject"

return (

name,

curr.type,

curr.start_point[0] + 1

)

curr = curr.parent

return None, None, None

def _get_node_text(self, node: Any) -> str:

if not node: return ""

return node.text.decode("utf-8")

def _parse_functions(self, captures: list, source_code: str, path: Path) -> list[Dict[str, Any]]:

functions = []

seen_nodes = set()

for node, capture_name in captures:

if capture_name == "function_node":

node_id =(node.start_byte, node.end_byte, node.type)

if node_id in seen_nodes:

continue

seen_nodes.add(node_id)

try:

start_line = node.start_point[0] + 1

end_line = node.end_point[0] + 1

#Manual child lookup

name_node = None

for child in node.children:

if child.type == "simple_identifier":

name_node = child

break

if name_node:

func_name= self._get_node_text(name_node)

params_node = None

for child in node.children:

if child.type == "function_value_parameters":

param_node = child

break

parameters = []

if params_node:

params_text = self._get_node_text(params_node)

parameters = self._extract_parameter_names(params_text)

source_text = self._get_node_text(node)

context_name, context_type, context_line = self._get_parent_context(node)

func_data = {

"name": func_name,

"args": parameters,

"line_number": start_line,

"end_line": end_line,

"path": str(path),

"lang": self.language_name,

"context": context_name,

"class_context": context_name if context_type and ("class" in context_type or "object" in context_type) else None

}

if self.index_source:

func_data["source"] = source_text

functions.append(func_data)

except Exception as e:

error_logger(f"Error parsing function in {path}: {e}")

continue

return functions

def _parse_classes(self, captures: list, source_code: str, path: Path) -> list[Dict[str, Any]]:

classes = []

seen_nodes = set()

for node, capture_name in captures:

if capture_name == "class":

node_id = (node.start_byte, node.end_byte, node.type)

if node_id in seen_nodes:

continue

seen_nodes.add(node_id)

try:

start_line = node.start_point[0] + 1

end_line = node.end_point[0] + 1

# Find name child (type_identifier or simple_identifier)

class_name = "Anonymous"

if node.type == "companion_object":

class_name = "Companion" # default name

for child in node.children:

if child.type in ("type_identifier","simple_identifier"):

class_name = self._get_node_text(child)

break

source_text = self._get_node_text(node)

bases = []

# Check for delegation specifiers

# class_declaration -> delegation_specifier

for child in node.children:

if child.type == "delegation_specifier":

# children: constructor_invocation or user_type -> type_identifier

# user_type -> type_identifier

for specifier in child.children:

# constructor_invocation -> user_type -> type_identifier

# user_type -> type_identifier

# We want text of the type

if specifier.type == "constructor_invocation":

# child 0 is typically user_type

for sub in specifier.children:

if sub.type == "user_type":

bases.append(self._get_node_text(sub))

elif specifier.type == "user_type":

bases.append(self._get_node_text(specifier))

elif specifier.type == "explicit_delegation":

# Not handling simple yet, uses 'by'

pass

class_data = {

"name": class_name,

"line_number": start_line,

"end_line": end_line,

"bases": bases,

"path": str(path),

"lang": self.language_name,

}

if self.index_source:

class_data["source"] = source_text

classes.append(class_data)

except Exception as e:

error_logger(f"Error parsing class in {path}: {e}")

continue

return classes

def _parse_variables(self, captures: list, source_code: str, path: Path) -> list[Dict[str, Any]]:

variables = []

seen_nodes = set()

for node, capture_name in captures:

if capture_name == "variable":

try:

start_line = node.start_point[0] + 1

ctx_name, ctx_type, ctx_line = self._get_parent_context(node)

# Destructuring declaration

if "destructuring" in node.type:

pass

#Regular property/variable

var_name = "unknown"

var_type = "Unknown"

var_decl = None

for child in node.children:

if child.type == "variable_declaration":

var_decl = child

break

if var_decl:

#Check for name and type in variable_declaration

for child in var_decl.children:

if child.type == "simple_identifier":

var_name = self._get_node_text(child)

if child.type == "user_type":

var_type = self._get_node_text(child)

# Attempt inference from initializer if type is unknown

if var_type == "Unknown":

# property_declaration -> expression (e.g. call_expression)

for child in node.children:

if child.type == "call_expression":

# call_expression -> simple_identifier (constructor)

for sub in child.children:

if sub.type == "simple_identifier":

var_type = self._get_node_text(sub)

break

if var_type != "Unknown": break

if var_name !=" unknown":

variables.append({

"name": var_name,

"type": var_type,

"line_number": start_line,

"path": str(path),

"lang": self.language_name,

"context": ctx_name,

"class_context": ctx_name if ctx_type and ("class" in ctx_type or "object" in ctx_type) else None

})

except Exception as e:

continue

return variables

def _parse_imports(self, captures:list, source_code: str) -> list[dict]:

imports = []

for node, capture_name in captures:

if capture_name == "import":

try:

text = self._get_node_text(node)

# remove 'import'

path = text.replace('import','').strip().split(' as ')[0].strip()

alias = None

if ' as ' in text:

alias = text.split(' as ')[1].strip()

imports.append({

"name": path,

"full_import_name": path,

"line_number": node.start_point[0]+1,

"alias": alias,

"context": (None, None),

"lang": self.language_name,

"is_dependency": False,

})

except Exception as e:

continue

return imports

def _parse_calls(self, captures: list, source_code: str, path: Path, variables: list[Dict[str,Any]]= []) -> list[Dict[str, Any]]:

calls = []

seen_calls = set()

# Index variables for fast lookup: (name, context) -> type

var_map = {}

for v in variables:

key = (v['name'], v['context'])

var_map[key] = v['type']

# Fallback for null context or partial match could be added

# For class props: (name, class_context) might work if local lookup fails?

for node, capture_name in captures:

if capture_name == "call_node":

try:

# navigation_expression check

start_line = node.start_point[0] + 1

# Get function name from call_expression

func_name = "unknown"

base_obj = None

# call_expression usually has children:

# simple_identifier (func name)

# or naviagtion_expression (obj.method)

# Heuristic for base object:

# If navigation_expression -> child[0] is base, child[1] is suffix (method)

# We need to look deeper into the call_expression structure.

# call_expression -> (simple_identifier)

# OR call_expression -> (navigation_expression (simple_identifier) (navigation_suffix (simple_identifier) ...))

# OR call_expression -> (navigation_expression (call_expression) ...) (chained)

# Simplified traversal to find the "function name" and "receiver"

#If it's a direct call: foo()

#If it's a method call: x.foo()

#Tree-sitter struct:

# (call_expression (simple_identifier) (call_suffix ...))

# (call_expression (navigation_expression (simple_identifier)(navigation_suffix (simple_identifier))) (call_suffix))

# -> name = 2nd simple_identifier, base = 1st simple_identifier

# Let's verify children

children = node.children

first_child = children[0]

if first_child.type == "simple_identifier":

call_name = self._get_node_text(first_child)

# No explicit base object

elif first_child.type == "navigation_expression":

# x.foo

# children: operand (x), operator (.), suffix (foo)

# Usually 3 children?

# Let's inspect nav expression children

nav_children = first_child.children

if len(nav_children) >= 2:

# operand is 0

operand = nav_children[0]

# last one is suffix?

suffix = nav_children[-1]

# Suffix usually contains the method name in a navigation_suffix node or directly?

# Suffix is (navigation_suffix (simple_identifier)) usually.

if suffix.type == "navigation_suffix":

for c in suffix.children:

if c.type == "simple_identifier":

call_name = self._get_node_text(c)

break

elif suffix.type == "simple_identifier":

call_name = self._get_node_text(suffix)

# Base object

base_obj = self._get_node_text(operand)

if call_name == "unknown":

continue

full_name = f"{base_obj}.{call_name}" if base_obj else call_name

ctx_name, ctx_type, ctx_line = self._get_parent_context(node)

# Interence

inferred_type = None

if base_obj:

# Lookup base_obj in variables

# Try exact context

inferred_type = var_map.get((base_obj, ctx_name))

if not inferred_type:

# Try class context if we are in a method

# This logic is approximate.

# If we are in method 'foo' of 'ClassA', and 'base_obj' refers to a property of 'ClassA',

# var_map entry would have context 'ClassA'

# But our 'variables' parsing puts context as 'ClassA' for props

# But 'ctx_name' here is 'foo'

# We need to know 'foo' is in 'ClassA'

#'get_parent_context' returns immediate parent.

pass

# Fallback: check global/file scope (context=None)

if not inferred_type:

inferred_type = var_map.get((base_obj, None))

# Fallback: check if any variable named base_obj exists

if not inferred_type:

for (vname, vctx), vtype in var_map.items():

if vname == base_obj:

inferred_type = vtype

break

calls.append({

"name": call_name,

"full_name": full_name,

"line_number": start_line,

"args": [], # Simplified

"inferred_obj_type": inferred_type,

"context": [None, ctx_type, ctx_line], # Keeping format compatible

"class_context": [None, None],

"lang": self.language_name,

"is_dependency": False

})

except Exception as e:

continue

return calls

def pre_scan_haskell(file: list[Path], parser_wrapper) -> dict:

name_to_files = {}

for path in files:

try:

with open(path, 'r', encoding='utf-8', errors='ignore') as f:

content = f.read()

# 1. Extract package

# package com.example.project

package_name = ""

pkg_match = re.search(r'^\s*package\s+([\w\.]+)', content, re.MULTILINE)

if pkg_match:

package_name = pkg_match.group(1)

# 2. Extract classes/objects/interfaces/typealiases

matches = re.finditer(

r'^\s*(class|object|interface|typealias)\s+(\w+)',

content,

)

for match in matches:

name = match.group(2)

# Map simple name

if name not in name_to_files:

name_to_files[name] = []

name_to_files[name].append(str(path))

# If package exists, map FQN

if package_name:

fqn = f"{package_name}.{name}"

if fqn not in name_to_files:

name_to_files[fqn] = []

name_to_files[fqn].append(str(path))

except Exception as e:

pass

return name_to_files