108 lines
4.6 KiB
Python
108 lines
4.6 KiB
Python
import importlib
|
||
import os
|
||
from typing import Optional, Type, List
|
||
from data.models.TwitchComponent import TwitchComponent
|
||
from twitchio import eventsub
|
||
from twitchio.ext import commands as twitch_commands
|
||
import twitch_bot.config as cfg
|
||
from libs.Db import Db
|
||
|
||
class TwitchBot(twitch_commands.Bot):
|
||
# def __init__(self, restart_controller=None, tokens: Optional[Dict[str, Dict[str, str]]] = None, db: Optional[Db] = None):
|
||
def __init__(self, restart_controller=None, db: Optional[Db] = None):
|
||
super().__init__(
|
||
client_id=cfg.TWITCH_CLIENT_ID,
|
||
client_secret=cfg.TWITCH_CLIENT_SECRET,
|
||
bot_id=cfg.TWITCH_BOT_ID,
|
||
owner_id=cfg.TWITCH_OWNER_ID,
|
||
prefix=cfg.TWITCH_COMMAND_PREFIX
|
||
)
|
||
# Store channel from config for reference/logging. TwitchIO v3 does not expose join_channels.
|
||
self._initial_channel: Optional[str] = cfg.TWITCH_CHANNEL
|
||
|
||
# Expose restart controller for commands (e.g., !restart)
|
||
self.restart_controller = restart_controller
|
||
self.db = db or Db(cfg.DB_CONN_STR, False)
|
||
|
||
|
||
async def setup_hook(self) -> None:
|
||
# Prefer explicit IDs from config; these should be numeric strings from Twitch.
|
||
broadcaster_id = cfg.TWITCH_OWNER_ID
|
||
bot_id = cfg.TWITCH_BOT_ID
|
||
|
||
# If broadcaster_id is missing, we cannot subscribe. Ask the user to provide it.
|
||
if not broadcaster_id:
|
||
print("⚠️ TWITCH_OWNER_ID is not set. Please set it to your broadcaster (channel) numeric user ID.")
|
||
if cfg.TWITCH_CHANNEL:
|
||
print(
|
||
"ℹ️ TWITCH_CHANNEL is set to '", cfg.TWITCH_CHANNEL,
|
||
"'. Provide TWITCH_OWNER_ID for that channel to enable chat subscription.",
|
||
)
|
||
return
|
||
|
||
if not bot_id:
|
||
print("⚠️ TWITCH_BOT_ID is not set. Please set it to the numeric user ID of the bot account.")
|
||
return
|
||
|
||
try:
|
||
payload = eventsub.ChatMessageSubscription(
|
||
broadcaster_user_id=broadcaster_id,
|
||
user_id=bot_id,
|
||
)
|
||
await self.subscribe_websocket(payload=payload)
|
||
channel_hint = cfg.TWITCH_CHANNEL or "<unknown>"
|
||
print(f"✅ Subscribed to EventSub ChatMessage for broadcaster {broadcaster_id} (channel hint: {channel_hint})")
|
||
except Exception as e:
|
||
print(f"❌ Failed to subscribe to EventSub ChatMessage: {e}")
|
||
|
||
|
||
async def load_commands(self) -> None:
|
||
components: List[str] = []
|
||
package_name = "twitch_bot.commands"
|
||
|
||
for filename in os.listdir(f"./{package_name.replace('.', '/')}"):
|
||
if filename.endswith(".py") and not filename.startswith("_"):
|
||
components.append(f"{package_name}.{filename[:-3]}")
|
||
|
||
for cmp in components:
|
||
try:
|
||
module = importlib.import_module(cmp)
|
||
except Exception as e:
|
||
print(f"⚠️ Failed to import {cmp}: {e}")
|
||
continue
|
||
|
||
# Find a component class to load
|
||
component_cls: Optional[Type[twitch_commands.Component]] = None
|
||
|
||
for attr in module.__dict__.values():
|
||
if isinstance(attr, type) and issubclass(attr, twitch_commands.Component):
|
||
component_cls = attr
|
||
break
|
||
|
||
if component_cls is None:
|
||
print(f"⚠️ No Component found in {cmp}; skipping.")
|
||
continue
|
||
|
||
component_record = self.db.session.query(TwitchComponent).filter_by(name=cmp).first()
|
||
if not component_record:
|
||
self.db.session.add(TwitchComponent(name=cmp))
|
||
self.db.session.commit()
|
||
component_record = self.db.session.query(TwitchComponent).filter_by(name=cmp).first()
|
||
|
||
if component_record.active:
|
||
instance = None
|
||
try:
|
||
instance = component_cls(self) # type: ignore[call-arg]
|
||
except Exception as e:
|
||
print(f"⚠️ Failed to instantiate {component_cls.__name__} in {cmp}: {e}")
|
||
continue
|
||
|
||
try:
|
||
await self.add_component(instance) # type: ignore[arg-type]
|
||
print(f"✅ Loaded Twitch component from {cmp}: {component_cls.__name__}")
|
||
except Exception as e:
|
||
print(f"⚠️ Failed to add component from {cmp}: {e}")
|
||
else:
|
||
print(f"⏭️ Component inactive in DB, skipping: {cmp}")
|
||
|
||
print(f"bot commands:", self.commands) |