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