mirror of
https://github.com/ParisNeo/lollms-webui.git
synced 2024-12-18 20:17:50 +00:00
upgraed execution code
This commit is contained in:
parent
1ddbac72d8
commit
e6bcfe5f86
30
tests/execution/test_execution.py
Normal file
30
tests/execution/test_execution.py
Normal file
@ -0,0 +1,30 @@
|
||||
import subprocess
|
||||
import platform
|
||||
from pathlib import Path
|
||||
from ascii_colors import get_trace_exception, trace_exception
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Create a temporary file.
|
||||
code = "print('Hello world');input('hi')"
|
||||
message_id=102
|
||||
root_folder = Path(r"E:\lollms\discussion_databases\html stuff\105")
|
||||
root_folder.mkdir(parents=True, exist_ok=True)
|
||||
tmp_file = root_folder / f"ai_code_{message_id}.py"
|
||||
with open(tmp_file, "w", encoding="utf8") as f:
|
||||
f.write(code)
|
||||
|
||||
try:
|
||||
# Determine the platform and open a terminal to execute the Python code.
|
||||
system = platform.system()
|
||||
if system == "Windows":
|
||||
subprocess.Popen(f"""start cmd /k "cd /d {root_folder} && python {tmp_file} && pause" """, shell=True)
|
||||
elif system == "Darwin": # macOS
|
||||
subprocess.Popen(["open", "-a", "Terminal", f'cd "{root_folder}" && python "{tmp_file}"'], shell=True)
|
||||
elif system == "Linux":
|
||||
subprocess.Popen(["x-terminal-emulator", "-e", f'bash -c "cd \\"{root_folder}\\" && python \\"{tmp_file}\\"; exec bash"'], shell=True)
|
||||
else:
|
||||
raise Exception(f"Unsupported platform: {system}")
|
||||
|
||||
except Exception as ex:
|
||||
error_message = f"Error executing Python code: {ex}"
|
||||
error_json = {"output": "<div class='text-red-500'>" + error_message + "\n" + get_trace_exception(ex) + "</div>", "execution_time": 0}
|
@ -13,10 +13,71 @@ import time
|
||||
import subprocess
|
||||
import json
|
||||
from lollms.client_session import Client
|
||||
|
||||
import platform
|
||||
from pathlib import Path
|
||||
lollmsElfServer:LOLLMSWebUI = LOLLMSWebUI.get_instance()
|
||||
|
||||
def execute_python(code, client:Client, message_id, build_file=True):
|
||||
|
||||
def execute_python(code, client, message_id, build_file=True):
|
||||
def spawn_process(code):
|
||||
"""Executes Python code in a new terminal and returns the output as JSON."""
|
||||
|
||||
# Start the timer.
|
||||
start_time = time.time()
|
||||
|
||||
# Create a temporary file.
|
||||
root_folder = client.discussion.discussion_folder
|
||||
root_folder.mkdir(parents=True, exist_ok=True)
|
||||
tmp_file = root_folder / f"ai_code_{message_id}.py"
|
||||
with open(tmp_file, "w", encoding="utf8") as f:
|
||||
f.write(code)
|
||||
|
||||
try:
|
||||
# Determine the platform and open a terminal to execute the Python code.
|
||||
system = platform.system()
|
||||
if system == "Windows":
|
||||
process = subprocess.Popen(f"""start cmd /k "cd /d "{root_folder}" && python "{tmp_file}" && pause" """, shell=True)
|
||||
elif system == "Darwin": # macOS
|
||||
process = subprocess.Popen(["open", "-a", "Terminal", f'cd "{root_folder}" && python "{tmp_file}"'], shell=True)
|
||||
elif system == "Linux":
|
||||
process = subprocess.Popen(["x-terminal-emulator", "-e", f'bash -c "cd \\"{root_folder}\\" && python \\"{tmp_file}\\"; exec bash"'], shell=True)
|
||||
else:
|
||||
raise Exception(f"Unsupported platform: {system}")
|
||||
|
||||
# Wait for the process to complete.
|
||||
process.wait()
|
||||
|
||||
# Get the output and error from the process.
|
||||
output, error = process.communicate()
|
||||
except Exception as ex:
|
||||
# Stop the timer.
|
||||
execution_time = time.time() - start_time
|
||||
error_message = f"Error executing Python code: {ex}"
|
||||
error_json = {"output": "<div class='text-red-500'>" + error_message + "\n" + get_trace_exception(ex) + "</div>", "execution_time": execution_time}
|
||||
return error_json
|
||||
|
||||
# Stop the timer.
|
||||
execution_time = time.time() - start_time
|
||||
|
||||
# Check if the process was successful.
|
||||
if process.returncode != 0:
|
||||
# The child process threw an exception.
|
||||
try:
|
||||
error_message = f"Output: {output.decode('utf-8', errors='ignore')}\nError executing Python code:\n{error.decode('utf-8', errors='ignore')}"
|
||||
except:
|
||||
error_message = f"Error executing Python code:\n{error}"
|
||||
error_json = {"output": "<div class='text-red-500'>" + error_message + "</div>", "execution_time": execution_time}
|
||||
return error_json
|
||||
|
||||
# The child process was successful.
|
||||
output_json = {"output": output.decode("utf8"), "execution_time": execution_time}
|
||||
return output_json
|
||||
|
||||
return spawn_process(code)
|
||||
|
||||
|
||||
|
||||
def execute_python_old(code, client:Client, message_id, build_file=True):
|
||||
def spawn_process(code):
|
||||
"""Executes Python code and returns the output as JSON."""
|
||||
|
||||
@ -65,3 +126,37 @@ def execute_python(code, client:Client, message_id, build_file=True):
|
||||
output_json = {"output": output.decode("utf8"), "execution_time": execution_time}
|
||||
return output_json
|
||||
return spawn_process(code)
|
||||
|
||||
|
||||
def create_and_execute_script(code, message_id, root_folder):
|
||||
try:
|
||||
# Ensure the root folder exists
|
||||
root_folder = Path(root_folder)
|
||||
root_folder.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create the temporary Python file
|
||||
tmp_file = root_folder / f"ai_code_{message_id}.py"
|
||||
with open(tmp_file, "w", encoding="utf8") as f:
|
||||
f.write(code)
|
||||
|
||||
# Determine the platform and open a terminal to execute the Python code
|
||||
system = platform.system()
|
||||
if system == "Windows":
|
||||
subprocess.Popen(f"""start cmd /k "cd /d "{root_folder}" && python "{tmp_file}" && pause" """, shell=True)
|
||||
elif system == "Darwin": # macOS
|
||||
subprocess.Popen(["open", "-a", "Terminal", f'cd "{root_folder}" && python "{tmp_file}"'], shell=True)
|
||||
elif system == "Linux":
|
||||
subprocess.Popen(["x-terminal-emulator", "-e", f'bash -c "cd \\"{root_folder}\\" && python \\"{tmp_file}\\"; exec bash"'], shell=True)
|
||||
else:
|
||||
raise Exception(f"Unsupported platform: {system}")
|
||||
|
||||
except Exception as ex:
|
||||
error_message = f"Error executing Python code: {ex}"
|
||||
error_json = {"output": "<div class='text-red-500'>" + error_message + "\n" + get_trace_exception(ex) + "</div>", "execution_time": 0}
|
||||
print(error_json)
|
||||
|
||||
if __name__ == "__main__":
|
||||
code = "print('Hello world');input('hi')"
|
||||
message_id = 102
|
||||
root_folder = r"E:\lollms\discussion_databases\html stuff\105"
|
||||
create_and_execute_script(code, message_id, root_folder)
|
Loading…
Reference in New Issue
Block a user