From f6f1358c1bd8d2042219eb3bca5b6c840e4657b5 Mon Sep 17 00:00:00 2001 From: zv Date: Sat, 18 Apr 2026 19:03:17 +0200 Subject: [PATCH] Initial commit --- .gitignore | 29 ++ README.md | 98 +++++ __init__.py | 5 + info.json | 23 ++ locales/pl-PL.po | 131 +++++++ tempvoice.py | 933 +++++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 1219 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 __init__.py create mode 100644 info.json create mode 100644 locales/pl-PL.po create mode 100644 tempvoice.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..23c994d --- /dev/null +++ b/.gitignore @@ -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 diff --git a/README.md b/README.md new file mode 100644 index 0000000..5988d16 --- /dev/null +++ b/README.md @@ -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 +[p]tempvoice setcategory +``` + +Check settings: + +```text +[p]tempvoice settings +``` + +## Commands + +### Admin (`manage_guild` or admin) + +- `[p]tempvoice settings` - show config and current state +- `[p]tempvoice setstart ` - set join-to-create channel +- `[p]tempvoice setcategory ` - 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 +``` + diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..f0dd67e --- /dev/null +++ b/__init__.py @@ -0,0 +1,5 @@ +from .tempvoice import TempVoice + + +async def setup(bot): + await bot.add_cog(TempVoice(bot)) diff --git a/info.json b/info.json new file mode 100644 index 0000000..b131c40 --- /dev/null +++ b/info.json @@ -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" +} diff --git a/locales/pl-PL.po b/locales/pl-PL.po new file mode 100644 index 0000000..3c65ce6 --- /dev/null +++ b/locales/pl-PL.po @@ -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 ` and `tempvoice setcategory `" +msgstr "- Dokończ konfigurację komendami: `tempvoice setstart ` oraz `tempvoice setcategory `" + +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." diff --git a/tempvoice.py b/tempvoice.py new file mode 100644 index 0000000..05f39f8 --- /dev/null +++ b/tempvoice.py @@ -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 ` " + "and `tempvoice setcategory `" + ) + ) + + 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)