# bot.py

import os as \_os import sys as \_sys import asyncio as \_asyncio import re as \_re from dotenv import load\_dotenv as \_load\_dotenv import openai as \_openai from twitchio.ext import commands as \_commands from gtts import gTTS as \_gTTS import platform as \_platform

## === Load Environment Variables ===

\_load\_dotenv()

## === Retrieve Environment Variables ===

OPENAI\_API\_KEY = \_os.getenv("OPENAI\_API\_KEY") TWITCH\_OAUTH\_TOKEN = \_os.getenv("TWITCH\_OAUTH\_TOKEN") TWITCH\_CHANNEL\_NAME = \_os.getenv("TWITCH\_CHANNEL\_NAME") BLOCKED\_WORDS = \_os.getenv("BLOCKED\_WORDS", "")

## === Validate Environment Variables ===

def \_validate\_env\_vars(): missing\_vars = \[] if not OPENAI\_API\_KEY: missing\_vars.append("OPENAI\_API\_KEY") if not TWITCH\_OAUTH\_TOKEN: missing\_vars.append("TWITCH\_OAUTH\_TOKEN") if not TWITCH\_CHANNEL\_NAME: missing\_vars.append("TWITCH\_CHANNEL\_NAME")

```
if missing_vars:
    _sys.exit(f"Error: Missing environment variables: {', '.join(missing_vars)}")
```

\_validate\_env\_vars()

## === Configure OpenAI ===

\_openai.api\_key = OPENAI\_API\_KEY

## === Configure Text-to-Speech ===

ENABLE\_TTS = True # Enable TTS

## === TTS Function Using gTTS ===

def \_speak(text: str): if text: try: print(f"\[TTS] Converting text to speech: {text}") tts = \_gTTS(text=text, lang='en', tld='co.uk') # Set to British English tts.save("output.mp3")

```
        if _platform.system() == "Darwin":  # macOS
            _os.system("afplay output.mp3")
        else:
            print("[TTS] Unsupported platform for afplay playback.")
    except Exception as _e:
        print(f"[TTS] Error during TTS: {_e}")
```

## === Content Filtering ===

def \_get\_blocked\_words(): return \[\_word.strip().lower() for \_word in BLOCKED\_WORDS.split(',')] if BLOCKED\_WORDS else \[]

BLOCKED\_WORDS\_LIST = \_get\_blocked\_words()

def \_is\_message\_allowed(message: str) -> bool: return not any(\_word in message.lower() for \_word in BLOCKED\_WORDS\_LIST)

## === Conversation History ===

conversation\_history = {}

def \_split\_text(text, max\_length=50): sentences = \_re.split(r'(?<=\[.!?]) +', text) chunks = \[] current\_chunk = ''

```
for sentence in sentences:
    if len(current_chunk) + len(sentence) + 1 <= max_length:
        current_chunk += ' ' + sentence if current_chunk else sentence
    else:
        if current_chunk:
            chunks.append(current_chunk)
        while len(sentence) > max_length:
            chunks.append(sentence[:max_length])
            sentence = sentence[max_length:]
        current_chunk = sentence

if current_chunk:
    chunks.append(current_chunk)

return chunks
```

async def \_generate\_openai\_response(channel: str, user: str, message: str) -> str: if not \_is\_message\_allowed(message): return "I'm sorry, OrbAI does not tolerate that."

```
you dont get this!
```

class \_TwitchBot(\_commands.Bot): def **init**(self): super().**init**( token=TWITCH\_OAUTH\_TOKEN, prefix="!", initial\_channels=\[f"#{TWITCH\_CHANNEL\_NAME.lower()}"] )

```
async def event_ready(self):
    print(f"Logged in as | {self.nick}")
    print(f"User ID is   | {self.user_id}")

async def event_message(self, message):
    if message.author is None or message.author.name.lower() == self.nick.lower():
        return

    if self.nick.lower() in message.content.lower():
        user = message.author.name.lower()
        channel = message.channel.name.lower()

       you dont get this!
       
                await message.channel.send(chunk)
            except Exception as _e:
                print(f"[Error] Unexpected error while sending message chunk: {_e}")

@staticmethod
async def _send_hello(ctx):
    response = f"Hello {ctx.author.name}! How can I assist you today?"
    if ENABLE_TTS and response:
        _speak(response)
    reply_chunks = _split_text(response, max_length=500)
    for chunk in reply_chunks:
        print(f"[Debug] Sending hello chunk: {chunk}")
        try:     
        except Exception as _e:
            print(f"[Error] Unexpected error while sending hello chunk: {_e}")

@commands.command(name='hello')
async def hello(self, ctx):
    await self._send_hello(ctx)

@commands.command(name='joke')
async def joke(self, ctx):
    prompt = "Tell me a funny joke."
   
    chunk: {_e}")

async def close(self):
    print("[Shutdown] Cleaning up resources.")
    await super().close()
```

## === Main Entry Point ===

if **name** == "**main**": \_bot = \_TwitchBot() try: \_bot.run() except KeyboardInterrupt: print("\n\[Shutdown] Bot is shutting down...") except Exception as \_e: print(f"\[Error] Unexpected error: {\_e}")