feat: AI Dungeon Master engine + web frontend
- Engine: stato SQLite, state machine, 10 approved tools, parser azioni - LLM: multi-provider (Ollama/OpenRouter/Local), system prompt DCC - DM cycle: LLM decide tool → engine esegue → narrazione + meccanica - Dati avventura: JSON machine-readable con 9 aree, mostri, enigmi, quest - Web: FastAPI + WebSocket backend, SPA frontend dark fantasy - Ref: architecture-patterns/neon-scratch-lounge (separazione reasoning/execution)
This commit is contained in:
truetype committed
1 parent
ae73fe43d3
commit
90f823fdb9
17 files changed
+4082
No files matched your search
@@ -0,0 +1,227 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
|
||||
from fastapi.responses import HTMLResponse, JSONResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from src.engine import db
|
||||
from src.engine.config import load_from_env
|
||||
from src.engine.dm import dm_cycle
|
||||
from src.engine.state_machine 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, {})
|
||||
enriched.append({
|
||||
"id": aid,
|
||||
"name": full.get("name", aid),
|
||||
"description": full.get("description", "")[:200],
|
||||
"connections": full.get("connections", []),
|
||||
"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)
|
||||
return {"campaign_id": camp_id}
|
||||
|
||||
|
||||
@app.post("/api/campaign/{campaign_id}/character")
|
||||
def add_character(
|
||||
campaign_id: int,
|
||||
name: str = "Eroe",
|
||||
char_class: str = "",
|
||||
occupation: str = "",
|
||||
):
|
||||
char_id = db.create_character(campaign_id, name, char_class, occupation)
|
||||
# Default stats for a lv0/lv1 character
|
||||
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}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WebSocket — real-time game loop
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@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:
|
||||
raw = await ws.receive_text()
|
||||
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:
|
||||
out = 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")
|
||||
await ws.send_json({"type": "error", "message": str(e)})
|
||||
|
||||
# Send updated state after action
|
||||
await ws.send_json({"type": "state", "campaign": _build_state(campaign_id)})
|
||||
|
||||
elif msg.get("type") == "ping":
|
||||
await ws.send_json({"type": "pong"})
|
||||
|
||||
|
||||
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, {})
|
||||
enriched.append({
|
||||
"id": aid,
|
||||
"name": full.get("name", aid),
|
||||
"visited": bool(sa.get("visited")),
|
||||
"connections": full.get("connections", []),
|
||||
})
|
||||
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("<h1>Frontend non trovato</h1>", 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()
|
||||
Reference in new issue
Block a user