"""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 import logging 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, get_keywords, add_keyword_points, clear_votes, check_writable, ) from cogs.moderation import ( get_or_create_xcom_role, get_or_create_xcom_channel, VOTE_XCOM_THRESHOLD, VOTE_RESET_DAYS, ) logger = logging.getLogger("excommunicado.bot") # 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") logger.info("Loaded 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) synced = await self.tree.sync(guild=guild) logger.info("Synced %d commands to guild %s: %s", len(synced), guild_id, ", ".join(c.name for c in synced)) else: synced = await self.tree.sync() logger.warning( "GUILD_ID not set: synced %d GLOBAL commands (may take up to 1h to appear): %s", len(synced), ", ".join(c.name for c in synced), ) log_id = os.getenv("LOG_CHANNEL_ID") if log_id: self.log_channel_id = int(log_id) logger.info("Log channel configured: %s", log_id) async def on_ready(self) -> None: ok, msg = check_writable() if ok: logger.info(msg) else: logger.error(msg) logger.info("Logged in as %s (ID: %s)", self.user, self.user.id) logger.info("Excommunicado bot is ready (XCOM stealth isolation mode).") print(f"✅ Logged in as {self.user} (ID: {self.user.id})") 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: content = message.content or "*(no text content)*" prefix = f"**[{message.channel.name}]** {message.author.mention}: " # Discord hard limit is 2000 chars; truncate to stay under it max_content = 2000 - len(prefix) - 1 if len(content) > max_content: content = content[: max_content - 1] + "…" await ch.send( prefix + content, allowed_mentions=discord.AllowedMentions.none(), ) except Exception: pass return # do not count floods or process commands from isolated users now = datetime.now(timezone.utc) # --- Admin-curated keyword / key-phrase auto-scoring --- # Captures critical abuse (e.g. trading cracked software/games, illegal # materials) even when no admin is online. Staff/owner are immune. author = message.author is_staff = isinstance(author, discord.Member) and ( author.guild_permissions.manage_roles or author.guild_permissions.administrator or author.id == message.guild.owner_id ) keywords = guild_data.get("keywords", {}) if keywords and not is_staff and message.content: content_low = message.content.lower() matched_weight = 0 matched_terms = [] for phrase, info in keywords.items(): if phrase and phrase in content_low: matched_weight += int(info.get("weight", 1)) matched_terms.append(info.get("display", phrase)) if matched_weight > 0: try: result = add_keyword_points( message.guild.id, author.id, matched_weight, reset_days=VOTE_RESET_DAYS ) print( f"[KEYWORD] {author} matched {matched_terms} (+{matched_weight}) " f"-> score {result['score']} in {message.guild.name}" ) me = message.guild.me can_manage = me is not None and author.top_role < me.top_role if result["score"] >= VOTE_XCOM_THRESHOLD and can_manage: cog = self.get_cog("ModerationCog") if cog is not None: reason = "Auto-flagged keywords: " + ", ".join(matched_terms[:5]) channel, already = await cog._perform_xcom( message.guild, author, reason, "keyword auto-detect" ) clear_votes(message.guild.id, author.id) log_embed = discord.Embed( title="🚩 Keyword Auto-XCOM", description=f"{author.mention} ({author.id}) was auto-XCOM'd into {channel.mention} after matching flagged keywords.", color=discord.Color.dark_red(), timestamp=now, ) log_embed.add_field(name="Matched", value=", ".join(matched_terms[:10]) or "—", inline=False) await self.log_action(message.guild, log_embed) return except Exception as kw_err: print(f"[KEYWORD] Auto-XCOM failed: {kw_err}") 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}") 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 as err: logger.warning("Failed to post log embed to channel %s: %s", self.log_channel_id, err) def _configure_logging() -> None: """Configure root logging from the LOG_LEVEL env var (default INFO).""" level_name = os.getenv("LOG_LEVEL", "INFO").upper() level = getattr(logging, level_name, logging.INFO) logging.basicConfig( level=level, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) # discord.py is noisy at DEBUG; keep it at INFO unless explicitly debugging. logging.getLogger("discord").setLevel( logging.DEBUG if level <= logging.DEBUG else logging.INFO ) logger.info("Logging configured at level %s", level_name) async def main() -> None: load_dotenv() _configure_logging() token = os.getenv("DISCORD_TOKEN") if not token: logger.error("DISCORD_TOKEN not set in .env") return ok, msg = check_writable() if not ok: logger.error(msg) else: logger.info(msg) bot = ExcommunicadoBot() async with bot: await bot.start(token) if __name__ == "__main__": asyncio.run(main())