import asyncio import logging from datetime import datetime import discord from discord import app_commands from discord.ext import commands logger = logging.getLogger("bcs.commands") class CommandsCog(commands.Cog): def __init__(self, bot: commands.Bot): self.bot = bot def _is_admin(self, interaction: discord.Interaction) -> bool: return interaction.user.guild_permissions.administrator @app_commands.command( name="dramareport", description="Show current drama scores for all tracked users.", ) @app_commands.default_permissions(administrator=True) async def dramareport(self, interaction: discord.Interaction): if not self._is_admin(interaction): await interaction.response.send_message( "Admin only.", ephemeral=True ) return scores = self.bot.drama_tracker.get_all_scores() if not scores: await interaction.response.send_message( "No drama tracked yet. Everyone's behaving... for now.", ephemeral=True, ) return sorted_scores = sorted(scores.items(), key=lambda x: x[1], reverse=True) lines = [] for user_id, score in sorted_scores: user = self.bot.get_user(user_id) name = user.display_name if user else f"Unknown ({user_id})" bar = self._score_bar(score) lines.append(f"{bar} **{score:.2f}** — {name}") embed = discord.Embed( title="Drama Report", description="\n".join(lines), color=discord.Color.orange(), ) await interaction.response.send_message(embed=embed, ephemeral=True) @app_commands.command( name="dramascore", description="Show a specific user's current drama score.", ) @app_commands.describe(user="The user to check") async def dramascore( self, interaction: discord.Interaction, user: discord.Member ): score = self.bot.drama_tracker.get_drama_score(user.id) user_data = self.bot.drama_tracker.get_user(user.id) embed = discord.Embed( title=f"Drama Score: {user.display_name}", color=self._score_color(score), ) embed.add_field(name="Score", value=f"{score:.2f}/1.0", inline=True) embed.add_field( name="Offenses", value=str(user_data.offense_count), inline=True ) embed.add_field( name="Immune", value="Yes" if user_data.immune else "No", inline=True, ) embed.add_field( name="Messages Tracked", value=str(len(user_data.entries)), inline=True, ) embed.add_field(name="Vibe Check", value=self._score_bar(score)) await interaction.response.send_message(embed=embed, ephemeral=True) @app_commands.command( name="bcs-status", description="Show bot monitoring status and monitored channels.", ) async def bcs_status(self, interaction: discord.Interaction): config = self.bot.config monitoring = config.get("monitoring", {}) sentiment = config.get("sentiment", {}) enabled = monitoring.get("enabled", True) channels = monitoring.get("channels", []) if channels: ch_mentions = [] for ch_id in channels: ch = self.bot.get_channel(ch_id) ch_mentions.append(ch.mention if ch else f"#{ch_id}") ch_text = ", ".join(ch_mentions) else: ch_text = "All channels" embed = discord.Embed( title="BCS Status", color=discord.Color.green() if enabled else discord.Color.greyple(), ) mode_config = self.bot.get_mode_config() mode_label = mode_config.get("label", self.bot.current_mode) moderation_level = mode_config.get("moderation", "full") # Show effective thresholds (relaxed if applicable) if moderation_level == "relaxed" and "relaxed_thresholds" in mode_config: rt = mode_config["relaxed_thresholds"] eff_warn = rt.get("warning_threshold", 0.80) eff_mute = rt.get("mute_threshold", 0.85) else: eff_warn = sentiment.get("warning_threshold", 0.6) eff_mute = sentiment.get("mute_threshold", 0.75) embed.add_field( name="Mode", value=f"{mode_label} ({moderation_level})", inline=True, ) embed.add_field( name="Monitoring", value="Active" if enabled else "Disabled", inline=True, ) embed.add_field(name="Channels", value=ch_text, inline=True) embed.add_field( name="Warning Threshold", value=str(eff_warn), inline=True, ) embed.add_field( name="Mute Threshold", value=str(eff_mute), inline=True, ) embed.add_field( name="Triage Model", value=f"`{self.bot.llm.model}`", inline=True, ) embed.add_field( name="Escalation Model", value=f"`{self.bot.llm_heavy.model}`", inline=True, ) embed.add_field( name="LLM Host", value=f"`{self.bot.llm.host}`", inline=True, ) await interaction.response.send_message(embed=embed, ephemeral=True) @app_commands.command( name="bcs-threshold", description="Adjust warning and mute thresholds. (Admin only)", ) @app_commands.default_permissions(administrator=True) @app_commands.describe( warning="Warning threshold (0.0-1.0)", mute="Mute threshold (0.0-1.0)", ) async def bcs_threshold( self, interaction: discord.Interaction, warning: float | None = None, mute: float | None = None, ): if not self._is_admin(interaction): await interaction.response.send_message( "Admin only.", ephemeral=True ) return sentiment = self.bot.config.setdefault("sentiment", {}) changes = [] if warning is not None: warning = max(0.0, min(1.0, warning)) sentiment["warning_threshold"] = warning changes.append(f"Warning: {warning:.2f}") if mute is not None: mute = max(0.0, min(1.0, mute)) sentiment["mute_threshold"] = mute changes.append(f"Mute: {mute:.2f}") if not changes: await interaction.response.send_message( "Provide at least one threshold to update.", ephemeral=True ) return await interaction.response.send_message( f"Thresholds updated: {', '.join(changes)}", ephemeral=True ) @app_commands.command( name="bcs-reset", description="Reset a user's drama score and offense count. (Admin only)", ) @app_commands.default_permissions(administrator=True) @app_commands.describe(user="The user to reset") async def bcs_reset( self, interaction: discord.Interaction, user: discord.Member ): if not self._is_admin(interaction): await interaction.response.send_message( "Admin only.", ephemeral=True ) return self.bot.drama_tracker.reset_user(user.id) asyncio.create_task(self.bot.db.delete_user_state(user.id)) await interaction.response.send_message( f"Reset drama data for {user.display_name}.", ephemeral=True ) @app_commands.command( name="bcs-immune", description="Toggle monitoring immunity for a user. (Admin only)", ) @app_commands.default_permissions(administrator=True) @app_commands.describe(user="The user to toggle immunity for") async def bcs_immune( self, interaction: discord.Interaction, user: discord.Member ): if not self._is_admin(interaction): await interaction.response.send_message( "Admin only.", ephemeral=True ) return is_immune = self.bot.drama_tracker.toggle_immunity(user.id) user_data = self.bot.drama_tracker.get_user(user.id) asyncio.create_task(self.bot.db.save_user_state( user_id=user.id, offense_count=user_data.offense_count, immune=user_data.immune, off_topic_count=user_data.off_topic_count, baseline_coherence=user_data.baseline_coherence, user_notes=user_data.notes or None, )) status = "now immune" if is_immune else "no longer immune" await interaction.response.send_message( f"{user.display_name} is {status} to monitoring.", ephemeral=True ) @app_commands.command( name="bcs-history", description="Show a user's recent drama incidents.", ) @app_commands.describe(user="The user to check history for") async def bcs_history( self, interaction: discord.Interaction, user: discord.Member ): incidents = self.bot.drama_tracker.get_recent_incidents(user.id) if not incidents: await interaction.response.send_message( f"No recent incidents for {user.display_name}.", ephemeral=True ) return lines = [] for entry in incidents: ts = datetime.fromtimestamp(entry.timestamp).strftime("%H:%M:%S") cats = ", ".join(c for c in entry.categories if c != "none") lines.append( f"`{ts}` — **{entry.toxicity_score:.2f}** | {cats or 'n/a'} | {entry.reasoning}" ) embed = discord.Embed( title=f"Drama History: {user.display_name}", description="\n".join(lines), color=discord.Color.orange(), ) await interaction.response.send_message(embed=embed, ephemeral=True) @app_commands.command( name="bcs-scan", description="Scan recent messages in this channel. (Admin only)", ) @app_commands.default_permissions(administrator=True) @app_commands.describe(count="Number of recent messages to scan (default 10, max 50)") async def bcs_scan( self, interaction: discord.Interaction, count: int = 10 ): if not self._is_admin(interaction): await interaction.response.send_message( "Admin only.", ephemeral=True ) return count = max(1, min(count, 50)) await interaction.response.defer() messages = [] async for msg in interaction.channel.history(limit=count): if not msg.author.bot and msg.content and msg.content.strip(): messages.append(msg) if not messages: await interaction.followup.send("No user messages found to scan.") return messages.reverse() # oldest first await interaction.followup.send( f"Scanning {len(messages)} messages... (first request may be slow while model loads)" ) for msg in messages: # Build context from the messages before this one idx = messages.index(msg) ctx_msgs = messages[max(0, idx - 3):idx] context = ( " | ".join(f"{m.author.display_name}: {m.content}" for m in ctx_msgs) if ctx_msgs else "(no prior context)" ) result = await self.bot.llm_heavy.analyze_message(msg.content, context) if result is None: embed = discord.Embed( title=f"Analysis: {msg.author.display_name}", description=f"> {msg.content[:200]}", color=discord.Color.greyple(), ) embed.add_field(name="Result", value="LLM returned no result", inline=False) else: score = result["toxicity_score"] categories = result["categories"] reasoning = result["reasoning"] cat_str = ", ".join(c for c in categories if c != "none") or "none" self.bot.drama_tracker.add_entry( msg.author.id, score, categories, reasoning ) drama_score = self.bot.drama_tracker.get_drama_score(msg.author.id) embed = discord.Embed( title=f"Analysis: {msg.author.display_name}", description=f"> {msg.content[:200]}", color=self._score_color(score), ) off_topic = result.get("off_topic", False) topic_cat = result.get("topic_category", "general_chat") topic_why = result.get("topic_reasoning", "") embed.add_field(name="Message Score", value=f"{score:.2f}", inline=True) embed.add_field(name="Rolling Drama", value=f"{drama_score:.2f}", inline=True) embed.add_field(name="Categories", value=cat_str, inline=True) embed.add_field(name="Reasoning", value=reasoning[:1024] or "n/a", inline=False) embed.add_field( name="Topic", value=f"{'OFF-TOPIC' if off_topic else 'On-topic'} ({topic_cat}){chr(10) + topic_why if topic_why else ''}", inline=False, ) await interaction.channel.send(embed=embed) await interaction.channel.send(f"Scan complete. Analyzed {len(messages)} messages.") @app_commands.command( name="bcs-test", description="Analyze a test message and show raw LLM response. (Admin only)", ) @app_commands.default_permissions(administrator=True) @app_commands.describe(message="The test message to analyze") async def bcs_test(self, interaction: discord.Interaction, message: str): if not self._is_admin(interaction): await interaction.response.send_message( "Admin only.", ephemeral=True ) return await interaction.response.defer(ephemeral=True) # Build channel context for game detection game_channels = self.bot.config.get("game_channels", {}) channel_context = "" if game_channels and hasattr(interaction.channel, "name"): ch_name = interaction.channel.name current_game = game_channels.get(ch_name) lines = [] if current_game: lines.append(f"Current channel: #{ch_name} ({current_game})") else: lines.append(f"Current channel: #{ch_name}") channel_list = ", ".join(f"#{ch} ({g})" for ch, g in game_channels.items()) lines.append(f"Game channels: {channel_list}") channel_context = "\n".join(lines) user_notes = self.bot.drama_tracker.get_user_notes(interaction.user.id) raw, parsed = await self.bot.llm_heavy.raw_analyze( message, user_notes=user_notes, channel_context=channel_context, ) embed = discord.Embed( title="BCS Test Analysis", color=discord.Color.blue() ) embed.add_field( name="Input Message", value=message[:1024], inline=False ) embed.add_field( name="Raw LLM Response", value=f"```json\n{raw[:1000]}\n```", inline=False, ) if parsed: embed.add_field( name="Parsed Score", value=f"{parsed['toxicity_score']:.2f}", inline=True, ) embed.add_field( name="Categories", value=", ".join(parsed["categories"]), inline=True, ) embed.add_field( name="Reasoning", value=parsed["reasoning"][:1024] or "n/a", inline=False, ) detected_game = parsed.get("detected_game") if detected_game: game_label = game_channels.get(detected_game, detected_game) embed.add_field( name="Detected Game", value=f"#{detected_game} ({game_label})", inline=True, ) else: embed.add_field( name="Parsing", value="Failed to parse response", inline=False ) await interaction.followup.send(embed=embed, ephemeral=True) @app_commands.command( name="bcs-notes", description="View, add, or clear per-user LLM notes. (Admin only)", ) @app_commands.default_permissions(administrator=True) @app_commands.describe( action="What to do with the notes", user="The user whose notes to manage", text="Note text to add (only used with 'add')", ) @app_commands.choices(action=[ app_commands.Choice(name="view", value="view"), app_commands.Choice(name="add", value="add"), app_commands.Choice(name="clear", value="clear"), ]) async def bcs_notes( self, interaction: discord.Interaction, action: app_commands.Choice[str], user: discord.Member, text: str | None = None, ): if not self._is_admin(interaction): await interaction.response.send_message( "Admin only.", ephemeral=True ) return if action.value == "view": notes = self.bot.drama_tracker.get_user_notes(user.id) embed = discord.Embed( title=f"Notes: {user.display_name}", description=notes or "_No notes yet._", color=discord.Color.blue(), ) await interaction.response.send_message(embed=embed, ephemeral=True) elif action.value == "add": if not text: await interaction.response.send_message( "Provide `text` when using the add action.", ephemeral=True ) return self.bot.drama_tracker.update_user_notes(user.id, f"[admin] {text}") user_data = self.bot.drama_tracker.get_user(user.id) asyncio.create_task(self.bot.db.save_user_state( user_id=user.id, offense_count=user_data.offense_count, immune=user_data.immune, off_topic_count=user_data.off_topic_count, baseline_coherence=user_data.baseline_coherence, user_notes=user_data.notes or None, )) await interaction.response.send_message( f"Note added for {user.display_name}.", ephemeral=True ) elif action.value == "clear": self.bot.drama_tracker.clear_user_notes(user.id) user_data = self.bot.drama_tracker.get_user(user.id) asyncio.create_task(self.bot.db.save_user_state( user_id=user.id, offense_count=user_data.offense_count, immune=user_data.immune, off_topic_count=user_data.off_topic_count, baseline_coherence=user_data.baseline_coherence, user_notes=None, )) await interaction.response.send_message( f"Notes cleared for {user.display_name}.", ephemeral=True ) @app_commands.command( name="bcs-mode", description="Switch the bot's personality mode.", ) @app_commands.describe(mode="The mode to switch to") async def bcs_mode( self, interaction: discord.Interaction, mode: str | None = None, ): modes_config = self.bot.config.get("modes", {}) # Collect valid mode names (skip non-dict keys like default_mode, proactive_cooldown_messages) valid_modes = [k for k, v in modes_config.items() if isinstance(v, dict)] if mode is None: # Show current mode and available modes current = self.bot.current_mode current_config = self.bot.get_mode_config() lines = [f"**Current mode:** {current_config.get('label', current)}"] lines.append(f"*{current_config.get('description', '')}*\n") lines.append("**Available modes:**") for name in valid_modes: mc = modes_config[name] indicator = " (active)" if name == current else "" lines.append(f"- `{name}` — {mc.get('label', name)}: {mc.get('description', '')}{indicator}") await interaction.response.send_message("\n".join(lines), ephemeral=True) return mode = mode.lower() if mode not in valid_modes: await interaction.response.send_message( f"Unknown mode `{mode}`. Available: {', '.join(f'`{m}`' for m in valid_modes)}", ephemeral=True, ) return old_mode = self.bot.current_mode self.bot.current_mode = mode new_config = self.bot.get_mode_config() # Persist mode to database asyncio.create_task(self.bot.db.save_setting("current_mode", mode)) # Update bot status to reflect the mode status_text = new_config.get("description", "Monitoring vibes...") await self.bot.change_presence( activity=discord.Activity( type=discord.ActivityType.watching, name=status_text ) ) await interaction.response.send_message( f"Mode switched: **{modes_config.get(old_mode, {}).get('label', old_mode)}** " f"-> **{new_config.get('label', mode)}**\n" f"*{new_config.get('description', '')}*" ) # Log mode change log_channel = discord.utils.get(interaction.guild.text_channels, name="bcs-log") if log_channel: try: await log_channel.send( f"**MODE CHANGE** | {interaction.user.mention} switched mode: " f"**{old_mode}** -> **{mode}**" ) except discord.HTTPException: pass logger.info( "Mode changed from %s to %s by %s", old_mode, mode, interaction.user.display_name, ) @bcs_mode.autocomplete("mode") async def _mode_autocomplete( self, interaction: discord.Interaction, current: str, ) -> list[app_commands.Choice[str]]: modes_config = self.bot.config.get("modes", {}) valid_modes = [k for k, v in modes_config.items() if isinstance(v, dict)] return [ app_commands.Choice(name=modes_config[m].get("label", m), value=m) for m in valid_modes if current.lower() in m.lower() ][:25] @staticmethod def _score_bar(score: float) -> str: filled = round(score * 10) return "\u2588" * filled + "\u2591" * (10 - filled) @staticmethod def _score_color(score: float) -> discord.Color: if score >= 0.75: return discord.Color.red() if score >= 0.6: return discord.Color.orange() if score >= 0.3: return discord.Color.yellow() return discord.Color.green() async def setup(bot: commands.Bot): await bot.add_cog(CommandsCog(bot))