1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
|
# This file is part of Senator Sperma
# Licensed under the GPL-3.0 License
# Copyright (C) 2026 Salmonik @ salmonik@tuta.io
# https://codeberg.org/salmonik/senator-sperma
import discord
from discord.ext import commands
import logging
import asyncio
logger = logging.getLogger(__name__)
class Voice(commands.Cog):
def __init__(self, bot):
self.bot = bot
async def _validate_user_voice_state(self, ctx):
"""Validate that user is in a voice channel."""
if not ctx.author.voice or not ctx.author.voice.channel:
await ctx.send("You need to be in a voice channel first!")
return False
return True
async def _create_voice_client(self, ctx):
"""Create a new voice client connection."""
try:
voice_client = await ctx.author.voice.channel.connect(timeout=30.0, reconnect=True)
logger.info(f"Connected to voice channel: {ctx.author.voice.channel}")
return voice_client
except asyncio.TimeoutError:
await ctx.send("Connection to voice channel timed out.")
return None
except discord.ClientException as e:
logger.error(f"Failed to connect to voice channel: {e}")
await ctx.send("Failed to connect to voice channel. Please try again.", ephemeral=True)
return None
async def _move_voice_client(self, voice_client, ctx):
"""Move existing voice client to user's channel."""
try:
await voice_client.move_to(ctx.author.voice.channel)
logger.info(f"Moved to voice channel: {ctx.author.voice.channel}")
return voice_client
except asyncio.TimeoutError:
await ctx.send("Connection to voice channel timed out.")
return None
except discord.ClientException as e:
logger.error(f"Failed to move to voice channel: {e}")
await ctx.send("Failed to move to your voice channel. Please try again.", ephemeral=True)
return None
async def _verify_connection(self, voice_client):
"""Verify and repair voice client connection if needed."""
if not voice_client.is_connected():
try:
await voice_client.connect(reconnect=True, timeout=10.0)
return voice_client
except Exception as e:
logger.error(f"Failed to reconnect to voice: {e}")
return None
return voice_client
async def ensure_voice_client(self, ctx):
"""Ensure we have a valid voice client connection."""
if not await self._validate_user_voice_state(ctx):
return None
voice_client = discord.utils.get(self.bot.voice_clients, guild=ctx.guild)
# If no voice client exists, create one
if not voice_client or not voice_client.is_connected():
return await self._create_voice_client(ctx)
# If we're in a different channel, move to the user's channel
if voice_client.channel != ctx.author.voice.channel:
return await self._move_voice_client(voice_client, ctx)
# Verify the connection is healthy
return await self._verify_connection(voice_client)
async def connect(self, ctx):
"""Connect to the user's voice channel with improved connection handling."""
voice_client = await self.ensure_voice_client(ctx)
if not voice_client:
return None
return voice_client
def setup(bot):
bot.add_cog(Voice(bot))
|
