| import subprocess |
| import os |
| import shutil |
| import uuid |
| import time |
| import tempfile |
| from typing import List, Dict, Any, Optional |
|
|
| class CodeExecutor: |
| def __init__(self): |
| |
| 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 = [] |
| |
| |
| temp_dir = tempfile.mkdtemp(prefix=f"exec_{uuid.uuid4().hex}_") |
| |
| try: |
| |
| 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) |
|
|
| |
| 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] |
| |
| 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 |
| }] |
|
|
| |
| 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: |
| 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 |
| ) |
| 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: |
| |
| shutil.rmtree(temp_dir, ignore_errors=True) |
|
|
| return results |
