Feat : better file system interaction
This commit is contained in:
@@ -16,6 +16,18 @@ class BashInterpreter(Tools):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.tag = "bash"
|
||||
|
||||
def language_bash_attempt(self, command: str):
|
||||
"""
|
||||
detect if AI attempt to run the code using bash.
|
||||
if so, return True, otherwise return False.
|
||||
The philosophy is that code written by the AI will be executed, so it should not use bash to run it.
|
||||
"""
|
||||
lang_interpreter = ["python3", "gcc", "g++", "go", "javac", "rustc", "clang", "clang++", "rustc", "rustc++", "rustc++"]
|
||||
for word in command.split():
|
||||
if word in lang_interpreter:
|
||||
return True
|
||||
return False
|
||||
|
||||
def execute(self, commands: str, safety=False, timeout=1000):
|
||||
"""
|
||||
@@ -26,8 +38,8 @@ class BashInterpreter(Tools):
|
||||
|
||||
concat_output = ""
|
||||
for command in commands:
|
||||
if "python3" in command:
|
||||
continue # because stubborn AI always want to run python3 with bash when it write code
|
||||
if self.language_bash_attempt(command):
|
||||
continue
|
||||
try:
|
||||
process = subprocess.Popen(
|
||||
command,
|
||||
|
||||
@@ -16,10 +16,6 @@ class FileFinder(Tools):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.tag = "file_finder"
|
||||
self.current_dir = os.path.dirname(os.getcwd())
|
||||
config = configparser.ConfigParser()
|
||||
config.read('./config.ini')
|
||||
self.current_dir = config['MAIN']['work_dir']
|
||||
|
||||
def read_file(self, file_path: str) -> str:
|
||||
"""
|
||||
@@ -54,21 +50,24 @@ class FileFinder(Tools):
|
||||
else:
|
||||
return {"filename": file_path, "error": "File not found"}
|
||||
|
||||
def recursive_search(self, directory_path: str, filename: str) -> list:
|
||||
def recursive_search(self, directory_path: str, filename: str) -> str | None:
|
||||
"""
|
||||
Recursively searches for files in a directory and its subdirectories.
|
||||
Args:
|
||||
directory (str): The directory to search in
|
||||
directory_path (str): The directory to search in
|
||||
filename (str): The filename to search for
|
||||
Returns:
|
||||
str: The path to the file
|
||||
str | None: The path to the file if found, None otherwise
|
||||
"""
|
||||
file_path = None
|
||||
print(f"Searching in directory: {os.path.abspath(directory_path)}")
|
||||
excluded_files = [".pyc", ".o", ".so", ".a", ".lib", ".dll", ".dylib", ".so", ".git"]
|
||||
for root, dirs, files in os.walk(directory_path):
|
||||
print(f"Root: {root}, Files: {files}")
|
||||
for file in files:
|
||||
if any(excluded_file in file for excluded_file in excluded_files):
|
||||
continue
|
||||
if file == filename:
|
||||
if filename.strip() in file.strip():
|
||||
file_path = os.path.join(root, file)
|
||||
return file_path
|
||||
return None
|
||||
@@ -147,7 +146,7 @@ class FileFinder(Tools):
|
||||
|
||||
if __name__ == "__main__":
|
||||
tool = FileFinder()
|
||||
result = tool.execute(["router.py:read"], False)
|
||||
result = tool.execute(["toto.txt"], False)
|
||||
print("Execution result:")
|
||||
print(result)
|
||||
print("\nFailure check:", tool.execution_failure_check(result))
|
||||
|
||||
+42
-2
@@ -23,6 +23,7 @@ HU787
|
||||
|
||||
import sys
|
||||
import os
|
||||
import configparser
|
||||
from abc import abstractmethod
|
||||
|
||||
sys.path.append('..')
|
||||
@@ -36,6 +37,43 @@ class Tools():
|
||||
self.api_key = None
|
||||
self.client = None
|
||||
self.messages = []
|
||||
self.config = configparser.ConfigParser()
|
||||
self.current_dir = self.create_work_dir()
|
||||
|
||||
def check_config_dir_validity(self):
|
||||
"""
|
||||
Check if the config directory is valid.
|
||||
"""
|
||||
path = self.config['MAIN']['work_dir']
|
||||
if path == "":
|
||||
print("WARNING: Work directory not set in config.ini")
|
||||
return False
|
||||
if path.lower() == "none":
|
||||
print("WARNING: Work directory set to none in config.ini")
|
||||
return False
|
||||
if not os.path.exists(path):
|
||||
print(f"WARNING: Work directory {path} does not exist")
|
||||
return False
|
||||
return True
|
||||
|
||||
def config_exists(self):
|
||||
"""
|
||||
Check if the config file exists.
|
||||
"""
|
||||
return os.path.exists('./config.ini')
|
||||
|
||||
def create_work_dir(self):
|
||||
"""
|
||||
Create the work directory if it does not exist.
|
||||
"""
|
||||
default_path = os.path.dirname(os.getcwd())
|
||||
if self.config_exists():
|
||||
self.config.read('./config.ini')
|
||||
config_path = self.config['MAIN']['work_dir']
|
||||
dir_path = default_path if not self.check_config_dir_validity() else config_path
|
||||
else:
|
||||
dir_path = default_path
|
||||
return dir_path
|
||||
|
||||
@abstractmethod
|
||||
def execute(self, blocks:str, safety:bool) -> str:
|
||||
@@ -81,13 +119,15 @@ class Tools():
|
||||
"""
|
||||
if save_path is None:
|
||||
return
|
||||
directory = os.path.dirname(save_path)
|
||||
save_path_dir = os.path.dirname(save_path)
|
||||
save_path_file = os.path.basename(save_path)
|
||||
directory = os.path.join(self.current_dir, save_path_dir)
|
||||
if directory and not os.path.exists(directory):
|
||||
print(f"Creating directory: {directory}")
|
||||
os.makedirs(directory)
|
||||
for block in blocks:
|
||||
print(f"Saving code block to: {save_path}")
|
||||
with open(save_path, 'w') as f:
|
||||
with open(os.path.join(directory, save_path_file), 'w') as f:
|
||||
f.write(block)
|
||||
|
||||
def load_exec_block(self, llm_text: str) -> tuple[list[str], str | None]:
|
||||
|
||||
Reference in New Issue
Block a user