Add lru-cashe

repo_agent/parsers/calls_parser.py

@@ -1,4 +1,5 @@
 import os
+from functools import lru_cache
 from collections import defaultdict
 from tree_sitter import Node  # type: ignore
 
@@ -55,13 +56,11 @@ def walk(node: Node, parent_func=None):
             }
 
             if node.type in call_node_types.get(language, set()):
-                if language == "java":
-                    call_name = self._extract_function_name(node, code)
-                elif language == "kotlin":
+                if language in {"java", "kotlin"}:
                     call_name = self._extract_function_name(node, code)
                 else:
-                    function_name_node = node.child_by_field_name("function")
-                    call_name = self._extract_function_name(function_name_node, code)
+                    func_node = node.child_by_field_name("function")
+                    call_name = self._extract_function_name(func_node, code)
 
                 if parent_func and call_name:
                     calls.append((parent_func, call_name))
@@ -70,13 +69,14 @@ def walk(node: Node, parent_func=None):
                 walk(child, parent_func)
 
         walk(root)
-        # Возвращаем список с функциями и вызовами + путь
+
         rel_path = os.path.relpath(file_path, self.repo_path)
         return [(name, start, end, rel_path) for name, start, end in functions], calls
 
-    def _extract_function_name(self, node: Node, code: bytes) -> str:
+    def _extract_function_name(self, node: Node, code: bytes) -> str | None:
         if node is None:
-            return None  # type: ignore
+            return None
+
         if node.type in ("selector_expression", "member_expression"):
             left = self._extract_function_name(
                 node.child_by_field_name("object")
@@ -87,65 +87,71 @@ def _extract_function_name(self, node: Node, code: bytes) -> str:
                 node.child_by_field_name("name") or node.child_by_field_name("field"),
                 code,
             )
-            return f"{left}.{right}" if left and right else None  # type: ignore
+            return f"{left}.{right}" if left and right else None
+
         elif node.type == "method_invocation":
             name_node = node.child_by_field_name("name")
-            return self._get_node_text(name_node, code) if name_node else None  # type: ignore
+            return self._get_node_text(name_node, code) if name_node else None
+
         elif node.type == "call_expression":
-            # Специальная обработка Kotlin вызова функции
             for child in node.children:
                 if child.type == "identifier":
-                    return self._get_node_text(child, code)  # type: ignore
-            return None  # type: ignore
+                    return self._get_node_text(child, code)
+            return None
+
         elif node.type == "identifier":
-            return self._get_node_text(node, code)  # type: ignore
+            return self._get_node_text(node, code)
+
         else:
-            return self._get_node_text(node, code)  # type: ignore
+            return self._get_node_text(node, code)
 
-    def _get_node_text(self, node, code: bytes):
+    def _get_node_text(self, node: Node, code: bytes) -> str | None:
         if not node:
             return None
-        # Берём срез из байтов и декодируем в строку
-        return code[node.start_byte : node.end_byte].decode("utf8")
+        return code[node.start_byte:node.end_byte].decode("utf-8")
 
-    def process_file(self, file_path: str, language: str):
+    @lru_cache(maxsize=256)
+    def _load_ast_and_code(self, file_path: str, language: str) -> tuple[Node, bytes]:
         parser = TreeSitterParser(language)
         root = parser.parse_file(file_path)
-        with open(file_path, "r", encoding="utf-8") as f:
+        with open(file_path, "rb") as f:
             code = f.read()
-        functions, calls = self.extract_functions_and_calls(
-            root, language, code.encode("utf-8"), file_path
-        )
-        for func_name, start_line, end_line, rel_path in functions:
-            self.call_graph[func_name]["location"] = {  # type: ignore
-                "file": rel_path,
-                "start_line": start_line,
-                "end_line": end_line,
-            }
-        for caller, callee in calls:
-            self.call_graph[caller]["calls"].add(callee)
-            self.call_graph[callee]["called_by"].add(caller)
+        return root, code
+
+    def process_file(self, file_path: str, language: str):
+        try:
+            root, code = self._load_ast_and_code(file_path, language)
+            functions, calls = self.extract_functions_and_calls(
+                root, language, code, file_path
+            )
+            for func_name, start_line, end_line, rel_path in functions:
+                self.call_graph[func_name]["location"] = {
+                    "file": rel_path,
+                    "start_line": start_line,
+                    "end_line": end_line,
+                }
+            for caller, callee in calls:
+                self.call_graph[caller]["calls"].add(callee)
+                self.call_graph[callee]["called_by"].add(caller)
+
+        except Exception as e:
+            print(f"[!] Failed to process {file_path}: {e}")
 
     def build_from_repo(self):
-        for root, _, files in os.walk(self.repo_path):
+        for root_dir, _, files in os.walk(self.repo_path):
             for file in files:
                 ext = os.path.splitext(file)[1]
-                if ext == ".py":
-                    lang = "python"
-                elif ext == ".go":
-                    lang = "go"
-                elif ext == ".java":
-                    lang = "java"
-                elif ext == ".kt":
-                    lang = "kotlin"
-                else:
+                lang = {
+                    ".py": "python",
+                    ".go": "go",
+                    ".java": "java",
+                    ".kt": "kotlin"
+                }.get(ext)
+                if not lang:
                     continue
 
-                path = os.path.join(root, file)
-                try:
-                    self.process_file(path, lang)
-                except Exception as e:
-                    print(f"Failed to process {path}: {e}")
+                path = os.path.join(root_dir, file)
+                self.process_file(path, lang)
 
     def get_call_graph(self):
         return self.call_graph

