"""
# ── Home ───────────────────────────────────────────────────────────────────────
@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
user = get_current_user(request)
if not user:
return HTMLResponse(_login_page())
kb = brain.kb_size if brain else 0
ok = brain is not None
content = f"""
Ciao! Sono GenerAI
Il tuo assistente AI specializzato in italiano.
Cerco sul web e apprendo autonomamente per darti la risposta migliore.
"""
return HTMLResponse(page("Account", content, ""))
# ── Auth Google ────────────────────────────────────────────────────────────────
@app.get("/auth/login")
def auth_login():
url = google_login_url()
if not url:
return HTMLResponse("
GOOGLE_CLIENT_ID non configurato.
Torna alla home", status_code=503)
return RedirectResponse(url)
@app.get("/auth/callback")
async def auth_callback(code: str = "", error: str = ""):
if error or not code:
return RedirectResponse("/?error=auth_failed")
token_data = await asyncio.to_thread(exchange_code, code)
access_token = token_data.get("access_token")
if not access_token:
return RedirectResponse("/?error=token_failed")
userinfo = await asyncio.to_thread(get_google_userinfo, access_token)
gid = userinfo.get("sub")
if not gid:
return RedirectResponse("/?error=userinfo_failed")
db = get_user_db()
db.get_or_create(
gid,
userinfo.get("email", ""),
userinfo.get("name", userinfo.get("email", "Utente")),
userinfo.get("picture", ""),
)
token = create_session(gid)
_schedule_mega_upload() # salva il nuovo utente su MEGA
resp = RedirectResponse("/ragno")
resp.set_cookie(COOKIE_NAME, token, max_age=86400 * 30,
httponly=True, samesite="lax", secure=True)
return resp
@app.get("/auth/logout")
def auth_logout():
resp = RedirectResponse("/")
resp.delete_cookie(COOKIE_NAME)
return resp
@app.get("/auth/me")
def auth_me(request: Request):
user = get_current_user(request)
if not user:
return {"logged_in": False}
return {
"logged_in": True,
"name": user["name"],
"email": user["email"],
"picture": user.get("picture", ""),
"credits": user.get("credits", 0),
}
# ── Crediti ────────────────────────────────────────────────────────────────────
@app.get("/credits")
def credits_endpoint():
ledger = CreditLedger()
return {"remaining": ledger.remaining, "total_used": ledger.total_used, "default": 10_000}
# ── Rete Ragno – SSE crawl ─────────────────────────────────────────────────────
@app.get("/ragno/crawl")
async def ragno_crawl_stream(request: Request,
q: str = Query(default=""),
budget: int = Query(default=10, ge=1, le=100)):
if not q.strip():
return JSONResponse({"error": "Query mancante"}, status_code=400)
user = get_current_user(request)
if not user:
return JSONResponse({"error": "login_required",
"message": "Devi accedere con Google per usare la Rete Ragno."},
status_code=401)
gid = user["id"]
db = get_user_db()
user_credits = db.remaining(gid)
if user_credits <= 0:
return JSONResponse({"error": "no_credits",
"message": "Crediti esauriti. Contatta l'amministratore."},
status_code=402)
query = q.strip()
budget = min(budget, 100, user_credits)
def _use_credit(n: int = 1) -> bool:
return db.use_credits(gid, n)
def _remaining() -> int:
return db.remaining(gid)
async def event_gen():
loop = asyncio.get_running_loop()
queue: asyncio.Queue = asyncio.Queue()
def _run():
try:
for ev in do_crawl(query, budget,
use_credit_fn=_use_credit,
credits_remaining_fn=_remaining):
loop.call_soon_threadsafe(queue.put_nowait, ev)
except Exception as exc:
loop.call_soon_threadsafe(queue.put_nowait,
{"type": "error", "message": str(exc), "credits_remaining": _remaining()})
finally:
loop.call_soon_threadsafe(queue.put_nowait, None)
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
pool.submit(_run)
while True:
item = await queue.get()
if item is None:
break
yield f"data: {json.dumps(item, ensure_ascii=False)}\n\n"
_schedule_mega_upload() # salva crediti aggiornati su MEGA
yield "data: [DONE]\n\n"
return StreamingResponse(
event_gen(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
# ── Rete Ragno – pagina ────────────────────────────────────────────────────────
@app.get("/ragno", response_class=HTMLResponse)
def ragno_page(request: Request):
if not get_current_user(request):
return RedirectResponse("/")
content = f"""
Rete Ragno
Motore di ricerca puro — il ragno esplora il web autonomamente, saltando di link in link. Nessun intermediario. Ogni pagina visitata costa 1 credito.
Crediti:
...
🧠 Riepilogo AI
"""
return HTMLResponse(page("Rete Ragno", content, "/ragno"))
# ── Stato cron job ─────────────────────────────────────────────────────────────
@app.get("/status")
async def cronjob_status():
api_key = os.environ.get("CRONJOB_API_KEY","")
if not api_key:
return {"error": "CRONJOB_API_KEY non impostata"}
def fetch():
hdrs = {"Authorization": f"Bearer {api_key}"}
jobs_r = req.get("https://api.cron-job.org/jobs", headers=hdrs, timeout=8)
if jobs_r.status_code != 200:
return {"error": f"HTTP {jobs_r.status_code}"}
result = []
for job in jobs_r.json().get("jobs",[]):
jid = job.get("jobId")
hist_r = req.get(f"https://api.cron-job.org/jobs/{jid}/history", headers=hdrs, timeout=8)
history = []
if hist_r.status_code == 200:
history = [{"date":e.get("date"),"status":e.get("status"),
"duration":e.get("duration"),"httpStatus":e.get("httpStatus")}
for e in hist_r.json().get("history",[])[:20]]
ok_n = sum(1 for h in history if h["status"]==1)
last = history[0] if history else {}
result.append({"jobId":jid,"title":job.get("title",""),"url":job.get("url",""),
"enabled":job.get("enabled",False),"last_run":last.get("date"),
"last_status":"ok" if last.get("status")==1 else "fail" if last else "unknown",
"last_http":last.get("httpStatus"),
"uptime_pct":round(ok_n/len(history)*100,1) if history else None,
"ok_count":ok_n,"fail_count":len(history)-ok_n,"history":history})
return {"jobs":result,"total":len(result)}
return await asyncio.to_thread(fetch)
@app.get("/status/ui", response_class=HTMLResponse)
def status_ui():
content = f"""
📊 Monitoraggio Operativo
Analisi in tempo reale delle attività automatiche e della stabilità dei servizi.
Architettura del sistema e licenze dei componenti usati.
Panoramica
GenerAI è un assistente AI in italiano con memoria locale e ricerca web autonoma.
Il backend è scritto in Python (FastAPI). Non usa un motore di ricerca esterno come Google o Bing tramite API a pagamento:
la ricerca web avviene tramite un browser headless (Chromium via Playwright) che naviga direttamente Bing, con Wikipedia
come fallback se non trova risultati utili.
Componenti principali
Componente
Cosa fa
Cervello (brain.py)
Decide se rispondere dalla memoria locale, cercare sul web, o usare un modello LLM esterno
Memoria (knowledge_base.py)
Database vettoriale locale (ChromaDB) che impara dalle domande frequenti
Ricerca web (scraper.py)
Chromium headless (Playwright) su Bing, con fallback Wikipedia
Rete Ragno (crawler.py)
Esplorazione autonoma del web salto per salto, a partire da Wikipedia, con crediti per utente
Login (auth.py)
Autenticazione Google OAuth, crediti per account
Backup (mega_sync.py)
Sincronizzazione del database su MEGA
World Monitor
Dashboard di intelligence globale (notizie, mappe, dati geopolitici) integrata come componente separato — vedi licenza sotto
Licenze
Il codice originale di GenerAI (questo assistente: brain, scraper, crawler, knowledge base, interfaccia) è
rilasciato con licenza MIT — puoi vederlo, copiarlo e modificarlo liberamente.
Il componente World Monitor, se presente in questo Space, è un progetto di terze parti di
koala73/worldmonitor,
distribuito con licenza AGPL-3.0-only. L'AGPL richiede che, essendo un servizio consultabile in rete, il codice
sorgente (comprese eventuali modifiche fatte per farlo girare qui) resti disponibile a chi lo usa — il codice sorgente
originale è pubblico al link sopra. GenerAI non rivendica alcuna proprietà su World Monitor: è integrato così com'è,
nel rispetto della sua licenza.
"""
blocks = {
"Stato": [
cmd_block("b0","Controlla se la AI e online",
f'Invoke-RestMethod "{BASE}/"',
f'curl "{BASE}/"'),
],
"Chat normale": [
cmd_block("b1","Fai una domanda (risposta completa)",
f'Invoke-RestMethod -Uri "{BASE}/ask" -Method POST -ContentType "application/json" -Body \'{"prompt":"chi ha inventato la lampadina?"}\' | ConvertTo-Json',
f'curl -s -X POST "{BASE}/ask" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"chi ha inventato la lampadina?\\"}}"'),
cmd_block("b2","Domanda con passi in tempo reale (streaming)",
f'curl -N -X POST "{BASE}/ask/stream" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"chi ha inventato la lampadina?\\"}}"',
f'curl -N -X POST "{BASE}/ask/stream" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"chi ha inventato la lampadina?\\"}}"'),
],
"Feedback": [
cmd_block("b11","Risposta utile",
f'Invoke-RestMethod -Uri "{BASE}/feedback?positive=true" -Method POST',
f'curl -X POST "{BASE}/feedback?positive=true"'),
cmd_block("b12","Risposta non utile",
f'Invoke-RestMethod -Uri "{BASE}/feedback?positive=false" -Method POST',
f'curl -X POST "{BASE}/feedback?positive=false"'),
],
}
sections = ""
for title, items in blocks.items():
sections += f'
{title}
{"".join(items)}
'
content = f"""
CMD Comandi e Script
Copia i comandi per PowerShell o CMD. Scarica gli script completi con menu interattivo.
{sections}
"""
return HTMLResponse(page("Comandi", content, "/commands"))
# ── Script downloads ───────────────────────────────────────────────────────────
@app.get("/scripts/powershell")
def script_ps():
s = f"""# GenerAI — Script PowerShell
# Come usarlo: apri PowerShell, vai nella cartella dove hai salvato il file, scrivi: .\\generai.ps1
$BASE = "{BASE}"
function Ask($d) {{
$body = @{{prompt = $d}} | ConvertTo-Json
$r = Invoke-RestMethod -Uri "$BASE/ask" -Method POST -ContentType "application/json" -Body $body
Write-Host ""
Write-Host "Risposta:" -ForegroundColor Cyan
Write-Host $r.result
Write-Host "(Fonte: $($r.status))" -ForegroundColor DarkGray
}}
function AskStream($d) {{
$body = @{{prompt = $d}} | ConvertTo-Json -Compress
Write-Host "Cosa sta facendo la AI:" -ForegroundColor Yellow
curl -s -N -X POST "$BASE/ask/stream" -H "Content-Type: application/json" -d $body
}}
function Stato {{
$r = Invoke-RestMethod "$BASE/"
Write-Host "AI accesa: $($r.status)" -ForegroundColor Green
Write-Host "Cose in memoria: $($r.kb_size)"
}}
function Feedback($positivo) {{
Invoke-RestMethod -Uri "$BASE/feedback?positive=$positivo" -Method POST | Out-Null
if ($positivo -eq "true") {{ Write-Host "Grazie! La AI imparera' da questa risposta." -ForegroundColor Green }}
else {{ Write-Host "Capito, la AI cerchera' di migliorare." -ForegroundColor Yellow }}
}}
while ($true) {{
Write-Host ""
Write-Host "========== GENERAI ==========" -ForegroundColor Magenta
Write-Host "1. Fai una domanda"
Write-Host "2. Fai una domanda (vedi i passi in diretta)"
Write-Host "3. Controlla se la AI e' accesa"
Write-Host "4. Questa risposta era utile"
Write-Host "5. Questa risposta non era utile"
Write-Host "0. Esci"
$s = Read-Host "Cosa vuoi fare?"
switch ($s) {{
"1" {{ Ask (Read-Host "Scrivi la tua domanda") }}
"2" {{ AskStream (Read-Host "Scrivi la tua domanda") }}
"3" {{ Stato }}
"4" {{ Feedback "true" }}
"5" {{ Feedback "false" }}
"0" {{ Write-Host "A presto!"; exit }}
default {{ Write-Host "Non ho capito, riprova." -ForegroundColor Red }}
}}
}}
"""
return Response(content=s, media_type="text/plain",
headers={"Content-Disposition": "attachment; filename=generai.ps1"})
@app.get("/scripts/batch")
def script_bat():
s = f"""@echo off
chcp 65001 >nul
set BASE={BASE}
:menu
echo.
echo ========== GENERAI ==========
echo 1. Fai una domanda
echo 2. Fai una domanda (vedi i passi in diretta)
echo 3. Controlla se la AI e' accesa
echo 0. Esci
set /p s=Cosa vuoi fare?
if "%s%"=="1" goto ask
if "%s%"=="2" goto stream
if "%s%"=="3" goto stato
if "%s%"=="0" goto fine
echo Non ho capito, riprova.
goto menu
:ask
set /p d=Scrivi la tua domanda:
curl -s -X POST "%BASE%/ask" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"%d%\\"}}"
echo.
goto menu
:stream
set /p d=Scrivi la tua domanda:
curl -N -X POST "%BASE%/ask/stream" -H "Content-Type: application/json" -d "{{\\"prompt\\":\\"%d%\\"}}"
echo.
goto menu
:stato
curl -s "%BASE%/"
echo.
goto menu
:fine
echo A presto!
exit /b
"""
return Response(content=s, media_type="text/plain",
headers={"Content-Disposition": "attachment; filename=generai.bat"})
# ── Ask / Feedback ─────────────────────────────────────────────────────────────
class AskRequest(BaseModel):
prompt: str
def _schedule_mega_upload():
email = os.environ.get("MEGA_EMAIL", "")
password = os.environ.get("MEGA_PASSWORD", "")
if email and password:
from mega_sync import upload_db
asyncio.create_task(asyncio.to_thread(upload_db, email, password))
@app.on_event("startup")
async def startup():
global brain
email = os.environ.get("MEGA_EMAIL", "")
password = os.environ.get("MEGA_PASSWORD", "")
if email and password:
from mega_sync import download_db
log.info("Scarico DB da MEGA...")
await asyncio.to_thread(download_db, email, password)
else:
log.warning("MEGA_EMAIL/MEGA_PASSWORD non impostati — DB solo locale.")
brain = Brain()
@app.post("/ask")
async def ask(request: AskRequest):
if not brain:
raise HTTPException(status_code=503, detail="AI non ancora pronta")
answer, status = await brain.ask(request.prompt)
if brain._last_doc_id:
_schedule_mega_upload()
return {"result": answer, "status": status}
@app.post("/ask/stream")
async def ask_stream(request: AskRequest):
if not brain:
raise HTTPException(status_code=503, detail="AI non ancora pronta")
async def generate():
queue: asyncio.Queue = asyncio.Queue()
async def on_status(msg): await queue.put(("status", msg))
async def run():
a, s = await brain.ask(request.prompt, on_status=on_status)
await queue.put(("result", a, s))
if brain._last_doc_id:
_schedule_mega_upload()
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",
headers={"Cache-Control":"no-cache","X-Accel-Buffering":"no","Connection":"keep-alive"})
@app.post("/feedback")
def feedback(positive: bool):
if brain: brain.give_feedback(positive)
return {"ok": True}
@app.post("/search")
async def search_raw(request: AskRequest):
from scraper import search_and_extract
try:
results = await asyncio.to_thread(search_and_extract, request.prompt, max_results=8)
except Exception as e:
log.warning("Search raw fallita: %s", fmt_exc(e))
results = []
return {"query": request.prompt, "results": results}
# ── WebSocket — sessione continua ──────────────────────────────────────────────
@app.websocket("/ws/chat")
async def ws_chat(ws: WebSocket):
"""
WebSocket per input/output continui con storia della conversazione.
Il client manda: {"prompt": "tua domanda"}
Il server risponde con messaggi di stato e poi la risposta finale.
"""
await ws.accept()
history = [] # storia della conversazione nella sessione
await ws.send_text(json.dumps({
"type": "connected",
"message": "Connesso a GenerAI. Manda {\"prompt\": \"la tua domanda\"}."
}))
try:
while True:
raw = await ws.receive_text()
try:
data = json.loads(raw)
except Exception:
await ws.send_text(json.dumps({"type":"error","message":"JSON non valido"}))
continue
prompt = data.get("prompt", "").strip()
if not prompt:
await ws.send_text(json.dumps({"type":"error","message":"Campo 'prompt' mancante"}))
continue
if not brain:
await ws.send_text(json.dumps({"type":"error","message":"AI non ancora pronta"}))
continue
# Manda aggiornamenti di stato in tempo reale
async def on_status(msg: str):
await ws.send_text(json.dumps({"type":"status","message":msg}))
answer, status = await brain.ask(prompt, on_status=on_status)
# Salva nella storia
history.append({"role":"user","content":prompt})
history.append({"role":"assistant","content":answer})
await ws.send_text(json.dumps({
"type": "result",
"message": answer,
"status": status,
"history_len": len(history) // 2,
}))
except WebSocketDisconnect:
pass
@app.get("/ws/ui", response_class=HTMLResponse)
def ws_ui():
content = f"""
Sessione continua
Connessione persistente — manda quante domande vuoi senza ricaricare la pagina. La sessione ricorda tutto.
