added API for run time input

app.py

@@ -1,4 +1,3 @@
-import gradio as gr
 import sys
 import io
 import traceback
@@ -7,9 +6,13 @@ import time
 import json
 import os
 from contextlib import redirect_stdout, redirect_stderr
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 import threading
+import asyncio
 import signal
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
+from pydantic import BaseModel
+from starlette.responses import HTMLResponse
 
 # Set up proper signal handling for Docker
 def signal_handler(signum, frame):
@@ -19,6 +22,83 @@ def signal_handler(signum, frame):
 signal.signal(signal.SIGTERM, signal_handler)
 signal.signal(signal.SIGINT, signal_handler)
 
+# --- Custom IO Redirection for Interactive Execution ---
+
+class WebSocketInputOutput:
+    """
+    Redirects stdin/stdout/stderr to/from a WebSocket connection.
+    Handles interactive input by waiting for messages from the client.
+    """
+    def __init__(self, websocket: WebSocket):
+        self.websocket = websocket
+        self.input_queue = asyncio.Queue()
+        self.output_buffer = io.StringIO()
+        self.error_buffer = io.StringIO()
+        self.closed = False
+
+    async def write(self, s: str):
+        """Writes to stdout/stderr and sends to WebSocket."""
+        self.output_buffer.write(s)
+        try:
+            # Send output in chunks or lines to prevent large messages
+            await self.websocket.send_text(json.dumps({"type": "output", "content": s}))
+        except WebSocketDisconnect:
+            self.closed = True
+            print("WebSocket disconnected during write.")
+        except Exception as e:
+            print(f"Error sending output over WebSocket: {e}")
+            self.closed = True
+
+    def flush(self):
+        """Flushes the buffer (no-op for now, as we send immediately)."""
+        pass
+
+    async def readline(self) -> str:
+        """Reads a line from stdin, waiting for input from WebSocket."""
+        try:
+            await self.websocket.send_text(json.dumps({"type": "input_request"}))
+            line = await self.input_queue.get() # Wait for input from client
+            return line
+        except WebSocketDisconnect:
+            self.closed = True
+            print("WebSocket disconnected during readline.")
+            raise EOFError("Input stream closed due to WebSocket disconnect")
+        except Exception as e:
+            print(f"Error requesting input over WebSocket: {e}")
+            self.closed = True
+            raise EOFError(f"Input stream error: {e}")
+
+    # For compatibility with sys.stdin, which expects a file-like object
+    def _get_input_line(self):
+        """Synchronous wrapper for readline for use with exec."""
+        # This is a bit tricky: exec is synchronous, but WebSocket is async.
+        # We need to bridge this. A simple way is to run the async part
+        # in the event loop, but it's usually better to refactor the executor
+        # to be fully async if input() is truly interactive.
+        # For this example, we'll use a blocking call to asyncio.run
+        # which is generally discouraged in a running event loop,
+        # but works for simple cases or if the exec is in a separate thread.
+        # A more robust solution involves a custom Future/Event loop integration.
+        try:
+            # This will block the current thread until input is available
+            return asyncio.run(self.readline())
+        except Exception as e:
+            print(f"Synchronous input wrapper error: {e}")
+            raise
+
+    def read(self, n=-1):
+        """Reads n characters. For simplicity, we'll treat it as readline."""
+        return self._get_input_line() # Or implement more complex buffering
+
+    def __getattr__(self, name):
+        """Delegate other attributes if needed, or raise AttributeError."""
+        # This is a placeholder; real implementation would be more robust
+        # For now, we only need write, flush, and readline.
+        raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
+
+
+# --- Python Syntax Validation FSM (unchanged) ---
+
 class PythonSyntaxFSM:
     """Finite State Machine for Python syntax validation"""
     
@@ -153,9 +233,10 @@ class PythonSyntaxFSM:
                     })
                 break
 
+# --- Secure Python Executor (modified for WebSockets) ---
 
 class SecurePythonExecutor:
-    """Secure Python code executor with sandboxing and timeout"""
+    """Secure Python code executor with sandboxing and timeout, now interactive."""
     
     def __init__(self, timeout: int = 10):
         self.timeout = timeout
@@ -167,6 +248,7 @@ class SecurePythonExecutor:
         }
         
         # Create safe builtins
