Spaces:
Sleeping
Sleeping
File size: 3,895 Bytes
0225ffc | 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 | import subprocess
import json
from fastapi import HTTPException
from sqlalchemy.orm import Session
from db import Question
from api.models.coding_models import RunCode
# =========================== Helper function to run user code =================
async def code_runner(body: RunCode, db: Session):
question = (
db.query(Question)
.filter(Question.id == body.question_id)
.first()
)
if not question:
raise HTTPException(
status_code=404,
detail="Question not found"
)
test_cases = question.test_cases
function_name = question.function_name
runner_code = body.code + "\n\n"
runner_code += """
import json
import inspect
# Helper to recursively convert PyTorch Tensors, NumPy arrays, etc. to clean Python structures
def to_list_recursive(val):
if hasattr(val, "tolist"):
return to_list_recursive(val.tolist())
if isinstance(val, tuple):
return [to_list_recursive(x) for x in val]
if isinstance(val, list):
return [to_list_recursive(x) for x in val]
if isinstance(val, dict):
return {k: to_list_recursive(v) for k, v in val.items()}
if hasattr(val, "item") and callable(getattr(val, "item")):
try:
return val.item()
except:
pass
return val
# Helper to automatically cast input list arguments based on function type hints
def prepare_args(func, raw_args):
try:
sig = inspect.signature(func)
params = list(sig.parameters.values())
except Exception:
return raw_args
prepared = []
for i, arg in enumerate(raw_args):
if i < len(params):
param = params[i]
annotation = param.annotation
annotation_str = str(annotation)
if "Tensor" in annotation_str or "tensor" in annotation_str:
try:
import torch
if isinstance(arg, (list, tuple)):
arg = torch.tensor(arg)
except Exception:
pass
elif "ndarray" in annotation_str or "array" in annotation_str:
try:
import numpy as np
if isinstance(arg, (list, tuple)):
arg = np.array(arg)
except Exception:
pass
prepared.append(arg)
return prepared
sol = Solution()
results = []
"""
for idx, tc in enumerate(test_cases):
args = repr(tc["test"])
expected = repr(tc["expected_output"])
runner_code += f"""
try:
test_input = {args}
prepared_input = prepare_args(sol.{function_name}, test_input)
output = sol.{function_name}(*prepared_input)
# Convert output recursively (handles single values, lists, and tuples of Tensors/Arrays)
converted_output = to_list_recursive(output)
passed = converted_output == {expected}
results.append({{
"test_case": {idx + 1},
"input": test_input,
"passed": passed,
"expected": {expected},
"got": converted_output
}})
except Exception as e:
results.append({{
"test_case": {idx + 1},
"input": {args},
"passed": False,
"error": str(e)
}})
"""
runner_code += """
print(json.dumps(results))
"""
with open("public/run.py", "w", encoding="utf-8") as f:
f.write(runner_code)
res = subprocess.run(
["python", "public/run.py"],
capture_output=True,
text=True,
timeout=10
)
if res.returncode != 0:
raise RuntimeError(res.stderr or res.stdout)
results = json.loads(res.stdout)
total = len(results)
passed = sum(1 for r in results if r.get("passed") is True)
status = "Accepted" if passed == total else "Failed"
return status, passed, total, results
|