from __future__ import annotations import json import html as html_utils import os import queue import random import csv import threading import time from pathlib import Path from typing import Any import requests from dotenv import load_dotenv from gradio_client import Client from huggingface_hub import hf_hub_download, upload_file from dod_logging import log_error, log_info, quiet_external_stdout from inference_mapper import EndpointConfig, get_endpoint_chain, mark_endpoint_failed, mark_endpoint_success load_dotenv(override=True) Card = dict[str, Any] GameState = dict[str, Any] ServerResponse = dict[str, Any] PLACEHOLDER_SECRET_VALUES = { "your_token", "your_huggingface_token", "your_hf_token", "hf_token", "token", } def get_optional_env_secret(name: str) -> str: """Return an environment secret while ignoring blank or placeholder values.""" value = os.getenv(name, "").strip().strip("\"'") if not value or value.lower() in PLACEHOLDER_SECRET_VALUES: return "" return value def get_int_env(name: str, default: int, minimum: int | None = None) -> int: """Read an integer environment value with optional lower bound.""" try: value = int(os.getenv(name, str(default))) except (TypeError, ValueError): value = default if minimum is not None: return max(minimum, value) return value def get_float_env(name: str, default: float, minimum: float | None = None) -> float: """Read a float environment value with optional lower bound.""" try: value = float(os.getenv(name, str(default))) except (TypeError, ValueError): value = default if minimum is not None: return max(minimum, value) return value # Maximum seconds an active player has to act during a normal turn. PLAYER_TURN_TIME_LIMIT_SECONDS = 30 # Seconds to pause between turns so Director audio and table feedback can breathe. TURN_HANDOFF_DELAY_SECONDS = get_float_env("DOD_TURN_HANDOFF_DELAY_SECONDS", 6.0, minimum=0.0) # Extra handoff multiplier before Nemotron acts, so the bot does not feel instant. BOT_TURN_HANDOFF_MULTIPLIER = get_float_env("DOD_BOT_TURN_HANDOFF_MULTIPLIER", 2.0, minimum=1.0) # Server-side grace window for the required Deploy shout. SERVER_SHOUT_WINDOW_BUFFER_SECONDS = 6 # Seconds to show end-game state before rotating/restarting the room. GAME_RESTART_COUNTDOWN_SECONDS = 10 # Minimum active seats required before the lobby can start a match countdown. MIN_PLAYERS_TO_START = get_int_env("DOD_MIN_PLAYERS_TO_START", 2, minimum=2) # Seconds to wait for more players once the minimum active seats are present. LOBBY_START_COUNTDOWN_SECONDS = get_int_env("DOD_LOBBY_START_COUNTDOWN_SECONDS", 30, minimum=0) # Seconds without heartbeat before a player is considered inactive. PLAYER_HEARTBEAT_KICK_LIMIT_SECONDS = 45.0 # Seconds without any table action before the room is force-closed. MAX_ROOM_INACTIVITY_TIMEOUT_SECONDS = 180.0 # Seconds during which duplicate active names are rejected. DUPLICATE_CHECK_WINDOW_SECONDS = 5.0 # Sync lobby during warmup TICK_LOBBY_WARMUP_SECONDS = 1 # Backend tick interval for turn timers and room cleanup. TICK_RATE_SERVER_SECONDS = 1 # Player board polling interval. SYNC_RATE_PLAYER_SECONDS = 1 # Spectator board polling interval. SYNC_RATE_SPECTATOR_SECONDS = 2 # Leaderboard polling interval. SYNC_RATE_LEADERBOARD_SECONDS = 15 # Maximum active players supported by the room, including the mandatory Nemotron bot. MAX_PLAYERS = get_int_env("DOD_MAX_PLAYERS", 3, minimum=2) # Built-in AI opponent name. BOT_NAME = "Nemotron" # Optional Hugging Face token used only for private dataset access. HF_DATASET_TOKEN = get_optional_env_secret("HF_TOKEN_DATASET") # Nemotron LLM server URL. LLM_URL = os.getenv("LLM_URL", "https://elismasilva-voxcpm2-nanovllm-service.hf.space") # Nemotron LLM server API key. LLM_API_KEY = os.getenv("LLM_API_KEY", "") # Hugging Face dataset that stores leaderboard persistence. LEADERBOARD_DATASET_REPO_ID = os.getenv("DOD_LEADERBOARD_DATASET_REPO_ID", "elismasilva/dod-leaderboard") # CSV filename used inside the leaderboard dataset and local data directory. LEADERBOARD_DATASET_PATH = "leaderboard.csv" # Development switch for local mapper and leaderboard files under the user home directory. LOCAL_DATA_ENABLED = os.getenv("DOD_USE_LOCAL_DATA", "").lower() in {"1", "true", "yes", "on"} # Local data folder used when `DOD_USE_LOCAL_DATA=True`. LOCAL_DATA_DIR = Path(os.getenv("DOD_LOCAL_DATA_DIR", Path.home() / ".dod")).expanduser() # Local leaderboard CSV used when `DOD_USE_LOCAL_DATA=True`. LOCAL_LEADERBOARD_PATH = Path( os.getenv("DOD_LOCAL_LEADERBOARD_PATH", LOCAL_DATA_DIR / LEADERBOARD_DATASET_PATH) ).expanduser() # XP thresholds for the five visible leaderboard stars. LEADERBOARD_STAR_XP_THRESHOLDS = (0, 2500, 7500, 15000, 30000) # External TTS endpoint used for director voice audio. TTS_API_URL = os.getenv("TTS_API_URL", "http://127.0.0.1:8000/generate_api") # Development switch that skips remote/local TTS downloads. TTS_DOWNLOAD_DISABLED = os.getenv("DOD_DISABLE_TTS", "").lower() in {"1", "true", "yes"} # Voice-control prompts for the TTS service. TTS_CONTROLS = { "pt": "Brazilian male broadcaster, speaking with a native Brazilian Portuguese accent. Deep but crisp voice. He sounds like an experienced, highly professional manager. 'No-nonsense', authoritative, and strategic tone. Clear studio recording.", "en": "American male executive, speaking with a standard US English accent. Deep but crisp voice. He sounds like an experienced, highly professional manager. 'No-nonsense', authoritative, and strategic tone. Clear studio recording." } # Reference voice id expected by the TTS service. TTS_VOICE_ID = "voz_1.wav" TTS_VOICE_SEED = 44 # Shared FIFO channel used by the app-level LLM worker. llm_queue: queue.Queue[dict[str, Any]] = queue.Queue() tts_gradio_clients: dict[str, Client] = {} def get_tts_gradio_client(endpoint: EndpointConfig, timeout_override: float | None = None) -> Client: """Return a cached Gradio client for a TTS endpoint.""" url = endpoint["url"] timeout = float(timeout_override if timeout_override is not None else endpoint.get("timeout", 120.0)) cache_key = f"{url}|{timeout}" if cache_key not in tts_gradio_clients: log_info(f"[TTS] Connecting Gradio client to {endpoint.get('name', 'endpoint')}: {url}", flush=True) with quiet_external_stdout(): tts_gradio_clients[cache_key] = Client(url, httpx_kwargs={"timeout": timeout}) return tts_gradio_clients[cache_key] APP_UI = { "en": { "title": "DOD - Deploy or Draw!
UNO GAME", "subtitle": "Select language, enter your name and join the queue.", "lang_label": "Language", "name_label": "Your Name", "btn_join": "Join Game", "btn_leave_queue": "Leave Queue", "hf_login_button": "Sign in with Hugging Face", "hf_logout_button": "Logout ({})", "hf_login_guest": "Optional: sign in with Hugging Face to save wins and XP on the leaderboard, or enter a guest name below.", "hf_login_authenticated": "Signed in with Hugging Face as **{name}**. This identity will be used for matches and leaderboard.", "status": "Waiting for action...", "tab_lobby": "🎮 Lobby & Spectator", "tab_player": "💻 Your Game", "welcome_play": "Welcome, {name}! You are in the game.", "welcome_queue": "Room is full. You are #{pos} in the queue.", "tab_leaderboard": "🏆 Leaderboard", "tab_manual": "📘 How to Play", "invalid_name": "⚠️ Enter a valid name!", "duplicate_name": "⚠️ This name is already active in another tab!", "warmup_status": "DOD UNO: Cooking cloud audio assets... Please wait about 30-50 seconds!" }, "pt": { "title": "DOD - Deploy or Draw!
JOGO DE UNO", "subtitle": "Selecione o idioma, digite seu nome e entre na fila.", "lang_label": "Idioma", "name_label": "Seu Nome", "btn_join": "Entrar na Partida", "btn_leave_queue": "Sair da Fila", "hf_login_button": "Entrar com Hugging Face", "hf_logout_button": "Sair ({})", "hf_login_guest": "Opcional: entre com Hugging Face para salvar vitórias e XP na classificação, ou digite um nome de convidado abaixo.", "hf_login_authenticated": "Conectado ao Hugging Face como **{name}**. Esta identidade será usada nas partidas e na classificação.", "status": "Aguardando ação...", "tab_lobby": "🎮 Lobby & Espectador", "tab_player": "💻 Sua Partida", "welcome_play": "Bem-vindo(a), {name}! Você está no jogo.", "welcome_queue": "Partida cheia. Você é o #{pos} na fila.", "tab_leaderboard": "🏆 Classificação", "tab_manual": "📘 Como Jogar", "invalid_name": "⚠️ Digite um nome válido!", "duplicate_name": "⚠️ Este nome já está ativo em outra aba!", "warmup_status": "DOD UNO: Cozinhando os assets de áudio na nuvem... Aguarde cerca de 30-50 segundos!" } } UI_I18N = { "en": { "waiting": "Waiting for {num} players to start...", "status": "SERVER STATUS:", "res": "Crisis Resolution", "panic": "IT Director Panic", "pick": "CHOOSE THE NEXT STACK (COLOR)", "draw": "DRAW PILE", "draw_btn": "DRAW", "table": "CARD ON TABLE", "stack_green": "FRONTEND", "stack_blue": "BACKEND", "stack_red": "DEVOPS", "stack_yellow": "A.I.", "turn": "Turn", "mute_audio": "Mute audio", "unmute_audio": "Unmute audio", "draw_pile_alt": "Draw pile provider", "final_deploy": "Final Deploy", "pass": "PASS TURN", "shout": "SHOUT DEPLOY!", "log": "📜 SERVER LOG", "you": "(YOU)", "leave": "LEAVE MATCH", "deploy_saved": "📢 DEPLOY SAVED!", "risk": "⚠️ RISK!", "accuse": "🚨 ACCUSE!", "player": "Player", "director": "Director", "queue_msg": "⏳ You are #{pos} in the queue.", "restarting": "🔄 Next match starting in {sec}s...", "toast_not_turn": "⛔ Not your turn!", "toast_wait_shout": "⚠️ 1 card left! Click SHOUT DEPLOY!", "toast_wait_color": "⚠️ Choose the Stack before continuing.", "toast_invalid": "❌ Invalid Move! Card doesn't match.", "toast_game_over": "💀 Director had a heart attack! GAME OVER.", "toast_victory": "🏆 VICTORY! Deploy completed successfully!", "toast_pick_shout": "🎨 Choose Stack. Then you can shout Deploy!", "toast_alert_deploy": "🚨 DEPLOY ALERT!", "toast_cant_accuse_self": "⛔ You cannot accuse yourself!", "toast_wait_accuse": "⏳ Player still has a chance to shout Deploy.", "toast_accuse_success": "🚨 You successfully accused {name}!", "toast_accuse_invalid": "❌ Invalid Accusation!", "toast_shout_protected": "📢 DEPLOY! You are protected from accusations.", "toast_shout_invalid": "⚠️ You can only shout DEPLOY! with 1 card left.", "toast_left_queue": "🚪 You left the queue.", "toast_left_game": "🚪 You abandoned the match.", "toast_game_over_abandon": "💀 Match ended due to lack of players.", "toast_game_not_started": "⚠️ The match has not started yet!", "lb_title": "🏆 PLAYERS LEADERBOARD", "start_countdown": "Starting with current players in {sec}s...", "warmup_status": "DOD UNO: Cooking cloud audio assets... Please wait about 30-50 seconds!", "lb_rank": "Rank", "lb_developer": "Player", "lb_wins": "Wins", "lb_games": "Games", "lb_xp": "Career XP", "lb_role": "Title", "role_intern": "Intern 👶", "role_junior": "Junior 🩹", "role_mid": "Mid-Level 🔨", "role_senior": "Senior 🚀", "role_lead": "Tech Lead 👑" }, "pt": { "waiting": "Aguardando {num} jogadores para iniciar...", "status": "STATUS DO SERVIDOR:", "res": "Resolução da Crise", "panic": "Pânico do Diretor de TI", "pick": "ESCOLHA A PRÓXIMA STACK (COR)", "draw": "MONTE", "draw_btn": "COMPRAR", "table": "CARTA NA MESA", "stack_green": "FRONTEND", "stack_blue": "BACKEND", "stack_red": "DEVOPS", "stack_yellow": "I.A.", "turn": "Turno", "mute_audio": "Silenciar áudio", "unmute_audio": "Ativar áudio", "draw_pile_alt": "Fornecedor do monte", "final_deploy": "Deploy Final", "pass": "PASSAR VEZ", "shout": "GRITAR DEPLOY!", "log": "📜 LOG DO SERVIDOR", "you": "(VOCÊ)", "leave": "SAIR DO JOGO", "deploy_saved": "📢 DEPLOY SALVO!", "risk": "⚠️ RISCO!", "accuse": "🚨 ACUSAR!", "player": "Jogador", "director": "Diretor", "queue_msg": "⏳ Você é o #{pos} na fila de espera.", "restarting": "🔄 Nova partida iniciando em {sec}s...", "toast_not_turn": "⛔ Não é a sua vez!", "toast_wait_shout": "⚠️ Você ficou com 1 carta! Confirme clicando em GRITAR DEPLOY!", "toast_wait_color": "⚠️ Escolha a Stack antes de continuar.", "toast_invalid": "❌ Jogada Inválida! A carta não combina na mesa.", "toast_game_over": "💀 O Diretor infartou! FIM DE JOGO.", "toast_victory": "🏆 VITÓRIA! Deploy concluído com sucesso!", "toast_pick_shout": "🎨 Escolha a Stack. Depois você poderá gritar Deploy!", "toast_alert_deploy": "🚨 ALERTA DE DEPLOY!", "toast_cant_accuse_self": "⛔ Você não pode acusar a si mesmo!", "toast_wait_accuse": "⏳ O jogador ainda tem chance de gritar Deploy.", "toast_accuse_success": "🚨 Você acusou {name} com sucesso!", "toast_accuse_invalid": "❌ Acusação inválida!", "toast_shout_protected": "📢 DEPLOY! Você está protegido contra acusações.", "toast_shout_invalid": "⚠️ Você só pode gritar DEPLOY! se tiver 1 carta na mão.", "toast_left_queue": "🚪 Você saiu da fila.", "toast_left_game": "🚪 Você abandonou a partida.", "toast_game_over_abandon": "💀 Partida encerrada por falta de jogadores.", "toast_game_not_started": "⚠️ A partida ainda não começou!", "lb_title": "🏆 CLASSIFICAÇÃO DOS DEVS", "start_countdown": "Iniciando com os jogadores atuais em {sec}s...", "warmup_status": "DOD UNO: Cozinhando os assets de áudio na nuvem... Aguarde cerca de 30-50 segundos!", "lb_rank": "Rank", "lb_developer": "Jogador", "lb_wins": "Vitórias", "lb_games": "Partidas", "lb_xp": "XP de Carreira", "lb_role": "Cargo", "role_intern": "Estagiário 👶", "role_junior": "Júnior 🩹", "role_mid": "Pleno 🔨", "role_senior": "Sênior 🚀", "role_lead": "Tech Lead 👑" } } LOG_I18N = { "en": { "lobby_wait": "[System]: Waiting for {num} players in the Lobby...", "connected": "🔌 {name} joined the server.", "queued": "⏳ {name} joined the waiting list.", "new_game": "✨ [New Game]: {crisis}! Solve the crisis!", "draw": "📥 [{name}] drew a card.", "play": "✅ [{name}] played '{card}': {feedback} (Res: {res}%, Pan: {pan}%){quote}", "win": "🏆 [VICTORY]: {name} executed the FINAL DEPLOY!", "empty_punish": "⚠️ [{name}] ran out of cards, but crisis continues. +2 tasks penalty!", "skip": "🛑 [Block]: {name} lost their turn in a call!", "reverse": "🔄 [Revert]: Game direction reversed!", "attack": "⚔️ [Attack]: {name} received +2 tasks and lost their turn!", "wild_pick": "🎨 [Wild]: Stack changed to {color}.", "wild_attack": "🔥 [Merge Conflict]: {name} got +4 conflicts and lost their turn!", "shout_alert": "⚠️ [{name}] has 1 card left and didn't shout Deploy!", "shout_success": "📢 [{name}] shouted DEPLOY!", "shout_fail": "🤐 [{name}] failed to shout Deploy. Vulnerable to accusations!", "accuse_success": "🚨 [ACCUSATION]: {name} didn't shout Deploy! Penalty: +2 cards.", "pass_turn": "⏭️ [{name}] ended their turn.", "game_over": "💀 [Game Over]: Panic reached 100%. Team fired.", "left_lobby": "🚪 [{name}] left the lobby.", "left_game": "🚪 [{name}] abandoned the match!", "game_over_abandon": "💀 [Game Over]: Match ended due to lack of players (W.O.).", "game_over_timeout": "💀 [Game Over]: Match expired due to 3 minutes of inactivity.", "afk_skip": "⏳ {name} is AFK. Passing turn...", "turn_timeout": "⏳ {name} took too long! Drew a card and lost their turn." }, "pt": { "lobby_wait": "[Sistema]: Aguardando {num} jogadores no Lobby...", "connected": "🔌 {name} conectou ao servidor.", "queued": "⏳ {name} entrou na fila de espera.", "new_game": "✨ [Novo Jogo]: {crisis}! Resolvam a crise!", "draw": "📥 [{name}] comprou uma carta.", "play": "✅ [{name}] jogou '{card}': {feedback} (Res: {res}%, Pan: {pan}%){quote}", "win": "🏆 [VITÓRIA]: {name} realizou o DEPLOY FINAL!", "empty_punish": "⚠️ [{name}] ficou sem cartas, mas a crise continua. +2 tarefas de punição!", "skip": "🛑 [Bloqueio]: {name} perdeu a vez na call!", "reverse": "🔄 [Revert]: A ordem do jogo foi invertida!", "attack": "⚔️ [Ataque]: {name} recebeu +2 tarefas e perdeu a vez!", "wild_pick": "🎨 [Coringa]: Stack alterada para {color}.", "wild_attack": "🔥 [Merge Conflict]: {name} recebeu +4 conflitos e perdeu a vez!", "shout_alert": "⚠️ [{name}] ainda tem 1 carta e não gritou Deploy!", "shout_success": "📢 [{name}] gritou DEPLOY!", "shout_fail": "🤐 [{name}] não gritou Deploy a tempo. Alvo vulnerável!", "accuse_success": "🚨 [ACUSAÇÃO]: {name} não gritou Deploy! Punição: +2 cartas.", "pass_turn": "⏭️ [{name}] encerrou o turno.", "game_over": "💀 [Fim de Jogo]: Pânico atingiu 100%. Equipe demitida.", "left_lobby": "🚪 [{name}] saiu do lobby.", "left_game": "🚪 [{name}] abandonou a partida!", "game_over_abandon": "💀 [Fim de Jogo]: Partida encerrada por falta de jogadores (W.O.).", "game_over_timeout": "💀 [Fim de Jogo]: Partida expirada por 3 minutos de inatividade.", "afk_skip": "⏳ {name} ficou ausente por muito tempo. Passando o turno...", "turn_timeout": "⏳ {name} demorou muito! Comprou uma carta e perdeu a vez." } } CRISES_DATABASE = [ { "title": {"en": "CRITICAL INCIDENT: DB CORRUPTED", "pt": "INCIDENTE CRÍTICO: DB CORROMPIDO"}, "desc": {"en": "MySQL is down. Front makes warning screen, Back cries, DevOps tries backup.", "pt": "O MySQL caiu. O Front faz uma tela de aviso, o Back chora e o DevOps tenta restaurar o backup."}, "quotes": { "good": [{"en": "Phew! DB is coming back!", "pt": "Ufa!! O banco tá voltando!"}, {"en": "Good job! Saved my job.", "pt": "Bom trabalho! Salvaram meu emprego."}, {"en": "Amen! A decent query!", "pt": "Amém! Finalmente uma query decente!"}], "bad": [{"en": "Holy shit! Did you DROP the wrong table?!", "pt": "Puta que o pariu! Vocês deram DROP na tabela errada?!"}, {"en": "CEO is calling, fix it now!", "pt": "Caralho! O CEO tá me ligando, volta esse banco logo!"}, {"en": "Are you crazy? Clients disappeared!", "pt": "Vocês são loucos? Meus clientes sumiram do sistema!"}], "mixed": [{"en": "Bro... this workaround will explode tomorrow, but works today.", "pt": "Mano... essa gambiarra no banco vai explodir amanhã, mas serve por hoje."}, {"en": "Indexed wrong, CPU at 99%, but it loaded. Phew.", "pt": "Indexou tudo errado e a CPU tá em 99%, mas a tela carregou. Ufa."}] } }, { "title": {"en": "HUGGING FACE HUB TRAFFIC SURGE", "pt": "PICO DE TRÁFEGO NO HUB DO HUGGING FACE"}, "desc": {"en": "Insane traffic! Models frozen, services bursting request limits.", "pt": "Tráfego insano! Modelos travados e serviços estourando limites de requisições."}, "quotes": { "good": [{"en": "Phew! Rate limit saved the server.", "pt": "Ufa! O limite de requisições salvou o servidor."}, {"en": "Yes! Scale the workers, fast!", "pt": "Isso! Escalem os servidores, rápido!"}, {"en": "Traffic is normalizing!", "pt": "O tráfego tá normalizando, continuem assim!"}], "bad": [{"en": "Are you crazy?! Cloud bandwidth will cost a fortune!", "pt": "Vocês são loucos?! Essa banda vai custar uma fortuna!"}, {"en": "Shit, the cluster died again!", "pt": "Puta merda, o cluster foi pro saco de novo!"}, {"en": "Damn, bots took down the entire home page!", "pt": "Caralho, os bots derrubaram a home page inteira!"}], "mixed": [{"en": "Did you cache the frontend poorly? At least it didn't crash.", "pt": "Fizeram cache no front-end de qualquer jeito? Tá, pelo menos o servidor não caiu."}, {"en": "Queue is huge, but gateway errors stopped. Acceptable.", "pt": "A fila tá gigante, mas os erros de acesso pararam. Aceitável."}] } }, { "title": {"en": "AWS KEYS LEAKED", "pt": "VAZAMENTO DE CHAVES AWS"}, "desc": {"en": "Someone committed the .env to a public repo. Bots mining crypto on our account!", "pt": "Alguém commitou o .env em um repositório público. Bots estão minerando cripto na nossa conta!"}, "quotes": { "good": [{"en": "Amen! Key revoked. Cold sweat here.", "pt": "Amém! Chave revogada. Suor frio aqui."}, {"en": "Great! Block billing before it hits $1M.", "pt": "Ótimo! Bloqueiem a cobrança antes de bater 1 milhão de dólares."}, {"en": "Phew! AWS account is safe for now.", "pt": "Ufa! A conta da AWS tá salva por enquanto."}], "bad": [{"en": "Holy crap, did they spin up 50 GPU instances in Asia?!", "pt": "Puta que pariu, já subiram 50 instâncias de GPU na Ásia?!"}, {"en": "Damn, which intern pushed this?!", "pt": "Caralho, quem foi o estagiário jumento que deu git push nisso?!"}, {"en": "We're bankrupt! Shut this all down now!", "pt": "Tamo falido! Desliga essa porra toda agora!"}], "mixed": [{"en": "Deleted the whole repo to hide the leak? My God...", "pt": "Deletaram o repositório inteiro pra esconder o vazamento? Meu Deus..."}, {"en": "Account is locked, but at least they stopped mining bitcoin.", "pt": "A conta tá travada por segurança, mas pelo menos pararam de minerar bitcoin."}] } }, { "title": {"en": "A.I. HALLUCINATING IN PROD", "pt": "I.A. ALUCINANDO NO PROD"}, "desc": {"en": "Support chatbot is swearing at clients. Turn it off now!", "pt": "O chatbot de suporte está xingando os clientes. Desliguem isso agora!"}, "quotes": { "good": [{"en": "Phew! Bot stopped swearing.", "pt": "Ufa! O bot parou de falar palavrão."}, {"en": "Adjusted the temperature? Finally sane answers.", "pt": "Ajustaram a temperatura? Finalmente respostas sãs."}, {"en": "Thank God they limited the model's prompt.", "pt": "Graças a Deus limitaram o prompt do modelo."}], "bad": [{"en": "Shit, the bot just insulted our biggest investor!", "pt": "Puta merda, o bot acabou de ofender o nosso maior investidor!"}, {"en": "Are you crazy? Who left this unfiltered model in prod?!", "pt": "Vocês são loucos? Quem deixou esse modelo sem filtro em produção?!"}, {"en": "Damn, AI is promising free products in chat!", "pt": "Caralho, a IA tá prometendo produto de graça no chat!"}], "mixed": [{"en": "Bot is answering in Latin now, but at least no insults.", "pt": "O bot tá dando respostas em latim agora, mas pelo menos não tá ofendendo ninguém."}, {"en": "Turned off AI and used a 1990s if/else? F*ck it, it works.", "pt": "Desligaram a IA e botaram um if/else de 1990? Foda-se, tá funcionando."}] } } ] def generate_full_deck() -> list[Card]: """Build and shuffle-ready the complete 108-card DOD UNO deck. Returns: List of card dictionaries containing stack, category, effects, and i18n text. """ deck = [] card_id = 1 stacks = [ {"color": "green", "name_en": "Frontend", "name_pt": "Frontend"}, {"color": "blue", "name_en": "Backend", "name_pt": "Backend"}, {"color": "red", "name_en": "DevOps", "name_pt": "DevOps"}, {"color": "yellow", "name_en": "A.I.", "name_pt": "I.A."} ] for stack in stacks: c = stack["color"] deck.append({ "id": card_id, "stack": c, "category": "SUPER", "categorySymbol": "🚀", "badge": "⭐", "res": 25, "panic": -10, "name": {"en": f"Deploy Friday 6PM ({stack['name_en']})", "pt": f"Deploy Sexta 18h ({stack['name_pt']})"}, "feedback": {"en": "Worked on first try! A miracle!", "pt": "Funcionou de primeira! Um milagre!"} }) card_id += 1 for _ in range(2): deck.append({"id": card_id, "stack": c, "category": "FIX", "categorySymbol": "🩹", "res": 15, "panic": -5, "name": {"en": "Fix Nasty Bug", "pt": "Corrigir Bug"}, "feedback": {"en": "Bug squashed successfully.", "pt": "Bug esmagado com sucesso."}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "REFACTOR", "categorySymbol": "🧹", "res": 10, "panic": -5, "name": {"en": "Refactor Code", "pt": "Refatorar Código"}, "feedback": {"en": "Clean and readable now.", "pt": "Código limpo e legível."}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "TECH_DEBT", "categorySymbol": "💸", "res": 10, "panic": -10, "name": {"en": "Clear Tech Debt", "pt": "Pagar Dívida Técnica"}, "feedback": {"en": "System is breathing better.", "pt": "O sistema está respirando melhor."}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "DOCS", "categorySymbol": "📝", "res": 5, "panic": -15, "name": {"en": "Update Docs", "pt": "Atualizar Docs"}, "feedback": {"en": "Director sighed in relief.", "pt": "Diretor respirou aliviado."}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "PATCH", "categorySymbol": "🔨", "res": 15, "panic": 10, "name": {"en": "Quick Workaround", "pt": "Gambiarra Rápida"}, "feedback": {"en": "It works... don't look at it.", "pt": "Funciona... só não olhe o código."}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "STACK_OVERFLOW", "categorySymbol": "📋", "res": 10, "panic": 5, "name": {"en": "StackOverflow Copy", "pt": "Copiar StackOverflow"}, "feedback": {"en": "No idea why it works, but it does.", "pt": "Não sei porque funciona, mas funciona."}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "SPAGHETTI", "categorySymbol": "🍝", "res": 5, "panic": 10, "name": {"en": "Spaghetti Code", "pt": "Código Espaguete"}, "feedback": {"en": "Nobody understands this architecture.", "pt": "Ninguém entende essa arquitetura."}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "BLIND_PR", "categorySymbol": "🙈", "res": 5, "panic": 15, "name": {"en": "Blind PR Merge", "pt": "Merge sem Review"}, "feedback": {"en": "Merged directly to main! Risky!", "pt": "Merge direto na main! Arriscado!"}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "BUG", "categorySymbol": "🐛", "res": -10, "panic": 15, "name": {"en": "Code Regression", "pt": "Regressão no Código"}, "feedback": {"en": "Broke what was already working!", "pt": "Quebrou o que já estava funcionando!"}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "SKIP", "categorySymbol": "🛑", "badge": "⭐", "res": 0, "panic": 0, "skip": True, "name": {"en": "Surprise Meeting", "pt": "Reunião Surpresa"}, "feedback": {"en": "Next dev pulled into a call, lost turn!", "pt": "Próximo perdeu a vez na call!"}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "REVERSE", "categorySymbol": "🔄", "badge": "⭐", "res": 5, "panic": 0, "reverse": True, "name": {"en": "Git Revert", "pt": "Git Revert"}, "feedback": {"en": "Project flow inverted!", "pt": "Fluxo invertido!"}}); card_id+=1 deck.append({"id": card_id, "stack": c, "category": "ATTACK", "categorySymbol": "➕", "badge": "+2", "res": 5, "panic": 5, "drawTwo": True, "name": {"en": "Code Review Hell", "pt": "Code Review Chato"}, "feedback": {"en": "Next dev redoes everything! (+2)", "pt": "Próximo dev refaz tudo! (+2)"}}); card_id+=1 for _ in range(4): deck.append({"id": card_id, "stack": "wild", "category": "WILD", "categorySymbol": "🎨", "badge": "🎨", "res": 10, "panic": -10, "name": {"en": "Senior Agile Consultant", "pt": "Consultor Ágil Sênior"}, "feedback": {"en": "Sprint reorganized! Choose the Stack.", "pt": "Sprint reorganizada! Escolha a Stack."}}); card_id+=1 deck.append({"id": card_id, "stack": "wild", "category": "NUKE", "categorySymbol": "☢️", "badge": "+4", "res": -10, "panic": 30, "drawFour": True, "name": {"en": "Drop Prod Database", "pt": "Apagar Banco Prod"}, "feedback": {"en": "Total chaos! Next dev suffers! (+4)", "pt": "Caos total! Próximo dev sofre! (+4)"}}); card_id+=1 return deck class GameManager: """Thread-shared game state machine for lobby, match, bot, audio, and leaderboard state.""" def __init__(self) -> None: """Initialize lobby, timers, caches, and persisted leaderboard state.""" self.players = [] self.queue = [] self.player_langs = {} self.authenticated_players: set[str] = set() self.player_pictures: dict[str, str] = {} self.player_ip_tokens: dict[str, str] = {} self.last_seen = {} self.game_started = False self.game_end_reason = "" self.restart_countdown = 0 self.lobby_start_countdown = -1 self.turn_start_shout_shown = False self.is_turn_start_shout = False self.direction = 1 self.current_crisis_idx = 0 self.last_move_time = 0.0 self.resolution = 0 self.panic = 20 self.active_card = None self.active_player = 0 self.hands = {} self.has_shouted_deploy = {} self.draw_pile = [] self.shout_countdown = SERVER_SHOUT_WINDOW_BUFFER_SECONDS self.turn_time_left = PLAYER_TURN_TIME_LIMIT_SECONDS self.turn_handoff_until = 0.0 self.display_active_player = 0 self.has_drawn_this_turn = False self.is_picking_color = False self.wild_draw_four_pending = False self.waiting_for_shout = False self.pending_skip_on_shout = False self._last_tick_time = 0.0 self._last_turn_tick = 0.0 self.events = [{"key": "lobby_wait", "kwargs": {"num": MAX_PLAYERS}}] self.audio_cache = {} self.audio_generation_id = 0 self.latest_director_audio = {"id": 0, "b64": ""} self.leaderboard_cache = {} self.match_stats = {} self.pending_audios = {} self.repo_id = LEADERBOARD_DATASET_REPO_ID self.leaderboard_path = LEADERBOARD_DATASET_PATH self.load_leaderboard_from_hf() self.modal_is_warm = False self.modal_is_warming_up = False def set_player_authenticated(self, player_name: str, is_authenticated: bool) -> None: """Mark whether a player identity is backed by Hugging Face OAuth.""" player_name = (player_name or "").strip() if not player_name or player_name == BOT_NAME: return if is_authenticated: self.authenticated_players.add(player_name) else: self.authenticated_players.discard(player_name) def is_leaderboard_eligible(self, player_name: str, has_authenticated_human: bool) -> bool: """Return whether a player can update the official leaderboard.""" if player_name == BOT_NAME: return has_authenticated_human return player_name in self.authenticated_players def set_player_ip_token(self, player_name: str, ip_token: str) -> None: """Store the Hugging Face ZeroGPU IP token associated with a player.""" player_name = (player_name or "").strip() ip_token = (ip_token or "").strip() if player_name and ip_token: self.player_ip_tokens[player_name] = ip_token def get_player_ip_token(self, player_name: str = "") -> str: """Return a player-specific ZeroGPU IP token or any active human token.""" player_name = (player_name or "").strip() if player_name and self.player_ip_tokens.get(player_name): return self.player_ip_tokens[player_name] for active_name in self.players: if active_name != BOT_NAME and self.player_ip_tokens.get(active_name): return self.player_ip_tokens[active_name] for queued_name in self.queue: if self.player_ip_tokens.get(queued_name): return self.player_ip_tokens[queued_name] return "" def set_player_picture(self, player_name: str, picture_url: str) -> None: """Store the authenticated profile image URL for a player.""" player_name = (player_name or "").strip() picture_url = (picture_url or "").strip() if not player_name or not picture_url: return self.player_pictures[player_name] = picture_url def init_game(self) -> None: """Start a fresh match for the current active players.""" self.game_started = True self.audio_generation_id += 1 self.game_end_reason = "" self.restart_countdown = 0 self.lobby_start_countdown = -1 self.direction = 1 self.current_crisis_idx = random.randint(0, len(CRISES_DATABASE)-1) self.resolution = 0 self.panic = 20 self.active_player = 0 self.display_active_player = 0 self.has_drawn_this_turn = False self.is_picking_color = False self.wild_draw_four_pending = False self.pending_skip_on_shout = False self.last_move_time = time.time() self.turn_time_left = PLAYER_TURN_TIME_LIMIT_SECONDS self.turn_handoff_until = 0.0 self.hands = {} self.has_shouted_deploy = {} self.discard_pile = [] self.events = [] for p_name in self.players: self.log_event("connected", name=p_name) self.shout_countdown = 0 self.waiting_for_shout = False crisis_data = CRISES_DATABASE[self.current_crisis_idx] self.log_event("new_game", crisis=crisis_data["title"]) self.draw_pile = generate_full_deck() random.shuffle(self.draw_pile) self.match_stats = {} for p_name in self.players: self.match_stats[p_name] = {"res_contrib": 0, "panic_mitigation": 0} self.hands[p_name] = self.get_unique_starting_hand(p_name, 7) self.sort_hand(p_name) self.has_shouted_deploy[p_name] = False self.last_seen[p_name] = time.time() starting_card = None for idx, card in enumerate(self.draw_pile): if card["stack"] != "wild": starting_card = self.draw_pile.pop(idx) break if not starting_card: starting_card = self.draw_pile.pop(0) self.active_card = starting_card self.trigger_bot_if_active() def load_leaderboard_from_hf(self) -> None: """Load persisted leaderboard rows from local disk or the configured Hugging Face dataset.""" try: if LOCAL_DATA_ENABLED: filepath = LOCAL_LEADERBOARD_PATH if not filepath.exists(): self.leaderboard_cache = {} log_info(f"[Leaderboard] Local leaderboard not found at {filepath}. Starting empty.", flush=True) return log_info(f"[Leaderboard] Loaded local leaderboard from {filepath}", flush=True) else: filepath = hf_hub_download( repo_id=self.repo_id, filename=self.leaderboard_path, repo_type="dataset", token=HF_DATASET_TOKEN or None, ) with open(filepath, mode="r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: self.leaderboard_cache[row["player_name"]] = { "wins": int(row["wins"]), "losses": int(row["losses"]), "xp": int(row["xp"]), "games": int(row["games_played"]), "picture_url": row.get("picture_url", "") } except Exception: self.leaderboard_cache = {} def log_event(self, key: str, **kwargs: Any) -> None: """Append a localized log event descriptor. Args: key: Event identifier resolved through `LOG_I18N`. **kwargs: Event formatting values. """ self.events.append({"key": key, "kwargs": kwargs}) def touch_presence(self, viewer_id: str) -> None: """Refresh the heartbeat for a player or queued user. Args: viewer_id: Player or queued user name associated with a browser tab. """ if viewer_id and (viewer_id in self.players or viewer_id in self.queue): self.last_seen[viewer_id] = time.time() def join_lobby(self, name: str, lang_code: str) -> str: """Join a player to the active room or queue. Args: name: User-facing player name. lang_code: Language code, usually `en` or `pt`. Returns: Accepted canonical player name, or `DUPLICATE_REJECT`. """ name_clean = name.strip() name_lower = name_clean.lower() if name_lower == BOT_NAME.lower(): return "DUPLICATE_REJECT" self.pending_audios[name_clean] = [] existing_players_lower = [p.lower() for p in self.players] existing_queue_lower = [q.lower() for q in self.queue] if name_lower in existing_players_lower or name_lower in existing_queue_lower: matched_name = next((p for p in self.players if p.lower() == name_lower), None) if not matched_name: matched_name = next((q for q in self.queue if q.lower() == name_lower), None) now = time.time() if self.game_started and now - self.last_seen.get(matched_name, 0) < DUPLICATE_CHECK_WINDOW_SECONDS: return "DUPLICATE_REJECT" self.player_langs[matched_name] = lang_code return matched_name self.player_langs[name_clean] = lang_code if self.game_started or len(self.players) >= MAX_PLAYERS: self.queue.append(name_clean) self.log_event("queued", name=name_clean) return name_clean if BOT_NAME in self.players: bot_index = self.players.index(BOT_NAME) self.players.insert(bot_index, name_clean) else: self.players.append(name_clean) self.log_event("connected", name=name_clean) if BOT_NAME not in self.players and len(self.players) < MAX_PLAYERS: self.players.append(BOT_NAME) self.player_langs[BOT_NAME] = "pt" if lang_code == "pt" else "en" self.log_event("connected", name=BOT_NAME) self.update_lobby_start_countdown(advance=False) return name_clean def rotate_queue_and_restart(self) -> None: """Promote queued players into the next match or reset the room to lobby state.""" for p_name in list(self.players): self.player_langs.pop(p_name, None) self.last_seen.pop(p_name, None) self.player_ip_tokens.pop(p_name, None) if p_name not in self.queue: self.authenticated_players.discard(p_name) human_slots = max(1, MAX_PLAYERS - 1) promoted_players = self.queue[:human_slots] self.queue = self.queue[human_slots:] self.players = promoted_players if self.players and len(self.players) < MAX_PLAYERS: self.players.append(BOT_NAME) first_human = promoted_players[0] self.player_langs[BOT_NAME] = "pt" if self.player_langs.get(first_human, "en") == "pt" else "en" self.game_started = False self.game_end_reason = "" self.reset_turn_flags() self.turn_handoff_until = 0.0 self.resolution = 0 self.panic = 20 self.active_card = None self.hands = {} self.has_shouted_deploy = {} self.discard_pile = [] self.draw_pile = [] self.events = [{"key": "lobby_wait", "kwargs": {"num": MAX_PLAYERS}}] self.pending_audios = {} self.last_move_time = time.time() self.modal_is_warm = False self.modal_is_warming_up = False self.lobby_start_countdown = -1 self.update_lobby_start_countdown(advance=False) def can_start_lobby_match(self) -> bool: """Return whether the current lobby is ready to warm up and start.""" return ( not self.game_started and self.restart_countdown == 0 and len(self.players) >= MIN_PLAYERS_TO_START and self.lobby_start_countdown == 0 ) def update_lobby_start_countdown(self, advance: bool = True) -> None: """Advance or reset the pre-match countdown based on active room size.""" if self.game_started or self.restart_countdown > 0: return if len(self.players) < MIN_PLAYERS_TO_START: self.lobby_start_countdown = -1 return if MAX_PLAYERS <= 2 or len(self.players) >= MAX_PLAYERS: self.lobby_start_countdown = 0 return if self.lobby_start_countdown < 0: self.lobby_start_countdown = LOBBY_START_COUNTDOWN_SECONDS elif advance and self.lobby_start_countdown > 0: self.lobby_start_countdown -= 1 def tick_countdown(self) -> None: """Advance server timers, inactivity cleanup, shout windows, and bot accusations.""" now = time.time() if now - self._last_tick_time < 0.9: return self._last_tick_time = now if self.game_started and now - self.last_move_time > MAX_ROOM_INACTIVITY_TIMEOUT_SECONDS: self.log_event("game_over_timeout") self.handle_game_over("timeout") return inactive_limit = PLAYER_HEARTBEAT_KICK_LIMIT_SECONDS for p_name in list(self.players): if p_name == BOT_NAME: continue if p_name not in self.last_seen: self.last_seen[p_name] = now elif now - self.last_seen[p_name] > inactive_limit: self.leave_game(p_name) for q_name in list(self.queue): if q_name not in self.last_seen: self.last_seen[q_name] = now elif now - self.last_seen[q_name] > inactive_limit: self.leave_game(q_name) if not self.game_started and self.restart_countdown > 0: self.restart_countdown -= 1 if self.restart_countdown == 0: self.rotate_queue_and_restart() return self.update_lobby_start_countdown() if self.game_started and self.is_turn_handoff_active(now): return if self.game_started and self.turn_handoff_until > 0: self.turn_handoff_until = 0.0 self.display_active_player = self.active_player if self.check_turn_start_deploy(): return self.trigger_bot_if_active() return if self.waiting_for_shout: if self.active_player < len(self.players) and self.players[self.active_player] == BOT_NAME: if random.random() < 0.90: self.shout_deploy(BOT_NAME) return if self.shout_countdown > 0: self.shout_countdown -= 1 if self.shout_countdown == 0: if self.active_player < len(self.players): p_name = self.players[self.active_player] self.log_event("shout_fail", name=p_name) self.waiting_for_shout = False was_turn_start = self.is_turn_start_shout self.is_turn_start_shout = False if not was_turn_start: skip = getattr(self, "pending_skip_on_shout", False) self.pending_skip_on_shout = False self.pass_turn(skip) return if self.game_started and not self.is_picking_color: for idx, p_name in enumerate(self.players): if p_name != BOT_NAME and p_name in self.hands: if self.can_accuse_player(idx): if random.random() < 0.25: self.accuse_player(idx, BOT_NAME) break if self.game_started and not self.is_picking_color: if self.active_player < len(self.players) and self.players[self.active_player] == BOT_NAME: self.turn_time_left = PLAYER_TURN_TIME_LIMIT_SECONDS return if self.turn_time_left > 0: self.turn_time_left -= 1 if self.turn_time_left == 0: if self.active_player < len(self.players): p_name = self.players[self.active_player] self.log_event("turn_timeout", name=p_name) card = self.get_unique_card(p_name) if card and p_name in self.hands: self.hands[p_name].append(card) self.sort_hand(p_name) self.pass_turn() def start_shout_window(self) -> None: """Open the short Deploy shout window for the active player.""" self.waiting_for_shout = True self.shout_countdown = SERVER_SHOUT_WINDOW_BUFFER_SECONDS def is_turn_handoff_active(self, now: float | None = None) -> bool: """Return whether the table is pausing between turns.""" if self.turn_handoff_until <= 0: return False now = time.time() if now is None else now return now < self.turn_handoff_until def get_turn_handoff_left(self) -> int: """Return whole seconds left in the current turn handoff pause.""" if not self.is_turn_handoff_active(): return 0 return max(0, int(round(self.turn_handoff_until - time.time()))) def block_during_handoff(self, caller_id: str) -> ServerResponse | None: """Block gameplay actions while the table is between turns.""" if not self.is_turn_handoff_active(): return None lang = self.player_langs.get(caller_id, "en") return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_not_turn"]} def clear_shout_window(self) -> None: """Close any active Deploy shout window without changing the turn.""" self.waiting_for_shout = False self.shout_countdown = 0 self.is_turn_start_shout = False self.pending_skip_on_shout = False def can_accuse_player(self, target_idx: int) -> bool: """Return whether a player is currently vulnerable to a Deploy accusation.""" if not self.game_started: return False if not 0 <= target_idx < len(self.players): return False if self.is_picking_color: return False if self.waiting_for_shout: return False target_name = self.players[target_idx] target_hand = self.hands.get(target_name, []) return len(target_hand) == 1 and not self.has_shouted_deploy.get(target_name, False) def get_unique_starting_hand(self, player_name: str, size: int) -> list[Card]: """Draw the initial hand from the shuffled deck. Args: player_name: Player receiving the cards. size: Number of cards to draw. Returns: Cards drawn from the top of the deck. """ hand = [] for _ in range(size): if self.draw_pile: hand.append(self.draw_pile.pop(0)) return hand def get_unique_card(self, player_name: str) -> Card | None: """Draw one card, rebuilding the draw pile when needed. Args: player_name: Player requesting the draw. Returns: The drawn card, or `None` if no card can be produced. """ if not self.draw_pile: if self.discard_pile: self.draw_pile = list(self.discard_pile) self.discard_pile = [] random.shuffle(self.draw_pile) else: self.draw_pile = generate_full_deck() random.shuffle(self.draw_pile) return self.draw_pile.pop(0) def draw_card(self, caller_id: str) -> ServerResponse: """Handle an active player's manual draw action. Args: caller_id: Player name making the draw request. Returns: State/toast response consumed by the custom HTML component. """ lang = self.player_langs.get(caller_id, "en") if not self.game_started: return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_game_not_started"]} if caller_id not in self.players: return {"state": None, "toast": ""} p_idx = self.players.index(caller_id) blocked = self.block_during_handoff(caller_id) if blocked: return blocked if p_idx != self.active_player: return {"state": None, "toast": UI_I18N[lang]["toast_not_turn"]} if self.waiting_for_shout: return {"state": None, "toast": UI_I18N[lang]["toast_wait_shout"]} if self.is_picking_color: return {"state": None, "toast": UI_I18N[lang]["toast_wait_color"]} card = self.get_unique_card(caller_id) if card: self.hands[caller_id].append(card) self.sort_hand(caller_id) self.has_drawn_this_turn = True self.has_shouted_deploy[caller_id] = False self.turn_time_left = min(PLAYER_TURN_TIME_LIMIT_SECONDS, self.turn_time_left + 10) self.log_event("draw", name=caller_id) self.last_move_time = time.time() return {"state": self.get_state(caller_id), "toast": ""} def is_valid_play(self, card: Card) -> bool: """Check whether a card can be played on the current active card. Args: card: Candidate card from a player's hand. Returns: True when the card matches by stack/category or is wild. """ if not self.active_card: return True if card["stack"] == "wild" or self.active_card["stack"] == "wild": return True return card["stack"] == self.active_card["stack"] or card["category"] == self.active_card["category"] def render_leaderboard_html(self, lang: str = "en") -> str: """Render the leaderboard as standalone HTML. Args: lang: Language code for labels. Returns: HTML string for the leaderboard tab. """ if lang not in ["en", "pt"]: lang = "en" t = UI_I18N[lang] sorted_board = sorted( self.leaderboard_cache.items(), key=lambda x: (x[1]["wins"], x[1]["xp"]), reverse=True ) html = f"""
{t["lb_title"]}
""" trophies = {1: "🥇", 2: "🥈", 3: "🥉"} role_keys = ["role_intern", "role_junior", "role_mid", "role_senior", "role_lead"] for idx, (p_name, stats) in enumerate(sorted_board[:20], 1): rank_display = trophies.get(idx, f"{idx}") xp_val = stats["xp"] active_stars = sum(1 for threshold in LEADERBOARD_STAR_XP_THRESHOLDS if xp_val >= threshold) active_stars = max(1, min(active_stars, len(LEADERBOARD_STAR_XP_THRESHOLDS))) role_key = role_keys[active_stars - 1] role_title = t[role_key] stars_visual = "".join( f'★' for star_idx in range(len(LEADERBOARD_STAR_XP_THRESHOLDS)) ) row_class = "" if idx == 1: row_class = "lb-row-gold" elif idx == 2: row_class = "lb-row-silver" elif idx == 3: row_class = "lb-row-bronze" wins_sub_en = f"({stats['wins']} wins / {stats['games']} matches)" wins_sub_pt = f"({stats['wins']} vitórias / {stats['games']} partidas)" wins_sub = wins_sub_en if lang == "en" else wins_sub_pt safe_player_name = html_utils.escape(p_name, quote=True) safe_wins_sub = html_utils.escape(wins_sub, quote=True) safe_role_title = html_utils.escape(role_title, quote=True) picture_url = "/gradio_api/file=assets/nemotron.jpg" if p_name == BOT_NAME else stats.get("picture_url", "") if picture_url: safe_picture_url = html_utils.escape(picture_url, quote=True) avatar_html = f'{safe_player_name}' else: avatar_html = """ """ html += f"""
{rank_display}
{avatar_html}
{safe_player_name} {safe_wins_sub} {stars_visual}
{xp_val} XP
""" html += """
""" return html def async_save_leaderboard_to_hf(self) -> None: """Persist the in-memory leaderboard to local disk or the configured Hugging Face dataset.""" try: if LOCAL_DATA_ENABLED: temp_path = LOCAL_LEADERBOARD_PATH temp_path.parent.mkdir(parents=True, exist_ok=True) else: temp_path = Path("./leaderboard.csv") with open(temp_path, mode="w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(["player_name", "wins", "losses", "xp", "games_played", "picture_url"]) for p_name, stats in self.leaderboard_cache.items(): writer.writerow([p_name, stats["wins"], stats["losses"], stats["xp"], stats["games"], stats.get("picture_url", "")]) if LOCAL_DATA_ENABLED: log_info(f"[Leaderboard] Saved local leaderboard to {temp_path}", flush=True) return upload_file( path_or_fileobj=str(temp_path), path_in_repo=self.leaderboard_path, repo_id=self.repo_id, repo_type="dataset", token=HF_DATASET_TOKEN or None, commit_message="Update Leaderboard Career Stats" ) temp_path.unlink(missing_ok=True) except Exception as e: log_error(f"Leaderboard sync failed: {e}") def sort_hand(self, player_name: str) -> None: """Sort a player's hand in-place by stack, category priority, then card id. Args: player_name: Player whose hand should be sorted. """ if player_name in self.hands: color_order = {"green": 1, "blue": 2, "red": 3, "yellow": 4, "wild": 5} category_order = { "SUPER": 1, "FIX": 2, "REFACTOR": 3, "TECH_DEBT": 4, "DOCS": 5, "PATCH": 6, "STACK_OVERFLOW": 7, "SPAGHETTI": 8, "BLIND_PR": 9, "BUG": 10, "SKIP": 11, "REVERSE": 12, "ATTACK": 13, "WILD": 14, "NUKE": 15 } self.hands[player_name].sort( key=lambda c: ( color_order.get(c.get("stack", "wild"), 99), category_order.get(c.get("category", ""), 99), c.get("id", 0) ) ) def handle_game_over(self, reason: str | None = None) -> None: """Finalize the current match, update scores, and schedule the next room rotation. Args: reason: End-state reason used by clients to select the right toast and audio. """ self.game_started = False self.audio_generation_id += 1 self.restart_countdown = GAME_RESTART_COUNTDOWN_SECONDS self.pending_audios = {} self.modal_is_warm = False winner_name = None if self.resolution >= 100: for p_name in self.players: if len(self.hands.get(p_name, [])) == 0: winner_name = p_name break if reason: self.game_end_reason = reason elif self.panic >= 100: self.game_end_reason = "game_over" elif winner_name: self.game_end_reason = "victory" else: self.game_end_reason = "game_over" leaderboard_updated = False has_authenticated_human = any( p_name != BOT_NAME and p_name in self.authenticated_players for p_name in self.players ) for p_name in self.players: if not self.is_leaderboard_eligible(p_name, has_authenticated_human): continue if p_name not in self.leaderboard_cache: self.leaderboard_cache[p_name] = {"wins": 0, "losses": 0, "xp": 0, "games": 0, "picture_url": ""} stats = self.leaderboard_cache[p_name] picture_url = "/gradio_api/file=assets/nemotron.jpg" if p_name == BOT_NAME else self.player_pictures.get(p_name, "") if picture_url: stats["picture_url"] = picture_url m_stats = self.match_stats.get(p_name, {"res_contrib": 0, "panic_mitigation": 0}) base_xp = 50 win_loss_xp = 0 if self.panic >= 100: stats["losses"] += 1 win_loss_xp = 0 elif winner_name: if p_name == winner_name: stats["wins"] += 1 win_loss_xp = 150 else: stats["losses"] += 1 win_loss_xp = 25 contrib_xp = int(m_stats["res_contrib"] * 1.0 + m_stats["panic_mitigation"] * 1.5) total_earned_xp = base_xp + win_loss_xp + contrib_xp stats["xp"] += total_earned_xp stats["games"] += 1 leaderboard_updated = True if leaderboard_updated: threading.Thread(target=self.async_save_leaderboard_to_hf).start() def play_card(self, player_index: int, card_index: int, caller_id: str) -> ServerResponse: """Play one card from a player's hand. Args: player_index: Index of the player whose hand is being used. card_index: Index of the card inside that player's hand. caller_id: Player name initiating the action. Returns: State/toast response consumed by the custom HTML component. """ lang = self.player_langs.get(caller_id, "en") if not self.game_started: return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_game_not_started"]} if caller_id not in self.players: return {"state": None, "toast": ""} p_idx = self.players.index(caller_id) blocked = self.block_during_handoff(caller_id) if blocked: return blocked if p_idx != player_index: return {"state": None, "toast": UI_I18N[lang]["toast_not_turn"]} if self.is_picking_color: return {"state": None, "toast": UI_I18N[lang]["toast_wait_color"]} if player_index != self.active_player: return {"state": None, "toast": UI_I18N[lang]["toast_not_turn"]} p_name = self.players[player_index] card = self.hands[p_name][card_index] if not self.is_valid_play(card): return {"state": None, "toast": UI_I18N[lang]["toast_invalid"]} if self.active_card: self.discard_pile.append(json.loads(json.dumps(self.active_card))) new_res = min(100, max(0, self.resolution + card["res"])) new_panic = min(100, max(0, self.panic + card["panic"])) contrib_res = max(0, card["res"]) mitigation_panic = abs(min(0, card["panic"])) if caller_id in self.match_stats: self.match_stats[caller_id]["res_contrib"] += contrib_res self.match_stats[caller_id]["panic_mitigation"] += mitigation_panic self.resolution = new_res self.panic = new_panic self.active_card = json.loads(json.dumps(card)) self.hands[p_name].pop(card_index) if len(self.hands[p_name]) > 1: self.has_shouted_deploy[p_name] = False res_sign = '+' if card['res'] >= 0 else '' panic_sign = '+' if card['panic'] >= 0 else '' has_quote = False card_type = "good" if card["category"] == "SKIP": has_quote = True card_type = "bad" elif card["res"] == 0 and card["panic"] == 0 and card["category"] not in ["NUKE", "SUPER"]: pass elif card["res"] < 0 or card["category"] == "NUKE": has_quote = True card_type = "bad" elif card["res"] > 0 and card["panic"] > 0: has_quote = True card_type = "bad" elif card["res"] >= 0 or card["panic"] < 0: has_quote = True card_type = "good" quote_id = time.time() self.log_event("play", name=caller_id, card=card['name'], feedback=card['feedback'], res=res_sign+str(card['res']), pan=panic_sign+str(card['panic']), has_quote=has_quote, quote=None, quote_id=quote_id) director_quote_task = None if has_quote: director_quote_task = { "type": "director_quote", "card_played": card['name'].get("en", ""), "card_type": card_type, "event_id": quote_id, "audio_generation_id": self.audio_generation_id, "ip_token": self.get_player_ip_token(caller_id), "card_context": { "name_en": card.get("name", {}).get("en", ""), "name_pt": card.get("name", {}).get("pt", ""), "feedback_en": card.get("feedback", {}).get("en", ""), "feedback_pt": card.get("feedback", {}).get("pt", ""), "category": card.get("category", ""), "stack": card.get("stack", ""), "resolution_delta": card["res"], "panic_delta": card["panic"], "resolution_after": self.resolution, "panic_after": self.panic, }, } def enqueue_director_quote() -> None: if director_quote_task and self.game_started: llm_queue.put(director_quote_task) self.last_move_time = time.time() if self.panic >= 100: self.log_event("game_over") self.handle_game_over("game_over") return {"state": self.get_state(caller_id), "toast": ""} if len(self.hands[p_name]) == 0: if self.resolution == 100: self.log_event("win", name=caller_id) self.handle_game_over("victory") return {"state": self.get_state(caller_id), "toast": ""} else: self.draw_cards_for_player(player_index, 2) self.log_event("empty_punish", name=caller_id) self.pass_turn() enqueue_director_quote() return {"state": self.get_state(caller_id), "toast": ""} if len(self.hands[p_name]) == 1: if card["stack"] == "wild": self.is_picking_color = True self.wild_draw_four_pending = card.get("drawFour", False) self.waiting_for_shout = False enqueue_director_quote() return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_pick_shout"]} else: self.start_shout_window() self.has_shouted_deploy[p_name] = False enqueue_director_quote() return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_alert_deploy"]} skip_next = False if card.get("skip"): next_player = (self.active_player + self.direction) % len(self.players) self.log_event("skip", name=self.players[next_player]) skip_next = True if card.get("reverse"): self.direction *= -1 self.log_event("reverse") if len(self.players) == 2: next_player = (self.active_player + self.direction) % len(self.players) self.log_event("skip", name=self.players[next_player]) skip_next = True if card.get("drawTwo"): next_player = (self.active_player + self.direction) % len(self.players) self.draw_cards_for_player(next_player, 2) self.log_event("attack", name=self.players[next_player]) skip_next = True if len(self.hands[p_name]) == 1: if card["stack"] == "wild": self.is_picking_color = True self.wild_draw_four_pending = card.get("drawFour", False) self.waiting_for_shout = False enqueue_director_quote() return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_pick_shout"]} else: self.start_shout_window() self.has_shouted_deploy[p_name] = False self.pending_skip_on_shout = skip_next enqueue_director_quote() return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_alert_deploy"]} if card["stack"] == "wild": self.is_picking_color = True self.wild_draw_four_pending = card.get("drawFour", False) enqueue_director_quote() return {"state": self.get_state(caller_id), "toast": ""} self.pass_turn(skip_next) enqueue_director_quote() return {"state": self.get_state(caller_id), "toast": ""} def reset_turn_flags(self) -> None: """Clear transient state associated with the current turn.""" self.has_drawn_this_turn = False self.waiting_for_shout = False self.shout_countdown = 0 self.is_picking_color = False self.turn_start_shout_shown = False self.is_turn_start_shout = False self.pending_skip_on_shout = False self.turn_handoff_until = 0.0 self.display_active_player = self.active_player def leave_game(self, caller_id: str) -> ServerResponse: """Remove a player from queue or active match. Args: caller_id: Player name leaving the room. Returns: State/toast response consumed by the custom HTML component. """ lang = self.player_langs.get(caller_id, "en") self.last_seen.pop(caller_id, None) if caller_id in self.queue: self.queue.remove(caller_id) self.authenticated_players.discard(caller_id) self.player_ip_tokens.pop(caller_id, None) return {"state": self.get_state(""), "toast": UI_I18N[lang]["toast_left_queue"]} if caller_id not in self.players: return {"state": None, "toast": ""} p_idx = self.players.index(caller_id) if not self.game_started: self.players.pop(p_idx) self.authenticated_players.discard(caller_id) self.player_ip_tokens.pop(caller_id, None) self.log_event("left_lobby", name=caller_id) return {"state": self.get_state(""), "toast": UI_I18N[lang]["toast_left_queue"]} self.players.pop(p_idx) hand = self.hands.pop(caller_id, []) self.has_shouted_deploy.pop(caller_id, None) self.authenticated_players.discard(caller_id) self.player_ip_tokens.pop(caller_id, None) self.draw_pile.extend(hand) random.shuffle(self.draw_pile) self.log_event("left_game", name=caller_id) if len(self.players) < 2: self.log_event("game_over_abandon") self.handle_game_over("abandon") return {"state": self.get_state(""), "toast": UI_I18N[lang]["toast_game_over_abandon"]} if self.active_player == p_idx: self.active_player = self.active_player % len(self.players) self.reset_turn_flags() self.turn_time_left = PLAYER_TURN_TIME_LIMIT_SECONDS self.check_turn_start_deploy() self.trigger_bot_if_active() elif self.active_player > p_idx: self.active_player -= 1 self.last_move_time = time.time() return {"state": self.get_state(""), "toast": UI_I18N[lang]["toast_left_game"]} def accuse_player(self, target_idx: int, caller_id: str) -> ServerResponse: """Accuse another player of missing the Deploy shout. Args: target_idx: Index of the accused player. caller_id: Player name making the accusation. Returns: State/toast response consumed by the custom HTML component. """ lang = self.player_langs.get(caller_id, "en") if not self.game_started: return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_game_not_started"]} if caller_id not in self.players: return {"state": None, "toast": ""} p_idx = self.players.index(caller_id) blocked = self.block_during_handoff(caller_id) if blocked: return blocked if p_idx == target_idx: return {"state": None, "toast": UI_I18N[lang]["toast_cant_accuse_self"]} if not 0 <= target_idx < len(self.players): return {"state": None, "toast": UI_I18N[lang]["toast_accuse_invalid"]} if self.is_picking_color: return {"state": None, "toast": UI_I18N[lang]["toast_wait_color"]} if self.waiting_for_shout: return {"state": None, "toast": UI_I18N[lang]["toast_wait_accuse"]} t_name = self.players[target_idx] if self.can_accuse_player(target_idx): self.draw_cards_for_player(target_idx, 2) self.clear_shout_window() self.log_event("accuse_success", name=t_name) self.last_move_time = time.time() return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_accuse_success"].replace("{name}", t_name)} return {"state": None, "toast": UI_I18N[lang]["toast_accuse_invalid"]} def draw_cards_for_player(self, player_idx: int, count: int) -> None: """Draw penalty cards into a player's hand. Args: player_idx: Target player index. count: Number of cards to draw. """ if 0 <= player_idx < len(self.players): p_name = self.players[player_idx] for _ in range(count): card = self.get_unique_card(p_name) if card: self.hands[p_name].append(card) self.sort_hand(p_name) self.has_shouted_deploy[p_name] = False def select_wild_color(self, color_name: str, caller_id: str) -> ServerResponse: """Resolve a pending wild-card color choice. Args: color_name: Selected stack color. caller_id: Player name choosing the color. Returns: State/toast response consumed by the custom HTML component. """ lang = self.player_langs.get(caller_id, "en") if not self.game_started: return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_game_not_started"]} if caller_id not in self.players: return {"state": None, "toast": ""} p_idx = self.players.index(caller_id) blocked = self.block_during_handoff(caller_id) if blocked: return blocked if p_idx != self.active_player: return {"state": None, "toast": UI_I18N[lang]["toast_not_turn"]} self.active_card["stack"] = color_name self.is_picking_color = False self.log_event("wild_pick", color=color_name.upper()) self.last_move_time = time.time() skip_next = False if self.wild_draw_four_pending: next_player = (self.active_player + self.direction) % len(self.players) self.draw_cards_for_player(next_player, 4) n_name = self.players[next_player] self.log_event("wild_attack", name=n_name) skip_next = True self.wild_draw_four_pending = False if len(self.hands[caller_id]) == 1: self.start_shout_window() self.has_shouted_deploy[caller_id] = False self.pending_skip_on_shout = skip_next return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_alert_deploy"]} self.pass_turn(skip_next) return {"state": self.get_state(caller_id), "toast": ""} def check_turn_start_deploy(self) -> bool: """Open a Deploy shout window if the incoming active player is vulnerable.""" p = self.active_player if p >= len(self.players): return False p_name = self.players[p] if p_name not in self.hands: return False if (len(self.hands[p_name]) == 1 and not self.has_shouted_deploy[p_name] and not self.waiting_for_shout and not self.turn_start_shout_shown): self.turn_start_shout_shown = True self.is_turn_start_shout = True self.start_shout_window() self.log_event("shout_alert", name=p_name) return True return False def shout_deploy(self, caller_id: str) -> ServerResponse: """Protect the active player from accusation by shouting Deploy. Args: caller_id: Active player name. Returns: State/toast response consumed by the custom HTML component. """ lang = self.player_langs.get(caller_id, "en") if not self.game_started: return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_game_not_started"]} if caller_id not in self.players: return {"state": None, "toast": ""} p_idx = self.players.index(caller_id) blocked = self.block_during_handoff(caller_id) if blocked: return blocked if p_idx != self.active_player: return {"state": None, "toast": UI_I18N[lang]["toast_not_turn"]} if len(self.hands[caller_id]) <= 2: self.has_shouted_deploy[caller_id] = True self.is_turn_start_shout = False self.log_event("shout_success", name=caller_id) self.last_move_time = time.time() if self.waiting_for_shout: self.waiting_for_shout = False self.shout_countdown = 0 if not self.is_turn_start_shout: skip = getattr(self, "pending_skip_on_shout", False) self.pending_skip_on_shout = False self.pass_turn(skip) return {"state": self.get_state(caller_id), "toast": UI_I18N[lang]["toast_shout_protected"]} return {"state": None, "toast": UI_I18N[lang]["toast_shout_invalid"]} def pass_turn_manual(self, caller_id: str) -> ServerResponse: """Let the active player manually pass after drawing or missing a shout. Args: caller_id: Active player name. Returns: State/toast response consumed by the custom HTML component. """ lang = self.player_langs.get(caller_id, "en") if caller_id not in self.players: return {"state": None, "toast": ""} p_idx = self.players.index(caller_id) blocked = self.block_during_handoff(caller_id) if blocked: return blocked if p_idx != self.active_player: return {"state": None, "toast": UI_I18N[lang]["toast_not_turn"]} if self.waiting_for_shout: self.waiting_for_shout = False p_name = self.players[self.active_player] self.log_event("shout_fail", name=p_name) skip = getattr(self, "pending_skip_on_shout", False) self.pending_skip_on_shout = False self.pass_turn(skip) return {"state": self.get_state(caller_id), "toast": ""} if self.has_drawn_this_turn: self.log_event("pass_turn", name=caller_id) self.pass_turn(False) self.last_move_time = time.time() return {"state": self.get_state(caller_id), "toast": ""} def pass_turn(self, skip_next: bool = False) -> None: """Advance the active player pointer. Args: skip_next: Whether to skip the next player in turn order. """ self.has_drawn_this_turn = False self.waiting_for_shout = False self.shout_countdown = 0 self.turn_start_shout_shown = False self.turn_time_left = PLAYER_TURN_TIME_LIMIT_SECONDS self.turn_handoff_until = 0.0 if not self.players: return base_step = 2 if skip_next else 1 step = base_step * self.direction previous_player = self.active_player self.active_player = (self.active_player + step) % len(self.players) if TURN_HANDOFF_DELAY_SECONDS > 0: next_player_name = self.players[self.active_player] if self.active_player < len(self.players) else "" delay_seconds = TURN_HANDOFF_DELAY_SECONDS if next_player_name == BOT_NAME: delay_seconds *= BOT_TURN_HANDOFF_MULTIPLIER self.display_active_player = -1 self.turn_handoff_until = time.time() + delay_seconds return self.display_active_player = self.active_player if self.check_turn_start_deploy(): return self.trigger_bot_if_active() def localize_card(self, card: Card | None, lang: str) -> Card | None: """Return a copy of a card localized for the requested language. Args: card: Card to localize. lang: Language code used to select localized fields. """ if not card: return None c = card.copy() c["name"] = c["name"].get(lang, c["name"]["en"]) c["feedback"] = c["feedback"].get(lang, c["feedback"]["en"]) return c def render_event(self, evt: dict[str, Any], lang: str) -> str: """Render a localized log event to HTML-ready text. Args: evt: Event descriptor from `self.events`. lang: Language code used to format the event. """ kwargs = {} for k, v in evt["kwargs"].items(): if isinstance(v, dict): kwargs[k] = v.get(lang, v.get("en", "")) else: kwargs[k] = v if evt["key"] == "play": quote_html = "" if evt["kwargs"].get("has_quote") and evt["kwargs"].get("quote") is not None: q_text = kwargs["quote"] director = UI_I18N[lang]["director"] quote_html = f"
   ↳ 👨‍💼 {director}: \"{q_text}\"" kwargs["quote"] = quote_html return LOG_I18N[lang][evt["key"]].format(**kwargs) def get_state(self, viewer_id: str = "") -> GameState: """Build the reactive state payload consumed by the custom HTML boards. Args: viewer_id: Player name for personalized hand, audio, and language state. Returns: Full board state dictionary. """ lang = self.player_langs.get(viewer_id, "en") if viewer_id and viewer_id != "": self.last_seen[viewer_id] = time.time() rendered_logs = [self.render_event(e, lang) for e in self.events] formatted_log = "

".join(rendered_logs) crisis_source = CRISES_DATABASE[self.current_crisis_idx] if self.game_started else {"title": {"en": "IDLE", "pt": "OCIOSO"}, "desc": {"en": "Waiting...", "pt": "Aguardando..."}} localized_crisis = { "title": crisis_source["title"].get(lang, crisis_source["title"]["en"]), "desc": crisis_source["desc"].get(lang, crisis_source["desc"]["en"]) } ordered_hands = [self.hands.get(p, []) for p in self.players] ordered_shouted = [self.has_shouted_deploy.get(p, False) for p in self.players] state_active_player = self.display_active_player if self.is_turn_handoff_active() else self.active_player active_player_name = self.players[state_active_player] if 0 <= state_active_player < len(self.players) else "" hand_length = len(self.hands.get(active_player_name, [])) if active_player_name else 0 inactivity_left = int(MAX_ROOM_INACTIVITY_TIMEOUT_SECONDS - (time.time() - self.last_move_time)) if self.game_started else 0 localized_audio = {"id": 0, "b64": ""} player_queue = self.pending_audios.get(viewer_id, []) if player_queue: next_audio = player_queue.pop(0) c_key = next_audio["cache_key"] if c_key in self.audio_cache: localized_audio["id"] = next_audio["id"] localized_audio["b64"] = self.audio_cache[c_key].get(lang, "") return { "game_started": self.game_started, "game_end_reason": self.game_end_reason, "players": self.players, "player_pictures": { p_name: ( "/gradio_api/file=assets/nemotron.jpg" if p_name == BOT_NAME else self.player_pictures.get(p_name, "") ) for p_name in self.players }, "queue": self.queue, "max_players": MAX_PLAYERS, "current_crisis": localized_crisis, "resolution": self.resolution, "panic": self.panic, "active_card": self.localize_card(self.active_card, lang), "active_player": state_active_player, "hands": [[self.localize_card(c, lang) for c in hand] for hand in ordered_hands], "game_log": formatted_log, "has_drawn_this_turn": self.has_drawn_this_turn, "is_picking_color": self.is_picking_color, "has_shouted_deploy": ordered_shouted, "waiting_for_shout": self.waiting_for_shout, "shout_countdown": self.shout_countdown, "restart_countdown": self.restart_countdown, "lobby_start_countdown": max(0, self.lobby_start_countdown), "is_warming_up": self.modal_is_warming_up, "inactivity_left": inactivity_left, "turn_left": self.turn_time_left, "turn_handoff_left": self.get_turn_handoff_left(), "pending_wild_shout": self.is_picking_color and hand_length == 1, "is_turn_start_shout": self.is_turn_start_shout, "i18n": UI_I18N[lang], "director_audio": localized_audio } def download_tts_language( self, cache_key: str, text: str, lang: str, store_cache: bool = True, use_warmup_timeout: bool = False, ) -> bool: """Download one TTS language variant into the shared audio cache. Args: cache_key: Audio cache bucket for the generated quote. text: Quote text to synthesize. lang: Language code for voice controls. store_cache: Keeps audio in local cache. use_warmup_timeout: Uses a longer timeout for cold-start warmup calls. Returns: True if the audio was successfully downloaded and cached, False otherwise. """ if TTS_DOWNLOAD_DISABLED: log_info(f"[TTS] Download skipped for {lang} because DOD_DISABLE_TTS=True.", flush=True) return False if store_cache and cache_key not in self.audio_cache: self.audio_cache[cache_key] = {"en": "", "pt": ""} if store_cache and self.audio_cache[cache_key].get(lang): return True if lang not in TTS_CONTROLS: return False payload = { "control": TTS_CONTROLS[lang], "text": text, "cfg_value": 4.0, "voice_id": TTS_VOICE_ID, "seed": TTS_VOICE_SEED } headers = {} tts_api_key = os.getenv("TTS_API_KEY") if tts_api_key: headers["Authorization"] = f"Bearer {tts_api_key}" for endpoint in get_endpoint_chain("tts"): url = endpoint.get("url", "") mode = endpoint.get("mode", "rest") api_name = endpoint.get("api_name") or "/generate_api" timeout = float(endpoint.get("warmup_timeout" if use_warmup_timeout else "timeout", 120.0)) try: log_info(f"[TTS] Requesting {lang} audio from {endpoint.get('name', 'endpoint')} ({mode}): {url}", flush=True) if mode == "gradio": client = get_tts_gradio_client(endpoint, timeout) result = client.predict(tts_api_key or "", payload, api_name=api_name) else: resp = requests.post(url, json=payload, headers=headers, timeout=timeout) if resp.status_code != 200: log_info(f"[TTS] REST endpoint failed with status code: {resp.status_code}", flush=True) mark_endpoint_failed("tts", endpoint, f"status {resp.status_code}") continue result = resp.json() if isinstance(result, str): result = json.loads(result) b64_audio = result.get("wav_base64", "") if isinstance(result, dict) else "" if b64_audio: if store_cache: self.audio_cache[cache_key][lang] = b64_audio mark_endpoint_success("tts", endpoint) return True log_info(f"[TTS] Endpoint returned no audio payload: {url}", flush=True) mark_endpoint_failed("tts", endpoint, "empty audio payload") except Exception as e: mark_endpoint_failed("tts", endpoint, str(e)) log_info(f"[TTS] Endpoint error ({lang}) at {url}: {e}", flush=True) log_info(f"[TTS] All mapped endpoints failed for language: {lang}", flush=True) return False def fetch_tts_async(self, quote_en: str, quote_pt: str, event_id: float) -> None: """Fetch director TTS audio for languages used by active players. Args: quote_en: English quote text. quote_pt: Portuguese quote text. event_id: Log-event identifier linked to the audio payload. """ cache_key = quote_en active_langs = {self.player_langs.get(p, "en") for p in self.players} if "en" in active_langs: self.download_tts_language(cache_key, quote_en, "en") if "pt" in active_langs: self.download_tts_language(cache_key, quote_pt, "pt") self.latest_director_audio = {"id": event_id, "cache_key": cache_key} def trigger_bot_if_active(self) -> None: """Queue a bot decision when the active player is Nemotron.""" if not self.game_started or not self.players: return if self.active_player < len(self.players): current_p = self.players[self.active_player] if current_p == BOT_NAME: llm_queue.put({ "type": "bot_decision", "bot_name": current_p, "ip_token": self.get_player_ip_token(), }) global_server = GameManager()