""" fallout.coyoteos.ovh — Venice of Wasteland — Visualisation simulation Flask app sur Ampère, lit PostgreSQL Vigile via SSH tunnel """ import subprocess, time, threading, os, json from flask import Flask, render_template_string, jsonify, abort app = Flask(__name__) # ── SSH tunnel → PostgreSQL Vigile ──────────────────────────────────────────── TUNNEL_LOCAL_PORT = 15432 _tunnel_proc = None def start_tunnel(): global _tunnel_proc cmd = [ "ssh", "-N", "-L", f"{TUNNEL_LOCAL_PORT}:localhost:5432", "-i", "/ssh/vigile.key", "-o", "StrictHostKeyChecking=no", "-o", "ServerAliveInterval=30", "-o", "ServerAliveCountMax=3", "ubuntu@79.72.30.231" ] _tunnel_proc = subprocess.Popen(cmd) time.sleep(3) print(f"SSH tunnel → Vigile:5432 via local:{TUNNEL_LOCAL_PORT}") def get_conn(): import psycopg2 return psycopg2.connect( host="localhost", port=TUNNEL_LOCAL_PORT, dbname="fallout", user="fallout", password="VeniceOfWasteland2026!", connect_timeout=5 ) def query(sql, params=None): try: conn = get_conn() cur = conn.cursor() cur.execute(sql, params or ()) cols = [d[0] for d in cur.description] rows = [dict(zip(cols, r)) for r in cur.fetchall()] conn.close() return rows except Exception as e: print(f"DB error: {e}") return [] def execute(sql, params=None): try: conn = get_conn() cur = conn.cursor() cur.execute(sql, params or ()) conn.commit() conn.close() return True except Exception as e: print(f"DB error: {e}") return False # ── Templates ───────────────────────────────────────────────────────────────── BASE_CSS = """ """ def nav(active=""): return f"""
""" SHELL = """ VaultCom — {{ title }} """ + BASE_CSS + """
{{ nav | safe }}

{{ title }}

{% if session %} Jour {{ session.day_current }}/{{ session.day_total }} {{ session.mode | upper }} {% else %} DB OFFLINE {% endif %}
{{ content | safe }}
""" def render(title, content, active="", session=None): return render_template_string(SHELL, title=title, content=content, nav=nav(active), session=session) # ── Helpers ─────────────────────────────────────────────────────────────────── def get_session(): rows = query("SELECT * FROM sessions ORDER BY id DESC LIMIT 1") return rows[0] if rows else None def hp_class(cur, mx): if mx == 0: return "warning" pct = cur / mx return "danger" if pct < 0.33 else ("warning" if pct < 0.66 else "") def bar_class(cur, mx): if mx == 0: return "warning" pct = cur / mx return "danger" if pct < 0.33 else ("warning" if pct < 0.66 else "") # ── Routes ──────────────────────────────────────────────────────────────────── @app.route("/") def index(): sess = get_session() chars = query("SELECT * FROM characters WHERE session_id = %s ORDER BY faction, name", (sess["id"],)) if sess else [] recent = query("""SELECT e.*, c.name as char_name FROM events e JOIN characters c ON c.id = e.character_id WHERE e.session_id = %s ORDER BY e.day_in_game DESC, e.hour_in_game DESC LIMIT 12""", (sess["id"],)) if sess else [] factions = query("""SELECT DISTINCT faction, COUNT(*) as nb, SUM(CASE WHEN status='alive' THEN 1 ELSE 0 END) as alive FROM characters WHERE session_id = %s GROUP BY faction ORDER BY faction""", (sess["id"],)) if sess else [] pending = query("SELECT COUNT(*) as n FROM action_queue WHERE session_id = %s AND status='pending'", (sess["id"],))[0]["n"] if sess else 0 lore_last = query("""SELECT * FROM lore_avancee WHERE session_id = %s ORDER BY day_start DESC LIMIT 1""", (sess["id"],)) if sess else [] # Characters grid char_cards = "" if chars: for c in chars: hp_pct = int(c["hp_current"] / max(c["hp_max"], 1) * 100) bc = bar_class(c["hp_current"], c["hp_max"]) nc = hp_class(c["hp_current"], c["hp_max"]) st_cls = f"status-{c['status']}" char_cards += f"""

{c['name']}

