379 lines
19 KiB
Python
379 lines
19 KiB
Python
"""
|
|
Moderation commands — ban, unban, mute, unmute, purge, mod edit/delete, warnings.
|
|
|
|
Updated purge command: treat `count` as a TOTAL across the network by default;
|
|
add optional per-channel mode, dry-run simulation, per-channel breakdown in the
|
|
mod log, and improved audit details. Uses per-network `mod_webhook` if set via
|
|
network config, otherwise falls back to existing `log_channel` behaviour.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
import discord
|
|
from redbot.core import commands
|
|
|
|
from ..models.permissions import Role, has_role, requires_role, role_name, get_role
|
|
from ..utils import ok_embed, err_embed, info_embed, warn_embed, truncate, COLOUR_INFO
|
|
from ._base import WormholeBase
|
|
|
|
|
|
class ModerationCommands(WormholeBase):
|
|
"""Mixin — moderation tools (per-network, role-gated)."""
|
|
|
|
# ── User bans ──────────────────────────────────────────────────────────
|
|
|
|
@WormholeBase.wh_mod.command(name="ban")
|
|
@requires_role(Role.MODERATOR)
|
|
async def wh_mod_ban(self, ctx: commands.Context, name: str, user: discord.User, *, reason: str = "No reason") -> None:
|
|
"""Ban a user from the network."""
|
|
nd = await self._net(name)
|
|
if user.id == nd["owner_id"]:
|
|
return await ctx.send(embed=err_embed("You can't ban the network owner."))
|
|
target_role = get_role(nd, user.id)
|
|
actor_role = get_role(nd, ctx.author.id)
|
|
if target_role >= actor_role and not await self.bot.is_owner(ctx.author):
|
|
return await ctx.send(embed=err_embed(f"You can't ban a {role_name(target_role)} — your role: {role_name(actor_role)}"))
|
|
async with self.config.networks() as ns:
|
|
bans = ns[name].setdefault("banned_users", [])
|
|
if user.id not in bans:
|
|
bans.append(user.id)
|
|
await ctx.send(embed=ok_embed(f"{user.mention} banned from `{name}`.\nReason: {reason}"))
|
|
await self._audit(name, "ban", str(ctx.author), str(user), reason)
|
|
await self._log(nd, warn_embed(f"🔨 **{user}** banned by {ctx.author}\nReason: {reason}"))
|
|
|
|
@WormholeBase.wh_mod.command(name="unban")
|
|
@requires_role(Role.MODERATOR)
|
|
async def wh_mod_unban(self, ctx: commands.Context, name: str, user: discord.User) -> None:
|
|
"""Unban a user from the network."""
|
|
nd = await self._net(name)
|
|
async with self.config.networks() as ns:
|
|
bans = ns[name].get("banned_users", [])
|
|
if user.id in bans:
|
|
bans.remove(user.id)
|
|
await ctx.send(embed=ok_embed(f"{user.mention} unbanned from `{name}`."))
|
|
await self._audit(name, "unban", str(ctx.author), str(user))
|
|
await self._log(nd, info_embed(f"✅ **{user}** unbanned by {ctx.author}", title="⚙️ Mod Action"))
|
|
|
|
# ── User mutes ─────────────────────────────────────────────────────────
|
|
|
|
@WormholeBase.wh_mod.command(name="mute")
|
|
@requires_role(Role.MODERATOR)
|
|
async def wh_mod_mute(self, ctx: commands.Context, name: str, user: discord.User, *, reason: str = "No reason") -> None:
|
|
"""Mute a user in the network (messages won't relay)."""
|
|
nd = await self._net(name)
|
|
target_role = get_role(nd, user.id)
|
|
actor_role = get_role(nd, ctx.author.id)
|
|
if target_role >= actor_role and not await self.bot.is_owner(ctx.author):
|
|
return await ctx.send(embed=err_embed(f"You can't mute a {role_name(target_role)}."))
|
|
async with self.config.networks() as ns:
|
|
mutes = ns[name].setdefault("muted_users", [])
|
|
if user.id not in mutes:
|
|
mutes.append(user.id)
|
|
await ctx.send(embed=ok_embed(f"{user.mention} muted in `{name}`.\nReason: {reason}"))
|
|
await self._audit(name, "mute", str(ctx.author), str(user), reason)
|
|
await self._log(nd, warn_embed(f"🔇 **{user}** muted by {ctx.author}\nReason: {reason}"))
|
|
|
|
@WormholeBase.wh_mod.command(name="unmute")
|
|
@requires_role(Role.MODERATOR)
|
|
async def wh_mod_unmute(self, ctx: commands.Context, name: str, user: discord.User) -> None:
|
|
"""Unmute a user in the network."""
|
|
nd = await self._net(name)
|
|
async with self.config.networks() as ns:
|
|
mutes = ns[name].get("muted_users", [])
|
|
if user.id in mutes:
|
|
mutes.remove(user.id)
|
|
await ctx.send(embed=ok_embed(f"{user.mention} unmuted in `{name}`."))
|
|
await self._audit(name, "unmute", str(ctx.author), str(user))
|
|
await self._log(nd, info_embed(f"✅ **{user}** unmuted by {ctx.author}", title="⚙️ Mod Action"))
|
|
|
|
# ── Server bans/mutes ──────────────────────────────────────────────────
|
|
|
|
@WormholeBase.wh_mod.command(name="banserver")
|
|
@requires_role(Role.ADMIN)
|
|
async def wh_mod_banserver(self, ctx: commands.Context, name: str, guild_id: int) -> None:
|
|
"""Ban a server from the network."""
|
|
nd = await self._net(name)
|
|
async with self.config.networks() as ns:
|
|
bans = ns[name].setdefault("banned_servers", [])
|
|
if guild_id not in bans:
|
|
bans.append(guild_id)
|
|
await ctx.send(embed=ok_embed(f"Server `{guild_id}` banned from `{name}`."))
|
|
await self._audit(name, "banserver", str(ctx.author), str(guild_id))
|
|
await self._log(nd, warn_embed(f"🔨 Server `{guild_id}` banned by {ctx.author}"))
|
|
|
|
@WormholeBase.wh_mod.command(name="unbanserver")
|
|
@requires_role(Role.ADMIN)
|
|
async def wh_mod_unbanserver(self, ctx: commands.Context, name: str, guild_id: int) -> None:
|
|
"""Unban a server from the network."""
|
|
nd = await self._net(name)
|
|
async with self.config.networks() as ns:
|
|
bans = ns[name].get("banned_servers", [])
|
|
if guild_id in bans:
|
|
bans.remove(guild_id)
|
|
await ctx.send(embed=ok_embed(f"Server `{guild_id}` unbanned from `{name}`."))
|
|
await self._audit(name, "unbanserver", str(ctx.author), str(guild_id))
|
|
await self._log(nd, info_embed(f"✅ Server `{guild_id}` unbanned by {ctx.author}", title="⚙️ Mod Action"))
|
|
|
|
@WormholeBase.wh_mod.command(name="muteserver")
|
|
@requires_role(Role.ADMIN)
|
|
async def wh_mod_muteserver(self, ctx: commands.Context, name: str, guild_id: int) -> None:
|
|
"""Mute a server in the network."""
|
|
nd = await self._net(name)
|
|
async with self.config.networks() as ns:
|
|
mutes = ns[name].setdefault("muted_servers", [])
|
|
if guild_id not in mutes:
|
|
mutes.append(guild_id)
|
|
await ctx.send(embed=ok_embed(f"Server `{guild_id}` muted in `{name}`."))
|
|
await self._audit(name, "muteserver", str(ctx.author), str(guild_id))
|
|
await self._log(nd, warn_embed(f"🔇 Server `{guild_id}` muted by {ctx.author}"))
|
|
|
|
@WormholeBase.wh_mod.command(name="unmuteserver")
|
|
@requires_role(Role.ADMIN)
|
|
async def wh_mod_unmuteserver(self, ctx: commands.Context, name: str, guild_id: int) -> None:
|
|
"""Unmute a server in the network."""
|
|
nd = await self._net(name)
|
|
async with self.config.networks() as ns:
|
|
mutes = ns[name].get("muted_servers", [])
|
|
if guild_id in mutes:
|
|
mutes.remove(guild_id)
|
|
await ctx.send(embed=ok_embed(f"Server `{guild_id}` unmuted in `{name}`."))
|
|
await self._audit(name, "unmuteserver", str(ctx.author), str(guild_id))
|
|
await self._log(nd, info_embed(f"✅ Server `{guild_id}` unmuted by {ctx.author}", title="⚙️ Mod Action"))
|
|
|
|
@WormholeBase.wh_mod.command(name="allowserver")
|
|
@requires_role(Role.ADMIN)
|
|
async def wh_mod_allowserver(self, ctx: commands.Context, name: str, guild_id: int) -> None:
|
|
"""Add a server to the allowlist."""
|
|
nd = await self._net(name)
|
|
async with self.config.networks() as ns:
|
|
allow = ns[name].setdefault("allowlist_servers", [])
|
|
if guild_id not in allow:
|
|
allow.append(guild_id)
|
|
await ctx.send(embed=ok_embed(f"Server `{guild_id}` added to allowlist for `{name}`."))
|
|
await self._audit(name, "allowserver", str(ctx.author), str(guild_id))
|
|
await self._log(nd, info_embed(f"Server `{guild_id}` added to allowlist by {ctx.author}", title="⚙️ Mod Action"))
|
|
|
|
@WormholeBase.wh_mod.command(name="removeallow")
|
|
@requires_role(Role.ADMIN)
|
|
async def wh_mod_removeallow(self, ctx: commands.Context, name: str, guild_id: int) -> None:
|
|
"""Remove a server from the allowlist."""
|
|
nd = await self._net(name)
|
|
async with self.config.networks() as ns:
|
|
allow = ns[name].get("allowlist_servers", [])
|
|
if guild_id in allow:
|
|
allow.remove(guild_id)
|
|
await ctx.send(embed=ok_embed(f"Server `{guild_id}` removed from allowlist for `{name}`."))
|
|
await self._audit(name, "removeallow", str(ctx.author), str(guild_id))
|
|
await self._log(nd, info_embed(f"Server `{guild_id}` removed from allowlist by {ctx.author}", title="⚙️ Mod Action"))
|
|
|
|
# ── Purge ──────────────────────────────────────────────────────────
|
|
|
|
@WormholeBase.wh_mod.command(name="purge")
|
|
@requires_role(Role.MODERATOR)
|
|
async def wh_mod_purge(self, ctx: commands.Context, name: str, count: int = 10, per_channel: bool = False, dry_run: bool = False) -> None:
|
|
"""Purge recent relayed messages across all channels.
|
|
|
|
By default `count` is treated as a TOTAL across the network. Set
|
|
`per_channel=True` to retain legacy per-channel behaviour.
|
|
Use `dry_run=True` to simulate without deleting.
|
|
"""
|
|
nd = await self._net(name)
|
|
if not nd:
|
|
return await ctx.send(embed=err_embed(f"Network `{name}` not found."))
|
|
requested = max(0, min(count, 50))
|
|
|
|
total_deleted = 0
|
|
per_channel_report = [] # list of (channel_id, attempted, deleted, failed)
|
|
|
|
# Iterate channels; if not per_channel, stop when total_deleted >= requested
|
|
for ch_id in nd.get("channels", []):
|
|
if not per_channel and total_deleted >= requested:
|
|
break
|
|
ch = self.bot.get_channel(ch_id)
|
|
if not ch:
|
|
continue
|
|
attempted = 0
|
|
deleted_here = 0
|
|
failed = 0
|
|
# Limit per-channel iteration to `requested` to avoid heavy scans
|
|
try:
|
|
async for msg in ch.history(limit=requested):
|
|
if msg.author.id == self.bot.user.id or (msg.webhook_id is not None):
|
|
attempted += 1
|
|
if dry_run:
|
|
# simulate - don't delete
|
|
continue
|
|
try:
|
|
await msg.delete()
|
|
deleted_here += 1
|
|
total_deleted += 1
|
|
except Exception:
|
|
failed += 1
|
|
# If honoring total cap, break when reached
|
|
if not per_channel and total_deleted >= requested:
|
|
break
|
|
except Exception:
|
|
# log and continue
|
|
try:
|
|
log = getattr(self, "log", None)
|
|
if log:
|
|
log.exception("Error scanning channel %s during purge", ch_id)
|
|
except Exception:
|
|
pass
|
|
per_channel_report.append((ch_id, attempted, deleted_here, failed))
|
|
|
|
# Build a concise human-friendly report
|
|
lines = []
|
|
for ch_id, attempted, deleted_here, failed in per_channel_report:
|
|
ch = self.bot.get_channel(ch_id)
|
|
name_str = f"#{ch.name}" if ch else f"<unknown {ch_id}>"
|
|
lines.append(f"• {name_str}: attempted {attempted}, deleted {deleted_here}, failed {failed}")
|
|
|
|
summary = f"Purged {total_deleted} relayed messages across `{name}` (requested {requested})" if not dry_run else f"Dry-run: {sum(p[1] for p in per_channel_report)} relayed messages would be affected across `{name}` (requested {requested})"
|
|
await ctx.send(embed=ok_embed(summary))
|
|
|
|
# Audit and network log with details
|
|
details = f"requested={requested} total_deleted={total_deleted} per_channel={len(per_channel_report)}"
|
|
await self._audit(name, "purge", str(ctx.author), details=details)
|
|
|
|
# Use the existing _log helper to send a network-visible message; include breakdown
|
|
breakdown = "\n".join(lines)[:1900] or "No relayed messages found."
|
|
title = f"🧹 Purge{' (dry-run)' if dry_run else ''}: {total_deleted}/{requested}"
|
|
await self._log(nd, warn_embed(f"{title} by {ctx.author}\n\n{breakdown}"))
|
|
|
|
# ── Cross-network mod edit/delete ──────────────────────────────────────
|
|
|
|
@WormholeBase.wh_mod.command(name="edit")
|
|
@requires_role(Role.MODERATOR)
|
|
async def wh_mod_edit(self, ctx: commands.Context, name: str, message_id: int, *, new_content: str) -> None:
|
|
"""Edit a relayed message across all channels."""
|
|
nd = await self._net(name)
|
|
if not nd:
|
|
return await ctx.send(embed=err_embed(f"Network `{name}` not found."))
|
|
orig_id = self.msg_map.get_original(name, message_id) or message_id
|
|
relayed = self.msg_map.get_relayed(name, orig_id)
|
|
edited = 0
|
|
for ch_id, msg_id in relayed.items():
|
|
ch = self.bot.get_channel(ch_id)
|
|
if not ch:
|
|
continue
|
|
try:
|
|
wh = await self._wh(ch)
|
|
await wh.edit_message(msg_id, content=new_content)
|
|
edited += 1
|
|
except Exception:
|
|
try:
|
|
msg = await ch.fetch_message(msg_id)
|
|
if msg.author.id == self.bot.user.id:
|
|
await msg.edit(content=new_content)
|
|
edited += 1
|
|
except Exception:
|
|
pass
|
|
await ctx.send(embed=ok_embed(f"Edited {edited} relayed copies."))
|
|
await self._audit(name, "mod_edit", str(ctx.author), str(orig_id), truncate(new_content, 100))
|
|
await self._log(nd, warn_embed(
|
|
f"✏️ Message `{orig_id}` edited by {ctx.author} ({edited} copies)\n"
|
|
f"New content: {truncate(new_content, 200)}"
|
|
))
|
|
|
|
@WormholeBase.wh_mod.command(name="nuke")
|
|
@requires_role(Role.MODERATOR)
|
|
async def wh_mod_nuke(self, ctx: commands.Context, name: str, message_id: int) -> None:
|
|
"""Delete a message and all its relayed copies across the network."""
|
|
nd = await self._net(name)
|
|
if not nd:
|
|
return await ctx.send(embed=err_embed(f"Network `{name}` not found."))
|
|
orig_id = self.msg_map.get_original(name, message_id) or message_id
|
|
relayed = self.msg_map.get_relayed(name, orig_id)
|
|
deleted = 0
|
|
for ch_id, msg_id in relayed.items():
|
|
ch = self.bot.get_channel(ch_id)
|
|
if ch:
|
|
try:
|
|
msg = await ch.fetch_message(msg_id)
|
|
await msg.delete()
|
|
deleted += 1
|
|
except Exception:
|
|
pass
|
|
# Also try to delete in source channels
|
|
for ch_id in nd.get("channels", []):
|
|
ch = self.bot.get_channel(ch_id)
|
|
if ch:
|
|
try:
|
|
msg = await ch.fetch_message(orig_id)
|
|
await msg.delete()
|
|
deleted += 1
|
|
except Exception:
|
|
pass
|
|
await ctx.send(embed=ok_embed(f"Nuked {deleted} message(s) across `{name}`."))
|
|
await self._audit(name, "nuke", str(ctx.author), str(orig_id))
|
|
await self._log(nd, warn_embed(f"💥 Message `{orig_id}` nuked by {ctx.author} ({deleted} copies deleted)"))
|
|
|
|
# ── Audit log ──────────────────────────────────────────────────────────
|
|
|
|
@WormholeBase.wh_mod.command(name="audit")
|
|
@requires_role(Role.HELPER)
|
|
async def wh_mod_audit(self, ctx: commands.Context, name: str, count: int = 20) -> None:
|
|
"""View the network audit log."""
|
|
nd = await self._net(name)
|
|
if not nd:
|
|
return await ctx.send(embed=err_embed(f"Network `{name}` not found."))
|
|
entries = nd.get("audit_log", [])[-count:]
|
|
if not entries:
|
|
return await ctx.send(embed=info_embed("Audit log is empty."))
|
|
lines = []
|
|
for e in reversed(entries):
|
|
ts = e.get("timestamp", "?")[:16]
|
|
act = e.get("action", "?")
|
|
usr = e.get("user", "?")
|
|
tgt = e.get("target", "")
|
|
det = e.get("details", "")
|
|
line = f"`{ts}` **{act}** by {usr}"
|
|
if tgt:
|
|
line += f" → {tgt}"
|
|
if det:
|
|
line += f" ({det})"
|
|
lines.append(line)
|
|
em = discord.Embed(
|
|
title=f"📋 Audit Log — {name}",
|
|
description="\n".join(lines),
|
|
colour=COLOUR_INFO,
|
|
)
|
|
await ctx.send(embed=em)
|
|
|
|
# ── Ban/mute lists ─────────────────────────────────────────────────────
|
|
|
|
@WormholeBase.wh_mod.command(name="bans")
|
|
@requires_role(Role.HELPER)
|
|
async def wh_mod_bans(self, ctx: commands.Context, name: str) -> None:
|
|
"""List banned users."""
|
|
nd = await self._net(name)
|
|
if not nd:
|
|
return await ctx.send(embed=err_embed(f"Network `{name}` not found."))
|
|
bans = nd.get("banned_users", [])
|
|
if not bans:
|
|
return await ctx.send(embed=info_embed(f"No banned users in `{name}`."))
|
|
lines = []
|
|
for uid in bans:
|
|
user = self.bot.get_user(uid)
|
|
lines.append(f"• {user or uid}")
|
|
await ctx.send(embed=info_embed("\n".join(lines), title=f"🔨 Banned Users — {name}"))
|
|
|
|
@WormholeBase.wh_mod.command(name="mutes")
|
|
@requires_role(Role.HELPER)
|
|
async def wh_mod_mutes(self, ctx: commands.Context, name: str) -> None:
|
|
"""List muted users."""
|
|
nd = await self._net(name)
|
|
if not nd:
|
|
return await ctx.send(embed=err_embed(f"Network `{name}` not found."))
|
|
mutes = nd.get("muted_users", [])
|
|
if not mutes:
|
|
return await ctx.send(embed=info_embed(f"No muted users in `{name}`."))
|
|
lines = []
|
|
for uid in mutes:
|
|
user = self.bot.get_user(uid)
|
|
lines.append(f"• {user or uid}")
|
|
await ctx.send(embed=info_embed("\n".join(lines), title=f"🔇 Muted Users — {name}"))
|