bot.py

Code used to for most of the backend of Orb AI.

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}")