Spaces:
Sleeping
Sleeping
| import os | |
| import uuid | |
| import asyncio | |
| import shutil | |
| from typing import List | |
| from fastapi import FastAPI, Request, Form, UploadFile, File, HTTPException | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.templating import Jinja2Templates | |
| from core.agent import StellarAgent | |
| from core.database import StellarDB | |
| app = FastAPI(title="Stellar Compute Agent") | |
| db = StellarDB() | |
| agent = StellarAgent() | |
| # 挂载静态文件 | |
| os.makedirs("static", exist_ok=True) | |
| os.makedirs("uploads", exist_ok=True) | |
| app.mount("/static", StaticFiles(directory="static"), name="static") | |
| templates = Jinja2Templates(directory="templates") | |
| async def index(request: Request): | |
| try: | |
| sats = db.get_satellites() | |
| tasks = db.get_tasks() | |
| return templates.TemplateResponse("index.html", { | |
| "request": request, | |
| "sats": sats, | |
| "tasks": tasks | |
| }) | |
| except Exception as e: | |
| return HTMLResponse(content=f"<html><body><h1>Internal Server Error</h1><p>{str(e)}</p></body></html>", status_code=500) | |
| async def create_mission(prompt: str = Form(...)): | |
| try: | |
| task_id = str(uuid.uuid4())[:8] | |
| # 在后台运行任务 | |
| result, logs = await agent.run_mission(task_id, prompt) | |
| return {"task_id": task_id, "result": result, "logs": logs} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def upload_file(file: UploadFile = File(...)): | |
| try: | |
| file_path = os.path.join("uploads", file.filename) | |
| with open(file_path, "wb") as buffer: | |
| shutil.copyfileobj(file.file, buffer) | |
| return {"filename": file.filename, "status": "success", "path": file_path} | |
| except Exception as e: | |
| return JSONResponse(status_code=500, content={"message": f"上传失败: {str(e)}"}) | |
| async def get_tasks(): | |
| return db.get_tasks() | |
| async def get_satellites(): | |
| return db.get_satellites() | |
| async def get_stats(): | |
| sats = db.get_satellites() | |
| labels = [s['name'] for s in sats] | |
| loads = [s['compute_load'] for s in sats] | |
| batteries = [s['battery'] for s in sats] | |
| return { | |
| "labels": labels, | |
| "loads": loads, | |
| "batteries": batteries | |
| } | |
| if __name__ == "__main__": | |
| import uvicorn | |
| port = int(os.environ.get("PORT", 7860)) | |
| uvicorn.run(app, host="0.0.0.0", port=port) | |