{c['status'].upper()}
{c['faction'] or '—'} · {c['location'] or '?'}
CAPS: {c['caps']} · XP: {c['xp']}
""" else: char_cards = '
◉
Aucun agent initialisé
' # Recent events ev_html = "" if recent: ev_html = '' else: ev_html = '
⊟
Journal vide — simulation non démarrée
' # Faction summary fac_html = "" for f in factions: fac_html += f'{f["faction"] or "?"}{f["alive"]}{f["nb"]}' # Lore entry lore_html = "" if lore_last: l = lore_last[0] lore_html = f"""

{l['title'] or f"Résumé jours {l['day_start']}–{l['day_end']}"}

Jours {l['day_start']}–{l['day_end']}
{l['content'][:600]}{'…' if len(l['content'])>600 else ''}
""" else: lore_html = '
◎
Pas encore de lore synthétisé
' content = f"""
▌ VAULTCOM ACTIVE ▌ SESSION: {sess['name'] if sess else 'N/A'} ▌ AGENTS: {len(chars)} ▌ ACTIONS EN ATTENTE: {pending} ▌ █
{char_cards}

Journal récent

{ev_html}

Factions actives

{fac_html if fac_html else ''}
FactionVivantsTotal
—

Dernière synthèse narrative

{lore_html}
""" return render("Mission Control", content, "home", sess) @app.route("/characters") def characters(): sess = get_session() chars = query("SELECT * FROM characters WHERE session_id = %s ORDER BY faction, name", (sess["id"],)) if sess else [] rows = "" for c in chars: hp_pct = int(c["hp_current"] / max(c["hp_max"], 1) * 100) bc = bar_class(c["hp_current"], c["hp_max"]) rows += f""" {c['name']} {c['faction'] or '—'} {c['location'] or '?'}
{c['hp_current']}/{c['hp_max']} {c['caps']} {c['status'].upper()} {c['xp']} XP """ content = f"""
{rows if rows else ''}
NomFactionLieuHPCapsStatusXP
Aucun personnage
""" return render("Agents", content, "chars", sess) @app.route("/character/") def character(cid): sess = get_session() chars = query("SELECT * FROM characters WHERE id = %s", (cid,)) if not chars: abort(404) c = chars[0] inv = query("SELECT * FROM inventory WHERE character_id = %s ORDER BY item_name", (cid,)) events = query("""SELECT * FROM events WHERE character_id = %s ORDER BY day_in_game, hour_in_game""", (cid,)) SPECIALS = [("S","strength"),("P","perception"),("E","endurance"), ("C","charisma"),("I","intelligence"),("A","agility"),("L","luck")] sp_html = "".join(f"""
{c[key]}
{abbr}
""" for abbr, key in SPECIALS) hp_pct = int(c["hp_current"] / max(c["hp_max"], 1) * 100) bc = bar_class(c["hp_current"], c["hp_max"]) nc = hp_class(c["hp_current"], c["hp_max"]) inv_rows = "".join(f"{i['item_name']}{i.get('quantity',1)}{i.get('description','')}" for i in inv) if inv else "Inventaire vide" ev_html = "" if events: ev_html = '' else: ev_html = '
Aucune action enregistrée
' bio_html = f'
{c["bio"] or "Dossier vide."}
' content = f"""

{c['name']}

{c['faction'] or '—'}
Localisation: {c['location'] or '?'} · Caps: {c['caps']} · XP: {c['xp']}
{c['status'].upper()}
{c['hp_current']}
/{c['hp_max']} HP
{sp_html}
{bio_html}

Inventaire

{inv_rows}
ObjetQtéDescription

Journal d'actions

