SubINaclS/Excommunicado
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added community voting and fixed bugs

Browse Source
This commit is contained in:
Subinacls
2026-06-09 06:50:38 -04:00
parent 9ca7d35ce0
commit a789fb131f
7 changed files with 997 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+145 -2
View File
@@ -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
Dockerfile
-5
View File
@@ -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 .
README.md
+109 -12
View File
@@ -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 <text>` – 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 <user>`
@@ -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 <user> <up|down>`
- **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 <user>`
- **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.
bot.py
+67 -4
View File
@@ -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:
cogs/moderation.py
+399 -60
View File
@@ -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):
docker-compose.yml
-2
View File
@@ -1,5 +1,3 @@
version: "3.9"
services:
excommunicado:
build: .
utils/config.py
+277 -12
View File
@@ -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", {})