- Handler WS type:dm — esegue dm_override bypassando il LLM - Pannello collassabile con selettore comandi e campi dinamici - Comandi: narrate, set_hp, add_item, force_roll, skip_combat - Stato aggiornato dopo DM override - Implementato narrate con testo personalizzato
524 lines
19 KiB
Python
524 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
from src.engine import dice
|
|
from src.engine import db as storage
|
|
|
|
|
|
ToolResult = dict[str, Any]
|
|
|
|
TOOL_REGISTRY: dict[str, type] = {}
|
|
|
|
def tool(fn):
|
|
TOOL_REGISTRY[fn.__name__] = fn
|
|
return fn
|
|
|
|
|
|
@tool
|
|
def roll_dice(die: str, count: int = 1, modifier: int = 0) -> ToolResult:
|
|
try:
|
|
r = dice.roll(die, count, modifier)
|
|
return {
|
|
"success": True,
|
|
"tool": "roll-dice",
|
|
"die": r["die"],
|
|
"count": r["count"],
|
|
"modifier": r["modifier"],
|
|
"results": r["results"],
|
|
"total": r["total"],
|
|
"crit": r["crit"],
|
|
"fumble": r["fumble"],
|
|
"message": f"Tiro: {r['raw']} → {r['results']} (totale: {r['total']})",
|
|
}
|
|
except ValueError as e:
|
|
return {"success": False, "tool": "roll-dice", "message": str(e)}
|
|
|
|
|
|
@tool
|
|
def dice_chain(die: str, steps: int) -> ToolResult:
|
|
res = dice.dice_chain(die, steps)
|
|
if "error" in res:
|
|
return {"success": False, "tool": "dice-chain", "message": res["error"]}
|
|
return {
|
|
"success": True,
|
|
"tool": "dice-chain",
|
|
"input_die": res["input_die"],
|
|
"steps": res["steps"],
|
|
"output_die": res["output_die"],
|
|
"message": f"Catena dadi: {die} {'+' if steps >= 0 else ''}{steps} → {res['output_die']}",
|
|
}
|
|
|
|
|
|
@tool
|
|
def combat(
|
|
character_id: int,
|
|
target_ac: int,
|
|
attack_bonus: int = 0,
|
|
damage_die: str = "d8",
|
|
damage_bonus: int = 0,
|
|
mighty_deed_die: str = "",
|
|
luck_mod: int = 0,
|
|
armor_type: str = "none",
|
|
monster_id: str = "",
|
|
campaign_id: int = 0,
|
|
) -> ToolResult:
|
|
char = storage.get_character(character_id)
|
|
if not char:
|
|
return {"success": False, "tool": "combat", "message": "Personaggio non trovato."}
|
|
|
|
init_roll = dice.roll("d20")
|
|
atk_roll = dice.roll("d20", modifier=attack_bonus)
|
|
raw_total = atk_roll["total"]
|
|
|
|
deed_roll = None
|
|
if mighty_deed_die:
|
|
deed_roll = dice.roll(mighty_deed_die)
|
|
raw_total += deed_roll["total"]
|
|
|
|
hit = raw_total >= target_ac
|
|
result_parts = []
|
|
crit_res = None
|
|
|
|
# HP persistenti: carica mostro da combat_state se disponibile
|
|
monster_name = ""
|
|
monster_hp_before = None
|
|
if monster_id and campaign_id:
|
|
state = storage.get_combat_state(campaign_id)
|
|
monster_state = next((m for m in state if m["monster_id"] == monster_id), None)
|
|
if monster_state:
|
|
monster_name = monster_state["monster_name"]
|
|
monster_hp_before = monster_state["current_hp"]
|
|
result_parts.append(f"{monster_name}: PF {monster_state['current_hp']}/{monster_state['max_hp']}")
|
|
|
|
if atk_roll["fumble"]:
|
|
fumble_die_name = storage.get_fumble_die(armor_type) or "d4"
|
|
fumble_roll = dice.roll(fumble_die_name)
|
|
fumble_res = storage.lookup_crit_fumble("fumble", fumble_roll["total"], luck_mod)
|
|
crit_res = fumble_res
|
|
result_parts.append(
|
|
f"Fallimento critico! Tiro fumble (d{armor_type}): {fumble_roll['total']} → "
|
|
+ (fumble_res["result"] if fumble_res else "Nessun risultato trovato.")
|
|
)
|
|
|
|
if atk_roll["crit"]:
|
|
crit_table_id = _crit_table_for(char)
|
|
crit_die_name = _crit_die_for(char)
|
|
crit_roll = dice.roll(crit_die_name)
|
|
crit_res = storage.lookup_crit_fumble(crit_table_id, crit_roll["total"], luck_mod)
|
|
result_parts.append(
|
|
f"Colpo critico! Tiro crit ({crit_die_name}): {crit_roll['total']} → "
|
|
+ (crit_res["result"] if crit_res else "Nessun risultato trovato.")
|
|
)
|
|
|
|
# Calcolo danno e persistenza
|
|
total_dmg = 0
|
|
new_hp = None
|
|
monster_defeated = False
|
|
if hit and not atk_roll["fumble"]:
|
|
dmg = dice.roll(damage_die, modifier=damage_bonus)
|
|
crit_dmg_bonus = 0
|
|
if atk_roll["crit"] and crit_res:
|
|
crit_dmg_bonus = _parse_extra_damage(crit_res.get("result", ""))
|
|
total_dmg = dmg["total"] + crit_dmg_bonus
|
|
result_parts.append(f"Colpito! Danni: {dmg['results']} + {crit_dmg_bonus} extra = {total_dmg}")
|
|
|
|
if monster_id and campaign_id and monster_hp_before is not None:
|
|
hp_result = storage.update_combat_hp(campaign_id, monster_id, total_dmg)
|
|
if hp_result:
|
|
new_hp = hp_result["new_hp"]
|
|
monster_defeated = hp_result["status"] == "defeated"
|
|
result_parts.append(f"PF {monster_name}: {monster_hp_before} → {new_hp}")
|
|
if monster_defeated:
|
|
result_parts.append(f"{monster_name} sconfitto!")
|
|
elif not atk_roll["fumble"]:
|
|
result_parts.append("Mancato!")
|
|
|
|
deed_msg = f" Deed: {deed_roll['total']}" if deed_roll else ""
|
|
|
|
return {
|
|
"success": True,
|
|
"tool": "combat",
|
|
"initiative": init_roll["total"],
|
|
"attack_raw": atk_roll["results"][0],
|
|
"attack_bonus": attack_bonus,
|
|
"attack_total": raw_total,
|
|
"target_ac": target_ac,
|
|
"hit": hit,
|
|
"crit": atk_roll["crit"],
|
|
"fumble": atk_roll["fumble"],
|
|
"deed": deed_roll["total"] if deed_roll else None,
|
|
"monster_hp_before": monster_hp_before,
|
|
"monster_hp_after": new_hp,
|
|
"monster_defeated": monster_defeated,
|
|
"message": f"Iniziativa: {init_roll['total']} | Attacco d20+{attack_bonus}: {atk_roll['results'][0]}+{attack_bonus}={atk_roll['total']} vs CA {target_ac} | {'COLPI TO!' if hit else 'Mancato.'}{deed_msg} | " + " | ".join(result_parts),
|
|
}
|
|
|
|
|
|
def _crit_table_for(char: dict) -> str:
|
|
cls = (char.get("char_class") or "").lower()
|
|
lv = char.get("level", 0)
|
|
if cls in ("guerriero",):
|
|
return "crit_V" if lv >= 5 else "crit_IV" if lv >= 3 else "crit_III"
|
|
if cls in ("nano",):
|
|
return "crit_V" if lv >= 6 else "crit_IV" if lv >= 4 else "crit_III"
|
|
if cls in ("ladro", "elfo"):
|
|
return "crit_II"
|
|
if cls in ("chierico", "halfling"):
|
|
return "crit_III"
|
|
if cls in ("mago",) or lv == 0:
|
|
return "crit_I"
|
|
return "crit_I"
|
|
|
|
|
|
def _crit_die_for(char: dict) -> str:
|
|
cls = (char.get("char_class") or "").lower()
|
|
lv = char.get("level", 0)
|
|
if cls in ("guerriero",):
|
|
return "d12" if lv >= 5 else "d10" if lv >= 3 else "d8"
|
|
if cls in ("nano",):
|
|
return "d12" if lv >= 6 else "d10" if lv >= 4 else "d8"
|
|
if cls in ("ladro", "elfo"):
|
|
return "d6"
|
|
if cls in ("chierico", "halfling"):
|
|
return "d8"
|
|
if cls in ("mago",) or lv == 0:
|
|
return "d4"
|
|
return "d4"
|
|
|
|
|
|
def _parse_extra_damage(result_text: str) -> int:
|
|
matches = re.findall(r'(\+?)(\d+)d\d+', result_text)
|
|
total = 0
|
|
for sign, num in matches:
|
|
if sign == "+":
|
|
total += int(num)
|
|
else:
|
|
total += int(num)
|
|
return total
|
|
|
|
|
|
@tool
|
|
def ability_check(character_id: int, dc: int, modifier: int = 0) -> ToolResult:
|
|
char = storage.get_character(character_id)
|
|
if not char:
|
|
return {"success": False, "tool": "ability-check", "message": "Personaggio non trovato."}
|
|
atk = dice.roll("d20", modifier=modifier)
|
|
passed = atk["total"] >= dc
|
|
return {
|
|
"success": True,
|
|
"tool": "ability-check",
|
|
"roll": atk["results"][0],
|
|
"modifier": modifier,
|
|
"total": atk["total"],
|
|
"dc": dc,
|
|
"passed": passed,
|
|
"crit": atk["crit"],
|
|
"fumble": atk["fumble"],
|
|
"message": f"Prova abilità: d20+{modifier} → {atk['results'][0]}+{modifier}={atk['total']} vs CD {dc} → {'Successo!' if passed else 'Fallimento.'}",
|
|
}
|
|
|
|
|
|
@tool
|
|
def saving_throw(character_id: int, save_type: str, dc: int, modifier: int = 0) -> ToolResult:
|
|
char = storage.get_character(character_id)
|
|
if not char:
|
|
return {"success": False, "tool": "saving-throw", "message": "Personaggio non trovato."}
|
|
valid = ("Tempra", "Riflessi", "Volontà")
|
|
if save_type not in valid:
|
|
return {"success": False, "tool": "saving-throw", "message": f"Tipo non valido: {save_type}. Usa: {', '.join(valid)}"}
|
|
atk = dice.roll("d20", modifier=modifier)
|
|
passed = atk["total"] >= dc
|
|
return {
|
|
"success": True,
|
|
"tool": "saving-throw",
|
|
"save_type": save_type,
|
|
"roll": atk["results"][0],
|
|
"modifier": modifier,
|
|
"total": atk["total"],
|
|
"dc": dc,
|
|
"passed": passed,
|
|
"crit": atk["crit"],
|
|
"fumble": atk["fumble"],
|
|
"message": f"Tiro Salvezza su {save_type}: d20+{modifier} → {atk['results'][0]}+{modifier}={atk['total']} vs CD {dc} → {'Superato!' if passed else 'Fallito.'}",
|
|
}
|
|
|
|
|
|
@tool
|
|
def movement(distance_m: int, terrain: str = "normale", stealth: bool = False) -> ToolResult:
|
|
speeds = {"normale": 12, "difficile": 6, "molto difficile": 3}
|
|
speed = speeds.get(terrain, 12)
|
|
rounds = max(1, -(-distance_m // speed))
|
|
stealth_msg = " (in furtivo)" if stealth else ""
|
|
return {
|
|
"success": True,
|
|
"tool": "movement",
|
|
"distance_m": distance_m,
|
|
"terrain": terrain,
|
|
"speed_m_per_round": speed,
|
|
"rounds": rounds,
|
|
"stealth": stealth,
|
|
"message": f"Movimento: {distance_m}m su terreno {terrain}{stealth_msg} → {rounds} round ({speed}m/round).",
|
|
}
|
|
|
|
|
|
@tool
|
|
def spellcasting(character_id: int, spell_name: str, spell_check: int, dc: int = 12) -> ToolResult:
|
|
char = storage.get_character(character_id)
|
|
if not char:
|
|
return {"success": False, "tool": "spellcasting", "message": "Personaggio non trovato."}
|
|
result_tier = "fallimento"
|
|
outcome = ""
|
|
if spell_check == 1:
|
|
outcome = "Fallimento critico! L'incantesimo si ritorce."
|
|
elif spell_check < dc:
|
|
outcome = f"L'incantesimo fallisce ({spell_check} < CD {dc})."
|
|
elif spell_check <= 11:
|
|
result_tier = "minore"
|
|
outcome = f"Successo minore ({spell_check}). Effetto base."
|
|
elif spell_check <= 17:
|
|
result_tier = "moderato"
|
|
outcome = f"Successo moderato ({spell_check}). Effetto migliorato."
|
|
elif spell_check <= 19:
|
|
result_tier = "maggiore"
|
|
outcome = f"Successo maggiore ({spell_check}). Effetto potente."
|
|
else:
|
|
result_tier = "massimo"
|
|
outcome = f"Successo massimo! ({spell_check}). Effetto devastante."
|
|
corruption_check = spell_check == 1 or (spell_check >= 12 and dice.roll("d20")["total"] <= 1)
|
|
return {
|
|
"success": True,
|
|
"tool": "spellcasting",
|
|
"spell": spell_name,
|
|
"spell_check": spell_check,
|
|
"dc": dc,
|
|
"result_tier": result_tier,
|
|
"corruption": corruption_check,
|
|
"message": f"Incantesimo: {spell_name} → Tiro {spell_check} vs CD {dc} → {outcome}" + (" (Corruzione!)" if corruption_check else ""),
|
|
}
|
|
|
|
|
|
@tool
|
|
def update_inventory(character_id: int, item: str = "", action: str = "add") -> ToolResult:
|
|
char = storage.get_character(character_id)
|
|
if not char:
|
|
return {"success": False, "tool": "update-inventory", "message": "Personaggio non trovato."}
|
|
|
|
if action == "add":
|
|
if not item:
|
|
return {"success": False, "tool": "update-inventory", "message": "Specificare un oggetto da aggiungere."}
|
|
storage.add_character_item(character_id, item)
|
|
elif action == "remove":
|
|
if not item:
|
|
return {"success": False, "tool": "update-inventory", "message": "Specificare un oggetto da rimuovere."}
|
|
if not storage.remove_character_item(character_id, item):
|
|
return {"success": False, "tool": "update-inventory", "message": f"Oggetto '{item}' non trovato."}
|
|
elif action == "equip":
|
|
if not item:
|
|
return {"success": False, "tool": "update-inventory", "message": "Specificare un oggetto da equipaggiare."}
|
|
if not storage.equip_character_item(character_id, item, True):
|
|
return {"success": False, "tool": "update-inventory", "message": f"Oggetto '{item}' non trovato."}
|
|
elif action == "unequip":
|
|
if not item:
|
|
return {"success": False, "tool": "update-inventory", "message": "Specificare un oggetto da rimuovere."}
|
|
if not storage.equip_character_item(character_id, item, False):
|
|
return {"success": False, "tool": "update-inventory", "message": f"Oggetto '{item}' non trovato."}
|
|
elif action == "list":
|
|
items = storage.get_character_items(character_id)
|
|
names = [f"{i['name']} x{i['quantity']}{' (eq.)' if i['equipped'] else ''}" for i in items]
|
|
return {
|
|
"success": True,
|
|
"tool": "update-inventory",
|
|
"action": action,
|
|
"inventory": [dict(i) for i in items],
|
|
"message": f"Inventario: {len(items)} oggetti — " + (", ".join(names) if names else "vuoto."),
|
|
}
|
|
else:
|
|
return {"success": False, "tool": "update-inventory", "message": f"Azione '{action}' non valida. Usa: add, remove, equip, unequip, list."}
|
|
|
|
# Sincronizza campo equipment legacy per backward compat
|
|
items = storage.get_character_items(character_id)
|
|
names = [i["name"] for i in items if i.get("quantity", 1) > 0]
|
|
storage.update_character(character_id, equipment=json.dumps(names))
|
|
|
|
return {
|
|
"success": True,
|
|
"tool": "update-inventory",
|
|
"action": action,
|
|
"item": item,
|
|
"inventory": [dict(i) for i in storage.get_character_items(character_id)],
|
|
"message": f"Inventario: {action} '{item}' → aggiornato.",
|
|
}
|
|
|
|
|
|
@tool
|
|
def award_xp(character_id: int, xp: int) -> ToolResult:
|
|
char = storage.get_character(character_id)
|
|
if not char:
|
|
return {"success": False, "tool": "award-xp", "message": "Personaggio non trovato."}
|
|
new_xp = (char.get("xp", 0) or 0) + xp
|
|
storage.update_character(character_id, xp=new_xp)
|
|
messages = [f"{char['name']} guadagna {xp} PE (totale: {new_xp})."]
|
|
leveled = False
|
|
if new_xp >= 10 and char.get("level", 0) == 0:
|
|
storage.update_character(character_id, level=1)
|
|
messages.append(f"{char['name']} sale a livello 1!")
|
|
leveled = True
|
|
return {
|
|
"success": True,
|
|
"tool": "award-xp",
|
|
"xp_gained": xp,
|
|
"xp_total": new_xp,
|
|
"leveled_up": leveled,
|
|
"message": " ".join(messages),
|
|
}
|
|
|
|
|
|
@tool
|
|
def update_quest(quest_id: int, status: str = "active", name: str = "", description: str = "") -> ToolResult:
|
|
VALID_STATUSES = ["active", "completed", "failed"]
|
|
if status not in VALID_STATUSES:
|
|
return {
|
|
"success": False,
|
|
"tool": "update-quest",
|
|
"message": f"Status '{status}' non valido. Usa: {', '.join(VALID_STATUSES)}",
|
|
}
|
|
quest = storage.get_quest(quest_id)
|
|
if not quest:
|
|
return {"success": False, "tool": "update-quest", "message": f"Missione #{quest_id} non trovata."}
|
|
|
|
changed = []
|
|
if status != quest["status"]:
|
|
changed.append(f"stato: {quest['status']}→{status}")
|
|
if name and name != quest.get("name"):
|
|
changed.append(f"nome: {quest.get('name')}→{name}")
|
|
if description and description != quest.get("description"):
|
|
changed.append("descrizione aggiornata")
|
|
|
|
storage.update_quest(quest_id, status=status, name=name, description=description)
|
|
display_name = name or quest["name"]
|
|
return {
|
|
"success": True,
|
|
"tool": "update-quest",
|
|
"quest_id": quest_id,
|
|
"name": display_name,
|
|
"status": status,
|
|
"message": f"Missione '{display_name}' → {status}." + (f" ({', '.join(changed)})" if changed else ""),
|
|
}
|
|
|
|
|
|
@tool
|
|
def dm_override(command: str, params: dict | None = None) -> ToolResult:
|
|
manual = {
|
|
"set_hp": "Imposta PF: {character_id, value}",
|
|
"force_roll": "Forza risultato dado: {die, forced_value}",
|
|
"skip_combat": "Salta un combattimento: {area_id}",
|
|
"add_item": "Aggiunge oggetto: {character_id, item}",
|
|
"narrate": "Testo narrativo forzato: {text}",
|
|
}
|
|
if command == "help":
|
|
return {
|
|
"success": True,
|
|
"tool": "dm-override",
|
|
"commands": manual,
|
|
"message": "Comandi DM disponibili: " + ", ".join(manual.keys()),
|
|
}
|
|
|
|
if command == "narrate":
|
|
text = (params or {}).get("text", "")
|
|
return {
|
|
"success": True,
|
|
"tool": "dm-override",
|
|
"command": command,
|
|
"params": params or {},
|
|
"message": text or "Override DM: narrate applicato.",
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"tool": "dm-override",
|
|
"command": command,
|
|
"params": params or {},
|
|
"message": f"Override DM: {command} applicato.",
|
|
}
|
|
|
|
|
|
def run(tool_name: str, **kwargs) -> ToolResult:
|
|
if tool_name not in TOOL_REGISTRY:
|
|
return {
|
|
"success": False,
|
|
"tool": tool_name,
|
|
"message": f"Tool '{tool_name}' sconosciuto. Disponibili: {', '.join(sorted(TOOL_REGISTRY))}",
|
|
}
|
|
try:
|
|
return TOOL_REGISTRY[tool_name](**kwargs)
|
|
except Exception as e:
|
|
return {"success": False, "tool": tool_name, "message": f"Errore: {e}"}
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("=== Tool Registry ===")
|
|
for name in sorted(TOOL_REGISTRY):
|
|
print(f" - {name}")
|
|
|
|
print("\n=== Test roll-dice ===")
|
|
r = run("roll_dice", die="d20")
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Test dice-chain ===")
|
|
r = run("dice_chain", die="d3", steps=2)
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Test unknown tool ===")
|
|
r = run("non_existent")
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Init DB and setup ===")
|
|
conn = storage.init_db()
|
|
camp_id = storage.create_campaign("Test", "Avventura di Prova")
|
|
char_id = storage.create_character(camp_id, "Bob il Contadino", "guerriero", "contadino")
|
|
storage.update_character(char_id,
|
|
level=3, str=14, agi=12, sta=13,
|
|
per=10, int=8, lck=9,
|
|
hp_current=18, hp_max=18,
|
|
ac=16, armor_type="medium"
|
|
)
|
|
print(f" Creata campagna #{camp_id} con personaggio #{char_id}")
|
|
|
|
print("\n=== Test combat ===")
|
|
r = run("combat", character_id=char_id, target_ac=15, attack_bonus=2,
|
|
damage_die="d8", mighty_deed_die="d3", luck_mod=0, armor_type="medium")
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Test ability-check ===")
|
|
r = run("ability_check", character_id=char_id, dc=15, modifier=2)
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Test saving-throw ===")
|
|
r = run("saving_throw", character_id=char_id, save_type="Tempra", dc=15, modifier=1)
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Test movement ===")
|
|
r = run("movement", distance_m=30, terrain="difficile")
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Test spellcasting ===")
|
|
r = run("spellcasting", character_id=char_id, spell_name="Dardo Incantato", spell_check=18, dc=12)
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Test update-inventory ===")
|
|
r = run("update_inventory", character_id=char_id, item="Pozione di Cura")
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Test award-xp ===")
|
|
r = run("award_xp", character_id=char_id, xp=5)
|
|
print(f" {r['message']}")
|
|
|
|
print("\n=== Test dm-override ===")
|
|
r = run("dm_override", command="narrate", params={"text": "Una luce accecante riempie la stanza."})
|
|
print(f" {r['message']}")
|