Last active
August 13, 2024 00:36
-
-
Save samwho/a4842454f976c80391a544ec76c95c99 to your computer and use it in GitHub Desktop.
tts.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import io | |
| import logging | |
| import os | |
| import time | |
| from datetime import datetime, timedelta | |
| import gtts | |
| import irc.bot | |
| import sounddevice as sd | |
| import soundfile as sf | |
| # Setup logging | |
| logging.basicConfig(level=logging.DEBUG) | |
| # Directory where MP3 files will be saved | |
| MP3_SAVE_DIR = r"C:\Users\simon\OneDrive\Desktop\Obs Pictures\MP3 Files TTS" | |
| # Ensure the directory exists | |
| os.makedirs(MP3_SAVE_DIR, exist_ok=True) | |
| # Retrieve WebSocket password from environment variable | |
| OBS_WS_PASSWORD = os.getenv("OBS_WS_PASSWORD") | |
| if not OBS_WS_PASSWORD: | |
| logging.error("OBS_WS_PASSWORD environment variable is not set") | |
| raise ValueError("OBS_WS_PASSWORD environment variable is not set") | |
| # A set to keep track of recent messages | |
| recent_messages = set() | |
| # Time period to consider for recent messages (e.g., 1 minute) | |
| recent_message_period = timedelta(minutes=1) | |
| class TwitchBot(irc.bot.SingleServerIRCBot): | |
| def __init__(self, server, port, channel, nickname, token): | |
| irc.bot.SingleServerIRCBot.__init__( | |
| self, [(server, port, "oauth:" + token)], nickname, nickname | |
| ) | |
| self.channel = channel | |
| self.token = token | |
| self.joined = False | |
| def on_welcome(self, connection, event): | |
| logging.info(f"Connected to server. Joining channel: {self.channel}") | |
| connection.join(self.channel) | |
| def on_join(self, connection, event): | |
| if event.target == self.channel: | |
| self.joined = True | |
| logging.info(f"Successfully joined channel: {self.channel}") | |
| else: | |
| logging.error( | |
| f"Failed to join channel: {self.channel}. Event target: {event.target}" | |
| ) | |
| def on_ready(self): | |
| if not self.joined: | |
| logging.error("Bot failed to join the channel.") | |
| else: | |
| logging.info("Bot is ready and joined the channel.") | |
| def on_pubmsg(self, connection, event): | |
| message = event.arguments[0] | |
| user = event.source.nick.lower() # Normalize the username to lowercase | |
| logging.info(f"Message from {user}: {message}") | |
| # Normalize all allowed users to lowercase | |
| normalized_allowed_users = [ | |
| allowed_user.lower() for allowed_user in allowed_users | |
| ] | |
| if user in normalized_allowed_users: | |
| logging.info(f"User {user} is allowed.") | |
| self.read_message(message, user) | |
| else: | |
| logging.warning(f"User {user} is not allowed.") | |
| def read_message(self, message, user): | |
| start_time = time.time() | |
| logging.info(f"Attempting to read message: {message}") | |
| # Check if the message was recently processed | |
| now = datetime.now() | |
| global recent_messages | |
| recent_messages = { | |
| (msg, timestamp) | |
| for msg, timestamp in recent_messages | |
| if now - timestamp < recent_message_period | |
| } | |
| if (message, now) in recent_messages: | |
| logging.info("Message already processed recently. Skipping.") | |
| return | |
| # Add current message to recent messages | |
| recent_messages.add((message, now)) | |
| lang = "en" # Default language | |
| if user == "ghastlywarchief": | |
| logging.info("Selecting language: German") | |
| lang = "de" | |
| elif user == "xeizzeth": | |
| logging.info("Selecting language: US English") | |
| lang = "en" | |
| else: | |
| logging.info("Selecting language: UK English") | |
| lang = "en-uk" | |
| try: | |
| tts = gtts.gTTS(text=message, lang=lang) | |
| with io.BytesIO() as mp3_buffer: | |
| tts.write_to_fp(mp3_buffer) | |
| mp3_buffer.seek(0) | |
| # Define the filename and path | |
| temp_filename = os.path.join( | |
| MP3_SAVE_DIR, f"{datetime.now().strftime('%Y%m%d%H%M%S')}.mp3" | |
| ) | |
| # Save the MP3 file to the specified directory | |
| with open(temp_filename, "wb") as mp3_file: | |
| mp3_file.write(mp3_buffer.getvalue()) | |
| data, fs = sf.read(temp_filename, dtype="float32") | |
| sd.play(data, fs, device="CABLE OUTPUT (VB-Audio Virtual Cable)") | |
| sd.wait() | |
| except Exception as e: | |
| logging.error(f"TTS Error: {e}") | |
| end_time = time.time() | |
| logging.info( | |
| f"Total processing time for message '{message}': {end_time - start_time:.2f} seconds" | |
| ) | |
| def on_error(self, connection, message): | |
| logging.error(f"Error received: {message}") | |
| if __name__ == "__main__": | |
| server = "irc.chat.twitch.tv" | |
| port = 6667 # Use SSL port for a secure connection | |
| channel = "#GhastlyWarChief" | |
| nickname = "Bot_Elias" | |
| allowed_users = ["GhastlyWarChief", "imBoloFake", "Xeizzeth"] | |
| token = os.getenv("TWITCH_OAUTH_TOKEN") | |
| if not token: | |
| logging.error("TWITCH_OAUTH_TOKEN environment variable is not set") | |
| raise ValueError("TWITCH_OAUTH_TOKEN environment variable is not set") | |
| try: | |
| bot = TwitchBot(server, port, channel, nickname, token) | |
| logging.info("Starting bot...") | |
| bot.start() | |
| except Exception as e: | |
| logging.error(f"Error: {e}") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment