slipstream/discordbot
slipstream/discordbot
1
1
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Add appeal system

Browse Source
This commit is contained in:
Codex 2025-06-07 03:27:27 +00:00 committed by Slipstream
parent cd6278ce52
commit fc4bf0e059
Signed by: slipstream
GPG Key ID: 13E498CE010AC6FD
2 changed files with 207 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

154
cogs/aimod_cog.py
View File

@ -39,12 +39,18 @@ from .aimod_config import (
GUILD_CONFIG_PATH,
USER_INFRACTIONS_PATH,
INFRACTION_BACKUP_DIR,
USER_APPEALS_PATH,
APPEAL_AI_MODEL,
APPEAL_AI_THINKING_BUDGET,
CONFIG_LOCK,
save_user_infractions,
save_user_appeals,
get_guild_config,
set_guild_config,
get_user_infraction_history,
add_user_infraction,
get_user_appeals,
add_user_appeal,
SERVER_RULES,
SUICIDAL_HELP_RESOURCES,
)
@ -508,6 +514,9 @@ class AIModerationCog(commands.Cog):
infractions_subgroup = app_commands.Group(
name="infractions", description="Manage user infractions.", parent=aimod_group
)
appeal_subgroup = app_commands.Group(
name="appeal", description="Appeal AI moderation actions.", parent=aimod_group
)
model_subgroup = app_commands.Group(
name="model",
description="Manage the AI model for moderation.",
@ -933,6 +942,102 @@ class AIModerationCog(commands.Cog):
f"Failed to restore infractions: {e}", ephemeral=True
)
@appeal_subgroup.command(name="submit", description="Submit a moderation appeal.")
@app_commands.describe(
action="The action you are appealing",
reason="Explain why you believe the action was incorrect",
guild_id="If using in DMs, provide the server ID",
)
async def appeal_submit(
self,
interaction: discord.Interaction,
action: str,
reason: str,
guild_id: int | None = None,
):
guild = interaction.guild or (
self.bot.get_guild(guild_id) if guild_id else None
)
if not guild:
await interaction.response.send_message(
"Invalid or missing guild ID.", ephemeral=True
)
return
log_channel_id = get_guild_config(guild.id, "MOD_LOG_CHANNEL_ID")
log_channel = self.bot.get_channel(log_channel_id) if log_channel_id else None
if not log_channel:
await interaction.response.send_message(
"Appeals are not enabled for this server.", ephemeral=True
)
return
ai_review = await self.run_appeal_ai(guild, interaction.user, action, reason)
timestamp = datetime.datetime.utcnow().isoformat()
await add_user_appeal(
guild.id, interaction.user.id, action, reason, timestamp, ai_review
)
embed = discord.Embed(title="New Appeal", color=discord.Color.blue())
embed.add_field(
name="User",
value=f"{interaction.user} ({interaction.user.id})",
inline=False,
)
embed.add_field(name="Action", value=action, inline=False)
embed.add_field(name="Appeal", value=reason, inline=False)
embed.add_field(name="AI Review", value=ai_review[:1000], inline=False)
embed.timestamp = discord.utils.utcnow()
await log_channel.send(embed=embed)
await interaction.response.send_message(
"Your appeal has been submitted.", ephemeral=True
)
@appeal_subgroup.command(
name="list", description="View a user's appeals (mods only)."
)
@app_commands.describe(user="The user to view appeals for")
async def appeal_list(self, interaction: discord.Interaction, user: discord.Member):
moderator_role_id = get_guild_config(interaction.guild.id, "MODERATOR_ROLE_ID")
moderator_role = (
interaction.guild.get_role(moderator_role_id) if moderator_role_id else None
)
has_permission = interaction.user.guild_permissions.administrator or (
moderator_role and moderator_role in interaction.user.roles
)
if not has_permission:
await interaction.response.send_message(
"You must be an administrator or have the moderator role to use this command.",
ephemeral=True,
)
return
appeals = get_user_appeals(interaction.guild.id, user.id)
if not appeals:
await interaction.response.send_message(
f"{user.mention} has no appeals.", ephemeral=False
)
return
embed = discord.Embed(
title=f"Appeals for {user.display_name}", color=discord.Color.blue()
)
for i, appeal in enumerate(appeals, 1):
ts = appeal.get("timestamp", "?")[:19].replace("T", " ")
summary = appeal.get("appeal_text", "")
ai_sum = appeal.get("ai_review", "")
if len(summary) > 150:
summary = summary[:147] + "..."
if len(ai_sum) > 150:
ai_sum = ai_sum[:147] + "..."
embed.add_field(
name=f"Appeal #{i} - {ts}",
value=f"Action: {appeal.get('action')}\nReason: {summary}\nAI: {ai_sum}",
inline=False,
)
await interaction.response.send_message(embed=embed, ephemeral=False)
@model_subgroup.command(
name="set", description="Change the AI model used for moderation (admin only)."
)
@ -2176,6 +2281,55 @@ CRITICAL: Do NOT output anything other than the required JSON response.
"FATAL: Bot lacks permission to send messages, even error notifications."
)
async def run_appeal_ai(
self, guild: discord.Guild, member: discord.User, action: str, appeal_text: str
) -> str:
"""Run the appeal text through the higher tier AI model."""
if not self.genai_client:
return "AI review unavailable."
history = get_user_infraction_history(guild.id, member.id)
history_text = json.dumps(history, indent=2) if history else "None"
system_prompt = (
"You are reviewing a user's appeal of a moderation action. "
"Think very extensively about the appeal, the provided history, and the server rules. "
"Return a short verdict (UPHOLD or OVERTURN) and your reasoning in plain text."
)
user_prompt = (
f"Server Rules:\n{SERVER_RULES}\n\n"
f"User History:\n{history_text}\n\n"
f"Action Appealed: {action}\n"
f"Appeal Text: {appeal_text}"
)
generation_config = types.GenerateContentConfig(
temperature=0.2,
max_output_tokens=8192,
safety_settings=STANDARD_SAFETY_SETTINGS,
thinking_config=types.ThinkingConfig(
thinking_budget=APPEAL_AI_THINKING_BUDGET
),
system_instruction=types.Content(
role="system", parts=[types.Part(text=system_prompt)]
),
)
try:
response = await self.genai_client.aio.models.generate_content(
model=f"publishers/google/models/{APPEAL_AI_MODEL}",
contents=[
types.Content(role="user", parts=[types.Part(text=user_prompt)])
],
config=generation_config,
)
result = self._get_response_text(response)
return result or "AI review failed to produce output."
except Exception as e: # noqa: BLE001
print(f"Appeal AI error: {e}")
return "AI review encountered an error."
@commands.Cog.listener(name="on_message")
async def message_listener(self, message: discord.Message):
"""Listens to messages and triggers moderation checks."""

