Initial commit

This commit is contained in:
zv
2026-04-18 19:03:17 +02:00
commit f6f1358c1b
6 changed files with 1219 additions and 0 deletions

29
.gitignore vendored Normal file
View File

@@ -0,0 +1,29 @@
# Python cache
__pycache__/
*.py[cod]
*$py.class
# Build / packaging
build/
dist/
*.egg-info/
.eggs/
# Virtual environments
.venv/
venv/
env/
# Tool caches
.mypy_cache/
.pytest_cache/
.ruff_cache/
# IDE / OS noise
.idea/
.vscode/
.DS_Store
Thumbs.db
# Runtime artifacts
*.log

98
README.md Normal file
View File

@@ -0,0 +1,98 @@
# TempVoice (Red-DiscordBot Cog)
Join-to-create temporary voice channels for Red-DiscordBot.
## Features
- Per-guild configuration (no hardcoded global channel IDs)
- Auto-create temporary voice channel when user joins a configured start channel
- Auto-move user to their own temporary channel
- Reuse existing temporary channel for the same owner if it still exists
- Auto-delete temporary channel when it becomes empty
- Owner command to set channel user limit (`0-99`)
- Admin commands for setup and cleanup
- i18n-ready runtime messages (English + Polish translation file)
## Requirements
- Red-DiscordBot `>= 3.5.0`
- Python `>= 3.8`
## Install
1. Put this cog folder in your custom cogs path.
2. In Discord:
```text
[p]load tempvoice
```
## Initial Setup (per server)
Use these commands on each guild:
```text
[p]tempvoice setstart <voice_channel>
[p]tempvoice setcategory <category>
```
Check settings:
```text
[p]tempvoice settings
```
## Commands
### Admin (`manage_guild` or admin)
- `[p]tempvoice settings` - show config and current state
- `[p]tempvoice setstart <voice_channel>` - set join-to-create channel
- `[p]tempvoice setcategory <category>` - set target category for temp channels
- `[p]tempvoice cleanup` - remove stale mappings and delete empty temp channels
### User (channel owner)
- `[p]tempvoice setlimit <0-99>` - set user limit on your own temp channel
- `[p]setlimit <0-99>` - hidden backward-compatible alias
## How It Works
1. User joins the configured start voice channel.
2. Cog creates (or reuses) an owner-mapped temporary voice channel in configured category.
3. User is moved to that channel.
4. When temporary channel is empty, it is deleted automatically.
## Data Stored
Per guild:
- `start_channel_id`
- `target_category_id`
- `channel_owners` (mapping: temporary channel ID -> owner user ID)
- `schema_version`
## Localization
- Runtime messages are localized via Red i18n.
- Included translation file: `locales/pl-PL.po`.
- Fallback language is English.
## Troubleshooting
- If setup commands succeed but still print command errors, check bot logs and reload cog:
```text
[p]reload tempvoice
```
- Ensure bot has:
- Guild permission: `Move Members`
- Category/channel permissions: `View Channel`, `Connect`, `Manage Channels`
- If state looks inconsistent:
```text
[p]tempvoice cleanup
```

5
__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from .tempvoice import TempVoice
async def setup(bot):
await bot.add_cog(TempVoice(bot))

23
info.json Normal file
View File

@@ -0,0 +1,23 @@
{
"author": [
"codex"
],
"install_msg": "TempVoice loaded. Use `[p]tempvoice settings` to review the current server configuration.",
"name": "TempVoice",
"short": "Join-to-create temporary voice channels",
"description": "Creates and manages join-to-create temporary voice channels per server, including owner user-limit control and admin configuration commands.",
"end_user_data_statement": "Stores per-server configuration and temporary channel ownership mappings (channel ID to owner ID).",
"requirements": [],
"tags": [
"voice",
"temporary",
"join-to-create"
],
"min_bot_version": "3.5.0",
"min_python_version": [
3,
8,
0
],
"type": "COG"
}

131
locales/pl-PL.po Normal file
View File

