# 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 datetime import json import logging import os import asyncio import discord from discord.ext import commands from gtts import gTTS from cogs.voice import get_voice_manager from cogs.utils.settings import get_volume, save_settings logger = logging.getLogger(__name__) ACTIVITY_FILE = 'temp/voice_activity.json' def load_activity_data(): if os.path.exists(ACTIVITY_FILE): with open(ACTIVITY_FILE, 'r') as f: return json.load(f) return {} def save_activity_data(data): with open(ACTIVITY_FILE, 'w') as f: json.dump(data, f, indent=4) def format_duration(seconds): hours, remainder = divmod(int(seconds), 3600) minutes, seconds = divmod(remainder, 60) if hours > 0: return f"{hours}h {minutes}m {seconds}s" elif minutes > 0: return f"{minutes}m {seconds}s" return f"{seconds}s" class VcLogs(commands.Cog): def __init__(self, bot): self.bot = bot self._voice_manager = None def get_voice_manager(self): if self._voice_manager is None: self._voice_manager = get_voice_manager(self.bot) return self._voice_manager @staticmethod async def handle_state_change(member, channel, state_type, is_enabled, log_channel): """Helper method to handle voice state changes""" state_messages = { 'deaf': { True: f"{member.display_name} jest bity przez ojca", False: f"{member.display_name} uciek od ojca" }, 'mute': { True: f"{member.display_name} idzie robić jajecznice", False: f"{member.display_name} zrobił jajecznice" } } tts_text = state_messages[state_type][is_enabled] return tts_text, channel async def play_tts(self, channel, text): voice_mgr = self.get_voice_manager() temp_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'temp') os.makedirs(temp_dir, exist_ok=True) tts_file = os.path.join(temp_dir, f"temp_tts_{os.getpid()}.mp3") tts = gTTS(text=text, lang='pl') logger.info("saving file") tts.save(tts_file) # Get voice status using our new method voice_status = await voice_mgr.get_voice_status(channel.guild) voice_client = voice_status[0] if voice_status else None is_connected = voice_status[1] if voice_status else False is_playing = voice_status[2] if voice_status else False # Ensure we have a connected voice client if not is_connected: voice_client = await voice_mgr.join_channel(channel) # Update status after joining voice_status = await voice_mgr.get_voice_status(channel.guild) voice_client = voice_status[0] if voice_status else None is_connected = voice_status[1] if voice_status else False is_playing = voice_status[2] if voice_status else False if not is_connected or voice_client is None: logger.error("Failed to join voice channel") if os.path.exists(tts_file): os.remove(tts_file) return volume = get_volume() logger.info(f"connected: {voice_client}") if voice_client and not is_playing: logger.info("preparing to play audio") audio_source = discord.FFmpegPCMAudio(tts_file, options=f'-filter:a "volume={volume}"') voice_client.play(audio_source) while voice_client.is_playing(): await asyncio.sleep(0.5) if os.path.exists(tts_file): os.remove(tts_file) @commands.slash_command( name="log_volume" ) async def volume(self, ctx: discord.ApplicationContext, volume: int): from cogs.utils.settings import get_settings settings = get_settings() settings['volume'] = volume await save_settings() await ctx.followup.send(f"Volume has been set to: {volume}") logger.info(f"Volume changed to: {volume}") @commands.Cog.listener() async def on_voice_state_update(self, member, before, after): log_channel = self.bot.get_channel(1213843195457183885) current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") current_ts = datetime.datetime.now().timestamp() tts_text = None tts_channel = None if member.bot: return user_id = str(member.id) activity_data = load_activity_data() if before.channel is None and after.channel is not None: is_afk = after.self_mute or after.self_deaf activity_data[user_id] = { 'join_time': current_ts, 'total_active': 0, 'total_afk': 0, 'is_afk': is_afk, 'afk_start': current_ts if is_afk else None, 'active_start': current_ts if not is_afk else None, 'channel_name': after.channel.name } save_activity_data(activity_data) message = f"[{current_time}] {member.name} dołączył na {after.channel.name}" tts_text = f"{member.display_name} sie zespawnił" tts_channel = after.channel await log_channel.send(message) logger.info(f"User joined voice channel: {message}") elif before.channel is not None and after.channel is None: if user_id in activity_data: data = activity_data[user_id] if data.get('is_afk') and data.get('afk_start'): afk_duration = current_ts - data['afk_start'] data['total_afk'] += afk_duration elif data.get('active_start'): active_since_start = current_ts - data['active_start'] data['total_active'] += active_since_start active_str = format_duration(data['total_active']) afk_str = format_duration(data['total_afk']) message = ( f"[{current_time}] {member.name} rozłączył się z {before.channel.name}\n" f"Aktywny: {active_str} | AFK: {afk_str}" ) tts_text = f"{member.display_name} jest chujem" tts_channel = before.channel await log_channel.send(message) logger.info(f"User left voice channel: {message}") del activity_data[user_id] save_activity_data(activity_data) else: message = f"[{current_time}] {member.name} rozłączył się z {before.channel.name}" tts_text = f"{member.display_name} jest chujem" tts_channel = before.channel await log_channel.send(message) logger.info(f"User left voice channel: {message}") elif user_id in activity_data: data = activity_data[user_id] if data.get('is_afk') and not (after.self_mute or after.self_deaf): afk_duration = current_ts - data['afk_start'] data['total_afk'] += afk_duration data['afk_start'] = None data['is_afk'] = False data['active_start'] = current_ts save_activity_data(activity_data) afk_str = format_duration(afk_duration) if after.self_mute: activity = "odciszył" else: activity = "odgłuchnął" message = ( f"[{current_time}] {member.name} {activity} się na {after.channel.name}\n" f"AFK: {afk_str}" ) await log_channel.send(message) logger.info(f"User returned from AFK: {message}") elif not data.get('is_afk') and (after.self_mute or after.self_deaf): active_since_start = 0 if data.get('active_start'): active_since_start = current_ts - data['active_start'] data['total_active'] += active_since_start data['afk_start'] = current_ts data['is_afk'] = True data['active_start'] = None save_activity_data(activity_data) if after.self_mute: activity = "wyciszył" else: activity = "ogłuchnął" active_str = format_duration(active_since_start) message = ( f"[{current_time}] {member.name} {activity} się na {after.channel.name}\n" f"Aktywny: {active_str}" ) await log_channel.send(message) logger.info(f"User went AFK: {message}") if before.self_deaf != after.self_deaf: tts_text, tts_channel = await self.handle_state_change( member, after.channel, 'deaf', after.self_deaf, log_channel ) if tts_text and tts_channel: await self.play_tts(tts_channel, tts_text) return if before.self_mute != after.self_mute: tts_text, tts_channel = await self.handle_state_change( member, after.channel, 'mute', after.self_mute, log_channel ) if tts_text and tts_channel: await self.play_tts(tts_channel, tts_text) def setup(bot): bot.add_cog(VcLogs(bot))