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