| import os |
| import asyncio |
| from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException |
| from fastapi.staticfiles import StaticFiles |
| from pydantic import BaseModel |
| from typing import Optional |
|
|
| |
| from backend.ai_model import generate, generate_with_debug |
| from backend.terminal import TerminalManager |
| from backend.file_manager import FileManager |
|
|
| app = FastAPI() |
|
|
| |
| term_manager = TerminalManager() |
| file_managers = {} |
|
|
| class GenerateRequest(BaseModel): |
| prompt: str |
| debug: bool = False |
|
|
| class SaveFileRequest(BaseModel): |
| session_id: str |
| filename: str |
| content: str |
|
|
| @app.post("/api/generate") |
| async def api_generate(req: GenerateRequest): |
| try: |
| if req.debug: |
| code = generate_with_debug(req.prompt) |
| else: |
| code = generate(req.prompt) |
| return {"code": code} |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
| @app.post("/api/files/save") |
| async def save_file(req: SaveFileRequest): |
| if req.session_id not in file_managers: |
| file_managers[req.session_id] = FileManager(req.session_id) |
| fm = file_managers[req.session_id] |
| fm.save(req.filename, req.content) |
| return {"status": "ok"} |
|
|
| @app.get("/api/files/{session_id}/{filename:path}") |
| async def load_file(session_id: str, filename: str): |
| if session_id not in file_managers: |
| file_managers[session_id] = FileManager(session_id) |
| fm = file_managers[session_id] |
| content = fm.load(filename) |
| if content is None: |
| raise HTTPException(status_code=404, detail="File not found") |
| return {"content": content} |
|
|
| @app.get("/api/files/{session_id}") |
| async def list_files(session_id: str): |
| if session_id not in file_managers: |
| file_managers[session_id] = FileManager(session_id) |
| fm = file_managers[session_id] |
| return {"files": fm.list_files()} |
|
|
| @app.websocket("/ws/terminal/{session_id}") |
| async def websocket_terminal(websocket: WebSocket, session_id: str): |
| await websocket.accept() |
| term = term_manager.get_session(session_id) |
| term.add_websocket(websocket) |
| try: |
| while True: |
| command = await websocket.receive_text() |
| term.write(command) |
| except WebSocketDisconnect: |
| term.remove_websocket(websocket) |
|
|
| |
| app.mount("/", StaticFiles(directory="static", html=True), name="static") |