repo_agent/parsers/file_parser.py

@@ -1,3 +1,4 @@
+from functools import lru_cache
 from tree_sitter import Parser, Language  # type: ignore
 import tree_sitter_python as tspython  # type: ignore
 import tree_sitter_java as tsjava  # type: ignore
@@ -58,7 +59,7 @@ def __init__(self, repo_path, file_path):
 
         setting = SettingsManager.get_setting()
         self.project_hierarchy = (
-            setting.project.target_repo / setting.project.hierarchy_name  # type: ignore
+                setting.project.target_repo / setting.project.hierarchy_name  # type: ignore
         )
 
         # Determine language from file extension
@@ -74,13 +75,15 @@ def __init__(self, repo_path, file_path):
         self.code = None
         self.root = None
 
+    @lru_cache(maxsize=100)
     def _detect_language(self, file_path):
-        """Detect programming language from file extension"""
+        """Detect programming language from file extension with caching"""
         _, ext = os.path.splitext(file_path)
         return EXTENSION_TO_LANGUAGE.get(ext.lower())
 
+    @lru_cache(maxsize=1)
     def read_file(self):
-        """Read file content"""
+        """Read file content with caching (file won't change during processing)"""
         abs_file_path = os.path.join(self.repo_path, self.file_path)
         with open(abs_file_path, "r", encoding="utf-8") as file:
             content = file.read()
@@ -119,17 +122,19 @@ def get_modified_file_versions(self):
 
         return current_version, previous_version
 
+    @lru_cache(maxsize=32)
     def parse_code(self, code: str):
-        """Parse code using tree-sitter"""
+        """Parse code using tree-sitter with caching"""
         if not self.parser:
             return None
         self.code = code
         tree = self.parser.parse(code.encode("utf8"))
         self.root = tree.root_node
         return self.root
 
+    @lru_cache(maxsize=100)
     def parse_file(self, filename: str):
-        """Parse file using tree-sitter"""
+        """Parse file using tree-sitter with caching"""
         with open(os.path.join(self.repo_path, filename), "r", encoding="utf-8") as f:
             code = f.read()
         return self.parse_code(code)
@@ -224,8 +229,9 @@ def _extract_kotlin_parameters(self, node):
                 return params
         return []
 
+    @lru_cache(maxsize=1)
     def get_functions_and_classes(self):
-        """Extract functions and classes from parsed code"""
+        """Extract functions and classes from parsed code with caching"""
         if not self.root or not self.language_name:
             return []
 
@@ -278,7 +284,7 @@ def walk(node, parent_name=None):
         return result
 
     def get_obj_code_info(
-        self, code_type, code_name, start_line, end_line, params, file_path=None
+            self, code_type, code_name, start_line, end_line, params, file_path=None
     ):
         code_info = {}
         code_info["type"] = code_type
@@ -291,12 +297,12 @@ def get_obj_code_info(
         target_file_path = file_path if file_path is not None else self.file_path
 
         with open(
-            os.path.join(self.repo_path, target_file_path),
-            "r",
-            encoding="utf-8",
+                os.path.join(self.repo_path, target_file_path),
+                "r",
+                encoding="utf-8",
         ) as code_file:
             lines = code_file.readlines()
-            code_content = "".join(lines[start_line - 1 : end_line])
+            code_content = "".join(lines[start_line - 1: end_line])
             name_column = lines[start_line - 1].find(code_name) if lines else 0
 
             have_return = "return" in code_content
@@ -307,8 +313,9 @@ def get_obj_code_info(
 
         return code_info
 
+    @lru_cache(maxsize=100)
     def generate_file_structure(self, file_path):
-        """Generate structure for a single file"""
+        """Generate structure for a single file with caching"""
         # Detect language for this specific file
         language = self._detect_language(file_path)
 
@@ -360,13 +367,16 @@ def _fallback_file_structure(self, file_path):
         return []
 
     def generate_overall_structure(self, file_path_reflections, jump_files) -> dict:
-        """Generate structure for entire repository"""
+        """Generate structure for entire repository with custom caching"""
         repo_structure = {}
         gitignore_checker = GitignoreChecker(
             directory=self.repo_path,
             gitignore_path=os.path.join(self.repo_path, ".gitignore"),
         )
 
+        # Custom cache for file processing results
+        file_cache = {}
+
         bar = tqdm(gitignore_checker.check_files_and_folders())
         for not_ignored_files in bar:
             normal_file_names = not_ignored_files
@@ -380,10 +390,17 @@ def generate_overall_structure(self, file_path_reflections, jump_files) -> dict:
                     f"{Fore.LIGHTYELLOW_EX}[TreeSitter-Parser] Skip Latest Version, Using Git-Status Version]: {Style.RESET_ALL}{normal_file_names}"
                 )
                 continue
+
+            # Check cache before processing
+            if normal_file_names in file_cache:
+                repo_structure[normal_file_names] = file_cache[normal_file_names]
+                continue
+
             try:
-                repo_structure[normal_file_names] = self.generate_file_structure(
-                    not_ignored_files
-                )
+                result = self.generate_file_structure(not_ignored_files)
+                repo_structure[normal_file_names] = result
+                # Cache the result
+                file_cache[normal_file_names] = result
             except Exception as e:
                 logger.error(
                     f"Alert: An error occurred while generating file structure for {not_ignored_files}: {e}"
@@ -444,4 +461,4 @@ def convert_to_markdown_file(self, file_path=None):
             markdown += f"{md_content[-1] if len(md_content) > 0 else ''}\n"
         markdown += "***\n"
 
-        return markdown
+        return markdown