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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
# 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 os
import json
import discord
import aiofiles
import logging
import tempfile
from discord.ext import commands
from gtts import gTTS
import asyncio
logger = logging.getLogger(__name__)
class TTS(commands.Cog):
def __init__(self, bot):
self.bot = bot
async def _load_settings(self):
"""Load volume settings from settings.json"""
volume = 1.0
try:
async with aiofiles.open('settings.json') as f:
content = await f.read()
data = json.loads(content)
volume = data.get('volume', 1.0)
except (json.JSONDecodeError, KeyError, FileNotFoundError) as e:
logger.error(f"Error loading settings: {e}")
return volume
async def _handle_file_output(self, ctx, temp_path, text):
"""Handle TTS file output"""
try:
await ctx.respond(file=discord.File(temp_path, filename='tts.mp3'))
logger.info(f"Sent TTS for text: {text}")
except Exception as e:
logger.error(f"Error sending file: {e}")
await ctx.respond("eror :c", ephemeral=True)
def _get_voice_manager(self):
from cogs.voice import get_voice_manager
return get_voice_manager(self.bot)
async def _get_or_connect_voice_client(self, channel):
"""Get existing voice client or connect to new channel"""
voice_mgr = self._get_voice_manager()
return await voice_mgr.join_channel(channel)
async def _handle_voice_output(self, ctx, temp_path, text, volume):
"""Handle TTS voice channel output"""
await ctx.respond(content="\u200b", ephemeral=True)
await ctx.interaction.delete_original_response()
# Check if the user is in a voice channel
if not (ctx.author.voice and ctx.author.voice.channel):
logger.info("User is not in a voice channel")
return
channel = ctx.author.voice.channel
try:
voice_client = await self._get_or_connect_voice_client(channel)
if not voice_client:
await ctx.respond("Failed to connect to voice channel", ephemeral=True)
return
# Play the TTS audio with volume adjustment
audio_source = discord.FFmpegPCMAudio(
temp_path,
options=f'-filter:a "volume={volume}"'
)
voice_client.play(audio_source)
logger.info(f"Playing TTS: {text}")
# Wait for the audio to finish playing
while voice_client.is_playing():
await asyncio.sleep(0.1)
except Exception as e:
logger.error(f"Error in voice channel handling: {e}")
await ctx.respond("eror :c", ephemeral=True)
async def _cleanup_temp_file(self, temp_path):
"""Clean up temporary file"""
try:
if temp_path and os.path.exists(temp_path):
os.unlink(temp_path)
except Exception as e:
logger.error(f"Error cleaning up temp file {temp_path}: {e}")
@commands.slash_command(name="tts", description="Generuje TTS dla danego tekstu")
async def tts(self, ctx,
text: str = discord.Option(str, "Tekst do przeczytania", required=True),
output: str = discord.Option(str, "Wybierz sposób odtwarzania", choices=["file", "vc"], default="vc")
):
await ctx.defer()
# Create temporary file in temp folder
temp_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'temp')
os.makedirs(temp_dir, exist_ok=True)
temp_path = os.path.join(temp_dir, f"tts_{os.getpid()}_{asyncio.get_event_loop().time()}.mp3")
try:
# Generate and save TTS to temp file
tts = gTTS(text=text, lang='pl')
tts.save(temp_path)
# Load settings
volume = await self._load_settings()
# Handle output based on user choice
if output == "file":
await self._handle_file_output(ctx, temp_path, text)
elif output == "vc":
await self._handle_voice_output(ctx, temp_path, text, volume)
finally:
await self._cleanup_temp_file(temp_path)
def setup(bot):
bot.add_cog(TTS(bot))
|
