Subinacls
270 lines
12 KiB
Python
270 lines
12 KiB
Python
"""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,
|
|
get_or_create_xcom_mod_channel,
|
|
_post_to_mod_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
|
|
)
|
|
logger.info(
|
|
"Keyword match: user=%s terms=%s (+%d) -> score=%d guild=%s",
|
|
author, matched_terms, matched_weight, result["score"], 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 _post_to_mod_channel(message.guild, None, log_embed)
|
|
await self.log_action(message.guild, log_embed)
|
|
logger.warning(
|
|
"Keyword auto-XCOM: user=%s guild=%s score=%d",
|
|
author, message.guild.name, result["score"],
|
|
)
|
|
return
|
|
except Exception as kw_err:
|
|
logger.exception("Keyword auto-XCOM failed: %s", 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())
|