| from fastapi import FastAPI, Request |
| from fastapi.responses import HTMLResponse |
| import subprocess |
| import sys |
| import io |
| import traceback |
|
|
| app = FastAPI() |
|
|
| notebook_globals = {} |
|
|
| @app.get("/", response_class=HTMLResponse) |
| async def read_root(): |
| with open("index.html", "r", encoding="utf-8") as f: |
| return f.read() |
|
|
| @app.post("/run_code") |
| async def run_code(request: Request): |
| data = await request.json() |
| code = data.get("code", "") |
| user_inputs = data.get("inputs", []) |
| |
| input_iterator = iter(user_inputs) |
| |
| |
| def mock_input(prompt=""): |
| print(prompt, end="") |
| try: |
| val = next(input_iterator) |
| print(val) |
| return val |
| except StopIteration: |
| return "" |
|
|
| old_stdout = sys.stdout |
| redirected_output = sys.stdout = io.StringIO() |
| |
| |
| import builtins |
| old_input = builtins.input |
| builtins.input = mock_input |
| |
| try: |
| exec(code, notebook_globals) |
| output = redirected_output.getvalue() |
| except Exception as e: |
| output = redirected_output.getvalue() + "\n--- ERROR ---\n" + traceback.format_exc() |
| finally: |
| |
| sys.stdout = old_stdout |
| builtins.input = old_input |
| |
| return {"output": output} |
|
|
| @app.post("/run_terminal") |
| async def run_terminal(request: Request): |
| data = await request.json() |
| cmd = data.get("cmd", "") |
| |
| try: |
| result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30) |
| return {"output": result.stdout + result.stderr} |
| except Exception as e: |
| return {"output": f"Command failed: {str(e)}"} |
|
|