53
cogs/aimod_config.py
View File

@ -31,6 +31,12 @@ GUILD_CONFIG_DIR = "data/"
GUILD_CONFIG_PATH = os.path.join(GUILD_CONFIG_DIR, "guild_config.json")
USER_INFRACTIONS_PATH = os.path.join(GUILD_CONFIG_DIR, "user_infractions.json")
INFRACTION_BACKUP_DIR = os.path.join(GUILD_CONFIG_DIR, "infraction_backups")
USER_APPEALS_PATH = os.path.join(GUILD_CONFIG_DIR, "user_appeals.json")
# AI model used for appeal reviews
APPEAL_AI_MODEL = "gemini-2.5-pro-preview-06-05"
# Thinking budget for appeal AI
APPEAL_AI_THINKING_BUDGET = 32768
os.makedirs(INFRACTION_BACKUP_DIR, exist_ok=True)
os.makedirs(GUILD_CONFIG_DIR, exist_ok=True)
@ -55,6 +61,16 @@ except Exception as e: # noqa: BLE001
print(f"Failed to load user infractions from {USER_INFRACTIONS_PATH}: {e}")
USER_INFRACTIONS = {}
if not os.path.exists(USER_APPEALS_PATH):
with open(USER_APPEALS_PATH, "w", encoding="utf-8") as f:
json.dump({}, f)
try:
with open(USER_APPEALS_PATH, "r", encoding="utf-8") as f:
USER_APPEALS = json.load(f)
except Exception as e: # noqa: BLE001
print(f"Failed to load user appeals from {USER_APPEALS_PATH}: {e}")
USER_APPEALS = {}
CONFIG_LOCK = asyncio.Lock()
@ -76,6 +92,15 @@ async def save_user_infractions():
print(f"Failed to save user infractions: {e}")
async def save_user_appeals():
async with CONFIG_LOCK:
try:
async with aiofiles.open(USER_APPEALS_PATH, "w", encoding="utf-8") as f:
await f.write(json.dumps(USER_APPEALS, indent=2))
except Exception as e: # noqa: BLE001
print(f"Failed to save user appeals: {e}")
def get_guild_config(guild_id: int, key: str, default=None):
guild_str = str(guild_id)
if guild_str in GUILD_CONFIG and key in GUILD_CONFIG[guild_str]:
@ -119,6 +144,34 @@ async def add_user_infraction(
await save_user_infractions()
def get_user_appeals(guild_id: int, user_id: int) -> list:
key = f"{guild_id}_{user_id}"
return USER_APPEALS.get(key, [])
async def add_user_appeal(
guild_id: int,
user_id: int,
action: str,
appeal_text: str,
timestamp: str,
ai_review: str,
):
key = f"{guild_id}_{user_id}"
if key not in USER_APPEALS:
USER_APPEALS[key] = []
appeal_record = {
"timestamp": timestamp,
"action": action,
"appeal_text": appeal_text,
"ai_review": ai_review,
}
USER_APPEALS[key].append(appeal_record)
USER_APPEALS[key] = USER_APPEALS[key][-10:]
await save_user_appeals()
DEFAULT_SERVER_RULES = """
# Server Rules