Spaces:
Sleeping
Sleeping
File size: 2,675 Bytes
89b6a21 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | import asyncio
from typing import Any
from bot.irc_connection import IrcConnectionMixin
from bot.irc_handlers import IrcLineHandlersMixin
from bot.irc_management import IrcChannelManagementMixin
from bot.irc_state import IrcChannelStateMixin
from bot.observability import observability
from bot.runtime_config import (
OWNER_ID,
TWITCH_TOKEN_REFRESH_TIMEOUT_SECONDS,
TWITCH_TOKEN_VALIDATE_TIMEOUT_SECONDS,
logger,
)
from bot.status_runtime import normalize_channel_login
from bot.twitch_tokens import TwitchTokenManager, TwitchTokenManagerSettings
class IrcByteBot(
IrcChannelManagementMixin,
IrcChannelStateMixin,
IrcLineHandlersMixin,
IrcConnectionMixin,
):
def __init__(
self,
*,
host: str,
port: int,
use_tls: bool,
bot_login: str,
channel_login: str | None = None,
channel_logins: list[str] | None = None,
user_token: str = "",
token_manager: TwitchTokenManager | None = None,
) -> None:
provided_channels = list(channel_logins or [])
if channel_login:
provided_channels.append(channel_login)
resolved_channels: list[str] = []
seen_channels: set[str] = set()
for candidate in provided_channels:
normalized = normalize_channel_login(candidate)
if not normalized or normalized in seen_channels:
continue
seen_channels.add(normalized)
resolved_channels.append(normalized)
if not resolved_channels:
raise RuntimeError("Defina ao menos 1 canal valido para o modo IRC.")
self.host = host
self.port = port
self.use_tls = use_tls
self.bot_login = bot_login.lower()
self.owner_id = OWNER_ID
self.channel_logins = resolved_channels
self.joined_channels = set(resolved_channels)
self.primary_channel_login = resolved_channels[0]
self.token_manager = token_manager or TwitchTokenManager(
access_token=user_token,
settings=TwitchTokenManagerSettings(
validate_timeout_seconds=TWITCH_TOKEN_VALIDATE_TIMEOUT_SECONDS,
refresh_timeout_seconds=TWITCH_TOKEN_REFRESH_TIMEOUT_SECONDS,
),
observability=observability,
logger=logger,
)
self.reader: asyncio.StreamReader | None = None
self.writer: Any = None
self._line_reader_running = False
self._line_reader_task: asyncio.Task[Any] | None = None
self._pending_join_events: dict[str, asyncio.Event] = {}
self._pending_part_events: dict[str, asyncio.Event] = {}
|