from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse, HTMLResponse, Response from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import uvicorn import asyncio import json from brain import Brain app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) brain: Brain | None = None BASE = "https://amogaddy-generai.hf.space" # ── Shared layout ────────────────────────────────────────────────────────────── def page(title: str, content: str, active: str = "") -> str: nav_items = [ ("/", "🏠", "Home"), ("/chat", "💬", "Chat"), ("/commands", "⚡", "Comandi"), ("/docs", "📖", "API Docs"), ] nav_html = "".join( f'{icon} {label}' for href, icon, label in nav_items ) return f""" GenerAI — {title}
{content}
""" # ── Home ─────────────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) async def home(): kb = brain.kb_size if brain else 0 status = "Online" if brain else "Avvio..." badge = "green" if brain else "red" content = f"""

GenerAI

Assistente AI in italiano con memoria locale e ricerca web.

Stato sistema

ComponenteStatoInfo
AI Brain{status}KB + Web search
Knowledge BaseOnline{kb} documenti in memoria
Ricerca WebChromiumBrowser headless (Playwright)

Accesso rapido

💬 Chat ⚡ Comandi terminale 📖 API Docs
""" return HTMLResponse(page("Home", content, "/")) # ── Chat ─────────────────────────────────────────────────────────────────────── @app.get("/chat", response_class=HTMLResponse) def chat_page(): content = f"""

💬 Chat

Fai una domanda alla AI — vedrai i passaggi in tempo reale.

""" return HTMLResponse(page("Chat", content, "/chat")) # ── Comandi ──────────────────────────────────────────────────────────────────── @app.get("/commands", response_class=HTMLResponse) def commands_page(): def cmd_block(bid, label, ps, cmd): return f"""
{label}
PowerShell CMD
{ps}
""" blocks = { "Stato AI": [ cmd_block("b0","Controlla che la AI sia online", f'Invoke-RestMethod "{BASE}/"', f'curl "{BASE}/"'), ], "Chat": [ cmd_block("b1","Domanda (risposta completa)", f'Invoke-RestMethod -Uri "{BASE}/ask" -Method POST -ContentType "application/json" -Body \'{"prompt":"chi e napoleone?"}\' | ConvertTo-Json', f'curl -s -X POST "{BASE}/ask" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"chi e napoleone?\\"}}"'), cmd_block("b2","Domanda in streaming (tempo reale)", f'curl -N -X POST "{BASE}/ask/stream" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"chi e napoleone?\\"}}"', f'curl -N -X POST "{BASE}/ask/stream" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"chi e napoleone?\\"}}"'), ], "Feedback": [ cmd_block("b6","Risposta utile", f'Invoke-RestMethod -Uri "{BASE}/feedback?positive=true" -Method POST', f'curl -X POST "{BASE}/feedback?positive=true"'), cmd_block("b7","Risposta non utile", f'Invoke-RestMethod -Uri "{BASE}/feedback?positive=false" -Method POST', f'curl -X POST "{BASE}/feedback?positive=false"'), ], } sections_html = "" for section, items in blocks.items(): icon = {"Stato AI":"🟢","Chat":"💬","Feedback":"👍"}.get(section,"") sections_html += f'

{icon} {section}

{"".join(items)}
' content = f"""

⚡ Comandi terminale

Scegli PowerShell o CMD, copia il comando. Oppure scarica lo script completo.

⬇ generai.ps1 ⬇ generai.bat
{sections_html} """ return HTMLResponse(page("Comandi", content, "/commands")) # ── Script downloads ─────────────────────────────────────────────────────────── @app.get("/scripts/powershell") def script_ps(): script = f"""# GenerAI — Script PowerShell | .\generai.ps1 $BASE = "{BASE}" function Ask($d) {{ $r = Invoke-RestMethod -Uri "$BASE/ask" -Method POST -ContentType "application/json" -Body (@{{prompt=$d}}|ConvertTo-Json) Write-Host "`n[$($r.status)]" -ForegroundColor Cyan; Write-Host $r.result }} function AskStream($d) {{ curl -s -N -X POST "$BASE/ask/stream" -H "Content-Type: application/json" -d (@{{prompt=$d}}|ConvertTo-Json -Compress) }} function Stato {{ Invoke-RestMethod "$BASE/" | ConvertTo-Json }} function Feedback($p) {{ Invoke-RestMethod -Uri "$BASE/feedback?positive=$p" -Method POST | Out-Null Write-Host "Feedback inviato!" -ForegroundColor Green }} while ($true) {{ Write-Host "`n===== GENERAI =====" -ForegroundColor Magenta Write-Host "1. Domanda 2. Streaming 3. Stato 4/5. Feedback 0. Esci" switch (Read-Host "Scelta") {{ "1" {{ Ask (Read-Host "Domanda") }} "2" {{ AskStream (Read-Host "Domanda") }} "3" {{ Stato }} "4" {{ Feedback "true" }} "5" {{ Feedback "false" }} "0" {{ exit }} default {{ Write-Host "Scelta non valida" -ForegroundColor Red }} }} }} """ return Response(content=script, media_type="text/plain", headers={"Content-Disposition": "attachment; filename=generai.ps1"}) @app.get("/scripts/batch") def script_bat(): script = f"""@echo off chcp 65001 >nul set BASE={BASE} :menu echo. echo ===== GENERAI ===== echo 1.Domanda 2.Streaming 3.Stato 0.Esci set /p s=Scelta: if "%s%"=="1" (set /p d=Domanda: & curl -s -X POST "%BASE%/ask" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"%d%\\"}}" & echo.) & goto menu if "%s%"=="2" (set /p d=Domanda: & curl -N -X POST "%BASE%/ask/stream" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"%d%\\"}}" & echo.) & goto menu if "%s%"=="3" (curl -s "%BASE%/" & echo.) & goto menu if "%s%"=="0" exit /b goto menu """ return Response(content=script, media_type="text/plain", headers={"Content-Disposition": "attachment; filename=generai.bat"}) # ── API endpoints ────────────────────────────────────────────────────────────── class AskRequest(BaseModel): prompt: str @app.post("/ask") async def ask(req: AskRequest): if not brain: raise HTTPException(status_code=503, detail="Brain non ancora pronto") answer, status = await brain.ask(req.prompt) return {"result": answer, "status": status} @app.post("/ask/stream") async def ask_stream(req: AskRequest): if not brain: raise HTTPException(status_code=503, detail="Brain non ancora pronto") async def generate(): queue: asyncio.Queue = asyncio.Queue() async def on_status(msg: str): await queue.put(("status", msg)) async def run(): answer, status = await brain.ask(req.prompt, on_status=on_status) await queue.put(("result", answer, status)) task = asyncio.create_task(run()) while True: item = await queue.get() if item[0] == "status": yield f"data: {json.dumps({'type':'status','message':item[1]})}\n\n" elif item[0] == "result": yield f"data: {json.dumps({'type':'result','message':item[1],'status':item[2]})}\n\n" break await task return StreamingResponse(generate(), media_type="text/event-stream") @app.post("/feedback") def feedback(positive: bool): if brain: brain.give_feedback(positive) return {"ok": True} @app.on_event("startup") async def startup(): global brain brain = Brain() if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)