+        # IMPORTANT: 'input' is now included and will be redirected!
         self.safe_builtins = {
             'print': print,
             'len': len,
@@ -226,22 +308,25 @@ class SecurePythonExecutor:
             'AttributeError': AttributeError,
             'NameError': NameError,
             'ZeroDivisionError': ZeroDivisionError,
+            'input': input # Now included for interactive use
         }
         
-    def execute_code(self, code: str) -> Dict[str, Any]:
-        """Execute Python code in a secure environment"""
+    async def execute_code(self, code: str, websocket_io: WebSocketInputOutput) -> Dict[str, Any]:
+        """Execute Python code in a secure environment with WebSocket IO."""
         start_time = time.time()
         
         # Input validation
         if not code or not code.strip():
+            error_msg = 'No code provided'
+            await websocket_io.write(f"Error: {error_msg}\n")
             return {
                 'success': False,
                 'output': '',
-                'error': 'No code provided',
+                'error': error_msg,
                 'execution_time': 0,
                 'validation': {
                     'valid': False,
-                    'errors': [{'message': 'Empty code input', 'line': 0}],
+                    'errors': [{'message': error_msg, 'line': 0}],
                     'warnings': [],
                     'total_issues': 1
                 }
@@ -254,16 +339,18 @@ class SecurePythonExecutor:
         # Check for restricted imports
         restricted_check = self.check_restricted_imports(code)
         if not restricted_check['allowed']:
+            error_msg = f"Security violation: Restricted module '{restricted_check['module']}' is not allowed"
+            await websocket_io.write(f"Error: {error_msg}\n")
             return {
                 'success': False,
                 'output': '',
-                'error': f"Security violation: Restricted module '{restricted_check['module']}' is not allowed",
+                'error': error_msg,
                 'execution_time': 0,
                 'validation': validation_result
             }
         
         # Execute code with timeout
-        result = self.run_with_timeout(code)
+        result = await self.run_with_timeout(code, websocket_io) # Pass websocket_io
         execution_time = time.time() - start_time
         
         return {
@@ -290,153 +377,176 @@ class SecurePythonExecutor:
         except Exception:
             return {'allowed': True, 'module': None}   # If parsing fails, let execution handle it
     
-    def run_with_timeout(self, code: str) -> Dict[str, Any]:
-        """Run code with timeout protection"""
-        output_buffer = io.StringIO()
-        error_buffer = io.StringIO()
+    async def run_with_timeout(self, code: str, websocket_io: WebSocketInputOutput) -> Dict[str, Any]:
+        """Run code with timeout protection, interacting via WebSocketIO."""
         result = {'success': False, 'output': '', 'error': ''}
         
+        # Store original stdin/stdout/stderr
+        original_stdin = sys.stdin
+        original_stdout = sys.stdout
+        original_stderr = sys.stderr
+
+        # Use a threading.Event to signal completion from the execution thread
+        execution_finished_event = threading.Event()
+
         def target():
+            nonlocal result
             try:
-                with redirect_stdout(output_buffer), redirect_stderr(error_buffer):
-                    # Create restricted execution environment
-                    restricted_globals = {
-                        '__builtins__': self.safe_builtins,
-                        '__name__': '__main__',
-                        '__doc__': None,
-                    }
-                    
-                    # Execute the code
-                    exec(code, restricted_globals)
-                    result['success'] = True
-                    
+                # Redirect sys.stdin, sys.stdout, sys.stderr for this thread
+                sys.stdin = websocket_io
+                sys.stdout = websocket_io
+                sys.stderr = websocket_io
+
+                # Create restricted execution environment
+                restricted_globals = {
+                    '__builtins__': self.safe_builtins,
+                    '__name__': '__main__',
+                    '__doc__': None,
+                }
+                
+                # Execute the code
+                exec(code, restricted_globals)
+                result['success'] = True
+                
+            except EOFError as e:
+                # This happens if the WebSocket disconnects during input()
+                result['error'] = f"Input stream closed: {str(e)}"
+                result['success'] = False
             except Exception as e:
-                error_msg = f"{type(e).__name__}: {str(e)}"
-                error_buffer.write(error_msg)
+                error_msg = f"{type(e).__name__}: {str(e)}\n{traceback.format_exc()}"
+                # Write error to WebSocket and internal buffer
+                try:
+                    asyncio.run(websocket_io.write(f"Error: {error_msg}\n"))
+                except Exception as write_e:
+                    print(f"Failed to write error to WebSocket: {write_e}")
+                websocket_io.error_buffer.write(error_msg)
                 result['success'] = False
-        
-        # Run with timeout
+            finally:
+                # Restore original stdin/stdout/stderr for the thread pool (if applicable)
+                sys.stdin = original_stdin
+                sys.stdout = original_stdout
+                sys.stderr = original_stderr
+                execution_finished_event.set() # Signal completion
+
+        # Run the execution in a separate thread to allow for timeout
         thread = threading.Thread(target=target)
-        thread.daemon = True
+        thread.daemon = True # Allow the program to exit even if thread is running
         thread.start()
+
+        # Wait for the thread to finish or timeout
         thread.join(timeout=self.timeout)
         
         if thread.is_alive():
             result['error'] = f"Code execution timed out after {self.timeout} seconds"
             result['success'] = False
+            # Attempt to send timeout message to client
+            try:
+                await websocket_io.write(f"Error: {result['error']}\n")
+            except Exception:
+                pass # Ignore if websocket is already closed
         else:
-            result['output'] = output_buffer.getvalue()
-            if error_buffer.getvalue():
-                result['error'] = error_buffer.getvalue()
+            # Execution finished within timeout
+            result['output'] = websocket_io.output_buffer.getvalue()
+            if websocket_io.error_buffer.getvalue():
+                result['error'] = websocket_io.error_buffer.getvalue()
                 result['success'] = False
             else:
                 result['success'] = True
-                
+        
+        # Send a final message to the client indicating execution is complete
+        try:
+            await websocket_io.send_text(json.dumps({"type": "execution_complete", "result": result}))
+        except WebSocketDisconnect:
+            pass # Ignore if websocket is already closed
+        except Exception as e:
+            print(f"Error sending execution_complete message: {e}")
+
         return result
 
 
 # Initialize the executor
 executor = SecurePythonExecutor(timeout=10)
 
-def execute_python_code_api(code: str) -> Dict[str, Any]:
-    """
-    API function to execute Python code.
-    This function is directly exposed as an API endpoint.
-    """
-    try:
-        result = executor.execute_code(code)
-        return result
-    except Exception as e:
-        return {
-            'success': False,
-            'output': '',
-            'error': f"Execution engine error: {str(e)}",
-            'execution_time': 0,
-            'validation': {
-                'valid': False,
-                'errors': [{'message': str(e), 'line': 0, 'type': 'engine_error'}],
-                'warnings': [],
-                'total_issues': 1
-            }
-        }
+# Initialize FastAPI app
+app = FastAPI(
+    title="Interactive Python Code Executor API",
+    description="A secure WebSocket API for executing Python code interactively.",
+    version="1.0.0",
+)
 
-def health_check_api() -> Dict[str, Any]:
-    """
-    Health check API endpoint.
-    """
-    try:
-        test_result = execute_python_code_api("print('Health check OK')")
-        return {
-            'status': 'healthy' if test_result['success'] else 'unhealthy',
-            'timestamp': time.time(),
-            'test_execution_success': test_result['success']
-        }
-    except Exception as e:
-        return {
-            'status': 'unhealthy',
-            'timestamp': time.time(),
-            'error': str(e),
-            'test_execution_success': False
-        }
+# Serve the HTML frontend
+@app.get("/", response_class=HTMLResponse)
+async def get_index():
+    """Serves the interactive frontend."""
+    # This assumes index.html is in the same directory as app.py
+    with open("index.html", "r") as f:
+        return HTMLResponse(content=f.read())
 
-# Create the Gradio interface specifically for API exposure
-# We use gr.Interface for direct API endpoint mapping, not gr.Blocks for this scenario.
-# We also create a separate interface for the health check.
+# WebSocket endpoint for interactive execution
+@app.websocket("/ws")
+async def websocket_endpoint(websocket: WebSocket):
+    await websocket.accept()
+    websocket_io = WebSocketInputOutput(websocket)
+    print(f"WebSocket connection established: {websocket.client}")
 
-# Main code execution API
-api_code_executor = gr.Interface(
-    fn=execute_python_code_api,
-    inputs=[
-        gr.Code(label="Python Code", language="python", lines=10, value="print('Hello from API!')")
-    ],
-    outputs=[
-        gr.JSON(label="Execution Result")
-    ],
-    # To enable direct API calls without relying on implicit UI interactions
-    api_name="predict",
-    title="Python Code Executor API (Main)",
-    description="Submits Python code for secure execution and returns the result."
-)
+    try:
+        while True:
+            message = await websocket.receive_text()
+            data = json.loads(message)
 
-# Health check API
-api_health_check = gr.Interface(
-    fn=health_check_api,
-    inputs=[],
-    outputs=[
-        gr.JSON(label="Health Status")
-    ],
-    # You can give this a different API name if you want a separate endpoint
-    api_name="health",
-    title="Service Health Check",
-    description="Checks the health and basic functionality of the API service."
-)
+            if data["type"] == "code":
+                code = data["content"]
+                print(f"Received code from {websocket.client}: {code[:50]}...")
+                # Execute code asynchronously in the background
+                asyncio.create_task(executor.execute_code(code, websocket_io))
+            elif data["type"] == "input":
+                user_input = data["content"]
+                print(f"Received input from {websocket.client}: {user_input.strip()[:50]}...")
+                await websocket_io.input_queue.put(user_input + "\n") # Add newline for readline
+            elif data["type"] == "ping":
+                # Respond to pings to keep connection alive
+                await websocket.send_text(json.dumps({"type": "pong"}))
+            else:
+                await websocket_io.write(f"Unknown message type: {data['type']}\n")
 
+    except WebSocketDisconnect:
+        print(f"WebSocket disconnected: {websocket.client}")
+    except json.JSONDecodeError:
+        print(f"Received invalid JSON from {websocket.client}: {message}")
+    except Exception as e:
+        print(f"WebSocket error for {websocket.client}: {e}")
+    finally:
+        websocket_io.closed = True
+        # Clean up any pending input requests if the client disconnects
+        while not websocket_io.input_queue.empty():
+            try:
+                websocket_io.input_queue.get_nowait()
+            except asyncio.QueueEmpty:
+                pass
+        print(f"WebSocket connection closed for {websocket.client}")
 
+# This block is for local development using `uvicorn`
+# Hugging Face Spaces will use `uvicorn` directly via the Dockerfile CMD
 if __name__ == "__main__":
-    print("Starting Python Code Executor API...")
+    import uvicorn
+    print("Starting Interactive Python Code Executor API (FastAPI)...")
     print(f"Python version: {sys.version}")
     print(f"Working directory: {os.getcwd()}")
     
-    server_name = os.getenv('GRADIO_SERVER_NAME', '0.0.0.0')
-    server_port = int(os.getenv('GRADIO_SERVER_PORT', 7860))
+    server_name = os.getenv('FASTAPI_SERVER_NAME', '0.0.0.0')
+    server_port = int(os.getenv('FASTAPI_SERVER_PORT', 7860))
     
     print(f"Starting server on {server_name}:{server_port}")
     
     try:
-        # Launch the main API interface
-        # Gradio will automatically create API endpoints for functions defined in gr.Interface
-        # The /api/predict endpoint will now directly map to execute_python_code_api
-        api_code_executor.launch(
-            server_name=server_name,
-            server_port=server_port,
-            share=False,
-            show_error=True,
-            quiet=False,
-            # show_api=True is default for Interface, and the API_NAME makes it explicit
+        uvicorn.run(
+            "app:app",  # app:app means the `app` object in `app.py`
+            host=server_name,
+            port=server_port,
+            reload=False, # Set to True for local dev to auto-reload on code changes
+            log_level="info"
         )
-        # Note: If you launch multiple interfaces, they might conflict on the same port
-        # For simplicity for this API-only case, we'll focus on the primary endpoint.
-        # If you truly need both, you'd integrate FastAPI more deeply or use separate ports.
-        
     except Exception as e:
-        print(f"Failed to start server: {e}")
+        print(f"Failed to start server: {e}")
+        sys.exit(1)
+

requirements.txt

@@ -1,4 +1,3 @@
-gradio==4.44.0
 fastapi==0.104.1
 uvicorn==0.24.0
 pydantic==2.5.0