feat: add Mumble server admin plugin (!mu)
Admin-only command with subcommand dispatch for server management: kick, ban, mute/unmute, deafen/undeafen, move, users, channels, mkchan, rmchan, rename, desc. Auto-loads on merlin via except_plugins.
This commit is contained in:
user
276
plugins/mumble_admin.py
Normal file
276
plugins/mumble_admin.py
Normal file
@@ -0,0 +1,276 @@
|
||||
"""Plugin: Mumble server administration via chat commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from derp.plugin import command
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# -- Helpers -----------------------------------------------------------------
|
||||
|
||||
|
||||
def _find_user(bot, name: str):
|
||||
"""Case-insensitive user lookup by name. Returns pymumble User or None."""
|
||||
mumble = getattr(bot, "_mumble", None)
|
||||
if mumble is None:
|
||||
return None
|
||||
lower = name.lower()
|
||||
for sid in list(mumble.users):
|
||||
user = mumble.users[sid]
|
||||
if user["name"].lower() == lower:
|
||||
return user
|
||||
return None
|
||||
|
||||
|
||||
def _find_channel(bot, name: str):
|
||||
"""Case-insensitive channel lookup by name. Returns pymumble Channel or None."""
|
||||
mumble = getattr(bot, "_mumble", None)
|
||||
if mumble is None:
|
||||
return None
|
||||
lower = name.lower()
|
||||
for cid in list(mumble.channels):
|
||||
chan = mumble.channels[cid]
|
||||
if chan["name"].lower() == lower:
|
||||
return chan
|
||||
return None
|
||||
|
||||
|
||||
def _channel_name(bot, channel_id: int) -> str:
|
||||
"""Resolve a channel ID to its name, or return the ID as string."""
|
||||
mumble = getattr(bot, "_mumble", None)
|
||||
if mumble is None:
|
||||
return str(channel_id)
|
||||
chan = mumble.channels.get(channel_id)
|
||||
if chan is None:
|
||||
return str(channel_id)
|
||||
return chan["name"]
|
||||
|
||||
|
||||
# -- Sub-handlers ------------------------------------------------------------
|
||||
|
||||
|
||||
async def _sub_kick(bot, message, args: list[str]) -> None:
|
||||
if not args:
|
||||
await bot.reply(message, "Usage: !mu kick <user> [reason]")
|
||||
return
|
||||
user = _find_user(bot, args[0])
|
||||
if user is None:
|
||||
await bot.reply(message, f"User not found: {args[0]}")
|
||||
return
|
||||
reason = " ".join(args[1:]) if len(args) > 1 else ""
|
||||
user.kick(reason)
|
||||
await bot.reply(message, f"Kicked {user['name']}")
|
||||
|
||||
|
||||
async def _sub_ban(bot, message, args: list[str]) -> None:
|
||||
if not args:
|
||||
await bot.reply(message, "Usage: !mu ban <user> [reason]")
|
||||
return
|
||||
user = _find_user(bot, args[0])
|
||||
if user is None:
|
||||
await bot.reply(message, f"User not found: {args[0]}")
|
||||
return
|
||||
reason = " ".join(args[1:]) if len(args) > 1 else ""
|
||||
user.ban(reason)
|
||||
await bot.reply(message, f"Banned {user['name']}")
|
||||
|
||||
|
||||
async def _sub_mute(bot, message, args: list[str]) -> None:
|
||||
if not args:
|
||||
await bot.reply(message, "Usage: !mu mute <user>")
|
||||
return
|
||||
user = _find_user(bot, args[0])
|
||||
if user is None:
|
||||
await bot.reply(message, f"User not found: {args[0]}")
|
||||
return
|
||||
user.mute()
|
||||
await bot.reply(message, f"Muted {user['name']}")
|
||||
|
||||
|
||||
async def _sub_unmute(bot, message, args: list[str]) -> None:
|
||||
if not args:
|
||||
await bot.reply(message, "Usage: !mu unmute <user>")
|
||||
return
|
||||
user = _find_user(bot, args[0])
|
||||
if user is None:
|
||||
await bot.reply(message, f"User not found: {args[0]}")
|
||||
return
|
||||
user.unmute()
|
||||
await bot.reply(message, f"Unmuted {user['name']}")
|
||||
|
||||
|
||||
async def _sub_deafen(bot, message, args: list[str]) -> None:
|
||||
if not args:
|
||||
await bot.reply(message, "Usage: !mu deafen <user>")
|
||||
return
|
||||
user = _find_user(bot, args[0])
|
||||
if user is None:
|
||||
await bot.reply(message, f"User not found: {args[0]}")
|
||||
return
|
||||
user.deafen()
|
||||
await bot.reply(message, f"Deafened {user['name']}")
|
||||
|
||||
|
||||
async def _sub_undeafen(bot, message, args: list[str]) -> None:
|
||||
if not args:
|
||||
await bot.reply(message, "Usage: !mu undeafen <user>")
|
||||
return
|
||||
user = _find_user(bot, args[0])
|
||||
if user is None:
|
||||
await bot.reply(message, f"User not found: {args[0]}")
|
||||
return
|
||||
user.undeafen()
|
||||
await bot.reply(message, f"Undeafened {user['name']}")
|
||||
|
||||
|
||||
async def _sub_move(bot, message, args: list[str]) -> None:
|
||||
if len(args) < 2:
|
||||
await bot.reply(message, "Usage: !mu move <user> <channel>")
|
||||
return
|
||||
user = _find_user(bot, args[0])
|
||||
if user is None:
|
||||
await bot.reply(message, f"User not found: {args[0]}")
|
||||
return
|
||||
chan = _find_channel(bot, " ".join(args[1:]))
|
||||
if chan is None:
|
||||
await bot.reply(message, f"Channel not found: {' '.join(args[1:])}")
|
||||
return
|
||||
user.move_in(chan["channel_id"])
|
||||
await bot.reply(message, f"Moved {user['name']} to {chan['name']}")
|
||||
|
||||
|
||||
async def _sub_users(bot, message, args: list[str]) -> None:
|
||||
mumble = getattr(bot, "_mumble", None)
|
||||
if mumble is None:
|
||||
return
|
||||
bots = getattr(bot.registry, "_bots", {})
|
||||
lines: list[str] = []
|
||||
for sid in sorted(mumble.users):
|
||||
user = mumble.users[sid]
|
||||
name = user["name"]
|
||||
flags: list[str] = []
|
||||
if name in bots:
|
||||
flags.append("bot")
|
||||
if user.get("mute") or user.get("self_mute"):
|
||||
flags.append("muted")
|
||||
if user.get("deaf") or user.get("self_deaf"):
|
||||
flags.append("deaf")
|
||||
chan = _channel_name(bot, user.get("channel_id", 0))
|
||||
tag = f" [{', '.join(flags)}]" if flags else ""
|
||||
lines.append(f" {name} in {chan}{tag}")
|
||||
header = f"Online: {len(lines)} user(s)"
|
||||
await bot.reply(message, header + "\n" + "\n".join(lines))
|
||||
|
||||
|
||||
async def _sub_channels(bot, message, args: list[str]) -> None:
|
||||
mumble = getattr(bot, "_mumble", None)
|
||||
if mumble is None:
|
||||
return
|
||||
lines: list[str] = []
|
||||
for cid in sorted(mumble.channels):
|
||||
chan = mumble.channels[cid]
|
||||
name = chan["name"]
|
||||
# Count users in this channel
|
||||
count = sum(
|
||||
1 for sid in mumble.users
|
||||
if mumble.users[sid].get("channel_id") == cid
|
||||
)
|
||||
lines.append(f" {name} ({count})")
|
||||
await bot.reply(message, "Channels:\n" + "\n".join(lines))
|
||||
|
||||
|
||||
async def _sub_mkchan(bot, message, args: list[str]) -> None:
|
||||
if not args:
|
||||
await bot.reply(message, "Usage: !mu mkchan <name> [temp]")
|
||||
return
|
||||
mumble = getattr(bot, "_mumble", None)
|
||||
if mumble is None:
|
||||
return
|
||||
name = args[0]
|
||||
temp = len(args) > 1 and args[1].lower() in ("temp", "temporary", "true")
|
||||
mumble.channels.new_channel(0, name, temporary=temp)
|
||||
label = " (temporary)" if temp else ""
|
||||
await bot.reply(message, f"Created channel: {name}{label}")
|
||||
|
||||
|
||||
async def _sub_rmchan(bot, message, args: list[str]) -> None:
|
||||
if not args:
|
||||
await bot.reply(message, "Usage: !mu rmchan <channel>")
|
||||
return
|
||||
chan = _find_channel(bot, " ".join(args))
|
||||
if chan is None:
|
||||
await bot.reply(message, f"Channel not found: {' '.join(args)}")
|
||||
return
|
||||
name = chan["name"]
|
||||
chan.remove()
|
||||
await bot.reply(message, f"Removed channel: {name}")
|
||||
|
||||
|
||||
async def _sub_rename(bot, message, args: list[str]) -> None:
|
||||
if len(args) < 2:
|
||||
await bot.reply(message, "Usage: !mu rename <channel> <new-name>")
|
||||
return
|
||||
chan = _find_channel(bot, args[0])
|
||||
if chan is None:
|
||||
await bot.reply(message, f"Channel not found: {args[0]}")
|
||||
return
|
||||
old = chan["name"]
|
||||
chan.rename_channel(args[1])
|
||||
await bot.reply(message, f"Renamed {old} to {args[1]}")
|
||||
|
||||
|
||||
async def _sub_desc(bot, message, args: list[str]) -> None:
|
||||
if len(args) < 2:
|
||||
await bot.reply(message, "Usage: !mu desc <channel> <text>")
|
||||
return
|
||||
chan = _find_channel(bot, args[0])
|
||||
if chan is None:
|
||||
await bot.reply(message, f"Channel not found: {args[0]}")
|
||||
return
|
||||
text = " ".join(args[1:])
|
||||
chan.set_channel_description(text)
|
||||
await bot.reply(message, f"Set description for {chan['name']}")
|
||||
|
||||
|
||||
# -- Dispatch table ----------------------------------------------------------
|
||||
|
||||
|
||||
_SUBS: dict[str, object] = {
|
||||
"kick": _sub_kick,
|
||||
"ban": _sub_ban,
|
||||
"mute": _sub_mute,
|
||||
"unmute": _sub_unmute,
|
||||
"deafen": _sub_deafen,
|
||||
"undeafen": _sub_undeafen,
|
||||
"move": _sub_move,
|
||||
"users": _sub_users,
|
||||
"channels": _sub_channels,
|
||||
"mkchan": _sub_mkchan,
|
||||
"rmchan": _sub_rmchan,
|
||||
"rename": _sub_rename,
|
||||
"desc": _sub_desc,
|
||||
}
|
||||
|
||||
_USAGE = (
|
||||
"Usage: !mu <action> [args]\n"
|
||||
"Actions: kick, ban, mute, unmute, deafen, undeafen, move, "
|
||||
"users, channels, mkchan, rmchan, rename, desc"
|
||||
)
|
||||
|
||||
|
||||
@command("mu", help="Mumble admin: !mu <action> [args]", tier="admin")
|
||||
async def cmd_mu(bot, message):
|
||||
"""Mumble server administration commands."""
|
||||
parts = message.text.split()
|
||||
if len(parts) < 2:
|
||||
await bot.reply(message, _USAGE)
|
||||
return
|
||||
sub = parts[1].lower()
|
||||
handler = _SUBS.get(sub)
|
||||
if handler is None:
|
||||
await bot.reply(message, _USAGE)
|
||||
return
|
||||
await handler(bot, message, parts[2:])
|
||||
Reference in New Issue
Block a user