From 45f7abc4023c7192f51651d90c898b97585309b2 Mon Sep 17 00:00:00 2001 From: Corback Date: Mon, 15 Jun 2026 23:30:51 +0000 Subject: [PATCH] feat: combat_engine + generate_pnj_batch + actions/encounter updated --- src/data/encounter_tables.json | 16 +- src/engine/actions.py | 3 +- src/engine/combat_engine.py | 393 +++++++++++++++++++++++++++++++ src/engine/encounter.py | 17 +- src/engine/generate_pnj_batch.py | 374 +++++++++++++++++++++++++++++ 5 files changed, 792 insertions(+), 11 deletions(-) create mode 100644 src/engine/combat_engine.py create mode 100644 src/engine/generate_pnj_batch.py diff --git a/src/data/encounter_tables.json b/src/data/encounter_tables.json index 01f7fbb..1a3163a 100644 --- a/src/data/encounter_tables.json +++ b/src/data/encounter_tables.json @@ -223,21 +223,21 @@ "outcomes": { "combat": { - "win": {"loot": true, "xp": true, "stress_delta": +2, "description": "Victoire au combat"}, - "loss": {"loot": false, "xp": false, "stress_delta": +5, "hp_loss_pct": 0.25, "description": "Défaite"}, - "draw": {"loot": false, "xp": true, "stress_delta": +3, "hp_loss_pct": 0.15, "description": "Combat indécis"} + "win": {"loot": true, "xp": true, "stress_delta": 2, "description": "Victoire au combat"}, + "loss": {"loot": false, "xp": false, "stress_delta": 5, "hp_loss_pct": 0.25, "description": "Défaite"}, + "draw": {"loot": false, "xp": true, "stress_delta": 3, "hp_loss_pct": 0.15, "description": "Combat indécis"} }, "fuite": { - "success": {"loot": false, "xp": false, "stress_delta": +1, "description": "Fuite réussie"}, - "failed": {"loot": false, "xp": false, "stress_delta": +3, "hp_loss_pct": 0.10, "description": "Fuite ratée, touché"} + "success": {"loot": false, "xp": false, "stress_delta": 1, "description": "Fuite réussie"}, + "failed": {"loot": false, "xp": false, "stress_delta": 3, "hp_loss_pct": 0.10, "description": "Fuite ratée, touché"} }, "evitement": { "success": {"loot": false, "xp": true, "stress_delta": 0, "description": "Discrétion réussie"}, - "failed": {"loot": false, "xp": false, "stress_delta": +2, "description": "Repéré, doit fuir"} + "failed": {"loot": false, "xp": false, "stress_delta": 2, "description": "Repéré, doit fuir"} }, "negociation": { - "success": {"loot": false, "xp": true, "stress_delta": +1, "caps_cost": true, "description": "Négociation réussie"}, - "failed": {"loot": false, "xp": false, "stress_delta": +4, "hp_loss_pct": 0.15, "description": "Négociation échouée, attaqué"} + "success": {"loot": false, "xp": true, "stress_delta": 1, "caps_cost": true, "description": "Négociation réussie"}, + "failed": {"loot": false, "xp": false, "stress_delta": 4, "hp_loss_pct": 0.15, "description": "Négociation échouée, attaqué"} } } } diff --git a/src/engine/actions.py b/src/engine/actions.py index b99a512..34f8528 100644 --- a/src/engine/actions.py +++ b/src/engine/actions.py @@ -16,6 +16,7 @@ Règles (Résumé Écran MJ + Résumé des Règles FR) : import random import db import inventory as inv_module +import combat_engine # --------------------------------------------------------------------------- # Rang de compétence par défaut (sans table skills complète en DB) @@ -437,7 +438,7 @@ def _execute_action(char: dict, decision: dict, target = chars_map.get(decision["target_id"]) if target and target.get("is_alive", True): weapon = _get_equipped_weapon(inv) - resolve_attack(char, target, weapon, session_id, day, tick) + combat_engine.resolve_attack(char, target, weapon, session_id, day, tick) elif action == "eat": food = next((i for i in inv if i["item_type"] == "food" and i["quantity"] >= 1), None) diff --git a/src/engine/combat_engine.py b/src/engine/combat_engine.py new file mode 100644 index 0000000..df8bed7 --- /dev/null +++ b/src/engine/combat_engine.py @@ -0,0 +1,393 @@ +""" +Combat Engine — Fallout : Venice of Wasteland +Résolution complète d'un round de combat 2D20. + +Règles (Résumé Écran MJ + Résumé des Règles FR) : + Test attaque : 2d20 ≤ Attribut + rang_compétence (1=crit=2 réussites, 20=complication) + Dés de combat : d6 — 1→1dmg, 2→2dmg, 3-4→0, 5-6→1dmg+1effet + Corps à corps : FOR + Corps à corps | Bonus melee : FOR<7→+0, 7-8→+1CD, 9-10→+2CD, 11+→+3CD + Distance légère: AGI + Arme légère | Distance lourde: END + Armes lourdes + Énergie : PER + Armes à énergie + Munitions : 1 de base (distance), +1 par effet "En rafale" + Couverture : réduit les dégâts (Résistance = n CD lancés, on soustrait le résultat) + Blessures crit : ≥5 dégâts après RD en un seul coup → jet d20 localisation + Effets d'armes : Brutal, Perforant X, De zone, Étourdissant, Persistant, Radioactif, En rafale +""" + +import random +import db + +# --------------------------------------------------------------------------- +# Table de localisation des blessures critiques (Résumé Écran MJ) +# Déclenchée si dégâts ≥ 5 après couverture en un seul coup +# --------------------------------------------------------------------------- +_CRIT_LOCATION_TABLE = [ + (2, "bras", "Laisse tomber l'objet porté. Plus d'action avec ce bras ce tour."), + (8, "bras", "Laisse tomber l'objet porté. Plus d'action avec ce bras ce tour."), + (11, "jambe", "Tombe à terre. Pas de sprint. Le déplacement devient une action capitale."), + (14, "jambe", "Tombe à terre. Pas de sprint. Le déplacement devient une action capitale."), + (17, "buste", "Blessure interne : subit 2 CD de dégâts ignorant la RD à chaque fin de tour."), + (20, "tete", "Perd ses actions normales au prochain tour. Difficulté +2 sur les tests basés sur la vue."), +] + +def _roll_crit_location(rng: random.Random) -> dict: + roll = rng.randint(1, 20) + for threshold, location, effect in _CRIT_LOCATION_TABLE: + if roll <= threshold: + return {"roll": roll, "location": location, "effect": effect} + return {"roll": roll, "location": "buste", "effect": "Blessure grave."} + + +# --------------------------------------------------------------------------- +# Modificateurs de portée (Résumé Écran MJ — table portées) +# diff_modifier selon portée optimale de l'arme et distance réelle +# --------------------------------------------------------------------------- +_RANGE_DIFF: dict[tuple[str, str], int] = { + # (portée_arme, distance_combat): modificateur difficulté + ("short", "short"): 0, + ("short", "medium"): 1, + ("short", "long"): 2, + ("short", "extreme"): 3, + ("medium", "short"): 1, + ("medium", "medium"): 0, + ("medium", "long"): 1, + ("medium", "extreme"): 2, + ("long", "short"): 2, + ("long", "medium"): 1, + ("long", "long"): 0, + ("long", "extreme"): 1, + ("extreme", "short"): 3, + ("extreme", "medium"): 2, + ("extreme", "long"): 1, + ("extreme", "extreme"): 0, +} + +# --------------------------------------------------------------------------- +# Rang de compétence par défaut (utilisé si skills non trackés en DB) +# --------------------------------------------------------------------------- +_DEFAULT_SKILL_RANK: dict[str, int] = { + "corps_a_corps": 2, + "arme_legere": 2, + "armes_lourdes": 2, + "arme_energie": 1, +} +_ROLE_SKILL_BONUS: dict[str, dict[str, int]] = { + "pillard": {"corps_a_corps": 1, "arme_legere": 1}, + "raider_boss":{"corps_a_corps": 2}, + "garde": {"arme_legere": 1, "corps_a_corps": 1}, + "eclaireur": {"arme_legere": 2}, + "technicien": {"arme_energie": 1}, +} + +def _skill_rank(char: dict, skill: str) -> int: + role = char.get("role", "survivant_generique") + return (_DEFAULT_SKILL_RANK.get(skill, 1) + + _ROLE_SKILL_BONUS.get(role, {}).get(skill, 0)) + +def _melee_bonus_cd(strength: int) -> int: + if strength < 7: return 0 + if strength <= 8: return 1 + if strength <= 10:return 2 + return 3 + + +# --------------------------------------------------------------------------- +# Jets de base +# --------------------------------------------------------------------------- + +def roll_2d20(attribute: int, skill_rank: int, n_dice: int = 2, + diff_modifier: int = 0) -> dict: + """ + Lance n_dice d20. + Réussite : dé ≤ attribute + skill_rank + Dé 1 = réussite critique (2 réussites). Dé 20 = complication. + diff_modifier : s'ajoute à la difficulté cible (portée, couverture partielle…) + """ + sr = attribute + skill_rank + rolls = [random.randint(1, 20) for _ in range(n_dice)] + successes = sum(2 if r == 1 else 1 for r in rolls if r <= sr) + complications = sum(1 for r in rolls if r == 20) + return { + "rolls": rolls, "sr": sr, + "successes": successes, "complications": complications, + "diff_modifier": diff_modifier, + } + + +def roll_cd(n_dice: int, rng: random.Random | None = None) -> dict: + """ + Lance n_dice dés de combat (d6). + 1→1dmg | 2→2dmg | 3-4→0 | 5-6→1dmg+1effet + """ + _r = rng or random + damage = effects = 0 + rolls = [] + for _ in range(max(1, n_dice)): + r = _r.randint(1, 6) + rolls.append(r) + if r == 1: damage += 1 + elif r == 2: damage += 2 + elif r >= 5: damage += 1; effects += 1 + return {"rolls": rolls, "damage": damage, "effects": effects} + + +def _cover_reduction(cover_rating: int, rng: random.Random) -> int: + """ + Lance cover_rating CD, retourne les dégâts absorbés. + (Résumé Écran MJ : Feuillage→1, Gravats/grillage→2, Béton/acier→3) + """ + if cover_rating <= 0: + return 0 + cd = roll_cd(cover_rating, rng) + return cd["damage"] + + +# --------------------------------------------------------------------------- +# Résolution des effets d'armes +# --------------------------------------------------------------------------- + +def _resolve_weapon_effects(effects_count: int, weapon_props: dict, + base_damage: int, rng: random.Random) -> dict: + """ + Applique les effets de l'arme selon le nombre d'Effets obtenus. + Retourne un dict avec les bonus/malus à appliquer. + """ + extra_damage = 0 + status_effects = [] + ammo_extra = 0 + + for _ in range(effects_count): + if weapon_props.get("Brutal"): + extra_damage += 1 + + if weapon_props.get("Etourdissant"): + status_effects.append("etourdi") + + if weapon_props.get("Persistant"): + status_effects.append("persistant") + + if weapon_props.get("Radioactif"): + status_effects.append("radioactif_1rad") + + if weapon_props.get("En_rafale"): + status_effects.append("en_rafale_cible_proche") + ammo_extra += 1 + + perforant = weapon_props.get("Perforant", 0) + + return { + "extra_damage": extra_damage, + "perforant": perforant, # ignore X pts de RD par Effet + "status_effects": status_effects, + "ammo_extra": ammo_extra, + } + + +# --------------------------------------------------------------------------- +# Point d'entrée principal +# --------------------------------------------------------------------------- + +def full_combat_round( + attacker: dict, + defender: dict, + weapon: dict | None, + combat_range: str = "medium", + cover_rating: int = 0, + rng: random.Random | None = None, +) -> dict: + """ + Résout un round de combat complet selon les règles Fallout 2D20. + + Args: + attacker : dict personnage attaquant (doit avoir les attributs SPECIAL) + defender : dict personnage défenseur + weapon : dict item équipé (None = mains nues) + combat_range : distance réelle du combat ('short'|'medium'|'long'|'extreme') + cover_rating : RD de la couverture (0-3) + rng : Random optionnel pour reproductibilité + + Returns: + dict complet avec test, dégâts, effets, blessure critique éventuelle, + munitions consommées, résumé narrative. + """ + _rng = rng or random.Random() + + # --- Attribut + compétence selon type d'arme --- + weapon_type = weapon.get("item_type", "weapon_melee") if weapon else "weapon_melee" + weapon_range = weapon.get("damage_range", "short") if weapon else "short" + weapon_props = weapon.get("properties") or {} + if isinstance(weapon_props, str): + import json + weapon_props = json.loads(weapon_props) + + if weapon_type == "weapon_melee" or weapon is None: + attr_val = attacker["strength"] + skill = "corps_a_corps" + is_melee = True + elif weapon_type == "weapon_energy": + attr_val = attacker["perception"] + skill = "arme_energie" + is_melee = False + else: + if weapon_props.get("Lourd"): + attr_val = attacker["endurance"] + skill = "armes_lourdes" + else: + attr_val = attacker["agility"] + skill = "arme_legere" + is_melee = False + + rank = _skill_rank(attacker, skill) + + # --- Modificateur de portée --- + range_mod = _RANGE_DIFF.get((weapon_range, combat_range), 0) if not is_melee else 0 + + # --- Défense cible --- + difficulty = (2 if defender.get("agility", 5) >= 9 else 1) + range_mod + + # --- Jet d'attaque --- + test = roll_2d20(attr_val, rank, diff_modifier=range_mod) + hit = test["successes"] >= difficulty + + result: dict = { + "hit": hit, + "test": test, + "difficulty": difficulty, + "skill": skill, + "weapon": weapon["item_slug"] if weapon else "mains_nues", + "weapon_name": weapon["item_name"] if weapon else "Mains nues", + "combat_range": combat_range, + "cover_rating": cover_rating, + "damage_raw": 0, + "damage_cover": 0, + "damage_final": 0, + "extra_damage": 0, + "effects": 0, + "weapon_effects_applied": [], + "critical_wound": None, + "status_effects": [], + "ammo_consumed": 0, + "killed": False, + "narrative": "", + } + + if not hit: + result["narrative"] = ( + f"{attacker['name']} attaque {defender['name']} " + f"({result['weapon_name']}) — RATÉ " + f"[SR={test['sr']}, dés={test['rolls']}, diff={difficulty}]" + ) + return result + + # --- Dégâts bruts --- + n_cd = (weapon.get("damage_dice", 1) if weapon else 1) + if is_melee: + n_cd += _melee_bonus_cd(attacker["strength"]) + + cd_result = roll_cd(n_cd, _rng) + raw_damage = cd_result["damage"] + effects_count = cd_result["effects"] + + # --- Effets d'arme --- + eff = _resolve_weapon_effects(effects_count, weapon_props, raw_damage, _rng) + raw_damage += eff["extra_damage"] + result["extra_damage"] = eff["extra_damage"] + result["effects"] = effects_count + result["weapon_effects_applied"] = eff["status_effects"] + result["ammo_consumed"] = (0 if is_melee else 1) + eff["ammo_extra"] + + # --- Couverture (RD) --- + cover_absorbed = _cover_reduction(cover_rating, _rng) + # Perforant réduit l'efficacité de la couverture + perforant_total = eff["perforant"] * effects_count + cover_absorbed = max(0, cover_absorbed - perforant_total) + + final_damage = max(0, raw_damage - cover_absorbed) + + result["damage_raw"] = raw_damage + result["damage_cover"] = cover_absorbed + result["damage_final"] = final_damage + + # --- Blessure critique (≥5 dégâts après couverture) --- + if final_damage >= 5: + crit = _roll_crit_location(_rng) + result["critical_wound"] = crit + + # --- Application en DB --- + new_hp = max(0, defender["hp"] - final_damage) + db.update_character(defender["id"], hp=new_hp) + + if new_hp == 0: + db.update_character(defender["id"], is_alive=False) + result["killed"] = True + + # --- Radiations (effets Radioactif) --- + rad_hits = result["weapon_effects_applied"].count("radioactif_1rad") + if rad_hits > 0 and not defender.get("immune_radiation", False): + new_rads = defender.get("rads", 0) + rad_hits + db.update_character(defender["id"], rads=new_rads) + result["rads_inflicted"] = rad_hits + + # --- Statuts persistants (stockés dans pnj_state.last_actions) --- + result["status_effects"] = [ + s for s in eff["status_effects"] + if s not in ("radioactif_1rad", "en_rafale_cible_proche") + ] + + # --- Narrative --- + crit_txt = ( + f" [BLESSURE CRITIQUE — {result['critical_wound']['location'].upper()} : " + f"{result['critical_wound']['effect']}]" + if result["critical_wound"] else "" + ) + killed_txt = " [MORT]" if result["killed"] else f" → {new_hp}/{defender['max_hp']} PV" + cover_txt = f" (couverture absorbe {cover_absorbed})" if cover_absorbed else "" + eff_txt = f" +{effects_count} effets" if effects_count else "" + + result["narrative"] = ( + f"{attacker['name']} touche {defender['name']} " + f"({result['weapon_name']}) : " + f"{raw_damage} brut{cover_txt} = {final_damage} dégâts{eff_txt}" + f"{killed_txt}{crit_txt}" + ) + + return result + + +# --------------------------------------------------------------------------- +# Intégration avec actions.py (remplace resolve_attack) +# --------------------------------------------------------------------------- + +def resolve_attack(attacker: dict, defender: dict, + weapon: dict | None, + session_id: int, day: int, tick: int, + combat_range: str = "medium", + cover_rating: int = 0, + rng: random.Random | None = None) -> dict: + """ + Wrapper appelable depuis actions.py et encounter.py. + Lance full_combat_round et logue le résultat en DB. + """ + result = full_combat_round(attacker, defender, weapon, + combat_range, cover_rating, rng) + + db.log_event( + session_id=session_id, day=day, tick=tick, + event_type="combat", + description=result["narrative"], + actor_id=attacker["id"], + location_slug=attacker.get("location_slug"), + mechanical_effect={ + "hit": result["hit"], + "damage_final": result["damage_final"], + "damage_raw": result["damage_raw"], + "damage_cover": result["damage_cover"], + "effects": result["effects"], + "weapon": result["weapon"], + "critical_wound":result["critical_wound"], + "status_effects":result["status_effects"], + "ammo_consumed": result["ammo_consumed"], + "killed": result["killed"], + } + ) + return result diff --git a/src/engine/encounter.py b/src/engine/encounter.py index c10a0fa..d3cdc32 100644 --- a/src/engine/encounter.py +++ b/src/engine/encounter.py @@ -16,6 +16,7 @@ import os import random import db import actions as act_module +import combat_engine _DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data") @@ -538,7 +539,19 @@ def resolve_encounter( outcomes_cfg = tables["outcomes"] if reaction == "combat": - result = _combat_roll(char, entity, rng) + # Utilise le moteur complet si l'entité est un PNJ en DB (actif/boss) + if entity.get("db_id") and entity.get("tier") in ("actif", "boss"): + defender_full = db.get_character_by_id(entity["db_id"]) + inv_att = [] + with db.cursor() as cur: + cur.execute("SELECT * FROM inventory WHERE character_id=%s AND session_id=%s", (char["id"], session_id)) + inv_att = [dict(r) for r in cur.fetchall()] + weapon = act_module._get_equipped_weapon(inv_att) + cr = combat_engine.full_combat_round(char, defender_full, weapon, rng=rng) + result = "win" if cr["hit"] and not cr["killed"] else ("win" if cr["killed"] else "loss") + outcome = result + else: + result = _combat_roll(char, entity, rng) outcome = result cfg = outcomes_cfg["combat"][result] @@ -675,7 +688,7 @@ def process_encounters_tick(session_id: int, day: int, tick: int, rng_seed: int day = day, tick = tick, event_type = f"encounter_{tier}", - character_id = char["id"], + actor_id = char["id"], location_slug = zone, description = result["narrative"], mechanical_effect = { diff --git a/src/engine/generate_pnj_batch.py b/src/engine/generate_pnj_batch.py new file mode 100644 index 0000000..9bf5356 --- /dev/null +++ b/src/engine/generate_pnj_batch.py @@ -0,0 +1,374 @@ +""" +Génération RAG+LLM d'un batch de PNJ pour la simulation. +Lance HORS simulation : python generate_pnj_batch.py + +Pipeline : + 1. Query ChromaDB (RAG) pour règles création personnage Fallout 2D20 + 2. Prompt qwen2.5:14b avec contexte + structure JSON attendue + 3. Parse + validation des PNJ générés + 4. Insertion en DB via character_creator + pnj_factory + +Ce script est one-shot — il ne tourne pas pendant la simulation. +""" + +import sys, os, json, re, time, requests +sys.path.insert(0, os.path.dirname(__file__)) + +import character_creator as cc +import pnj_factory as factory +import db +from config import SESSION_ID + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- + +CHROMA_URL = "http://localhost:8000" # ChromaDB REST API +OLLAMA_URL = "http://localhost:11434" +LLM_MODEL = "qwen2.5:14b" +COLLECTION = "fallout_vst" + +# Combien de PNJ à générer par appel +N_PNJ_TIER2 = 5 # PNJ+ nommés (lieutenants, personnages récurrents) +N_PNJ_TIER3 = 10 # PNJ de passage (archétypes instanciés) + +# --------------------------------------------------------------------------- +# RAG — Récupération contexte depuis ChromaDB +# --------------------------------------------------------------------------- + +def rag_query(query: str, n_results: int = 5) -> str: + """Interroge ChromaDB et retourne le contexte textuel concatené.""" + try: + resp = requests.post( + f"{CHROMA_URL}/api/v2/collections/{COLLECTION}/query", + json={"query_texts": [query], "n_results": n_results}, + timeout=15, + ) + resp.raise_for_status() + data = resp.json() + docs = data.get("documents", [[]])[0] + return "\n\n---\n\n".join(docs) if docs else "" + except Exception as e: + print(f" [RAG WARN] ChromaDB inaccessible : {e}") + return "" + + +def build_rag_context() -> str: + """Combine plusieurs queries RAG pour un contexte riche.""" + queries = [ + "création personnage Fallout 2D20 SPECIAL attributs statistiques", + "origines survivant habitant abri initié confrerie goule super mutant", + "rôles narratifs pillard marchand garde éclaireur docteur technicien", + "factions Louisiana Nouvelle-Orléans post-apocalyptique", + "équipements armes armures Fallout terres désolées bayou", + ] + parts = [] + for q in queries: + ctx = rag_query(q, n_results=3) + if ctx: + parts.append(ctx) + time.sleep(0.2) + return "\n\n=====\n\n".join(parts) + + +# --------------------------------------------------------------------------- +# LLM — Génération des PNJ +# --------------------------------------------------------------------------- + +PROMPT_TIER2 = """Tu es le Maître de Jeu de "Fallout : Venice of Wasteland", un JdR 2D20 post-apocalyptique en Louisiane. + +CONTEXTE DES RÈGLES (extrait du manuel) : +{rag_context} + +FACTIONS EXISTANTES : union, cda, syndicat_capitole, regie, dynaste_oak, ecumeurs, consortium, grand_krewe + +ZONES : independance, laplace, baton_rouge, oak_plantation, la_paroisse, nola_vieux_carre, nola_cbd, nola_fleuve, pearl_river, donaldsonville + +RÔLES DISPONIBLES : garde, marchand, eclaireur, docteur, technicien, pillard, lieutenant, chasseur_primes, passeur, scavenger, survivant_generique + +ORIGINES DISPONIBLES : survivant, habitant_abri, initie_confrerie, goule, super_mutant + +Génère exactement {n} PNJ de TIER 2 (personnages nommés secondaires, lieutenants et personnages récurrents importants). +Chaque PNJ doit avoir : +- Un nom louisianais immersif avec un surnom entre guillemets +- Un rôle cohérent avec sa faction +- Des stats SPECIAL réalistes (1-10, super_mutant peut avoir FOR/END jusqu'à 12) +- Une courte description de personnalité (1 phrase) + +Réponds UNIQUEMENT avec un JSON valide, aucun texte avant ou après : +{{ + "pnj": [ + {{ + "name": "Prénom 'Surnom' Nom", + "role": "garde", + "origin": "survivant", + "faction_slug": "union", + "location_slug": "independance", + "strength": 7, + "perception": 5, + "endurance": 6, + "charisma": 4, + "intelligence": 5, + "agility": 6, + "luck": 5, + "caps": 150, + "description": "Une phrase de description" + }} + ] +}}""" + +PROMPT_TIER3 = """Tu es le Maître de Jeu de "Fallout : Venice of Wasteland". + +ARCHÉTYPES DISPONIBLES (clés pour instanciation) : +{archetypes} + +ZONES : independance, laplace, baton_rouge, oak_plantation, la_paroisse, nola_vieux_carre, nola_cbd, nola_fleuve, pearl_river, donaldsonville + +FACTIONS : union, cda, syndicat_capitole, regie, dynaste_oak, ecumeurs, consortium, grand_krewe (ou null pour sans faction) + +Génère exactement {n} PNJ de TIER 3 (personnages de passage, vendeurs, voyageurs, rencontres). +Chaque PNJ est instancié depuis un archétype. Choisis des archétypes variés et des localisations cohérentes. + +Réponds UNIQUEMENT avec un JSON valide : +{{ + "pnj": [ + {{ + "archetype": "vendeur_ambulant", + "name": "Prénom 'Surnom' Nom", + "faction_slug": "consortium", + "location_slug": "laplace" + }} + ] +}}""" + + +def call_llm(prompt: str, timeout: int = 120) -> str: + """Appelle Ollama et retourne le texte généré.""" + resp = requests.post( + f"{OLLAMA_URL}/api/generate", + json={"model": LLM_MODEL, "prompt": prompt, "stream": False}, + timeout=timeout, + ) + resp.raise_for_status() + return resp.json().get("response", "") + + +def parse_json_response(text: str) -> dict | None: + """Extrait le JSON d'une réponse LLM.""" + match = re.search(r"\{.*\}", text, re.DOTALL) + if not match: + return None + try: + return json.loads(match.group()) + except json.JSONDecodeError as e: + print(f" [PARSE ERR] {e}") + return None + + +# --------------------------------------------------------------------------- +# Validation et normalisation +# --------------------------------------------------------------------------- + +_VALID_ROLES = set(cc._ROLES.keys()) +_VALID_ORIGINS = set(cc.ORIGINS.keys()) + +def validate_pnj_tier2(pnj: dict) -> dict | None: + """Valide et normalise un PNJ tier 2 généré par le LLM.""" + required = ["name", "role", "origin", "faction_slug", "location_slug", + "strength", "perception", "endurance", "charisma", + "intelligence", "agility", "luck"] + for field in required: + if field not in pnj: + print(f" [SKIP] Champ manquant '{field}' pour {pnj.get('name','?')}") + return None + + if pnj["role"] not in _VALID_ROLES: + pnj["role"] = "survivant_generique" + if pnj["origin"] not in _VALID_ORIGINS: + pnj["origin"] = "survivant" + + # Clamper les stats 1-10 (12 pour super_mutant) + cap = 12 if pnj["origin"] == "super_mutant" else 10 + for attr in ["strength","perception","endurance","charisma","intelligence","agility","luck"]: + pnj[attr] = max(1, min(cap, int(pnj[attr]))) + + pnj.setdefault("caps", 200) + return pnj + + +def validate_pnj_tier3(pnj: dict, valid_archetypes: set) -> dict | None: + """Valide un spec PNJ tier 3.""" + if "archetype" not in pnj or pnj["archetype"] not in valid_archetypes: + print(f" [SKIP] Archétype invalide '{pnj.get('archetype','?')}'") + return None + if "name" not in pnj: + print(f" [SKIP] Pas de nom pour archétype {pnj['archetype']}") + return None + return pnj + + +# --------------------------------------------------------------------------- +# Insertion en DB +# --------------------------------------------------------------------------- + +def insert_tier2(pnj_data: dict, seed: int) -> dict | None: + """Insère un PNJ tier 2 généré par LLM en DB avec équipement.""" + try: + # Calcul HP selon règles (END + LCK) + max_hp = pnj_data["endurance"] + pnj_data["luck"] + char_dict = { + "session_id": SESSION_ID, + "name": pnj_data["name"], + "character_type": "pnj_sim", + "faction_slug": pnj_data.get("faction_slug"), + "location_slug": pnj_data.get("location_slug"), + "seed_base": seed, + "role": pnj_data["role"], + "origin": pnj_data["origin"], + "strength": pnj_data["strength"], + "perception": pnj_data["perception"], + "endurance": pnj_data["endurance"], + "charisma": pnj_data["charisma"], + "intelligence": pnj_data["intelligence"], + "agility": pnj_data["agility"], + "luck": pnj_data["luck"], + "hp": max_hp, + "max_hp": max_hp, + "caps": pnj_data.get("caps", 200), + "rads": 0, + "is_alive": True, + "is_in_shelter": False, + "needs_food": cc.ORIGINS[pnj_data["origin"]]["needs_food"], + "needs_water": cc.ORIGINS[pnj_data["origin"]]["needs_water"], + "needs_sleep": cc.ORIGINS[pnj_data["origin"]]["needs_sleep"], + "immune_radiation": cc.ORIGINS[pnj_data["origin"]]["immune_radiation"], + "immune_poison": cc.ORIGINS[pnj_data["origin"]]["immune_poison"], + "description": pnj_data.get("description", ""), + "tier": 2, + "archetype_key": None, + } + char_id = cc._insert_character(char_dict) + char_dict["id"] = char_id + items = factory.equip_character(char_id, pnj_data["role"], SESSION_ID, seed=seed) + return {"char": char_dict, "items": items} + except Exception as e: + print(f" [DB ERR] {pnj_data.get('name','?')} : {e}") + return None + + +def insert_tier3(spec: dict, seed: int) -> dict | None: + """Instancie un archétype tier 3 en DB.""" + try: + char = factory.spawn_from_archetype( + archetype_key = spec["archetype"], + session_id = SESSION_ID, + name = spec["name"], + faction_slug = spec.get("faction_slug"), + location_slug = spec.get("location_slug"), + seed = seed, + tier_override = 3, + ) + return char + except Exception as e: + print(f" [DB ERR] {spec.get('name','?')} : {e}") + return None + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +def main(): + print("=== Génération RAG+LLM de PNJ ===\n") + + # Récupération contexte RAG + print("1. Query RAG (ChromaDB)...") + rag_ctx = build_rag_context() + if rag_ctx: + print(f" Contexte récupéré : {len(rag_ctx)} caractères\n") + else: + print(" Aucun contexte RAG — le LLM utilisera ses connaissances internes\n") + + # Archetypes disponibles pour tier 3 + archetypes_list = factory.list_archetypes() + archetypes_str = "\n".join( + f" - {a['key']} : {a['description']}" for a in archetypes_list + ) + valid_archetype_keys = {a["key"] for a in archetypes_list} + + # ---- Tier 2 ---- + print(f"2. Génération {N_PNJ_TIER2} PNJ+ tier 2 (qwen2.5:14b)...") + prompt2 = PROMPT_TIER2.format( + rag_context=rag_ctx[:3000] if rag_ctx else "(manuel non disponible, utilise tes connaissances)", + n=N_PNJ_TIER2, + ) + try: + t0 = time.time() + resp2 = call_llm(prompt2, timeout=180) + print(f" LLM répondu en {time.time()-t0:.1f}s") + data2 = parse_json_response(resp2) + except Exception as e: + print(f" [LLM ERR] {e}") + data2 = None + + tier2_created = [] + if data2 and "pnj" in data2: + for i, pnj in enumerate(data2["pnj"]): + validated = validate_pnj_tier2(pnj) + if not validated: + continue + result = insert_tier2(validated, seed=3000 + i) + if result: + char = result["char"] + tier2_created.append(char) + print(f" ✓ [T2] {char['name']:<40} {char.get('faction_slug','—'):<20} @ {char.get('location_slug','—')}") + print(f" Items: {', '.join(result['items'])}") + else: + print(" [WARN] Pas de PNJ tier 2 générés") + + # ---- Tier 3 ---- + print(f"\n3. Génération {N_PNJ_TIER3} PNJ de passage tier 3 (qwen2.5:14b)...") + prompt3 = PROMPT_TIER3.format( + archetypes=archetypes_str, + n=N_PNJ_TIER3, + ) + try: + t0 = time.time() + resp3 = call_llm(prompt3, timeout=180) + print(f" LLM répondu en {time.time()-t0:.1f}s") + data3 = parse_json_response(resp3) + except Exception as e: + print(f" [LLM ERR] {e}") + data3 = None + + tier3_created = [] + if data3 and "pnj" in data3: + for i, spec in enumerate(data3["pnj"]): + validated = validate_pnj_tier3(spec, valid_archetype_keys) + if not validated: + continue + char = insert_tier3(validated, seed=4000 + i) + if char: + tier3_created.append(char) + print(f" ✓ [T3] {char.get('name','?'):<40} archetype={spec['archetype']:<20} @ {spec.get('location_slug','—')}") + else: + print(" [WARN] Pas de PNJ tier 3 générés") + + # ---- Résumé ---- + print(f"\n=== Résumé ===") + print(f" Tier 2 créés : {len(tier2_created)}") + print(f" Tier 3 créés : {len(tier3_created)}") + + all_chars = db.get_all_characters(SESSION_ID) + by_tier = {} + for c in all_chars: + t = c.get("tier") or 1 + by_tier[t] = by_tier.get(t, 0) + 1 + labels = {0:"BOSS", 1:"ACTIFS SIM", 2:"PNJ+", 3:"PASSAGE"} + print("\n DB total par tier :") + for t in sorted(by_tier): + print(f" Tier {t} ({labels.get(t,'?')}) : {by_tier[t]} PNJ") + + +if __name__ == "__main__": + main()