Spaces:
Sleeping
Sleeping
File size: 5,931 Bytes
89b6a21 127bfda 89b6a21 37ee895 89b6a21 37ee895 89b6a21 37ee895 89b6a21 3074b3f 89b6a21 37ee895 89b6a21 127bfda 89b6a21 37ee895 89b6a21 37ee895 89b6a21 37ee895 89b6a21 37ee895 89b6a21 37ee895 89b6a21 3074b3f 89b6a21 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 2641e6e 127bfda 89b6a21 2641e6e 89b6a21 127bfda 37ee895 89b6a21 37ee895 89b6a21 37ee895 | 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 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | import asyncio
import os
from bot.autonomy_runtime import autonomy_runtime
from bot.clip_jobs_runtime import clip_jobs
from bot.eventsub_runtime import ByteBot
from bot.irc_runtime import IrcByteBot
from bot.observability import observability
from bot.persistence_layer import persistence
from bot.runtime_config import (
TWITCH_BOT_LOGIN,
TWITCH_CHANNEL_LOGIN,
TWITCH_CHANNEL_LOGINS_RAW,
TWITCH_CLIENT_ID,
TWITCH_CLIENT_SECRET_INLINE,
TWITCH_CLIENT_SECRET_NAME,
TWITCH_IRC_HOST,
TWITCH_IRC_PORT,
TWITCH_IRC_TLS,
TWITCH_TOKEN_REFRESH_MARGIN_SECONDS,
logger,
)
from bot.sentiment_engine import sentiment_engine
from bot.status_runtime import parse_channel_logins
from bot.twitch_tokens import TwitchTokenManager, TwitchTokenManagerSettings
def get_secret(secret_name: str = "twitch-client-secret") -> str:
value = os.environ.get("TWITCH_CLIENT_SECRET") or ""
if not value:
raise RuntimeError(f"Secret não configurado: {secret_name}")
return value
def require_env(name: str) -> str:
value = (os.environ.get(name) or "").strip()
if not value:
raise RuntimeError(f"Variavel obrigatoria ausente: {name}")
return value
async def resolve_irc_channel_logins() -> list[str]:
"""Resolve a lista de canais para entrar (Supabase com Fallback para ENV)."""
# Tentativa 1: Supabase (PersistenceLayer)
db_channels = await persistence.get_active_channels()
if db_channels:
return db_channels
# Tentativa 2: Variáveis de Ambiente (Legado/Fallback)
explicit_channels = parse_channel_logins(TWITCH_CHANNEL_LOGINS_RAW)
if explicit_channels:
return explicit_channels
fallback_channels = parse_channel_logins(TWITCH_CHANNEL_LOGIN)
if fallback_channels:
return fallback_channels
required_single_channel = require_env("TWITCH_CHANNEL_LOGIN")
required_channels = parse_channel_logins(required_single_channel)
if not required_channels:
raise RuntimeError(
"TWITCH_CHANNEL_LOGIN invalido. Use login Twitch sem # e com 3-25 caracteres."
)
return required_channels
def resolve_client_secret_for_irc_refresh() -> str:
if TWITCH_CLIENT_SECRET_INLINE:
return TWITCH_CLIENT_SECRET_INLINE
secret_name = TWITCH_CLIENT_SECRET_NAME or "twitch-client-secret"
try:
return get_secret(secret_name=secret_name)
except RuntimeError as error:
logger.warning("Falha ao carregar secret '%s': %s", secret_name, error)
return ""
except Exception as error:
logger.warning("Falha ao carregar secret '%s': %s", secret_name, error)
return ""
def build_irc_token_manager() -> TwitchTokenManager:
user_token = require_env("TWITCH_USER_TOKEN")
refresh_token = os.environ.get("TWITCH_REFRESH_TOKEN")
client_id = TWITCH_CLIENT_ID or require_env("TWITCH_CLIENT_ID")
client_secret = resolve_client_secret_for_irc_refresh()
return TwitchTokenManager(
access_token=user_token,
refresh_token=refresh_token,
client_id=client_id,
client_secret=client_secret or "",
refresh_margin_seconds=TWITCH_TOKEN_REFRESH_MARGIN_SECONDS,
settings=TwitchTokenManagerSettings(),
observability=observability,
logger=logger,
)
def run_irc_mode() -> None:
try:
from bot.logic import context_manager
from bot.runtime_config import irc_channel_control
token_manager = build_irc_token_manager()
async def run_with_channel_control() -> None:
# Injeta o loop para persistência cross-thread (Dashboard)
running_loop = asyncio.get_running_loop()
context_manager.set_main_loop(running_loop)
# Sincroniza canais permitidos (Supabase ou ENV)
channel_logins = await resolve_irc_channel_logins()
bot = IrcByteBot(
host=TWITCH_IRC_HOST,
port=TWITCH_IRC_PORT,
use_tls=TWITCH_IRC_TLS,
bot_login=TWITCH_BOT_LOGIN or require_env("TWITCH_BOT_LOGIN"),
channel_logins=channel_logins,
token_manager=token_manager,
)
irc_channel_control.bind(loop=running_loop, bot=bot)
async def verify_clips_auth_loop() -> None:
while True:
try:
await token_manager.validate_clips_auth()
except Exception as e:
logger.error("Falha ao validar auth de clips: %s", e)
await asyncio.sleep(3600)
# Start background tasks
asyncio.create_task(verify_clips_auth_loop())
# Fase 4: Loop de limpeza de memoria (Context + Sentiment)
asyncio.create_task(context_manager.start_cleanup_loop())
try:
await bot.run_forever()
finally:
clip_jobs.stop()
autonomy_runtime.unbind()
irc_channel_control.unbind()
asyncio.run(run_with_channel_control())
except Exception as error:
logger.critical("Fatal Byte Error (IRC): %s", error)
observability.record_error(category="fatal_irc", details=str(error))
def run_eventsub_mode() -> None:
async def run_async() -> None:
from bot.logic import context_manager
context_manager.set_main_loop(asyncio.get_running_loop())
require_env("TWITCH_CLIENT_ID")
require_env("TWITCH_BOT_ID")
require_env("TWITCH_CHANNEL_ID")
bot = ByteBot(client_secret=get_secret())
try:
await bot.run()
finally:
autonomy_runtime.unbind()
try:
asyncio.run(run_async())
except Exception as error:
logger.critical("Fatal Byte Error (EventSub): %s", error)
observability.record_error(category="fatal_eventsub", details=str(error))
|