Initial commit - Excommunicado Discord bot (XCOM stealth isolation)
This commit is contained in:
@@ -0,0 +1,159 @@
|
||||
"""Main entry point for the Excommunicado Discord bot.
|
||||
|
||||
Excommunicado prevents flooding by isolating (excommunicating) offenders into
|
||||
private 'xcom-lounge' channels without banning them from the server.
|
||||
"""
|
||||
import os
|
||||
import asyncio
|
||||
from datetime import datetime, timezone
|
||||
from collections import defaultdict, deque
|
||||
from typing import Deque, Dict
|
||||
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
from discord import app_commands
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from utils.config import (
|
||||
get_guild_data,
|
||||
update_guild_data,
|
||||
add_isolated_user,
|
||||
remove_isolated_user,
|
||||
get_isolated_users,
|
||||
increment_absolution_points,
|
||||
)
|
||||
from cogs.moderation import get_or_create_xcom_role, get_or_create_xcom_channel
|
||||
|
||||
# Simple in-memory rate limiting for flood detection (user_id -> deque of timestamps)
|
||||
USER_MESSAGE_TIMES: Dict[int, Deque[datetime]] = defaultdict(lambda: deque(maxlen=20))
|
||||
FLOOD_THRESHOLD = 5 # messages
|
||||
FLOOD_WINDOW_SECONDS = 8 # within this many seconds
|
||||
|
||||
|
||||
class ExcommunicadoBot(commands.Bot):
|
||||
def __init__(self):
|
||||
intents = discord.Intents.default()
|
||||
intents.message_content = True
|
||||
intents.members = True
|
||||
intents.guilds = True
|
||||
super().__init__(command_prefix="!", intents=intents, help_command=None)
|
||||
|
||||
self.log_channel_id = None # set from env if provided
|
||||
|
||||
async def setup_hook(self) -> None:
|
||||
"""Load cogs and sync slash commands."""
|
||||
await self.load_extension("cogs.moderation")
|
||||
|
||||
guild_id = os.getenv("GUILD_ID")
|
||||
if guild_id:
|
||||
guild = discord.Object(id=int(guild_id))
|
||||
self.tree.copy_global_to(guild=guild)
|
||||
await self.tree.sync(guild=guild)
|
||||
print(f"[INFO] Synced commands to guild {guild_id}")
|
||||
else:
|
||||
await self.tree.sync()
|
||||
print("[INFO] Synced global commands (may take up to 1h)")
|
||||
|
||||
log_id = os.getenv("LOG_CHANNEL_ID")
|
||||
if log_id:
|
||||
self.log_channel_id = int(log_id)
|
||||
|
||||
async def on_ready(self) -> None:
|
||||
print(f"✅ Logged in as {self.user} (ID: {self.user.id})")
|
||||
print("Excommunicado bot is ready (XCOM stealth isolation mode).")
|
||||
|
||||
async def on_message(self, message: discord.Message) -> None:
|
||||
"""Stealth isolation + flood detection.
|
||||
|
||||
Isolated users keep seeing the normal server (no view permission changes).
|
||||
Their public messages are silently deleted and forwarded to the shared
|
||||
xcom-lounge room so they do not realize they have been moved.
|
||||
Trusted staff can also hold the Excommunicado role to share the room with them.
|
||||
"""
|
||||
if message.author.bot or not message.guild:
|
||||
return
|
||||
|
||||
guild_data = get_guild_data(message.guild.id)
|
||||
isolated_users = guild_data.get("isolated_users", {})
|
||||
xcom_role_id = guild_data.get("xcom_role_id")
|
||||
xcom_channel_id = guild_data.get("xcom_channel_id")
|
||||
|
||||
is_isolated = str(message.author.id) in isolated_users or (
|
||||
xcom_role_id and any(r.id == xcom_role_id for r in getattr(message.author, "roles", []))
|
||||
)
|
||||
|
||||
if is_isolated:
|
||||
# Award absolution points for positive participation in the xcom-lounge
|
||||
if xcom_channel_id and message.channel.id == xcom_channel_id:
|
||||
increment_absolution_points(message.guild.id, message.author.id, 1)
|
||||
|
||||
# Stealth mode: silently relocate message to the shared room
|
||||
if message.channel.id != xcom_channel_id:
|
||||
try:
|
||||
await message.delete()
|
||||
if xcom_channel_id:
|
||||
ch = message.guild.get_channel(xcom_channel_id)
|
||||
if ch:
|
||||
await ch.send(
|
||||
f"**[{message.channel.name}]** {message.author.mention}: {message.content}"
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
return # do not count floods or process commands from isolated users
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
times = USER_MESSAGE_TIMES[message.author.id]
|
||||
times.append(now)
|
||||
|
||||
# Clean old entries
|
||||
while times and (now - times[0]).total_seconds() > FLOOD_WINDOW_SECONDS:
|
||||
times.popleft()
|
||||
|
||||
if len(times) >= FLOOD_THRESHOLD:
|
||||
# Flood detected - auto XCOM the offender into xcom-lounge
|
||||
print(f"[FLOOD] Detected from {message.author} in {message.guild.name} - auto-XCOM'ing")
|
||||
try:
|
||||
role = await get_or_create_xcom_role(message.guild)
|
||||
channel = await get_or_create_xcom_channel(message.guild, role)
|
||||
if role not in message.author.roles:
|
||||
await message.author.add_roles(role, reason="Auto-XCOM'd: flooding detected")
|
||||
add_isolated_user(
|
||||
message.guild.id,
|
||||
message.author.id,
|
||||
"Auto-detected flooding (rapid messages)",
|
||||
now.isoformat()
|
||||
)
|
||||
await channel.send(
|
||||
f"{message.author.mention} has been auto-XCOM'd for flooding the channel.\n"
|
||||
"You are now chatting in the moderated shared room. Staff are present."
|
||||
)
|
||||
except Exception as flood_err:
|
||||
print(f"[FLOOD] Auto-XCOM failed: {flood_err}")
|
||||
|
||||
await self.process_commands(message)
|
||||
|
||||
async def log_action(self, guild: discord.Guild, embed: discord.Embed) -> None:
|
||||
"""Send log to configured log channel if set."""
|
||||
if self.log_channel_id:
|
||||
log_channel = guild.get_channel(self.log_channel_id)
|
||||
if log_channel:
|
||||
try:
|
||||
await log_channel.send(embed=embed)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
load_dotenv()
|
||||
token = os.getenv("DISCORD_TOKEN")
|
||||
if not token:
|
||||
print("ERROR: DISCORD_TOKEN not set in .env")
|
||||
return
|
||||
|
||||
bot = ExcommunicadoBot()
|
||||
async with bot:
|
||||
await bot.start(token)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user