diff --git a/.gitignore b/.gitignore index cf8d5ef..e9899e8 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,149 @@ -__pycache__/ +# Byte-compiled / optimized / DLL files +__pycache__/ *.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +.python-version + +# pipenv +Pipfile.lock + +# poetry +poetry.lock + +# pdm +.pdm.toml + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments .env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# Project specific +# Discord bot +.env +.env.* data/ *.log -.venv/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo + +# OS generated files +.DS_Store +Thumbs.db diff --git a/Dockerfile b/Dockerfile index 3f757e2..73190c9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -4,11 +4,6 @@ FROM python:3.12-slim # Set working directory WORKDIR /app -# Install system dependencies (if needed) -RUN apt-get update && apt-get install -y --no-install-recommends \ - build-essential \ - && rm -rf /var/lib/apt/lists/* - # Copy requirements first (better layer caching) COPY requirements.txt . diff --git a/README.md b/README.md index d51aa04..ca46dce 100644 --- a/README.md +++ b/README.md @@ -23,14 +23,24 @@ In **stealth mode**, offending users **do not realize** they have been moved. Th - **Slash commands only** (modern Discord UX) - `/xcom-setup` – one-time server initialization - `/xcom @user [reason]` – excommunicate a user (manual or via auto-detection) - - `/xcom-release @user` – restore full access + - `/xcom-release @user` – restore full access (the `[XCOM]` mark remains) + - `/xcom-unmark @user` – remove the permanent `[XCOM]` nickname mark + - `/xcom-vow ` – let an isolated user post a vow of redemption to staff + - `/xcom-witness @staff @target` – add a staff member to the shared lounge + - `/xcom-absolve @user` – release a user early once they earn enough points + - `/xcom-vote @user up|down` – community vote to excommunicate (or defend) a user + - `/xcom-votes @user` – staff view of a user's current vote tally + - `/xcom-keywords` – admin panel to curate auto-score keywords/phrases - `/xcom-list` – view all currently XCOM users with reasons and timestamps - **Automatic flood protection** – detects ≥5 messages within 8 seconds and auto-excommunicates +- **Community voting** – members up/down-vote a user; reaching a net **+10** (with quorum) auto-XCOMs them; tallies reset weekly +- **Keyword auto-scoring** – admin-curated phrases (e.g. cracked-software/warez trading) add weighted points to a sender's score and can auto-XCOM critical abuse even when no admin is online - **Absolution Points** – users earn points for good behavior in the xcom-lounge; staff can absolve early - **Shared hidden channel** `xcom-lounge` with 5-second slowmode -- **Persistent state** – JSON storage survives bot restarts +- **Persistent state** – atomic, corruption-safe JSON storage survives bot restarts +- **Mention-safe forwarding** – relocated messages never ping (`@everyone`/`@here`/roles are neutralized) - **Rich embeds + DM notifications** for offenders and staff logs -- **Graceful permission handling** and role hierarchy checks +- **Graceful permission handling** and role-hierarchy safety checks - Zero bans/kicks – users stay in the server ## Philosophy & Use Cases @@ -70,7 +80,7 @@ The bot runs as a non-root user with healthchecks and log rotation. To view logs: ```powershell -docker compose logs -f excommunicado-bot +docker compose logs -f excommunicado ``` To stop: @@ -149,10 +159,13 @@ The user is stealthily XCOM'd. Their messages are forwarded to the shared room. - `reason` (optional) – default "Flooding the channel" - **Effects**: - Assigns `Excommunicado` role + - Applies the permanent `[XCOM]` nickname mark (best-effort; needs Manage Nicknames) - Records in `data/xcom_config.json` - Posts welcome embed in `xcom-lounge` - DMs the user - Logs to staff channel (if set) +- **Role-hierarchy safety**: refuses to XCOM the server owner, a member whose top role is equal to or higher than the invoker's, or a member ranked above the bot — each returns a clear ephemeral error instead of failing silently. +- **Cooldown**: max 5 uses per 15 seconds per moderator (per guild) to prevent accidental mass-spam. - **Edge cases**: Already under XCOM, bot, self-target, missing hierarchy → clear ephemeral error ### `/xcom-release ` @@ -173,12 +186,35 @@ The user is stealthily XCOM'd. Their messages are forwarded to the shared room. ### `/xcom-absolve @user` - **Permission**: Manage Roles +- **Cooldown**: max 5 uses per 15 seconds per moderator (per guild) - **Effects**: If the user has ≥10 absolution points (earned by positive participation in the lounge), removes the role early. The mark remains. ### `/xcom-list` - **Permission**: Manage Roles - **Output**: Embed listing up to 15 users with reason + date (truncated if more) +### `/xcom-vote ` +- **Permission**: None (any member) — light cooldown of 10 uses / 60s per member +- **Effects**: Casts a weighted community vote on the target's running score. + - `up` = excommunicate, `down` = defend. + - Admins/moderators count **±2**; regular members **±1**. + - **One vote per voter per target per 24h** (re-voting replaces your previous vote). + - When the **net score ≥ 10** *and* at least **5 distinct voters** have participated, the user is automatically XCOM'd and the tally is cleared. + - Tallies **reset weekly**. +- **Guardrails**: cannot vote on yourself, bots, the server owner, staff/moderators, or anyone ranked above the bot. Confirmations are ephemeral (preserves stealth, avoids public pile-on). + +### `/xcom-votes ` +- **Permission**: Manage Roles +- **Output**: Ephemeral embed showing the target's current score, distinct-voter count, threshold, and quorum, with a breakdown of points from human votes vs. keyword flags. + +### `/xcom-keywords` +- **Permission**: Manage Server (admin) +- **Effects**: Opens an **interactive ephemeral panel** to curate the auto-score keyword/phrase list: + - **Add / Edit** — a modal to enter a phrase and its score weight (1–100). Re-adding an existing phrase updates its weight. + - **Remove** — a modal to delete a phrase. + - **Refresh** — re-render the current list. +- **How it works**: any non-staff message containing a listed phrase (case-insensitive substring match) adds that phrase's weight to the sender's community score. When the score reaches the XCOM threshold, the user is auto-isolated — **no quorum required**, since this path is meant to capture abuse when no admin is online. Staff, the server owner, and members ranked above the bot are immune. + All commands are ephemeral where possible to keep chat clean. ## Configuration @@ -192,13 +228,40 @@ All commands are ephemeral where possible to keep chat clean. ### Tunable Constants (in `bot.py`) ```python -FLOOD_THRESHOLD = 5 -FLOOD_WINDOW_SECONDS = 5 +FLOOD_THRESHOLD = 5 # messages +FLOOD_WINDOW_SECONDS = 8 # within this many seconds ``` -Change and restart to adjust sensitivity. +A user who sends `FLOOD_THRESHOLD` messages within `FLOOD_WINDOW_SECONDS` is auto-excommunicated. Change and restart to adjust sensitivity. + +### Community Vote Constants (in `cogs/moderation.py`) +```python +VOTE_XCOM_THRESHOLD = 10 # net score required to auto-XCOM a user +VOTE_ADMIN_WEIGHT = 2 # admins / moderators count double +VOTE_USER_WEIGHT = 1 # regular members +VOTE_COOLDOWN_HOURS = 24 # one vote per voter per target per 24h +VOTE_RESET_DAYS = 7 # tallies reset weekly +VOTE_MIN_UNIQUE_VOTERS = 5 # quorum: distinct voters required before XCOM fires +``` + +**Recommended guardrails (already enforced):** +- **Quorum** (`VOTE_MIN_UNIQUE_VOTERS`) prevents a handful of admins (±2 each) or a tiny brigade from instantly triggering an XCOM — a genuine community consensus is required. +- **Staff immunity** — moderators/admins and the server owner can't be vote-targeted. +- **Hierarchy check** — the bot refuses votes it couldn't enforce. +- **Ephemeral results** keep the target unaware (stealth) and discourage public harassment campaigns. +- **Auto-clear on trigger** stops a target from being re-XCOM'd on the very next vote. + +Tune `VOTE_XCOM_THRESHOLD` and `VOTE_MIN_UNIQUE_VOTERS` together to match your server size — larger servers should raise both. + +### Keyword Auto-Scoring +Admins curate the phrase list live via `/xcom-keywords` (no restart needed). Each phrase has an integer **weight**; a matching message adds that weight to the sender's community score (the same score `/xcom-vote` feeds). Notes: +- Matching is **case-insensitive substring**, so curate carefully — this is a cybersecurity channel where some terms are legitimate in context. Prefer specific phrases (`"selling cracked"`) over broad single words (`"crack"`). +- Multiple distinct phrases matched in one message stack their weights. +- A high weight (e.g. `10`) makes a single message enough to XCOM instantly; low weights (`1`–`3`) require repeated abuse. +- Keyword points are part of the weekly-resetting score and are shown separately in `/xcom-votes`. +- The keyword path **bypasses the human voter quorum** by design, but still respects staff/owner immunity and bot role hierarchy. ### Data File -`data/xcom_config.json` (auto-created): +`data/xcom_config.json` (auto-created). Writes are **atomic** (written to a temp file, then `os.replace`d) and the file is **cached in memory**, so the bot does not re-read disk on every message: ```json { "123456789": { @@ -207,12 +270,37 @@ Change and restart to adjust sensitivity. "isolated_users": { "333333333": { "reason": "Auto-detected flooding...", - "timestamp": "2026-06-08T12:34:56.789012+00:00" + "timestamp": "2026-06-08T12:34:56.789012+00:00", + "absolution_points": 0, + "marked": true + } + }, + "votes": { + "444444444": { + "week_start": "2026-06-08T00:00:00+00:00", + "voters": { + "555555555": { "direction": 1, "weight": 2, "timestamp": "2026-06-08T12:00:00+00:00" } + }, + "keyword_points": 3 + } + }, + "keywords": { + "selling cracked software": { + "weight": 5, + "added_by": 555555555, + "added_at": "2026-06-08T10:00:00+00:00", + "display": "selling cracked software" } } } } ``` +- `absolution_points` – earned by positive participation in the lounge; `/xcom-absolve` needs ≥10. +- `marked` – whether the permanent `[XCOM]` nickname mark is applied. +- `votes` – per-target community-vote tallies; each voter's `direction` (±1) × `weight` is summed, plus `keyword_points` from auto-scoring, into the net score; the whole record is discarded once `week_start` is older than a week. +- `keywords` – admin-curated auto-score phrases (managed via `/xcom-keywords`). + +If the file is ever corrupted, the bot backs it up to `xcom_config.json.corrupt` and starts fresh instead of crashing. ## How Isolation Works (Technical) — Stealth Mode 1. On `/xcom-setup`, the bot creates: @@ -223,7 +311,7 @@ Change and restart to adjust sensitivity. - **No view permissions are removed** from public channels — the user continues to see the full server and does not realize anything changed. 3. On every message the isolated user sends outside `xcom-lounge`: - The message is **instantly deleted** (public never sees the spam) - - The content is forwarded to the shared `xcom-lounge` room + - The content is forwarded to the shared `xcom-lounge` room **with all mentions neutralized** (no `@everyone`/`@here`/role pings) and **truncated** to Discord's 2000-character limit - Staff members who also hold the `Excommunicado` role can see and reply in the same room 4. The offender experiences normal server browsing + "their messages just go to the moderated lounge where staff are" 5. `/xcom-release` removes the role — full normal behavior is restored instantly @@ -262,15 +350,24 @@ Key flows: - **Commands not appearing** → Restart bot after `GUILD_ID` change; wait 1h for global sync - **Isolated user messages still appear publicly** → The stealth suppression logic in on_message is not triggering (check that the user has the Excommunicado role or is in the isolated_users JSON) - **Auto-detection too sensitive** → Increase `FLOOD_WINDOW_SECONDS` or threshold -- **JSON corrupted** → Delete `data/xcom_config.json` (bot will recreate) +- **JSON corrupted** → The bot auto-recovers: the bad file is moved to `data/xcom_config.json.corrupt` and a fresh one is created. Restore from the backup if you need the old data. +- **`[XCOM]` mark not applied** → Bot is missing `Manage Nicknames`, or the target outranks the bot (role assignment still succeeds; the mark is best-effort) ## Security Best Practices -- Never commit `.env` or the token +- Never commit `.env` or the token (the provided `.gitignore` already excludes `.env`, `.env.*`, and `data/`) - Use a dedicated bot account (not your personal account) - Give the bot the minimum permissions required - Regularly audit `data/xcom_config.json` for old entries - Give trusted staff the `Excommunicado` role so they can share the moderated room with XCOM users and respond naturally +### Built-in hardening +- **No mention abuse**: forwarded messages are sent with mentions disabled, so an isolated user cannot make the bot ping `@everyone`, `@here`, or any role. +- **Length-safe forwarding**: relocated messages are truncated to Discord's 2000-character limit; image/sticker-only messages forward a placeholder instead of failing. +- **Atomic persistence**: config writes can't half-complete and corrupt state; a corrupt file is quarantined automatically. +- **Role-hierarchy enforcement**: `/xcom` will not target the owner, equal/higher-ranked members, or members above the bot. +- **Command cooldowns**: `/xcom` and `/xcom-absolve` are rate-limited (5 per 15s per moderator) and surface a friendly "slow down" message instead of erroring. +- **Runs as non-root** in Docker with healthchecks and log rotation. + ## Extending the Bot Example: add a `/xcom-stats` command counting total excommunications. diff --git a/bot.py b/bot.py index c3cf31b..97b8f8c 100644 --- a/bot.py +++ b/bot.py @@ -21,8 +21,16 @@ from utils.config import ( remove_isolated_user, get_isolated_users, increment_absolution_points, + get_keywords, + add_keyword_points, + clear_votes, +) +from cogs.moderation import ( + get_or_create_xcom_role, + get_or_create_xcom_channel, + VOTE_XCOM_THRESHOLD, + VOTE_RESET_DAYS, ) -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)) @@ -94,14 +102,71 @@ class ExcommunicadoBot(commands.Bot): 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( - f"**[{message.channel.name}]** {message.author.mention}: {message.content}" + 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) @@ -130,8 +195,6 @@ class ExcommunicadoBot(commands.Bot): 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: diff --git a/cogs/moderation.py b/cogs/moderation.py index 69593d1..658e889 100644 --- a/cogs/moderation.py +++ b/cogs/moderation.py @@ -19,8 +19,23 @@ from utils.config import ( add_isolated_user, remove_isolated_user, get_isolated_users, + cast_vote, + get_vote_score, + get_keyword_points, + clear_votes, + add_keyword, + remove_keyword, + get_keywords, ) +# --- Community vote configuration --- +VOTE_XCOM_THRESHOLD = 10 # net score required to auto-XCOM a user +VOTE_ADMIN_WEIGHT = 2 # admins / moderators count double +VOTE_USER_WEIGHT = 1 # regular members +VOTE_COOLDOWN_HOURS = 24 # one vote per voter per target per 24h +VOTE_RESET_DAYS = 7 # tallies reset weekly +VOTE_MIN_UNIQUE_VOTERS = 5 # quorum: distinct voters required before XCOM fires + async def get_or_create_xcom_role(guild: discord.Guild) -> discord.Role: """Get existing Excommunicado role or create a new one.""" @@ -72,12 +87,184 @@ async def get_or_create_xcom_channel(guild: discord.Guild, role: discord.Role) - return channel +def _keyword_list_embed(guild: discord.Guild) -> discord.Embed: + """Build an embed listing the guild's auto-score keywords.""" + keywords = get_keywords(guild.id) + embed = discord.Embed( + title="🚩 Auto-Score Keywords", + description=( + "Messages matching these phrases add points to the sender's community score.\n" + "Reaching the XCOM threshold auto-isolates them — even when no admin is online.\n" + "_Staff and the server owner are immune._" + ), + color=discord.Color.orange(), + ) + if not keywords: + embed.add_field(name="No keywords yet", value="Use **Add / Edit** to create one.", inline=False) + else: + lines = [] + for _, info in sorted(keywords.items(), key=lambda kv: -int(kv[1].get("weight", 0))): + lines.append(f"• `{info.get('display', '?')}` — **+{int(info.get('weight', 1))}**") + # Embed field values cap at 1024 chars; chunk if needed + chunk = "" + idx = 1 + for line in lines: + if len(chunk) + len(line) + 1 > 1024: + embed.add_field(name=f"Phrases ({idx})", value=chunk, inline=False) + chunk = "" + idx += 1 + chunk += line + "\n" + if chunk: + embed.add_field(name=f"Phrases ({idx})" if idx > 1 else "Phrases", value=chunk, inline=False) + embed.set_footer(text="Curate carefully — this is a cybersecurity channel; some terms are legitimate in context.") + return embed + + +class KeywordAddModal(discord.ui.Modal, title="Add / Edit Auto-Score Keyword"): + phrase = discord.ui.TextInput( + label="Keyword or phrase", + placeholder="e.g. cracked software, free nitro generator, warez", + max_length=200, + required=True, + ) + weight = discord.ui.TextInput( + label="Score weight (integer, e.g. 1-10)", + placeholder="Higher = closer to instant XCOM. Threshold is 10.", + max_length=4, + required=True, + default="1", + ) + + def __init__(self, parent_view: "KeywordManagerView"): + super().__init__() + self.parent_view = parent_view + + async def on_submit(self, interaction: discord.Interaction): + try: + weight_val = int(str(self.weight.value).strip()) + except ValueError: + await interaction.response.send_message("Weight must be a whole number.", ephemeral=True) + return + if weight_val < 1 or weight_val > 100: + await interaction.response.send_message("Weight must be between 1 and 100.", ephemeral=True) + return + entry = add_keyword(interaction.guild.id, str(self.phrase.value), weight_val, interaction.user.id) + if entry is None: + await interaction.response.send_message("That phrase was empty after normalization.", ephemeral=True) + return + await interaction.response.edit_message( + embed=_keyword_list_embed(interaction.guild), view=self.parent_view + ) + + +class KeywordRemoveModal(discord.ui.Modal, title="Remove Auto-Score Keyword"): + phrase = discord.ui.TextInput( + label="Keyword or phrase to remove", + placeholder="Type the phrase exactly as listed", + max_length=200, + required=True, + ) + + def __init__(self, parent_view: "KeywordManagerView"): + super().__init__() + self.parent_view = parent_view + + async def on_submit(self, interaction: discord.Interaction): + removed = remove_keyword(interaction.guild.id, str(self.phrase.value)) + if not removed: + await interaction.response.send_message("No matching keyword was found.", ephemeral=True) + return + await interaction.response.edit_message( + embed=_keyword_list_embed(interaction.guild), view=self.parent_view + ) + + +class KeywordManagerView(discord.ui.View): + """Interactive admin panel for curating auto-score keywords.""" + + def __init__(self, author_id: int): + super().__init__(timeout=180) + self.author_id = author_id + + async def interaction_check(self, interaction: discord.Interaction) -> bool: + # Only the invoking admin (with Manage Server) may operate the panel. + if interaction.user.id != self.author_id or not interaction.user.guild_permissions.manage_guild: + await interaction.response.send_message( + "Only the admin who opened this panel can use it.", ephemeral=True + ) + return False + return True + + @discord.ui.button(label="Add / Edit", style=discord.ButtonStyle.success, emoji="➕") + async def add_edit(self, interaction: discord.Interaction, button: discord.ui.Button): + await interaction.response.send_modal(KeywordAddModal(self)) + + @discord.ui.button(label="Remove", style=discord.ButtonStyle.danger, emoji="🗑️") + async def remove(self, interaction: discord.Interaction, button: discord.ui.Button): + await interaction.response.send_modal(KeywordRemoveModal(self)) + + @discord.ui.button(label="Refresh", style=discord.ButtonStyle.secondary, emoji="🔄") + async def refresh(self, interaction: discord.Interaction, button: discord.ui.Button): + await interaction.response.edit_message(embed=_keyword_list_embed(interaction.guild), view=self) + + class ModerationCog(commands.Cog): """XCOM commands and logic.""" def __init__(self, bot: commands.Bot): self.bot = bot + async def _perform_xcom(self, guild: discord.Guild, user: discord.Member, reason: str, applied_by: str): + """Shared XCOM logic used by /xcom and the community vote trigger. + + Returns (channel, already_isolated). Raises discord.Forbidden if the + Excommunicado role cannot be assigned (caller should handle it). + """ + role = await get_or_create_xcom_role(guild) + channel = await get_or_create_xcom_channel(guild, role) + + if role in user.roles: + return channel, True + + await user.add_roles(role, reason=f"XCOM ({applied_by}): {reason}") + + # Apply permanent XCOM Mark (nickname prefix) - best effort + try: + current_nick = user.nick or user.name + if not current_nick.startswith("[XCOM]"): + new_nick = f"[XCOM] {current_nick}"[:32] + await user.edit(nick=new_nick, reason="XCOM permanent mark applied") + except discord.Forbidden: + pass + + timestamp = datetime.now(timezone.utc).isoformat() + add_isolated_user(guild.id, user.id, reason, timestamp) + + welcome = discord.Embed( + title="🪐 XCOM Lounge", + description=( + f"**Reason for moderation:** {reason}\n\n" + "You are now in the shared XCOM discussion area with staff.\n" + "Your messages in other channels are being handled privately here.\n" + "This helps keep the main server clean while we resolve the issue.\n\n" + "Staff are present in this room to assist you." + ), + color=discord.Color.dark_red() + ) + welcome.set_footer(text=f"Isolated at {timestamp[:19]}") + + try: + await channel.send(f"{user.mention}", embed=welcome) + except Exception: + pass + + try: + await user.send(embed=welcome) + except discord.Forbidden: + pass + + return channel, False + @app_commands.command(name="xcom-setup", description="One-time setup: create XCOM role and xcom-lounge shared room") @app_commands.checks.has_permissions(manage_guild=True) async def xcom_setup(self, interaction: discord.Interaction): @@ -106,6 +293,7 @@ class ModerationCog(commands.Cog): @app_commands.command(name="xcom", description="Excommunicate a user into the shared moderated room (anti-violation)") @app_commands.checks.has_permissions(manage_roles=True) + @app_commands.checks.cooldown(5, 15.0, key=lambda i: (i.guild_id, i.user.id)) @app_commands.describe(user="The user who violated rules", reason="Reason for XCOM") async def xcom(self, interaction: discord.Interaction, user: discord.Member, reason: Optional[str] = "Flooding the channel"): """Excommunicate the user by assigning the Excommunicado role so they are contained in the xcom-lounge.""" @@ -115,58 +303,42 @@ class ModerationCog(commands.Cog): if user.id == interaction.user.id: await interaction.response.send_message("You cannot XCOM yourself.", ephemeral=True) return + if user.id == interaction.guild.owner_id: + await interaction.response.send_message("You cannot XCOM the server owner.", ephemeral=True) + return + + # Role-hierarchy safety: the invoker must outrank the target, + # and the bot must be able to manage the target. + invoker = interaction.user + if isinstance(invoker, discord.Member) and invoker.id != interaction.guild.owner_id: + if user.top_role >= invoker.top_role: + await interaction.response.send_message( + "You cannot XCOM a member with an equal or higher role than yours.", + ephemeral=True, + ) + return + me = interaction.guild.me + if me is not None and user.top_role >= me.top_role: + await interaction.response.send_message( + "I cannot XCOM this member — their highest role is above mine. Move my role higher.", + ephemeral=True, + ) + return + + # Defer the response because this command performs multiple async operations + await interaction.response.defer(ephemeral=True) guild = interaction.guild - role = await get_or_create_xcom_role(guild) - channel = await get_or_create_xcom_channel(guild, role) - - if role in user.roles: - await interaction.response.send_message(f"{user.mention} is already under XCOM.", ephemeral=True) + try: + channel, already = await self._perform_xcom(guild, user, reason, f"by {interaction.user}") + except discord.Forbidden: + await interaction.followup.send("Missing permissions to assign role. Check bot role hierarchy.", ephemeral=True) return - try: - await user.add_roles(role, reason=f"XCOM by {interaction.user}: {reason}") - except discord.Forbidden: - await interaction.response.send_message("Missing permissions to assign role. Check bot role hierarchy.", ephemeral=True) + if already: + await interaction.followup.send(f"{user.mention} is already under XCOM.", ephemeral=True) return - # Apply permanent XCOM Mark (nickname prefix) - survives release - try: - current_nick = user.nick or user.name - if not current_nick.startswith("[XCOM]"): - new_nick = f"[XCOM] {current_nick}"[:32] # Discord nick limit - await user.edit(nick=new_nick, reason="XCOM permanent mark applied") - except discord.Forbidden: - pass # Bot lacks permission to change nickname - - timestamp = datetime.now(timezone.utc).isoformat() - add_isolated_user(guild.id, user.id, reason, timestamp) - - # Welcome message in the isolation channel - welcome = discord.Embed( - title="🪐 XCOM Lounge", - description=( - f"**Reason for moderation:** {reason}\n\n" - "You are now in the shared XCOM discussion area with staff.\n" - "Your messages in other channels are being handled privately here.\n" - "This helps keep the main server clean while we resolve the issue.\n\n" - "Staff are present in this room to assist you." - ), - color=discord.Color.dark_red() - ) - welcome.set_footer(text=f"Isolated at {timestamp[:19]}") - - try: - await channel.send(f"{user.mention}", embed=welcome) - except Exception: - pass - - # DM the user - try: - await user.send(embed=welcome) - except discord.Forbidden: - pass # DMs disabled - # Log action log_embed = discord.Embed( title="🪐 User XCOM'd", @@ -179,7 +351,7 @@ class ModerationCog(commands.Cog): if hasattr(self.bot, "log_action"): await self.bot.log_action(guild, log_embed) - await interaction.response.send_message( + await interaction.followup.send( f"✅ {user.mention} has been XCOM'd into the shared room {channel.mention}.\nThey continue to see the server normally.", ephemeral=True ) @@ -189,16 +361,21 @@ class ModerationCog(commands.Cog): @app_commands.describe(user="User to release from XCOM") async def xcom_release(self, interaction: discord.Interaction, user: discord.Member): """Remove the Excommunicado role. The [XCOM] nickname mark stays until an admin runs /xcom-unmark.""" + await interaction.response.defer(ephemeral=True) guild = interaction.guild data = get_guild_data(guild.id) role_id = data.get("xcom_role_id") role = guild.get_role(role_id) if role_id else None if not role or role not in user.roles: - await interaction.response.send_message(f"{user.mention} is not currently under XCOM.", ephemeral=True) + await interaction.followup.send(f"{user.mention} is not currently under XCOM.", ephemeral=True) return - await user.remove_roles(role, reason=f"Released from XCOM by {interaction.user}") + try: + await user.remove_roles(role, reason=f"Released from XCOM by {interaction.user}") + except discord.Forbidden: + await interaction.followup.send("Missing permissions to remove role. Check bot role hierarchy.", ephemeral=True) + return remove_isolated_user(guild.id, user.id) @@ -209,10 +386,10 @@ class ModerationCog(commands.Cog): ) try: await user.send(embed=release_embed) - except: + except discord.HTTPException: pass - await interaction.response.send_message( + await interaction.followup.send( f"✅ {user.mention} has been released from XCOM and restored to normal access. The [XCOM] mark remains until an admin removes it.", ephemeral=True ) @@ -288,6 +465,7 @@ class ModerationCog(commands.Cog): @app_commands.command(name="xcom-absolve", description="Check absolution points and release a user early (admin only)") @app_commands.checks.has_permissions(manage_roles=True) + @app_commands.checks.cooldown(5, 15.0, key=lambda i: (i.guild_id, i.user.id)) @app_commands.describe(user="XCOM user to evaluate for absolution") async def xcom_absolve(self, interaction: discord.Interaction, user: discord.Member): """If the user has enough absolution points, remove role and optionally the mark.""" @@ -300,12 +478,15 @@ class ModerationCog(commands.Cog): await interaction.response.send_message(f"{user.mention} is not under XCOM.", ephemeral=True) return + # Defer because this command performs async operations + await interaction.response.defer(ephemeral=True) + isolated = get_isolated_users(guild.id) user_data = isolated.get(str(user.id), {}) points = user_data.get("absolution_points", 0) if points < 10: # Threshold for absolution - await interaction.response.send_message( + await interaction.followup.send( f"{user.mention} has {points} absolution points (need 10). Not ready for absolution yet.", ephemeral=True ) @@ -317,7 +498,7 @@ class ModerationCog(commands.Cog): # Optional: remove from isolated list but keep the mark flag remove_isolated_user(guild.id, user.id) - await interaction.response.send_message( + await interaction.followup.send( f"✅ {user.mention} has been absolved ({points} points). The [XCOM] mark remains until manually removed.", ephemeral=True ) @@ -348,6 +529,143 @@ class ModerationCog(commands.Cog): embed.set_footer(text=f"Total: {len(isolated)} | Use /xcom-release to restore access") await interaction.response.send_message(embed=embed, ephemeral=True) + @app_commands.command(name="xcom-vote", description="Cast a community vote to excommunicate (up) or defend (down) a user") + @app_commands.checks.cooldown(10, 60.0, key=lambda i: (i.guild_id, i.user.id)) + @app_commands.describe(user="The user to vote on", direction="up = excommunicate, down = defend") + @app_commands.choices(direction=[ + app_commands.Choice(name="up (excommunicate)", value="up"), + app_commands.Choice(name="down (defend)", value="down"), + ]) + async def xcom_vote(self, interaction: discord.Interaction, user: discord.Member, direction: app_commands.Choice[str]): + """Community-driven XCOM. Net score >= threshold (with quorum) auto-isolates the target.""" + guild = interaction.guild + voter = interaction.user + + # --- Guardrails --- + if user.bot: + await interaction.response.send_message("You cannot vote on bots.", ephemeral=True) + return + if user.id == voter.id: + await interaction.response.send_message("You cannot vote on yourself.", ephemeral=True) + return + if user.id == guild.owner_id: + await interaction.response.send_message("You cannot vote against the server owner.", ephemeral=True) + return + # Staff / moderator immunity + if user.guild_permissions.manage_roles or user.guild_permissions.administrator: + await interaction.response.send_message("You cannot vote against staff or moderators.", ephemeral=True) + return + + data = get_guild_data(guild.id) + role_id = data.get("xcom_role_id") + role = guild.get_role(role_id) if role_id else None + if role and role in user.roles: + await interaction.response.send_message(f"{user.mention} is already under XCOM.", ephemeral=True) + return + + # The bot must be able to enforce the result + me = guild.me + if me is not None and user.top_role >= me.top_role: + await interaction.response.send_message( + "That member ranks above me, so a community XCOM could not be enforced.", ephemeral=True + ) + return + + await interaction.response.defer(ephemeral=True) + + is_staff = voter.guild_permissions.administrator or voter.guild_permissions.manage_roles + weight = VOTE_ADMIN_WEIGHT if is_staff else VOTE_USER_WEIGHT + dir_val = 1 if direction.value == "up" else -1 + + result = cast_vote( + guild.id, user.id, voter.id, dir_val, weight, + cooldown_hours=VOTE_COOLDOWN_HOURS, reset_days=VOTE_RESET_DAYS, + ) + + if not result["ok"]: + if result["reason"] == "cooldown": + await interaction.followup.send( + f"⏳ You've already voted on this user. You can vote again in ~{result['retry_hours']:.1f}h " + "(one vote per user per 24h).", + ephemeral=True, + ) + else: + await interaction.followup.send("Your vote could not be recorded.", ephemeral=True) + return + + score = result["score"] + unique_voters = result["unique_voters"] + + # --- Threshold reached: trigger the community XCOM --- + if score >= VOTE_XCOM_THRESHOLD and unique_voters >= VOTE_MIN_UNIQUE_VOTERS: + try: + channel, already = await self._perform_xcom( + guild, user, "Community vote threshold reached", "community vote" + ) + except discord.Forbidden: + await interaction.followup.send( + "Your vote was recorded, but I lack permission to XCOM this user.", ephemeral=True + ) + return + + clear_votes(guild.id, user.id) + + log_embed = discord.Embed( + title="🪐 Community XCOM Triggered", + description=f"{user.mention} ({user.id}) reached the community vote threshold and was placed in {channel.mention}.", + color=discord.Color.dark_red(), + timestamp=datetime.now(timezone.utc), + ) + log_embed.add_field(name="Final score", value=str(score), inline=True) + log_embed.add_field(name="Distinct voters", value=str(unique_voters), inline=True) + if hasattr(self.bot, "log_action"): + await self.bot.log_action(guild, log_embed) + + await interaction.followup.send( + f"✅ The community has spoken — the score reached **{score}** and the user has been excommunicated.", + ephemeral=True, + ) + return + + # --- Not yet --- + arrow = "⬆️" if dir_val > 0 else "⬇️" + remaining = VOTE_XCOM_THRESHOLD - score + progress = f" — {remaining} more to trigger XCOM" if remaining > 0 else "" + quorum_note = ( + "" if unique_voters >= VOTE_MIN_UNIQUE_VOTERS + else f"\n_Needs ≥{VOTE_MIN_UNIQUE_VOTERS} distinct voters (currently {unique_voters})._" + ) + await interaction.followup.send( + f"{arrow} Vote recorded. Current score: **{score}/{VOTE_XCOM_THRESHOLD}**{progress}." + f"{quorum_note}\n_Tallies reset weekly. One vote per user per 24h._", + ephemeral=True, + ) + + @app_commands.command(name="xcom-votes", description="View the current community vote tally for a user (staff)") + @app_commands.checks.has_permissions(manage_roles=True) + @app_commands.describe(user="User to inspect") + async def xcom_votes(self, interaction: discord.Interaction, user: discord.Member): + """Show the running community vote score and quorum status for a user.""" + score, unique_voters = get_vote_score(interaction.guild.id, user.id, reset_days=VOTE_RESET_DAYS) + kw_points = get_keyword_points(interaction.guild.id, user.id, reset_days=VOTE_RESET_DAYS) + embed = discord.Embed(title="🗳️ Community Vote Standing", color=discord.Color.orange()) + embed.add_field(name="Target", value=user.mention, inline=True) + embed.add_field(name="Score", value=f"{score}/{VOTE_XCOM_THRESHOLD}", inline=True) + embed.add_field(name="Distinct voters", value=f"{unique_voters}/{VOTE_MIN_UNIQUE_VOTERS}", inline=True) + embed.add_field(name="From human votes", value=str(score - kw_points), inline=True) + embed.add_field(name="From keyword flags", value=str(kw_points), inline=True) + embed.set_footer(text=f"Threshold {VOTE_XCOM_THRESHOLD} • quorum {VOTE_MIN_UNIQUE_VOTERS} • resets weekly") + await interaction.response.send_message(embed=embed, ephemeral=True) + + @app_commands.command(name="xcom-keywords", description="Manage auto-score keywords/phrases (admin panel)") + @app_commands.checks.has_permissions(manage_guild=True) + async def xcom_keywords(self, interaction: discord.Interaction): + """Open the interactive panel to add, edit, or remove auto-score keywords.""" + view = KeywordManagerView(interaction.user.id) + await interaction.response.send_message( + embed=_keyword_list_embed(interaction.guild), view=view, ephemeral=True + ) + @xcom.error @xcom_release.error @xcom_unmark.error @@ -356,14 +674,35 @@ class ModerationCog(commands.Cog): @xcom_absolve.error @xcom_setup.error @xcom_list.error + @xcom_vote.error + @xcom_votes.error + @xcom_keywords.error async def permission_error(self, interaction: discord.Interaction, error: app_commands.AppCommandError): - if isinstance(error, app_commands.MissingPermissions): - await interaction.response.send_message( - "You need the 'Manage Roles' (or Manage Server for setup) permission to use this command.", - ephemeral=True - ) - else: - await interaction.response.send_message(f"Error: {error}", ephemeral=True) + try: + if isinstance(error, app_commands.MissingPermissions): + if not interaction.response.is_done(): + await interaction.response.send_message( + "You need the 'Manage Roles' (or Manage Server for setup) permission to use this command.", + ephemeral=True + ) + else: + await interaction.followup.send( + "You need the 'Manage Roles' (or Manage Server for setup) permission to use this command.", + ephemeral=True + ) + elif isinstance(error, app_commands.CommandOnCooldown): + msg = f"⏳ Slow down — try again in {error.retry_after:.1f}s." + if not interaction.response.is_done(): + await interaction.response.send_message(msg, ephemeral=True) + else: + await interaction.followup.send(msg, ephemeral=True) + else: + if not interaction.response.is_done(): + await interaction.response.send_message(f"Error: {error}", ephemeral=True) + else: + await interaction.followup.send(f"Error: {error}", ephemeral=True) + except discord.errors.InteractionResponded: + pass # Interaction was already responded to async def setup(bot: commands.Bot): diff --git a/docker-compose.yml b/docker-compose.yml index 7af471f..e351563 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,5 +1,3 @@ -version: "3.9" - services: excommunicado: build: . diff --git a/utils/config.py b/utils/config.py index f8a9051..0ae808b 100644 --- a/utils/config.py +++ b/utils/config.py @@ -1,31 +1,70 @@ """Configuration and persistence for the Excommunicado bot.""" import json +import os +import tempfile +import threading +from datetime import datetime, timezone, timedelta from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Tuple DATA_DIR = Path(__file__).parent.parent / "data" CONFIG_FILE = DATA_DIR / "xcom_config.json" +# In-memory cache so we don't re-read the file on every message. +# Protected by a lock because discord.py callbacks can interleave. +_lock = threading.RLock() +_cache: Optional[Dict[str, Any]] = None + def _ensure_data_dir() -> None: DATA_DIR.mkdir(parents=True, exist_ok=True) def load_config() -> Dict[str, Any]: - """Load the entire config dict.""" - _ensure_data_dir() - if not CONFIG_FILE.exists(): - save_config({}) - return {} - with open(CONFIG_FILE, "r", encoding="utf-8") as f: - return json.load(f) + """Load the entire config dict (cached after first read).""" + global _cache + with _lock: + if _cache is not None: + return _cache + _ensure_data_dir() + if not CONFIG_FILE.exists(): + _cache = {} + save_config(_cache) + return _cache + try: + with open(CONFIG_FILE, "r", encoding="utf-8") as f: + _cache = json.load(f) + except (json.JSONDecodeError, OSError) as err: + # Corrupt or unreadable file: back it up and start fresh + print(f"[CONFIG] Failed to load config ({err}); starting empty.") + try: + if CONFIG_FILE.exists(): + CONFIG_FILE.replace(CONFIG_FILE.with_suffix(".json.corrupt")) + except OSError: + pass + _cache = {} + save_config(_cache) + return _cache def save_config(data: Dict[str, Any]) -> None: - """Save the config dict.""" - _ensure_data_dir() - with open(CONFIG_FILE, "w", encoding="utf-8") as f: - json.dump(data, f, indent=2, ensure_ascii=False) + """Atomically persist the config dict and refresh the cache.""" + global _cache + with _lock: + _ensure_data_dir() + # Write to a temp file in the same dir, then atomically replace. + fd, tmp_path = tempfile.mkstemp(dir=str(DATA_DIR), suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_path, CONFIG_FILE) + except BaseException: + if os.path.exists(tmp_path): + os.remove(tmp_path) + raise + _cache = data def get_guild_data(guild_id: int) -> Dict[str, Any]: @@ -100,3 +139,229 @@ def get_isolated_users(guild_id: int) -> Dict[str, Dict]: """Return all isolated users for guild.""" data = get_guild_data(guild_id) return data.get("isolated_users", {}) + + +# --------------------------------------------------------------------------- +# Community vote system +# --------------------------------------------------------------------------- +# Stored under config[gid]["votes"][target_id] = { +# "week_start": iso, +# "voters": { voter_id: {"direction": 1|-1, "weight": int, "timestamp": iso} }, +# "keyword_points": int # auto-accrued from admin-curated keyword matches +# } +# Admin-curated keyword list lives under config[gid]["keywords"][phrase] = { +# "weight": int, "added_by": int, "added_at": iso, "display": str +# } + + +def _parse_iso(value: Optional[str]) -> Optional[datetime]: + if not value: + return None + try: + dt = datetime.fromisoformat(value) + except (ValueError, TypeError): + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def _vote_score(voters: Dict[str, Dict]) -> int: + return sum(int(v.get("direction", 0)) * int(v.get("weight", 0)) for v in voters.values()) + + +def _record_score(rec: Dict[str, Any]) -> int: + """Combined net score: human votes + auto-accrued keyword points.""" + return _vote_score(rec.get("voters", {})) + int(rec.get("keyword_points", 0)) + + +def cast_vote( + guild_id: int, + target_id: int, + voter_id: int, + direction: int, + weight: int, + cooldown_hours: int = 24, + reset_days: int = 7, +) -> Dict[str, Any]: + """Record a community vote atomically. + + direction: +1 (up / excommunicate) or -1 (down / defend). + Returns: {ok, reason, score, unique_voters, retry_hours}. + A voter may only vote on a given target once per ``cooldown_hours``; the + whole tally for a target resets after ``reset_days``. + """ + direction = 1 if direction >= 0 else -1 + with _lock: + config = load_config() + gid, tid, vid = str(guild_id), str(target_id), str(voter_id) + guild = config.setdefault( + gid, {"xcom_role_id": None, "xcom_channel_id": None, "isolated_users": {}} + ) + votes = guild.setdefault("votes", {}) + now = datetime.now(timezone.utc) + + rec = votes.get(tid) + # Weekly reset + if rec is not None: + week_start = _parse_iso(rec.get("week_start")) + if week_start is None or now - week_start >= timedelta(days=reset_days): + rec = None + if rec is None: + rec = {"week_start": now.isoformat(), "voters": {}, "keyword_points": 0} + votes[tid] = rec + + voters = rec["voters"] + prev = voters.get(vid) + if prev: + last = _parse_iso(prev.get("timestamp")) + if last is not None and now - last < timedelta(hours=cooldown_hours): + retry_hours = cooldown_hours - (now - last).total_seconds() / 3600.0 + return { + "ok": False, + "reason": "cooldown", + "score": _record_score(rec), + "unique_voters": len(voters), + "retry_hours": max(retry_hours, 0.0), + } + + voters[vid] = { + "direction": direction, + "weight": int(weight), + "timestamp": now.isoformat(), + } + save_config(config) + return { + "ok": True, + "reason": None, + "score": _record_score(rec), + "unique_voters": len(voters), + "retry_hours": None, + } + + +def get_vote_score(guild_id: int, target_id: int, reset_days: int = 7) -> Tuple[int, int]: + """Return (net_score, unique_voters) for a target, applying weekly reset.""" + with _lock: + config = load_config() + gid, tid = str(guild_id), str(target_id) + rec = config.get(gid, {}).get("votes", {}).get(tid) + if not rec: + return 0, 0 + now = datetime.now(timezone.utc) + week_start = _parse_iso(rec.get("week_start")) + if week_start is None or now - week_start >= timedelta(days=reset_days): + config[gid]["votes"].pop(tid, None) + save_config(config) + return 0, 0 + return _record_score(rec), len(rec.get("voters", {})) + + +def get_keyword_points(guild_id: int, target_id: int, reset_days: int = 7) -> int: + """Return the auto-accrued keyword points for a target (with weekly reset).""" + with _lock: + config = load_config() + gid, tid = str(guild_id), str(target_id) + rec = config.get(gid, {}).get("votes", {}).get(tid) + if not rec: + return 0 + week_start = _parse_iso(rec.get("week_start")) + if week_start is None or datetime.now(timezone.utc) - week_start >= timedelta(days=reset_days): + return 0 + return int(rec.get("keyword_points", 0)) + + +def add_keyword_points( + guild_id: int, target_id: int, amount: int, reset_days: int = 7 +) -> Dict[str, Any]: + """Add auto-accrued keyword points to a target's tally (atomic). + + Returns {score, unique_voters, keyword_points}. Applies the weekly reset. + """ + with _lock: + config = load_config() + gid, tid = str(guild_id), str(target_id) + guild = config.setdefault( + gid, {"xcom_role_id": None, "xcom_channel_id": None, "isolated_users": {}} + ) + votes = guild.setdefault("votes", {}) + now = datetime.now(timezone.utc) + + rec = votes.get(tid) + if rec is not None: + week_start = _parse_iso(rec.get("week_start")) + if week_start is None or now - week_start >= timedelta(days=reset_days): + rec = None + if rec is None: + rec = {"week_start": now.isoformat(), "voters": {}, "keyword_points": 0} + votes[tid] = rec + + rec["keyword_points"] = int(rec.get("keyword_points", 0)) + int(amount) + save_config(config) + return { + "score": _record_score(rec), + "unique_voters": len(rec.get("voters", {})), + "keyword_points": rec["keyword_points"], + } + + +def clear_votes(guild_id: int, target_id: int) -> None: + """Remove the vote tally for a target (e.g. after an XCOM is triggered).""" + with _lock: + config = load_config() + gid, tid = str(guild_id), str(target_id) + if config.get(gid, {}).get("votes", {}).pop(tid, None) is not None: + save_config(config) + + +# --------------------------------------------------------------------------- +# Admin-curated keyword / key-phrase list +# --------------------------------------------------------------------------- + + +def _normalize_phrase(phrase: str) -> str: + return " ".join(phrase.lower().split()) + + +def add_keyword(guild_id: int, phrase: str, weight: int, added_by: int) -> Optional[Dict[str, Any]]: + """Add or update an auto-score keyword/phrase. Returns the stored entry or None if invalid.""" + key = _normalize_phrase(phrase) + if not key: + return None + with _lock: + config = load_config() + gid = str(guild_id) + guild = config.setdefault( + gid, {"xcom_role_id": None, "xcom_channel_id": None, "isolated_users": {}} + ) + keywords = guild.setdefault("keywords", {}) + entry = { + "weight": int(weight), + "added_by": int(added_by), + "added_at": datetime.now(timezone.utc).isoformat(), + "display": phrase.strip(), + } + keywords[key] = entry + save_config(config) + return entry + + +def remove_keyword(guild_id: int, phrase: str) -> bool: + """Remove a keyword/phrase. Returns True if something was removed.""" + key = _normalize_phrase(phrase) + with _lock: + config = load_config() + gid = str(guild_id) + keywords = config.get(gid, {}).get("keywords", {}) + if key in keywords: + keywords.pop(key, None) + save_config(config) + return True + return False + + +def get_keywords(guild_id: int) -> Dict[str, Dict]: + """Return the guild's keyword map: {normalized_phrase: {weight, display, ...}}.""" + config = load_config() + return config.get(str(guild_id), {}).get("keywords", {}) +