from __future__ import annotations import json import asyncio import logging import os from pathlib import Path from concurrent.futures import ThreadPoolExecutor import uvicorn # noqa: F401 — needed by main() from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from src.engine import db from src.engine.config import load_from_env from src.engine.dm import dm_cycle from src.engine.game_context import GameContext log = logging.getLogger(__name__) STATIC_DIR = Path(__file__).resolve().parent / "static" app = FastAPI(title="DCC RPG — AI Dungeon Master") # --------------------------------------------------------------------------- # REST endpoints # --------------------------------------------------------------------------- @app.get("/api/campaigns") def list_campaigns(): conn = db.get_conn() rows = conn.execute( "SELECT id, name, adventure, turn, created_at FROM campaigns ORDER BY id DESC" ).fetchall() return [dict(r) for r in rows] @app.get("/api/campaign/{campaign_id}") def get_campaign_state(campaign_id: int): try: ctx = GameContext.load(campaign_id) except ValueError: return JSONResponse({"error": "Campagna non trovata"}, status_code=404) # Full area details from adventure data all_areas = db.get_all_areas() area_map = {a["id"]: a for a in all_areas} visited_ids = {a["area_id"] for a in ctx.areas if a.get("visited")} enriched = [] for sa in ctx.areas: aid = sa["area_id"] full = area_map.get(aid, {}) raw_conns = full.get("connections", []) conn_names = [] for cid in raw_conns: linked = area_map.get(cid) conn_names.append(linked["name"] if linked else cid) enriched.append({ "id": aid, "name": full.get("name", aid), "description": full.get("description", "")[:200], "connections": conn_names, "visited": bool(sa.get("visited")), "features": full.get("features", [])[:3], }) return { "campaign": dict(ctx.campaign) if ctx.campaign else {}, "characters": ctx.characters, "quests": ctx.quests, "areas": enriched, "turn": ctx.turn, } @app.post("/api/campaign") def create_campaign(name: str = "Nuova Avventura", adventure: str = ""): camp_id = db.create_campaign(name, adventure) db.seed_adventure_state(camp_id) db.seed_quests_from_adventure(camp_id) # Mark first area as visited areas = db.get_all_areas() if areas: db.set_area_state(camp_id, areas[0]["id"], visited=True) for pg in genera_funnel(4): _create_lv0_character(camp_id, pg) return {"campaign_id": camp_id, "funnel": True} def _create_lv0_character(campaign_id: int, pg: dict) -> int: char_id = db.create_character( campaign_id, name=pg["name"], char_class=f"Lv0 ({pg['occupation']})", occupation=pg["occupation"], ) db.update_character( char_id, level=0, str=pg["str"], agi=pg["agi"], sta=pg["sta"], per=pg["per"], int=pg["int"], lck=pg["lck"], hp_current=pg["hp_current"], hp_max=pg["hp_max"], ac=pg["ac"], armor_type=pg["armor_type"], weapon=pg["weapon"], equipment=f'["{pg["weapon"]}", "{pg["belonging"]}"]', ) return char_id @app.post("/api/campaign/{campaign_id}/character") def add_character( campaign_id: int, name: str = "", char_class: str = "", occupation: str = "", ): if name and char_class: char_id = db.create_character(campaign_id, name, char_class, occupation) db.update_character(char_id, level=1, str=12, agi=11, sta=12, per=10, int=10, lck=10, hp_current=8, hp_max=8, ac=12) return {"character_id": char_id} pg = genera_pg_lv0(name) char_id = _create_lv0_character(campaign_id, pg) return {"character_id": char_id, "level": 0, "name": pg["name"], "occupation": pg["occupation"]} # --------------------------------------------------------------------------- # WebSocket — real-time game loop # --------------------------------------------------------------------------- _executor = ThreadPoolExecutor(max_workers=4) @app.websocket("/ws/{campaign_id}") async def game_ws(ws: WebSocket, campaign_id: int): await ws.accept() try: ctx = GameContext.load(campaign_id) except ValueError: await ws.send_json({"type": "error", "message": "Campagna non trovata"}) await ws.close() return # Send initial state await ws.send_json({"type": "state", "campaign": _build_state(campaign_id)}) while True: try: raw = await ws.receive_text() except WebSocketDisconnect: break try: msg = json.loads(raw) except json.JSONDecodeError: msg = {"type": "action", "payload": raw} if msg.get("type") == "action": player_action = msg.get("payload", "").strip() if not player_action: continue try: # dm_cycle è bloccante (chiamata HTTP al LLM) — # la eseguiamo in un thread worker per non bloccare # il loop asyncio e tenere vivi i ping WebSocket out = await asyncio.get_event_loop().run_in_executor( _executor, dm_cycle, campaign_id, player_action ) await ws.send_json({ "type": "response", "turn": out["turn"], "tool": out["tool"], "params": out["params"], "narration": out["narration"], "mechanics": out["mechanics"], "full_output": out["full_output"], }) except Exception as e: log.exception("DM cycle failed") try: await ws.send_json({"type": "error", "message": str(e)}) except WebSocketDisconnect: break # Send updated state after action try: await ws.send_json({"type": "state", "campaign": _build_state(campaign_id)}) except WebSocketDisconnect: break elif msg.get("type") == "ping": try: await ws.send_json({"type": "pong"}) except WebSocketDisconnect: break def _build_state(campaign_id: int) -> dict: ctx = GameContext.load(campaign_id) all_areas = db.get_all_areas() area_map = {a["id"]: a for a in all_areas} enriched = [] for sa in ctx.areas: aid = sa["area_id"] full = area_map.get(aid, {}) raw_conns = full.get("connections", []) conn_names = [] for cid in raw_conns: linked = area_map.get(cid) conn_names.append(linked["name"] if linked else cid) enriched.append({ "id": aid, "name": full.get("name", aid), "visited": bool(sa.get("visited")), "connections": conn_names, }) return { "turn": ctx.turn, "characters": ctx.characters, "quests": ctx.quests, "areas": enriched, } # --------------------------------------------------------------------------- # Serve frontend # --------------------------------------------------------------------------- @app.get("/health") def health(): return {"status": "ok", "provider": load_from_env().kind} @app.get("/") async def index(): html_path = STATIC_DIR / "index.html" if html_path.exists(): content = html_path.read_text(encoding="utf-8") return HTMLResponse(content) return HTMLResponse("

Frontend non trovato

", status_code=404) # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main(): logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) db.init_db() cfg = load_from_env() log.info( "DM config: provider=%s model=%s", cfg.kind, getattr(cfg, f"{cfg.kind}_model", "?"), ) import uvicorn uvicorn.run(app, host="0.0.0.0", port=8300, log_level="info") if __name__ == "__main__": main()