CodeGraphContext

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 JAVA_QUERIES = { "functions": """ (method_declaration name: (identifier) @name parameters: (formal_parameters) @params ) @function_node (constructor_declaration name: (identifier) @name parameters: (formal_parameters) @params ) @function_node """, "classes": """ [ (class_declaration name: (identifier) @name) (interface_declaration name: (identifier) @name) (enum_declaration name: (identifier) @name) (annotation_type_declaration name: (identifier) @name) ] @class """, "imports": """ (import_declaration) @import """, "calls": """ (method_invocation name: (identifier) @name ) (object_creation_expression type: (type_identifier) @name ) """, } class JavaTreeSitterParser: def __init__(self, generic_parser_wrapper: Any): self.generic_parser_wrapper = generic_parser_wrapper self.language_name = "java" self.language = generic_parser_wrapper.language self.parser = generic_parser_wrapper.parser self.queries = { name: self.language.query(query_str) for name, query_str in JAVA_QUERIES.items() } def parse(self, file_path: Path, is_dependency: bool = False) -> Dict[str, Any]: try: with open(file_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: {file_path}") return { "file_path": str(file_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_imports = [] parsed_calls = [] for capture_name, query in self.queries.items(): captures = query.captures(tree.root_node) if capture_name == "functions": parsed_functions = self._parse_functions(captures, source_code, file_path) elif capture_name == "classes": parsed_classes = self._parse_classes(captures, source_code, file_path) elif capture_name == "imports": parsed_imports = self._parse_imports(captures, source_code) elif capture_name == "calls": parsed_calls = self._parse_calls(captures, source_code) return { "file_path": str(file_path), "functions": parsed_functions, "classes": parsed_classes, "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 Java file {file_path}: {e}") return { "file_path": str(file_path), "functions": [], "classes": [], "variables": [], "imports": [], "function_calls": [], "is_dependency": is_dependency, "lang": self.language_name, } def _parse_functions(self, captures: list, source_code: str, file_path: Path) -> list[Dict[str, Any]]: functions = [] source_lines = source_code.splitlines() for node, capture_name in captures: if capture_name == "function_node": try: start_line = node.start_point[0] + 1 end_line = node.end_point[0] + 1 name_captures = [ (n, cn) for n, cn in captures if cn == "name" and n.parent == node ] if name_captures: name_node = name_captures[0][0] func_name = source_code[name_node.start_byte:name_node.end_byte] params_captures = [ (n, cn) for n, cn in captures if cn == "params" and n.parent == node ] parameters = [] if params_captures: params_node = params_captures[0][0] params_text = source_code[params_node.start_byte:params_node.end_byte] parameters = self._extract_parameter_names(params_text) # Extract annotations applied to this function annotations = [] for n, cn in captures: if cn == "annotation" and n.parent == node: ann_name_node = n.child_by_field_name("name") if ann_name_node: ann_name = source_code[ann_name_node.start_byte:ann_name_node.end_byte] annotations.append(ann_name) source_text = source_code[node.start_byte:node.end_byte] functions.append({ "name": func_name, "parameters": parameters, "annotations":annotations, "line_number": start_line, "end_line": end_line, "source": source_text, "file_path": str(file_path), "lang": self.language_name, }) except Exception as e: error_logger(f"Error parsing function in {file_path}: {e}") continue return functions def _parse_classes(self, captures: list, source_code: str, file_path: Path) -> list[Dict[str, Any]]: classes = [] for node, capture_name in captures: if capture_name == "class": try: start_line = node.start_point[0] + 1 end_line = node.end_point[0] + 1 name_captures = [ (n, cn) for n, cn in captures if cn == "name" and n.parent == node ] if name_captures: name_node = name_captures[0][0] class_name = source_code[name_node.start_byte:name_node.end_byte] source_text = source_code[node.start_byte:node.end_byte] classes.append({ "name": class_name, "line_number": start_line, "end_line": end_line, "source": source_text, "file_path": str(file_path), "lang": self.language_name, }) except Exception as e: error_logger(f"Error parsing class in {file_path}: {e}") continue return classes def _parse_imports(self, captures: list, source_code: str) -> list[dict]: imports = [] for node, capture_name in captures: if capture_name == "import": try: import_text = source_code[node.start_byte:node.end_byte] import_match = re.search(r'import\s+(?:static\s+)?([^;]+)', import_text) if import_match: import_path = import_match.group(1).strip() import_data = { "name": import_path, "full_import_name": import_path, "line_number": node.start_point[0] + 1, "alias": None, "context": (None, None), "lang": self.language_name, "is_dependency": False, } imports.append(import_data) except Exception as e: error_logger(f"Error parsing import: {e}") continue return imports def _parse_calls(self, captures: list, source_code: str) -> list[dict]: calls = [] seen_calls = set() for node, capture_name in captures: if capture_name == "name": try: call_name = source_code[node.start_byte:node.end_byte] line_number = node.start_point[0] + 1 # Avoid duplicates call_key = f"{call_name}_{line_number}" if call_key in seen_calls: continue seen_calls.add(call_key) call_data = { "name": call_name, "full_name": call_name, "line_number": line_number, "args": [], "inferred_obj_type": None, "context": (None, None), "class_context": (None, None), "lang": self.language_name, "is_dependency": False, } calls.append(call_data) except Exception as e: error_logger(f"Error parsing call: {e}") continue return calls def _extract_parameter_names(self, params_text: str) -> list[str]: params = [] if not params_text or params_text.strip() == "()": return params params_content = params_text.strip("()") if not params_content: return params for param in params_content.split(","): param = param.strip() if param: parts = param.split() if len(parts) >= 2: param_name = parts[-1] params.append(param_name) return params def pre_scan_java(files: list[Path], parser_wrapper) -> dict: name_to_files = {} for file_path in files: try: with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: content = f.read() class_matches = re.finditer(r'\b(?:public\s+|private\s+|protected\s+)?(?:static\s+)?(?:abstract\s+)?(?:final\s+)?class\s+(\w+)', content) for match in class_matches: class_name = match.group(1) if class_name not in name_to_files: name_to_files[class_name] = [] name_to_files[class_name].append(str(file_path)) interface_matches = re.finditer(r'\b(?:public\s+|private\s+|protected\s+)?interface\s+(\w+)', content) for match in interface_matches: interface_name = match.group(1) if interface_name not in name_to_files: name_to_files[interface_name] = [] name_to_files[interface_name].append(str(file_path)) except Exception as e: error_logger(f"Error pre-scanning Java file {file_path}: {e}") return name_to_files def _find_annotations(tree): # Detect annotation definitions like @interface CustomAnnotation annotations = [] for node in tree.find_all("annotation_type_declaration"): name = node.child_by_field_name("name").text annotations.append({ "type": "Annotation", "name": name, "location": node.start_point }) return annotations def _find_applied_annotations(tree): # Detect usages like @Entity, @Override applied = [] for node in tree.find_all("marker_annotation"): name = node.child_by_field_name("name").text applied.append({ "type": "AnnotationUsage", "name": name, "location": node.start_point }) return applied