{ev_html}
""" return render(c["name"], content, "chars", sess) @app.route("/actions") def actions(): sess = get_session() rows = query(""" SELECT aq.*, c.name as char_name, t.name as target_name FROM action_queue aq LEFT JOIN characters c ON c.id = aq.character_id LEFT JOIN characters t ON t.id = aq.target_char_id WHERE aq.session_id = %s ORDER BY aq.day_in_game DESC, aq.hour_in_game DESC, aq.id DESC LIMIT 100""", (sess["id"],)) if sess else [] def status_badge(s): cls = {"pending":"amber","resolving":"green","resolved":"dim","cancelled":"red"}.get(s,"dim") return f'{s.upper()}' trs = "".join(f""" J{r['day_in_game']}h{r['hour_in_game']:02d} {r['char_name'] or '?'} {r['location'] or '?'} {r['action_text'][:80]}{'…' if len(r['action_text'])>80 else ''} {r['target_name'] or '—'} {r['scene_id'] or '—'} {status_badge(r['status'])} {(r['result_text'] or '')[:60]} """ for r in rows) if rows else "Aucune action" content = f"""
{trs}
MomentAgentLieuActionCibleScèneStatusRésultat
""" return render("File d'Actions", content, "actions", sess) @app.route("/world") def world(): sess = get_session() if not sess: return render("État du Monde", '
DB offline
', "world") day = sess["day_current"] states = query("""SELECT * FROM world_state WHERE session_id = %s AND day_in_game = %s ORDER BY location""", (sess["id"], day)) factions = query("""SELECT * FROM faction_relations WHERE session_id = %s ORDER BY faction_a, faction_b""", (sess["id"],)) def stab_color(s): if s >= 70: return "var(--green)" if s >= 40: return "var(--amber)" return "var(--red)" world_rows = "".join(f""" {s['location']} {s['controlling_faction'] or '—'}
{s['stability']}% {s['population'] or '?'} {s['notes'] or ''} """ for s in states) if states else "Aucune donnée" fac_rows = "".join(f""" {f['faction_a']} {f['faction_b']} {'+' if f['relation_score']>0 else ''}{f['relation_score']} {f.get('notes','')} """ for f in factions) if factions else "Aucune donnée" content = f"""

Zones — Jour {day}

{world_rows}
LieuFactionStabilitéPop.Notes

Relations entre factions

{fac_rows}
Faction AFaction BScoreNotes
""" return render("État du Monde", content, "world", sess) @app.route("/lore") def lore(): sess = get_session() entries = query("""SELECT * FROM lore_avancee WHERE session_id = %s ORDER BY day_start DESC""", (sess["id"],)) if sess else [] cards = "" if entries: for e in entries: facs = ", ".join(e["factions_involved"] or []) or "—" locs = ", ".join(e["locations_involved"] or []) or "—" cards += f"""

{e['title'] or f"Chronique jours {e['day_start']}–{e['day_end']}"}

J{e['day_start']}–{e['day_end']}
Factions: {facs} · Lieux: {locs}
{e['content']}
""" else: cards = '
◎
Aucune synthèse narrative générée
Le MJ produira des entrées toutes les 24h de jeu simulé
' return render("Lore Avancée", cards, "lore", sess) @app.route("/events") def events(): sess = get_session() evs = query("""SELECT e.*, c.name as char_name FROM events e JOIN characters c ON c.id = e.character_id WHERE e.session_id = %s ORDER BY e.day_in_game DESC, e.hour_in_game DESC LIMIT 200""", (sess["id"],)) if sess else [] rows = "".join(f""" J{e['day_in_game']}h{e['hour_in_game']:02d} {e['char_name']} {e['location'] or '?'} {e['action'][:80]}{'…' if len(e['action'])>80 else ''} {(e['result'] or '')[:100]}{'…' if e['result'] and len(e['result'])>100 else ''} """ for e in evs) if evs else "Journal vide" content = f"""
{rows}
MomentAgentLieuActionRésultat
""" return render("Journal des Événements", content, "events", sess) @app.route("/api/summary") def api_summary(): """Pour le widget dashboard dash.coyoteos.ovh""" sess = get_session() if not sess: return jsonify({"ok": False}) chars = query("SELECT status, COUNT(*) n FROM characters WHERE session_id=%s GROUP BY status", (sess["id"],)) pending = query("SELECT COUNT(*) n FROM action_queue WHERE session_id=%s AND status='pending'", (sess["id"],)) return jsonify({ "ok": True, "session": sess["name"], "day": sess["day_current"], "day_total": sess["day_total"], "chars": {r["status"]: r["n"] for r in chars}, "pending_actions": pending[0]["n"] if pending else 0, }) # ── Main ────────────────────────────────────────────────────────────────────── if __name__ == "__main__": t = threading.Thread(target=start_tunnel, daemon=True) t.start() print("VaultCom :5000") app.run(host="0.0.0.0", port=5000, debug=False)