@@ -0,0 +1,131 @@
msgid ""
msgstr ""
"Project-Id-Version: TempVoice 1.0\n"
"POT-Creation-Date: 2026-04-18 00:00+0000\n"
"PO-Revision-Date: 2026-04-18 00:00+0000\n"
"Last-Translator: codex\n"
"Language-Team: Polish\n"
"Language: pl_PL\n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=UTF-8\n"
"Content-Transfer-Encoding: 8bit\n"
msgid "{mention} your temporary channel was created: **{channel_name}**."
msgstr "{mention} utworzono Twój kanał tymczasowy: **{channel_name}**."
msgid "TempVoice is not fully configured for this server. An administrator should set the start channel and target category with `tempvoice setstart` and `tempvoice setcategory`."
msgstr "TempVoice nie jest w pełni skonfigurowany dla tego serwera. Administrator powinien ustawić kanał startowy i kategorię docelową komendami `tempvoice setstart` oraz `tempvoice setcategory`."
msgid "I can't create a temporary channel because the target category no longer exists."
msgstr "Nie mogę utworzyć kanału tymczasowego, bo kategoria docelowa już nie istnieje."
msgid "I can't move you because I am missing the `move_members` permission."
msgstr "Nie mogę Cię przenieść, bo brakuje mi uprawnienia `move_members`."
msgid "I can't create your temporary channel. I need `view_channel`, `connect`, and `manage_channels` in the target category."
msgstr "Nie mogę utworzyć Twojego kanału tymczasowego. Potrzebuję uprawnień `view_channel`, `connect` oraz `manage_channels` w kategorii docelowej."
msgid "I couldn't move you to your temporary channel."
msgstr "Nie udało mi się przenieść Cię na Twój kanał tymczasowy."
msgid "I can't create your temporary channel because I am missing permissions."
msgstr "Nie mogę utworzyć Twojego kanału tymczasowego z powodu brakujących uprawnień."
msgid "There was an error while creating your temporary channel."
msgstr "Wystąpił błąd podczas tworzenia Twojego kanału tymczasowego."
msgid "I couldn't move you to your newly created temporary channel."
msgstr "Nie udało mi się przenieść Cię na nowo utworzony kanał tymczasowy."
msgid "Invalid value. Please provide a number from 0 to 99 (0 = unlimited)."
msgstr "Nieprawidłowa wartość. Podaj liczbę od 0 do 99 (0 = brak limitu)."
msgid "You must be connected to your own temporary voice channel."
msgstr "Musisz być połączony ze swoim kanałem głosowym tymczasowym."
msgid "This is not a temporary channel managed by TempVoice."
msgstr "To nie jest kanał tymczasowy zarządzany przez TempVoice."
msgid "This temporary channel belongs to another user."
msgstr "Ten kanał tymczasowy należy do innego użytkownika."
msgid "I couldn't verify my permissions right now."
msgstr "Nie mogę teraz zweryfikować swoich uprawnień."
msgid "I can't change the limit because I am missing `manage_channels`."
msgstr "Nie mogę zmienić limitu, bo brakuje mi uprawnienia `manage_channels`."
msgid "I don't have permission to edit this channel's user limit."
msgstr "Nie mam uprawnień do edycji limitu użytkowników tego kanału."
msgid "There was an error while changing the channel user limit."
msgstr "Wystąpił błąd podczas zmiany limitu użytkowników kanału."
msgid "User limit for {channel_mention} is now set to unlimited."
msgstr "Limit użytkowników dla {channel_mention} został ustawiony na brak limitu."
msgid "User limit for {channel_mention} is now set to {limit}."
msgstr "Limit użytkowników dla {channel_mention} został ustawiony na {limit}."
msgid "TempVoice configuration:"
msgstr "Konfiguracja TempVoice:"
msgid "- Configured: {configured}"
msgstr "- Skonfigurowane: {configured}"
msgid "yes"
msgstr "tak"
msgid "no"
msgstr "nie"
msgid "- Start channel: {channel}"
msgstr "- Kanał startowy: {channel}"
msgid "not set"
msgstr "nie ustawiono"
msgid "- Target category: {category}"
msgstr "- Kategoria docelowa: {category}"
msgid "- Active temporary channels: {count}"
msgstr "- Aktywne kanały tymczasowe: {count}"
msgid "- Stale mappings (cleanup candidate): {count}"
msgstr "- Osierocone mapowania (do cleanup): {count}"
msgid "- Complete setup with: `tempvoice setstart <voice_channel>` and `tempvoice setcategory <category>`"
msgstr "- Dokończ konfigurację komendami: `tempvoice setstart <kanał_voice>` oraz `tempvoice setcategory <kategoria>`"
msgid "That channel does not belong to this server."
msgstr "Ten kanał nie należy do tego serwera."
msgid "I can't use this start channel. Missing bot permissions: {missing}."
msgstr "Nie mogę użyć tego kanału startowego. Brakujące uprawnienia bota: {missing}."
msgid "Now set the target category with `tempvoice setcategory`."
msgstr "Teraz ustaw kategorię docelową komendą `tempvoice setcategory`."
msgid "Start channel set to {channel_mention} (`{channel_id}`)."
msgstr "Kanał startowy ustawiono na {channel_mention} (`{channel_id}`)."
msgid "That category does not belong to this server."
msgstr "Ta kategoria nie należy do tego serwera."
msgid "I can't use this category. Missing bot permissions: {missing}."
msgstr "Nie mogę użyć tej kategorii. Brakujące uprawnienia bota: {missing}."
msgid "Now set the start channel with `tempvoice setstart`."
msgstr "Teraz ustaw kanał startowy komendą `tempvoice setstart`."
msgid "Target category set to {category_mention} (`{category_id}`)."
msgstr "Kategorię docelową ustawiono na {category_mention} (`{category_id}`)."
msgid "Cleanup finished. Deleted empty channels: {deleted}. Removed stale mappings: {removed}. Failed deletions: {failed}."
msgstr "Cleanup zakończony. Usunięte puste kanały: {deleted}. Usunięte osierocone mapowania: {removed}. Nieudane usunięcia: {failed}."
msgid "An unexpected error occurred while setting the start channel. Check logs for details."
msgstr "Wystąpił nieoczekiwany błąd podczas ustawiania kanału startowego. Sprawdź logi po szczegóły."
msgid "An unexpected error occurred while setting the target category. Check logs for details."
msgstr "Wystąpił nieoczekiwany błąd podczas ustawiania kategorii docelowej. Sprawdź logi po szczegóły."

