798 lines
30 KiB
Python
798 lines
30 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-pause",
|
||
description="Pause or resume bot monitoring. (Admin only)",
|
||
)
|
||
@app_commands.default_permissions(administrator=True)
|
||
async def bcs_pause(self, interaction: discord.Interaction):
|
||
if not self._is_admin(interaction):
|
||
await interaction.response.send_message(
|
||
"Admin only.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
monitoring = self.bot.config.setdefault("monitoring", {})
|
||
currently_enabled = monitoring.get("enabled", True)
|
||
monitoring["enabled"] = not currently_enabled
|
||
|
||
if monitoring["enabled"]:
|
||
await interaction.response.send_message(
|
||
"Monitoring **resumed**.", ephemeral=True
|
||
)
|
||
else:
|
||
await interaction.response.send_message(
|
||
"Monitoring **paused**.", 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,
|
||
aliases=",".join(user_data.aliases) if user_data.aliases else 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 idx, msg in enumerate(messages):
|
||
# Build context from the messages before this one
|
||
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,
|
||
aliases=",".join(user_data.aliases) if user_data.aliases else 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,
|
||
aliases=",".join(user_data.aliases) if user_data.aliases else None,
|
||
))
|
||
await interaction.response.send_message(
|
||
f"Notes cleared for {user.display_name}.", ephemeral=True
|
||
)
|
||
|
||
@app_commands.command(
|
||
name="bcs-alias",
|
||
description="Manage nicknames/aliases for a user. (Admin only)",
|
||
)
|
||
@app_commands.default_permissions(administrator=True)
|
||
@app_commands.describe(
|
||
action="What to do with aliases",
|
||
user="The user whose aliases to manage",
|
||
text="Comma-separated aliases (only used with 'set')",
|
||
)
|
||
@app_commands.choices(action=[
|
||
app_commands.Choice(name="view", value="view"),
|
||
app_commands.Choice(name="set", value="set"),
|
||
app_commands.Choice(name="clear", value="clear"),
|
||
])
|
||
async def bcs_alias(
|
||
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":
|
||
aliases = self.bot.drama_tracker.get_user_aliases(user.id)
|
||
desc = ", ".join(aliases) if aliases else "_No aliases set._"
|
||
embed = discord.Embed(
|
||
title=f"Aliases: {user.display_name}",
|
||
description=desc,
|
||
color=discord.Color.blue(),
|
||
)
|
||
await interaction.response.send_message(embed=embed, ephemeral=True)
|
||
|
||
elif action.value == "set":
|
||
if not text:
|
||
await interaction.response.send_message(
|
||
"Provide `text` with comma-separated aliases (e.g. `Glam, G`).", ephemeral=True
|
||
)
|
||
return
|
||
aliases = [a.strip() for a in text.split(",") if a.strip()]
|
||
self.bot.drama_tracker.set_user_aliases(user.id, aliases)
|
||
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,
|
||
aliases=",".join(aliases),
|
||
))
|
||
await interaction.response.send_message(
|
||
f"Aliases for {user.display_name} set to: {', '.join(aliases)}", ephemeral=True
|
||
)
|
||
|
||
elif action.value == "clear":
|
||
self.bot.drama_tracker.set_user_aliases(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,
|
||
aliases=None,
|
||
))
|
||
await interaction.response.send_message(
|
||
f"Aliases 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,
|
||
)
|
||
|
||
@app_commands.command(
|
||
name="drama-leaderboard",
|
||
description="Show the all-time drama leaderboard for the server.",
|
||
)
|
||
@app_commands.describe(period="Time period to rank (default: 30d)")
|
||
@app_commands.choices(period=[
|
||
app_commands.Choice(name="Last 7 days", value="7d"),
|
||
app_commands.Choice(name="Last 30 days", value="30d"),
|
||
app_commands.Choice(name="Last 90 days", value="90d"),
|
||
app_commands.Choice(name="All time", value="all"),
|
||
])
|
||
async def drama_leaderboard(
|
||
self, interaction: discord.Interaction, period: app_commands.Choice[str] | None = None,
|
||
):
|
||
await interaction.response.defer()
|
||
|
||
period_val = period.value if period else "30d"
|
||
if period_val == "all":
|
||
days = None
|
||
period_label = "All Time"
|
||
else:
|
||
days = int(period_val.rstrip("d"))
|
||
period_label = f"Last {days} Days"
|
||
|
||
rows = await self.bot.db.get_drama_leaderboard(interaction.guild.id, days)
|
||
if not rows:
|
||
await interaction.followup.send(
|
||
f"No drama data for **{period_label}**. Everyone's been suspiciously well-behaved."
|
||
)
|
||
return
|
||
|
||
# Compute composite score for each user
|
||
scored = []
|
||
for r in rows:
|
||
avg_tox = r["avg_toxicity"]
|
||
max_tox = r["max_toxicity"]
|
||
msg_count = r["messages_analyzed"]
|
||
action_weight = r["warnings"] + r["mutes"] * 2 + r["off_topic"] * 0.5
|
||
action_rate = min(1.0, action_weight / msg_count * 10) if msg_count > 0 else 0.0
|
||
composite = avg_tox * 0.4 + max_tox * 0.2 + action_rate * 0.4
|
||
scored.append({**r, "composite": composite, "action_rate": action_rate})
|
||
|
||
scored.sort(key=lambda x: x["composite"], reverse=True)
|
||
top = scored[:10]
|
||
|
||
medals = ["🥇", "🥈", "🥉"]
|
||
lines = []
|
||
for i, entry in enumerate(top):
|
||
rank = medals[i] if i < 3 else f"`{i + 1}.`"
|
||
|
||
# Resolve display name from guild if possible
|
||
member = interaction.guild.get_member(entry["user_id"])
|
||
name = member.display_name if member else entry["username"]
|
||
|
||
lines.append(
|
||
f"{rank} **{entry['composite']:.2f}** — {name}\n"
|
||
f" Avg: {entry['avg_toxicity']:.2f} | "
|
||
f"Peak: {entry['max_toxicity']:.2f} | "
|
||
f"⚠️ {entry['warnings']} | "
|
||
f"🔇 {entry['mutes']} | "
|
||
f"📢 {entry['off_topic']}"
|
||
)
|
||
|
||
embed = discord.Embed(
|
||
title=f"Drama Leaderboard — {period_label}",
|
||
description="\n".join(lines),
|
||
color=discord.Color.orange(),
|
||
)
|
||
embed.set_footer(text=f"{len(rows)} users tracked | {sum(r['messages_analyzed'] for r in rows)} messages analyzed")
|
||
|
||
await interaction.followup.send(embed=embed)
|
||
|
||
@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))
|