Create executor.py

executor.py

@@ -0,0 +1,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