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

Applying previous commit.

Browse Source
This commit is contained in:
Slipstream 2025-06-07 04:01:03 +00:00
parent 22d22e4452
commit d845418443
Signed by: slipstream
GPG Key ID: 13E498CE010AC6FD
2 changed files with 278 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

231
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.",
@ -844,6 +853,59 @@ class AIModerationCog(commands.Cog):
await interaction.response.send_message(embed=embed, ephemeral=False)
@appeal_subgroup.command(
name="human_review",
description="Request a human moderator to review your case.",
)
@app_commands.describe(
reason="Explain why you want a human to review the AI decision",
guild_id="If using in DMs, provide the server ID",
)
async def appeal_human_review(
self,
interaction: discord.Interaction,
reason: str,
guild_id: int | None = None,
):
"""Let a user request a manual moderator review."""
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
timestamp = datetime.datetime.utcnow().isoformat()
await add_user_appeal(
guild.id, interaction.user.id, "HUMAN_REVIEW", reason, timestamp, ""
)
embed = discord.Embed(
title="Human Review Requested", color=discord.Color.orange()
)
embed.add_field(
name="User",
value=f"{interaction.user} ({interaction.user.id})",
inline=False,
)
embed.add_field(name="Request", value=reason, inline=False)
embed.timestamp = discord.utils.utcnow()
await log_channel.send(embed=embed)
await interaction.response.send_message(
"Your request for a human review has been sent.", ephemeral=True
)
@infractions_subgroup.command(
name="clear",
description="Clear a user's AI moderation infraction history (admin only).",
@ -933,6 +995,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)."
)
@ -1480,9 +1638,7 @@ CRITICAL: Do NOT output anything other than the required JSON response.
else:
model_path = model_id_to_use
thinking_config = types.ThinkingConfig(
thinking_budget=0
)
thinking_config = types.ThinkingConfig(thinking_budget=0)
generation_config = types.GenerateContentConfig(
temperature=0.2,
@ -1985,10 +2141,24 @@ CRITICAL: Do NOT output anything other than the required JSON response.
)
try:
dm_channel = await message.author.create_dm()
await dm_channel.send(
f"Your recent message in **{message.guild.name}** was removed for violating Rule **{rule_violated}**. "
f"Reason: _{reasoning}_. Please review the server rules. This is a formal warning."
warn_embed = discord.Embed(
title="⚠️ Moderation Warning",
description=(
f"Your recent message in **{message.guild.name}** was removed for violating **Rule {rule_violated}**."
),
color=discord.Color.orange(),
)
if message.content:
warn_embed.add_field(
name="Message Content",
value=message.content[:1024],
inline=False,
)
warn_embed.add_field(name="Reason", value=reasoning, inline=False)
warn_embed.set_footer(
text="Please review the server rules. This is a formal warning."
)
await dm_channel.send(embed=warn_embed)
action_taken_message += " User notified via DM with warning."
except discord.Forbidden:
print(
@ -2164,6 +2334,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