AJ Isaacs
6e1a73847d
Adds a BotSettings key-value table. The active mode is saved when changed via /bcs-mode and restored on startup. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
625 lines
23 KiB
Python
625 lines
23 KiB
Python
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))
|