Cloud-Based IDE Backend

from hopx_ai import Sandbox
import os
from typing import Dict, List, Any

class IDEFileSystem:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.sandbox = None
    
    def initialize_project(self, project_id: str) -> Dict[str, Any]:
        """Initialize a new project sandbox"""
        try:
            self.sandbox = Sandbox.create(
                template="code-interpreter",
                api_key=self.api_key,
                timeout_seconds=3600  # 1 hour for IDE session
            )
            
            # Create project structure
            self.sandbox.files.mkdir("/workspace/project")
            self.sandbox.files.write("/workspace/project/.gitignore", "__pycache__/\n*.pyc\n")
            
            return {
                "success": True,
                "sandbox_id": self.sandbox.sandbox_id,
                "project_path": "/workspace/project"
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    def list_files(self, path: str = "/workspace/project") -> List[Dict[str, Any]]:
        """List files and directories"""
        try:
            files = self.sandbox.files.list(path)
            return [
                {
                    "name": f.name,
                    "path": f.path,
                    "is_dir": f.is_dir,
                    "size": f.size,
                    "modified": f.modified_time.isoformat() if hasattr(f.modified_time, 'isoformat') else str(f.modified_time)
                }
                for f in files
            ]
        except Exception as e:
            return []
    
    def read_file(self, file_path: str) -> Dict[str, Any]:
        """Read file contents"""
        try:
            content = self.sandbox.files.read(file_path)
            return {
                "success": True,
                "content": content,
                "path": file_path
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "path": file_path
            }
    
    def write_file(self, file_path: str, content: str) -> Dict[str, Any]:
        """Write file contents"""
        try:
            self.sandbox.files.write(file_path, content)
            return {
                "success": True,
                "path": file_path,
                "size": len(content)
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "path": file_path
            }
    
    def create_directory(self, dir_path: str) -> Dict[str, Any]:
        """Create a directory"""
        try:
            self.sandbox.files.mkdir(dir_path)
            return {
                "success": True,
                "path": dir_path
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "path": dir_path
            }
    
    def delete_file(self, file_path: str) -> Dict[str, Any]:
        """Delete a file or directory"""
        try:
            self.sandbox.files.remove(file_path)
            return {
                "success": True,
                "path": file_path
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "path": file_path
            }
    
    def cleanup(self):
        """Clean up sandbox"""
        if self.sandbox:
            self.sandbox.kill()
            self.sandbox = None

# Usage
fs = IDEFileSystem(api_key=os.getenv("HOPX_API_KEY"))
result = fs.initialize_project("my-project")
print(result)

# List files
files = fs.list_files()
print(files)

# Create a file
fs.write_file("/workspace/project/main.py", "print('Hello, World!')")

# Read the file
content = fs.read_file("/workspace/project/main.py")
print(content)

fs.cleanup()

import asyncio
from typing import AsyncIterator, Dict, Any

class IDETerminal:
    def __init__(self, sandbox: Sandbox):
        self.sandbox = sandbox
        self.terminal_session = None
    
    async def connect(self) -> Dict[str, Any]:
        """Connect to terminal"""
        try:
            self.terminal_session = await self.sandbox.terminal.connect()
            return {
                "success": True,
                "session_id": str(self.terminal_session)
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    async def send_command(self, command: str) -> None:
        """Send command to terminal"""
        if self.terminal_session:
            await self.sandbox.terminal.send_input(
                self.terminal_session,
                command + "\n"
            )
    
    async def receive_output(self) -> AsyncIterator[Dict[str, Any]]:
        """Receive terminal output"""
        if not self.terminal_session:
            return
        
        async for message in self.sandbox.terminal.iter_output(self.terminal_session):
            yield {
                "type": message.get("type", "output"),
                "data": message.get("data", ""),
                "timestamp": message.get("timestamp")
            }
    
    async def resize(self, rows: int, cols: int) -> None:
        """Resize terminal"""
        if self.terminal_session:
            await self.sandbox.terminal.resize(
                self.terminal_session,
                rows=rows,
                cols=cols
            )
    
    async def disconnect(self):
        """Disconnect from terminal"""
        self.terminal_session = None

# Usage
sandbox = Sandbox.create(template="code-interpreter", api_key=os.getenv("HOPX_API_KEY"))
terminal = IDETerminal(sandbox)

await terminal.connect()
await terminal.send_command("pwd")
await terminal.send_command("ls -la")

async for output in terminal.receive_output():
    print(output)
    # Break after some outputs
    break

await terminal.disconnect()
sandbox.kill()

class IDEProjectExecutor:
    def __init__(self, sandbox: Sandbox):
        self.sandbox = sandbox
    
    def execute_project(self, entry_point: str, working_dir: str = "/workspace/project") -> Dict[str, Any]:
        """Execute a multi-file project"""
        try:
            # Determine language from file extension
            language = self._detect_language(entry_point)
            
            # Execute in project directory
            result = self.sandbox.run_code(
                self._get_execution_code(entry_point, language),
                language=language,
                working_dir=working_dir,
                timeout=30
            )
            
            return {
                "success": result.success,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "exit_code": result.exit_code,
                "execution_time": result.execution_time
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    def _detect_language(self, file_path: str) -> str:
        """Detect language from file extension"""
        ext = file_path.split('.')[-1].lower()
        language_map = {
            'py': 'python',
            'js': 'javascript',
            'ts': 'typescript',
            'java': 'java',
            'go': 'go'
        }
        return language_map.get(ext, 'python')
    
    def _get_execution_code(self, entry_point: str, language: str) -> str:
        """Get code to execute based on language"""
        if language == 'python':
            return f"exec(open('{entry_point}').read())"
        elif language == 'javascript':
            return f"require('{entry_point.replace('.js', '')}')"
        else:
            # For other languages, use appropriate execution method
            return f"# Execute {entry_point}"

# Usage
sandbox = Sandbox.create(template="code-interpreter", api_key=os.getenv("HOPX_API_KEY"))
executor = IDEProjectExecutor(sandbox)

# Create a multi-file project
sandbox.files.write("/workspace/project/utils.py", """
def greet(name):
    return f"Hello, {name}!"
""")

sandbox.files.write("/workspace/project/main.py", """
from utils import greet
print(greet("World"))
""")

# Execute
result = executor.execute_project("main.py")
print(result)

sandbox.kill()

class CloudIDEBackend:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.fs = IDEFileSystem(api_key)
        self.sandbox = None
        self.terminal = None
        self.executor = None
    
    def initialize(self, project_id: str) -> Dict[str, Any]:
        """Initialize IDE session"""
        result = self.fs.initialize_project(project_id)
        if result["success"]:
            self.sandbox = self.fs.sandbox
            self.terminal = IDETerminal(self.sandbox)
            self.executor = IDEProjectExecutor(self.sandbox)
        return result
    
    async def handle_file_operation(self, operation: str, **kwargs) -> Dict[str, Any]:
        """Handle file operations"""
        if operation == "list":
            return {"files": self.fs.list_files(kwargs.get("path", "/workspace/project"))}
        elif operation == "read":
            return self.fs.read_file(kwargs["path"])
        elif operation == "write":
            return self.fs.write_file(kwargs["path"], kwargs["content"])
        elif operation == "create_dir":
            return self.fs.create_directory(kwargs["path"])
        elif operation == "delete":
            return self.fs.delete_file(kwargs["path"])
        else:
            return {"success": False, "error": f"Unknown operation: {operation}"}
    
    async def handle_terminal_command(self, command: str) -> Dict[str, Any]:
        """Handle terminal command"""
        if not self.terminal.terminal_session:
            await self.terminal.connect()
        
        await self.terminal.send_command(command)
        
        # Collect output
        outputs = []
        async for output in self.terminal.receive_output():
            outputs.append(output)
            if len(outputs) >= 10:  # Limit output
                break
        
        return {"outputs": outputs}
    
    def handle_execution(self, entry_point: str) -> Dict[str, Any]:
        """Handle code execution"""
        return self.executor.execute_project(entry_point)
    
    def cleanup(self):
        """Clean up all resources"""
        if self.terminal:
            asyncio.run(self.terminal.disconnect())
        self.fs.cleanup()

# Usage
ide = CloudIDEBackend(api_key=os.getenv("HOPX_API_KEY"))
ide.initialize("my-project")

# File operations
result = asyncio.run(ide.handle_file_operation("write", path="/workspace/project/app.py", content="print('Hello')"))
print(result)

# Terminal
result = asyncio.run(ide.handle_terminal_command("ls -la"))
print(result)

# Execution
result = ide.handle_execution("app.py")
print(result)

ide.cleanup()