Spaces:
Runtime error
Runtime error
File size: 2,823 Bytes
00226c3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
import subprocess, tempfile, shutil, traceback, importlib.util
import concurrent.futures
import os
class CodeExecutor:
def __init__(self):
self.temp_dir = tempfile.TemporaryDirectory()
self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
def _install_packages(self, packages):
for package in packages:
package = package.strip()
if not package:
continue
try:
spec = importlib.util.find_spec(package)
if spec is None:
subprocess.check_call(["pip", "install", package], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=15)
except subprocess.CalledProcessError as e:
raise Exception(f"❌ Error installing package {package}: {e}")
except subprocess.TimeoutExpired:
raise Exception(f"⏰ Timed out installing package {package}")
def _execute_code(self, code, inputs):
temp_file = f"{self.temp_dir.name}/user_code.py"
output_file = f"{self.temp_dir.name}/stdout.txt"
error_file = f"{self.temp_dir.name}/stderr.txt"
with open(temp_file, "w") as f:
f.write(code)
with open(output_file, "w") as out, open(error_file, "w") as err:
try:
process = subprocess.Popen(
["python", temp_file],
stdin=subprocess.PIPE,
stdout=out,
stderr=err,
text=True
)
if inputs:
for line in inputs:
process.stdin.write(line + "\n")
process.stdin.close()
process.wait(timeout=15)
except subprocess.TimeoutExpired:
process.kill()
err.write("⏰ Execution timed out.\n")
except Exception:
err.write(traceback.format_exc())
with open(output_file, "r") as out, open(error_file, "r") as err:
output_text = out.read().strip()
error_text = err.read().strip()
if error_text:
return f"⚠️ Error:\n{error_text}"
return f"✅ Output:\n{output_text}"
def execute(self, code, inputs, packages):
if packages:
self._install_packages(packages.split(","))
future = self.executor.submit(self._execute_code, code, inputs)
try:
return future.result(timeout=20)
except concurrent.futures.TimeoutError:
return "⏳ Code execution exceeded time limit."
except Exception as e:
return f"❌ Execution Error: {e}"
def __del__(self):
self.executor.shutdown(wait=False)
shutil.rmtree(self.temp_dir.name) |