933
tempvoice.py Normal file
View File

@@ -0,0 +1,933 @@
import asyncio
import logging
import re
import time
from typing import Dict, Iterable, List, Optional, Set, Tuple
import discord
from redbot.core import Config, commands
from redbot.core.bot import Red
from redbot.core.i18n import Translator, cog_i18n, set_contextual_locales_from_guild
_ = Translator("TempVoice", __file__)
log = logging.getLogger("red.tempvoice")
@cog_i18n(_)
class TempVoice(commands.Cog):
"""Join-to-create temporary voice channels."""
CONFIG_SCHEMA_VERSION = 2
UNCONFIGURED_NOTICE_COOLDOWN = 600
def __init__(self, bot: Red) -> None:
self.bot = bot
self.config = Config.get_conf(self, identifier=1495065310565236910, force_registration=True)
self.config.register_guild(
start_channel_id=None,
target_category_id=None,
channel_owners={},
schema_version=self.CONFIG_SCHEMA_VERSION,
)
self._member_locks: Dict[Tuple[int, int], asyncio.Lock] = {}
self._guild_locks: Dict[int, asyncio.Lock] = {}
self._schema_ready_guilds: Set[int] = set()
self._unconfigured_notice_ts: Dict[int, float] = {}
def cog_unload(self) -> None:
self._member_locks.clear()
self._guild_locks.clear()
self._schema_ready_guilds.clear()
self._unconfigured_notice_ts.clear()
@staticmethod
def _coerce_int(value: object) -> Optional[int]:
try:
if value is None:
return None
return int(value)
except (TypeError, ValueError):
return None
@staticmethod
def _safe_voice_name(raw_name: str, fallback_id: int) -> str:
name = raw_name.strip()
name = re.sub(r"\s+", " ", name)
name = re.sub(r"[^\w\- .]", "", name, flags=re.UNICODE)
name = name.strip(" .")
if not name:
name = "user-{}".format(fallback_id)
return name[:32]
@staticmethod
def _normalize_channel_owners(raw: object) -> Dict[str, int]:
normalized: Dict[str, int] = {}
if not isinstance(raw, dict):
return normalized
for raw_channel_id, raw_owner_id in raw.items():
channel_id = TempVoice._coerce_int(raw_channel_id)
owner_id = TempVoice._coerce_int(raw_owner_id)
if channel_id is None or owner_id is None:
continue
normalized[str(channel_id)] = owner_id
return normalized
@staticmethod
def _dedupe_owner_mappings(channel_owners: Dict[str, int]) -> Dict[str, int]:
deduped: Dict[str, int] = {}
seen_owners: Set[int] = set()
for channel_id, owner_id in channel_owners.items():
if owner_id in seen_owners:
continue
deduped[channel_id] = owner_id
seen_owners.add(owner_id)
return deduped
@staticmethod
def _filter_existing_voice_channels(
guild: discord.Guild, channel_owners: Dict[str, int]
) -> Dict[str, int]:
filtered: Dict[str, int] = {}
for channel_id_str, owner_id in channel_owners.items():
channel = guild.get_channel(int(channel_id_str))
if isinstance(channel, discord.VoiceChannel):
filtered[channel_id_str] = owner_id
return filtered
@staticmethod
def _validate_start_channel_id(guild: discord.Guild, channel_id: Optional[int]) -> Optional[int]:
if channel_id is None:
return None
channel = guild.get_channel(channel_id)
if isinstance(channel, discord.VoiceChannel):
return channel_id
return None
@staticmethod
def _validate_target_category_id(guild: discord.Guild, category_id: Optional[int]) -> Optional[int]:
if category_id is None:
return None
category = guild.get_channel(category_id)
if isinstance(category, discord.CategoryChannel):
return category_id
return None
@staticmethod
def _bot_member(guild: discord.Guild) -> Optional[discord.Member]:
return guild.me
def _get_member_lock(self, guild_id: int, member_id: int) -> asyncio.Lock:
key = (guild_id, member_id)
lock = self._member_locks.get(key)
if lock is None:
lock = asyncio.Lock()
self._member_locks[key] = lock
return lock
def _release_member_lock(self, guild_id: int, member_id: int, lock: asyncio.Lock) -> None:
key = (guild_id, member_id)
if self._member_locks.get(key) is lock and not lock.locked():
self._member_locks.pop(key, None)
def _get_guild_lock(self, guild_id: int) -> asyncio.Lock:
lock = self._guild_locks.get(guild_id)
if lock is None:
lock = asyncio.Lock()
self._guild_locks[guild_id] = lock
return lock
async def _set_locale_context(self, guild: discord.Guild) -> None:
try:
await set_contextual_locales_from_guild(self.bot, guild)
except Exception:
log.debug("Failed to set locale context for guild %s (%s).", guild, guild.id)
def _migrate_legacy_channel_owners(self, guild: discord.Guild, data: dict) -> Dict[str, int]:
migrated: Dict[str, int] = {}
owners_seen: Set[int] = set()
raw_owner_to_channel = data.get("owner_to_channel")
if isinstance(raw_owner_to_channel, dict):
for raw_owner_id, raw_channel_id in raw_owner_to_channel.items():
owner_id = self._coerce_int(raw_owner_id)
channel_id = self._coerce_int(raw_channel_id)
if owner_id is None or channel_id is None:
continue
channel = guild.get_channel(channel_id)
if not isinstance(channel, discord.VoiceChannel):
continue
if owner_id in owners_seen:
continue
migrated[str(channel_id)] = owner_id
owners_seen.add(owner_id)
raw_channel_to_owner = data.get("channel_to_owner")
if isinstance(raw_channel_to_owner, dict):
for raw_channel_id, raw_owner_id in raw_channel_to_owner.items():
channel_id = self._coerce_int(raw_channel_id)
owner_id = self._coerce_int(raw_owner_id)
if channel_id is None or owner_id is None:
continue
if owner_id in owners_seen:
continue
channel = guild.get_channel(channel_id)
if not isinstance(channel, discord.VoiceChannel):
continue
migrated[str(channel_id)] = owner_id
owners_seen.add(owner_id)
raw_channel_owners = self._normalize_channel_owners(data.get("channel_owners"))
for channel_id_str, owner_id in raw_channel_owners.items():
if owner_id in owners_seen:
continue
channel = guild.get_channel(int(channel_id_str))
if not isinstance(channel, discord.VoiceChannel):
continue
migrated[channel_id_str] = owner_id
owners_seen.add(owner_id)
return migrated
async def _ensure_guild_schema(self, guild: discord.Guild) -> None:
if guild.id in self._schema_ready_guilds:
return
async with self._get_guild_lock(guild.id):
if guild.id in self._schema_ready_guilds:
return
conf = self.config.guild(guild)
data = await conf.all()
schema_version = self._coerce_int(data.get("schema_version"))
if schema_version is None:
schema_version = 0
start_channel_id = self._validate_start_channel_id(
guild, self._coerce_int(data.get("start_channel_id"))
)
target_category_id = self._validate_target_category_id(
guild, self._coerce_int(data.get("target_category_id"))
)
if schema_version < self.CONFIG_SCHEMA_VERSION:
channel_owners = self._migrate_legacy_channel_owners(guild, data)
else:
channel_owners = self._normalize_channel_owners(data.get("channel_owners"))
channel_owners = self._filter_existing_voice_channels(guild, channel_owners)
channel_owners = self._dedupe_owner_mappings(channel_owners)
await conf.start_channel_id.set(start_channel_id)
await conf.target_category_id.set(target_category_id)
await conf.channel_owners.set(channel_owners)
await conf.schema_version.set(self.CONFIG_SCHEMA_VERSION)
self._schema_ready_guilds.add(guild.id)
async def _get_runtime_config(self, guild: discord.Guild) -> Tuple[Optional[int], Optional[int]]:
await self._ensure_guild_schema(guild)
conf = self.config.guild(guild)
raw_start_channel_id = self._coerce_int(await conf.start_channel_id())
raw_target_category_id = self._coerce_int(await conf.target_category_id())
start_channel_id = self._validate_start_channel_id(guild, raw_start_channel_id)
target_category_id = self._validate_target_category_id(guild, raw_target_category_id)
if start_channel_id != raw_start_channel_id or target_category_id != raw_target_category_id:
async with self._get_guild_lock(guild.id):
await conf.start_channel_id.set(start_channel_id)
await conf.target_category_id.set(target_category_id)
return start_channel_id, target_category_id
async def _get_channel_owners_snapshot(self, guild: discord.Guild) -> Dict[str, int]:
await self._ensure_guild_schema(guild)
raw = await self.config.guild(guild).channel_owners()
normalized = self._normalize_channel_owners(raw)
normalized = self._filter_existing_voice_channels(guild, normalized)
normalized = self._dedupe_owner_mappings(normalized)
return normalized
async def _set_owner_channel_mapping(self, guild: discord.Guild, owner_id: int, channel_id: int) -> None:
await self._ensure_guild_schema(guild)
async with self._get_guild_lock(guild.id):
conf = self.config.guild(guild)
channel_owners = self._normalize_channel_owners(await conf.channel_owners())
channel_owners = self._filter_existing_voice_channels(guild, channel_owners)
channel_owners = self._dedupe_owner_mappings(channel_owners)
for existing_channel_id_str, mapped_owner_id in list(channel_owners.items()):
if mapped_owner_id == owner_id and int(existing_channel_id_str) != channel_id:
channel_owners.pop(existing_channel_id_str, None)
channel_owners[str(channel_id)] = owner_id
channel_owners = self._dedupe_owner_mappings(channel_owners)
await conf.channel_owners.set(channel_owners)
async def _remove_channel_mappings_bulk(self, guild: discord.Guild, channel_ids: Iterable[int]) -> int:
ids = {int(channel_id) for channel_id in channel_ids}
if not ids:
return 0
await self._ensure_guild_schema(guild)
async with self._get_guild_lock(guild.id):
conf = self.config.guild(guild)
channel_owners = self._normalize_channel_owners(await conf.channel_owners())
removed = 0
for channel_id in ids:
if channel_owners.pop(str(channel_id), None) is not None:
removed += 1
if removed:
channel_owners = self._dedupe_owner_mappings(channel_owners)
await conf.channel_owners.set(channel_owners)
return removed
async def _remove_mapping_by_owner(self, guild: discord.Guild, owner_id: int) -> None:
await self._ensure_guild_schema(guild)
async with self._get_guild_lock(guild.id):
conf = self.config.guild(guild)
channel_owners = self._normalize_channel_owners(await conf.channel_owners())
changed = False
for channel_id_str, mapped_owner_id in list(channel_owners.items()):
if mapped_owner_id == owner_id:
channel_owners.pop(channel_id_str, None)
changed = True
if changed:
channel_owners = self._dedupe_owner_mappings(channel_owners)
await conf.channel_owners.set(channel_owners)
async def _get_channel_for_owner(self, guild: discord.Guild, owner_id: int) -> Optional[int]:
channel_owners = await self._get_channel_owners_snapshot(guild)
stale_channels: Set[int] = set()
for channel_id_str, mapped_owner_id in channel_owners.items():
if mapped_owner_id != owner_id:
continue
channel_id = int(channel_id_str)
channel = guild.get_channel(channel_id)
if isinstance(channel, discord.VoiceChannel):
return channel_id
stale_channels.add(channel_id)
if stale_channels:
await self._remove_channel_mappings_bulk(guild, stale_channels)
return None
async def _get_owner_for_channel(self, guild: discord.Guild, channel_id: int) -> Optional[int]:
channel_owners = await self._get_channel_owners_snapshot(guild)
owner_id = channel_owners.get(str(channel_id))
if owner_id is None:
return None
channel = guild.get_channel(channel_id)
if channel is None:
await self._remove_channel_mappings_bulk(guild, {channel_id})
return None
return owner_id
def _find_fallback_text_channel(self, guild: discord.Guild) -> Optional[discord.TextChannel]:
bot_member = self._bot_member(guild)
if bot_member is None:
return None
if guild.system_channel:
perms = guild.system_channel.permissions_for(bot_member)
if perms.view_channel and perms.send_messages:
return guild.system_channel
for text_channel in guild.text_channels:
perms = text_channel.permissions_for(bot_member)
if perms.view_channel and perms.send_messages:
return text_channel
return None
async def _send_guild_message(
self,
guild: discord.Guild,
content: str,
preferred_channel: Optional[discord.abc.GuildChannel] = None,
) -> bool:
await self._set_locale_context(guild)
candidates: List[discord.abc.GuildChannel] = []
if preferred_channel is not None and callable(getattr(preferred_channel, "send", None)):
candidates.append(preferred_channel)
fallback = self._find_fallback_text_channel(guild)
if fallback is not None:
candidates.append(fallback)
used_ids: Set[int] = set()
for channel in candidates:
channel_id = getattr(channel, "id", None)
if channel_id is not None and channel_id in used_ids:
continue
if channel_id is not None:
used_ids.add(channel_id)
try:
await channel.send(content)
return True
except (discord.Forbidden, discord.HTTPException):
continue
return False
async def _send_creation_message(self, member: discord.Member, voice_channel: discord.VoiceChannel) -> None:
content = _("{mention} your temporary channel was created: **{channel_name}**.").format(
mention=member.mention, channel_name=voice_channel.name
)
await self._send_guild_message(member.guild, content, preferred_channel=voice_channel)
async def _send_error_message(self, guild: discord.Guild, user_mention: str, message: str) -> None:
await self._send_guild_message(guild, "{mention} {message}".format(mention=user_mention, message=message))
async def _maybe_notify_unconfigured(self, guild: discord.Guild) -> None:
now = time.monotonic()
last = self._unconfigured_notice_ts.get(guild.id, 0.0)
if now - last < self.UNCONFIGURED_NOTICE_COOLDOWN:
return
self._unconfigured_notice_ts[guild.id] = now
await self._send_guild_message(
guild,
_(
"TempVoice is not fully configured for this server. "
"An administrator should set the start channel and target category "
"with `tempvoice setstart` and `tempvoice setcategory`."
),
)
async def _move_member_to_channel(self, member: discord.Member, channel: discord.VoiceChannel) -> bool:
try:
await member.move_to(channel, reason="TempVoice: move user to temporary channel")
return True
except discord.Forbidden:
log.warning(
"Missing permissions to move user %s (%s) to channel %s (%s).",
member,
member.id,
channel,
channel.id,
)
except discord.HTTPException:
log.exception(
"HTTP error while moving user %s (%s) to channel %s (%s).",
member,
member.id,
channel,
channel.id,
)
return False
async def _handle_join_to_create(self, member: discord.Member, target_category_id: int) -> None:
guild = member.guild
bot_member = self._bot_member(guild)
if bot_member is None:
return
target_category = guild.get_channel(target_category_id)
if not isinstance(target_category, discord.CategoryChannel):
await self._send_error_message(
guild,
member.mention,
_("I can't create a temporary channel because the target category no longer exists."),
)
return
if not bot_member.guild_permissions.move_members:
await self._send_error_message(
guild,
member.mention,
_("I can't move you because I am missing the `move_members` permission."),
)
return
category_perms = target_category.permissions_for(bot_member)
if not (category_perms.view_channel and category_perms.connect and category_perms.manage_channels):
await self._send_error_message(
guild,
member.mention,
_(
"I can't create your temporary channel. I need `view_channel`, `connect`, "
"and `manage_channels` in the target category."
),
)
return
existing_channel_id = await self._get_channel_for_owner(guild, member.id)
if existing_channel_id is not None:
existing_channel = guild.get_channel(existing_channel_id)
if (
isinstance(existing_channel, discord.VoiceChannel)
and existing_channel.category_id == target_category_id
):
moved = await self._move_member_to_channel(member, existing_channel)
if not moved:
await self._send_error_message(
guild, member.mention, _("I couldn't move you to your temporary channel.")
)
return
await self._remove_mapping_by_owner(guild, member.id)
channel_name = self._safe_voice_name(member.display_name or member.name, member.id)
try:
new_channel = await guild.create_voice_channel(
name=channel_name,
category=target_category,
reason="TempVoice: create temporary voice for {} ({})".format(member, member.id),
)
except discord.Forbidden:
await self._send_error_message(
guild,
member.mention,
_("I can't create your temporary channel because I am missing permissions."),
)
return
except discord.HTTPException:
log.exception(
"HTTP error while creating a temporary channel for user %s (%s).",
member,
member.id,
)
await self._send_error_message(
guild,
member.mention,
_("There was an error while creating your temporary channel."),
)
return
await self._set_owner_channel_mapping(guild, member.id, new_channel.id)
moved = await self._move_member_to_channel(member, new_channel)
if not moved:
await self._send_error_message(
guild,
member.mention,
_("I couldn't move you to your newly created temporary channel."),
)
try:
await new_channel.delete(reason="TempVoice: cleanup after failed member move")
except (discord.Forbidden, discord.HTTPException):
log.debug(
"Failed to delete temporary channel %s (%s) after failed member move.",
new_channel,
new_channel.id,
)
await self._remove_channel_mappings_bulk(guild, {new_channel.id})
return
await self._send_creation_message(member, new_channel)
async def _delete_temp_channel_if_empty(self, channel: discord.abc.GuildChannel) -> None:
if not isinstance(channel, discord.VoiceChannel):
return
owner_id = await self._get_owner_for_channel(channel.guild, channel.id)
if owner_id is None:
return
if channel.members:
return
try:
await channel.delete(reason="TempVoice: deleting empty temporary channel")
except discord.NotFound:
await self._remove_channel_mappings_bulk(channel.guild, {channel.id})
except discord.Forbidden:
log.warning(
"Missing permissions to delete temporary channel %s (%s).",
channel,
channel.id,
)
except discord.HTTPException:
log.exception(
"HTTP error while deleting temporary channel %s (%s).",
channel,
channel.id,
)
else:
await self._remove_channel_mappings_bulk(channel.guild, {channel.id})
async def _setlimit_impl(self, ctx: commands.Context, limit: int) -> None:
await self._set_locale_context(ctx.guild)
if limit < 0 or limit > 99:
await ctx.send(_("Invalid value. Please provide a number from 0 to 99 (0 = unlimited)."))
return
if ctx.author.voice is None or not isinstance(ctx.author.voice.channel, discord.VoiceChannel):
await ctx.send(_("You must be connected to your own temporary voice channel."))
return
channel = ctx.author.voice.channel
owner_id = await self._get_owner_for_channel(ctx.guild, channel.id)
if owner_id is None:
await ctx.send(_("This is not a temporary channel managed by TempVoice."))
return
if owner_id != ctx.author.id:
await ctx.send(_("This temporary channel belongs to another user."))
return
bot_member = self._bot_member(ctx.guild)
if bot_member is None:
await ctx.send(_("I couldn't verify my permissions right now."))
return
channel_perms = channel.permissions_for(bot_member)
if not channel_perms.manage_channels:
await ctx.send(_("I can't change the limit because I am missing `manage_channels`."))
return
try:
await channel.edit(
user_limit=limit,
reason="TempVoice setlimit by {} ({})".format(ctx.author, ctx.author.id),
)
except discord.Forbidden:
await ctx.send(_("I don't have permission to edit this channel's user limit."))
return
except discord.HTTPException:
log.exception(
"HTTP error while changing user limit in channel %s (%s) by %s (%s).",
channel,
channel.id,
ctx.author,
ctx.author.id,
)
await ctx.send(_("There was an error while changing the channel user limit."))
return
if limit == 0:
await ctx.send(
_("User limit for {channel_mention} is now set to unlimited.").format(
channel_mention=channel.mention
)
)
else:
await ctx.send(
_("User limit for {channel_mention} is now set to {limit}.").format(
channel_mention=channel.mention, limit=limit
)
)
@commands.Cog.listener()
async def on_voice_state_update(
self,
member: discord.Member,
before: discord.VoiceState,
after: discord.VoiceState,
) -> None:
if member.bot or before.channel == after.channel:
return
guild = member.guild
member_lock = self._get_member_lock(guild.id, member.id)
try:
start_channel_id, target_category_id = await self._get_runtime_config(guild)
if start_channel_id is None or target_category_id is None:
await self._maybe_notify_unconfigured(guild)
elif after.channel is not None and after.channel.id == start_channel_id:
async with member_lock:
current_voice = member.voice.channel if member.voice is not None else None
if current_voice is not None and current_voice.id == start_channel_id:
await self._handle_join_to_create(member, target_category_id)
if before.channel is not None:
await self._delete_temp_channel_if_empty(before.channel)
except Exception:
log.exception(
"Unexpected exception in on_voice_state_update for member %s (%s).",
member,
member.id,
)
finally:
self._release_member_lock(guild.id, member.id, member_lock)
@commands.Cog.listener()
async def on_guild_channel_delete(self, channel: discord.abc.GuildChannel) -> None:
try:
guild = channel.guild
await self._ensure_guild_schema(guild)
conf = self.config.guild(guild)
if isinstance(channel, discord.VoiceChannel):
await self._remove_channel_mappings_bulk(guild, {channel.id})
start_channel_id = self._coerce_int(await conf.start_channel_id())
if start_channel_id == channel.id:
await conf.start_channel_id.set(None)
if isinstance(channel, discord.CategoryChannel):
target_category_id = self._coerce_int(await conf.target_category_id())
if target_category_id == channel.id:
await conf.target_category_id.set(None)
except Exception:
log.exception(
"Unexpected exception in on_guild_channel_delete for channel %s (%s).",
channel,
channel.id,
)
@commands.Cog.listener()
async def on_guild_remove(self, guild: discord.Guild) -> None:
self._schema_ready_guilds.discard(guild.id)
self._unconfigured_notice_ts.pop(guild.id, None)
self._guild_locks.pop(guild.id, None)
@commands.group(name="tempvoice", invoke_without_command=True)
@commands.guild_only()
@commands.admin_or_permissions(manage_guild=True)
async def tempvoice_group(self, ctx: commands.Context) -> None:
"""Administrative commands for TempVoice."""
await ctx.send_help()
@tempvoice_group.command(name="settings")
async def tempvoice_settings(self, ctx: commands.Context) -> None:
"""Show TempVoice configuration for this server."""
await self._set_locale_context(ctx.guild)
start_channel_id, target_category_id = await self._get_runtime_config(ctx.guild)
channel_owners = await self._get_channel_owners_snapshot(ctx.guild)
start_channel = ctx.guild.get_channel(start_channel_id) if start_channel_id is not None else None
target_category = ctx.guild.get_channel(target_category_id) if target_category_id is not None else None
active_channels = 0
stale_mappings = 0
for channel_id_str in channel_owners:
channel = ctx.guild.get_channel(int(channel_id_str))
if isinstance(channel, discord.VoiceChannel):
active_channels += 1
else:
stale_mappings += 1
configured = start_channel is not None and target_category is not None
lines = [
_("TempVoice configuration:"),
_("- Configured: {configured}").format(configured=_("yes") if configured else _("no")),
_("- Start channel: {channel}").format(
channel=start_channel.mention if isinstance(start_channel, discord.VoiceChannel) else _("not set")
),
_("- Target category: {category}").format(
category=target_category.mention
if isinstance(target_category, discord.CategoryChannel)
else _("not set")
),
_("- Active temporary channels: {count}").format(count=active_channels),
_("- Stale mappings (cleanup candidate): {count}").format(count=stale_mappings),
]
if not configured:
lines.append(
_(
"- Complete setup with: `tempvoice setstart <voice_channel>` "
"and `tempvoice setcategory <category>`"
)
)
await ctx.send("\n".join(lines))
@tempvoice_group.command(name="setstart")
async def tempvoice_setstart(self, ctx: commands.Context, channel: discord.VoiceChannel) -> None:
"""Set the join-to-create start voice channel for this server."""
await self._set_locale_context(ctx.guild)
try:
if channel.guild.id != ctx.guild.id:
await ctx.send(_("That channel does not belong to this server."))
return
bot_member = self._bot_member(ctx.guild)
if bot_member is None:
await ctx.send(_("I couldn't verify my permissions right now."))
return
channel_perms = channel.permissions_for(bot_member)
missing = []
if not channel_perms.view_channel:
missing.append("view_channel")
if not channel_perms.connect:
missing.append("connect")
if not bot_member.guild_permissions.move_members:
missing.append("move_members")
if missing:
await ctx.send(
_("I can't use this start channel. Missing bot permissions: {missing}.").format(
missing=", ".join(missing)
)
)
return
await self._ensure_guild_schema(ctx.guild)
await self.config.guild(ctx.guild).start_channel_id.set(channel.id)
suffix = ""
try:
current_start_channel_id, target_category_id = await self._get_runtime_config(ctx.guild)
if target_category_id is None:
suffix = " " + _("Now set the target category with `tempvoice setcategory`.")
except Exception:
log.exception(
"TempVoice setstart post-save validation failed for guild %s (%s).",
ctx.guild,
ctx.guild.id,
)
await ctx.send(
_("Start channel set to {channel_mention} (`{channel_id}`).").format(
channel_mention=channel.mention,
channel_id=channel.id,
)
+ suffix
)
except Exception:
log.exception(
"Unexpected exception in tempvoice setstart for guild %s (%s).",
ctx.guild,
ctx.guild.id,
)
await ctx.send(
_(
"An unexpected error occurred while setting the start channel. "
"Check logs for details."
)
)
@tempvoice_group.command(name="setcategory")
async def tempvoice_setcategory(self, ctx: commands.Context, category: discord.CategoryChannel) -> None:
"""Set the target category for temporary channels in this server."""
await self._set_locale_context(ctx.guild)
try:
if category.guild.id != ctx.guild.id:
await ctx.send(_("That category does not belong to this server."))
return
bot_member = self._bot_member(ctx.guild)
if bot_member is None:
await ctx.send(_("I couldn't verify my permissions right now."))
return
perms = category.permissions_for(bot_member)
missing = []
if not perms.view_channel:
missing.append("view_channel")
if not perms.connect:
missing.append("connect")
if not perms.manage_channels:
missing.append("manage_channels")
if not bot_member.guild_permissions.move_members:
missing.append("move_members")
if missing:
await ctx.send(
_("I can't use this category. Missing bot permissions: {missing}.").format(
missing=", ".join(missing)
)
)
return
await self._ensure_guild_schema(ctx.guild)
await self.config.guild(ctx.guild).target_category_id.set(category.id)
suffix = ""
try:
start_channel_id, current_target_category_id = await self._get_runtime_config(ctx.guild)
if start_channel_id is None:
suffix = " " + _("Now set the start channel with `tempvoice setstart`.")
except Exception:
log.exception(
"TempVoice setcategory post-save validation failed for guild %s (%s).",
ctx.guild,
ctx.guild.id,
)
await ctx.send(
_("Target category set to {category_mention} (`{category_id}`).").format(
category_mention=category.mention,
category_id=category.id,
)
+ suffix
)
except Exception:
log.exception(
"Unexpected exception in tempvoice setcategory for guild %s (%s).",
ctx.guild,
ctx.guild.id,
)
await ctx.send(
_(
"An unexpected error occurred while setting the target category. "
"Check logs for details."
)
)
@tempvoice_group.command(name="cleanup")
async def tempvoice_cleanup(self, ctx: commands.Context) -> None:
"""Clean stale mappings and remove empty temporary channels."""
await self._set_locale_context(ctx.guild)
guild = ctx.guild
current_start_channel_id, target_category_id = await self._get_runtime_config(guild)
channel_owners = await self._get_channel_owners_snapshot(guild)
stale_channel_ids: Set[int] = set()
channels_to_delete: List[discord.VoiceChannel] = []
for channel_id_str in channel_owners:
channel_id = int(channel_id_str)
channel = guild.get_channel(channel_id)
if not isinstance(channel, discord.VoiceChannel):
stale_channel_ids.add(channel_id)
continue
if target_category_id is not None and channel.category_id != target_category_id:
stale_channel_ids.add(channel_id)
continue
if not channel.members:
channels_to_delete.append(channel)
deleted_channel_ids: Set[int] = set()
failed_deletions = 0
for channel in channels_to_delete:
try:
await channel.delete(reason="TempVoice cleanup command: delete empty temporary channel")
deleted_channel_ids.add(channel.id)
except (discord.Forbidden, discord.HTTPException):
failed_deletions += 1
removed_mapping_count = await self._remove_channel_mappings_bulk(
guild,
stale_channel_ids.union(deleted_channel_ids),
)
await ctx.send(
_(
"Cleanup finished. Deleted empty channels: {deleted}. "
"Removed stale mappings: {removed}. Failed deletions: {failed}."
).format(
deleted=len(deleted_channel_ids),
removed=removed_mapping_count,
failed=failed_deletions,
)
)
@tempvoice_group.command(name="setlimit")
@commands.guild_only()
async def tempvoice_setlimit(self, ctx: commands.Context, limit: int) -> None:
"""Set user limit (0-99) on your own temporary voice channel."""
await self._setlimit_impl(ctx, limit)
@commands.command(name="setlimit", hidden=True)
@commands.guild_only()
async def setlimit_alias(self, ctx: commands.Context, limit: int) -> None:
"""Backward-compatible alias for tempvoice setlimit."""
await self._setlimit_impl(ctx, limit)