| import os |
| import subprocess |
| import asyncio |
| from fastapi import FastAPI, WebSocket |
| from fastapi.responses import HTMLResponse |
| from pydantic import BaseModel |
|
|
| app = FastAPI() |
|
|
| class CommandRequest(BaseModel): |
| command: str |
| cwd: str = "/home/coder/workspace" |
|
|
| @app.post("/terminal/api/execute") |
| async def execute_command(req: CommandRequest): |
| try: |
| result = subprocess.run( |
| req.command, |
| shell=True, |
| capture_output=True, |
| text=True, |
| timeout=30, |
| cwd=req.cwd, |
| ) |
| return { |
| "stdout": result.stdout, |
| "stderr": result.stderr, |
| "returncode": result.returncode, |
| } |
| except subprocess.TimeoutExpired: |
| return {"error": "命令执行超时（30秒限制）"} |
| except Exception as e: |
| return {"error": str(e)} |
|
|
| @app.websocket("/terminal/ws/terminal") |
| async def websocket_terminal(websocket: WebSocket): |
| await websocket.accept() |
| try: |
| while True: |
| data = await websocket.receive_text() |
| if not data.strip(): |
| await websocket.send_text("\r\n") |
| continue |
| process = await asyncio.create_subprocess_shell( |
| data, |
| stdout=asyncio.subprocess.PIPE, |
| stderr=asyncio.subprocess.PIPE, |
| cwd="/home/coder/workspace" |
| ) |
| stdout, stderr = await process.communicate() |
| output = stdout.decode("utf-8", errors="replace") |
| error = stderr.decode("utf-8", errors="replace") |
| response = output |
| if error: |
| response += f"\n[stderr]\n{error}" |
| if not response: |
| response = "(无输出)\n" |
| await websocket.send_text(response) |
| except Exception as e: |
| await websocket.send_text(f"\r\n[连接错误: {e}]\r\n") |
|
|
| @app.get("/terminal/", response_class=HTMLResponse) |
| async def terminal_page(): |
| with open("/app/static/index.html", "r") as f: |
| return HTMLResponse(content=f.read()) |
