"""Discord inbound adapter — translates Discord events into ChatService calls. Key design decisions vs the old app/discord_bot.py: - No module-level singleton: the bot is constructed via create_bot() factory - ChatService is injected directly — no HTTP roundtrip to the FastAPI API - Pure functions (build_answer_embed, parse_conversation_id, etc.) are extracted and independently testable - All logging, no print() - Error embeds never leak exception details """ import logging from typing import Optional import discord from discord import app_commands from discord.ext import commands from domain.models import ChatResult from domain.services import ChatService logger = logging.getLogger(__name__) CONFIDENCE_THRESHOLD = 0.4 FOOTER_PREFIX = "conv:" MAX_EMBED_DESCRIPTION = 4000 # --------------------------------------------------------------------------- # Pure helper functions (testable without Discord) # --------------------------------------------------------------------------- def build_answer_embed( result: ChatResult, title: str = "Rules Answer", color: discord.Color | None = None, ) -> discord.Embed: """Build a Discord embed from a ChatResult. Handles truncation, cited rules, confidence warnings, and footer. """ if color is None: color = discord.Color.blue() # Truncate long responses with a notice text = result.response if len(text) > MAX_EMBED_DESCRIPTION: text = ( text[: MAX_EMBED_DESCRIPTION - 60] + "\n\n*(Response truncated — ask a more specific question)*" ) embed = discord.Embed(title=title, description=text, color=color) # Cited rules if result.cited_rules: embed.add_field( name="📋 Cited Rules", value=", ".join(f"`{rid}`" for rid in result.cited_rules), inline=False, ) # Low confidence warning if result.confidence < CONFIDENCE_THRESHOLD: embed.add_field( name="⚠️ Confidence", value=f"Low ({result.confidence:.0%}) — a human review has been requested", inline=False, ) # Footer with full conversation ID for follow-ups embed.set_footer( text=f"{FOOTER_PREFIX}{result.conversation_id} | Reply to ask a follow-up" ) return embed def build_error_embed(error: Exception) -> discord.Embed: """Build a safe error embed that never leaks exception internals.""" _ = error # logged by the caller, not exposed to users return discord.Embed( title="❌ Error", description=( "Something went wrong while processing your request. " "Please try again later." ), color=discord.Color.red(), ) def parse_conversation_id(footer_text: Optional[str]) -> Optional[str]: """Extract conversation UUID from embed footer text. Expected format: "conv: | Reply to ask a follow-up" Returns None if the footer is missing, malformed, or empty. """ if not footer_text or FOOTER_PREFIX not in footer_text: return None try: raw = footer_text.split(FOOTER_PREFIX)[1].split(" ")[0].strip() return raw if raw else None except (IndexError, AttributeError): return None # --------------------------------------------------------------------------- # Bot class # --------------------------------------------------------------------------- class StratChatbot(commands.Bot): """Discord bot that answers Strat-O-Matic rules questions. Unlike the old implementation, this bot calls ChatService directly instead of going through the HTTP API, eliminating the roundtrip. """ def __init__( self, chat_service: ChatService, guild_id: Optional[str] = None, ): intents = discord.Intents.default() intents.message_content = True super().__init__(command_prefix="!", intents=intents) self.chat_service = chat_service self.guild_id = guild_id # Register commands and events self._register_commands() def _register_commands(self) -> None: """Register slash commands and event handlers.""" @self.tree.command( name="ask", description="Ask a question about Strat-O-Matic league rules", ) @app_commands.describe( question="Your rules question (e.g., 'Can a runner steal on a 2-2 count?')" ) async def ask_command(interaction: discord.Interaction, question: str): await self._handle_ask(interaction, question) @self.event async def on_ready(): if not self.user: return logger.info("Bot logged in as %s (ID: %s)", self.user, self.user.id) @self.event async def on_message(message: discord.Message): await self._handle_follow_up(message) async def setup_hook(self) -> None: """Sync slash commands on startup.""" if self.guild_id: guild = discord.Object(id=int(self.guild_id)) self.tree.copy_global_to(guild=guild) await self.tree.sync(guild=guild) logger.info("Slash commands synced to guild %s", self.guild_id) else: await self.tree.sync() logger.info("Slash commands synced globally") # ------------------------------------------------------------------ # /ask command handler # ------------------------------------------------------------------ async def _handle_ask( self, interaction: discord.Interaction, question: str ) -> None: """Handle the /ask slash command.""" await interaction.response.defer(ephemeral=False) try: result = await self.chat_service.answer_question( message=question, user_id=str(interaction.user.id), channel_id=str(interaction.channel_id), ) embed = build_answer_embed(result, title="Rules Answer") await interaction.followup.send(embed=embed) except Exception as e: logger.error( "Error in /ask from user %s: %s", interaction.user.id, e, exc_info=True, ) await interaction.followup.send(embed=build_error_embed(e)) # ------------------------------------------------------------------ # Follow-up reply handler # ------------------------------------------------------------------ async def _handle_follow_up(self, message: discord.Message) -> None: """Handle reply-based follow-up questions.""" if message.author.bot: return if not message.reference or message.reference.message_id is None: return # Use cached resolved message first, fetch only if needed referenced = message.reference.resolved if referenced is None or not isinstance(referenced, discord.Message): referenced = await message.channel.fetch_message( message.reference.message_id ) if referenced.author != self.user: return # Extract conversation ID from the referenced embed footer embed = referenced.embeds[0] if referenced.embeds else None footer_text = embed.footer.text if embed and embed.footer else None conversation_id = parse_conversation_id(footer_text) if conversation_id is None: await message.reply( "❓ Could not find conversation context. Use `/ask` to start fresh.", mention_author=True, ) return parent_message_id = str(referenced.id) loading_msg = await message.reply( "🔍 Looking into that follow-up...", mention_author=True ) try: result = await self.chat_service.answer_question( message=message.content, user_id=str(message.author.id), channel_id=str(message.channel.id), conversation_id=conversation_id, parent_message_id=parent_message_id, ) response_embed = build_answer_embed( result, title="Follow-up Answer", color=discord.Color.green() ) await loading_msg.edit(content=None, embed=response_embed) except Exception as e: logger.error( "Error in follow-up from user %s: %s", message.author.id, e, exc_info=True, ) await loading_msg.edit(content=None, embed=build_error_embed(e)) # --------------------------------------------------------------------------- # Factory + entry point # --------------------------------------------------------------------------- def create_bot( chat_service: ChatService, guild_id: Optional[str] = None, ) -> StratChatbot: """Construct a StratChatbot with injected dependencies.""" return StratChatbot(chat_service=chat_service, guild_id=guild_id) def run_bot( token: str, chat_service: ChatService, guild_id: Optional[str] = None, ) -> None: """Construct and run the Discord bot (blocking call).""" if not token: raise ValueError("Discord bot token must not be empty") bot = create_bot(chat_service=chat_service, guild_id=guild_id) bot.run(token)