File size: 5,709 Bytes
32880eb | 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 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | import subprocess
import os
import shutil
import uuid
import time
import tempfile
from typing import List, Dict, Any, Optional
class CodeExecutor:
def __init__(self):
# Mapping languages to their extensions and execution commands
self.languages = {
"python": {"ext": "py", "cmd": "python3"},
"javascript": {"ext": "js", "cmd": "node"},
"cpp": {"ext": "cpp", "compiler": "g++", "out": "code_out"},
"c": {"ext": "c", "compiler": "gcc", "out": "code_out"},
"java": {"ext": "java", "compiler": "javac", "class_name": "Main"}
}
def _resolve_binary(self, binary: str) -> str:
"""Checks if a binary exists in the system path."""
path = shutil.which(binary)
if not path:
raise FileNotFoundError(f"Compiler/Runtime missing in container: {binary}")
return path
def run(self, language: str, code: str, test_cases: List[Dict[str, str]], time_limit: float = 2.0) -> List[Dict[str, Any]]:
if language not in self.languages:
return [{"status": "Error", "message": f"Language {language} not supported"}]
lang_config = self.languages[language]
results = []
# Create a unique sandbox directory for this execution
temp_dir = tempfile.mkdtemp(prefix=f"exec_{uuid.uuid4().hex}_")
try:
# 1. Write the code to a file
file_name = f"Main.{lang_config['ext']}" if language == "java" else f"solution.{lang_config['ext']}"
file_path = os.path.join(temp_dir, file_name)
with open(file_path, "w") as f:
f.write(code)
# 2. Compile Phase (if applicable)
exec_path = file_path
if "compiler" in lang_config:
compiler_bin = self._resolve_binary(lang_config["compiler"])
if language in ["cpp", "c"]:
out_path = os.path.join(temp_dir, lang_config["out"])
compile_cmd = [compiler_bin, file_path, "-o", out_path]
exec_path = out_path
elif language == "java":
compile_cmd = [compiler_bin, file_path]
# Java exec path remains the class name, but we run from temp_dir
exec_path = "Main"
compilation = subprocess.run(
compile_cmd,
capture_output=True,
text=True,
timeout=10
)
if compilation.returncode != 0:
return [{
"status": "Compilation Error",
"stderr": compilation.stderr,
"stdout": compilation.stdout
}]
# 3. Execution Phase (Loop through test cases)
for i, tc in enumerate(test_cases):
input_data = tc.get("input", "")
expected_output = tc.get("expected", "").strip()
run_cmd = []
if language == "python":
run_cmd = [self._resolve_binary("python3"), file_path]
elif language == "javascript":
run_cmd = [self._resolve_binary("node"), file_path]
elif language == "java":
run_cmd = [self._resolve_binary("java"), "-cp", temp_dir, "Main"]
else: # C/C++
run_cmd = [exec_path]
start_time = time.perf_counter()
try:
process = subprocess.run(
run_cmd,
input=input_data,
capture_output=True,
text=True,
timeout=time_limit,
cwd=temp_dir # Run inside the sandbox
)
end_time = time.perf_counter()
actual_output = process.stdout.strip()
execution_time = round(end_time - start_time, 4)
if process.returncode != 0:
results.append({
"test_case": i,
"status": "Runtime Error",
"stderr": process.stderr,
"time": execution_time
})
elif actual_output == expected_output:
results.append({
"test_case": i,
"status": "Accepted",
"time": execution_time
})
else:
results.append({
"test_case": i,
"status": "Wrong Answer",
"expected": expected_output,
"actual": actual_output,
"time": execution_time
})
except subprocess.TimeoutExpired:
results.append({
"test_case": i,
"status": "Time Limit Exceeded",
"time": time_limit
})
except Exception as e:
results.append({
"test_case": i,
"status": "Internal Error",
"message": str(e)
})
finally:
# Cleanup sandbox
shutil.rmtree(temp_dir, ignore_errors=True)
return results |