Coverage for bot/bootstrap_runtime.py: 71%

1import asyncio 

2import os 

3 

4from bot.autonomy_runtime import autonomy_runtime 

5from bot.clip_jobs_runtime import clip_jobs 

6from bot.eventsub_runtime import ByteBot 

7from bot.irc_runtime import IrcByteBot 

8from bot.observability import observability 

9from bot.runtime_config import ( 

10 CLIENT_ID, 

11 TWITCH_BOT_LOGIN, 

12 TWITCH_CHANNEL_LOGIN, 

13 TWITCH_CHANNEL_LOGINS_RAW, 

14 TWITCH_CLIENT_SECRET_INLINE, 

15 TWITCH_CLIENT_SECRET_NAME, 

16 TWITCH_IRC_HOST, 

17 TWITCH_IRC_PORT, 

18 TWITCH_IRC_TLS, 

19 TWITCH_REFRESH_TOKEN, 

20 TWITCH_TOKEN_REFRESH_MARGIN_SECONDS, 

21 TWITCH_TOKEN_REFRESH_TIMEOUT_SECONDS, 

22 TWITCH_TOKEN_VALIDATE_TIMEOUT_SECONDS, 

23 TWITCH_USER_TOKEN, 

24 irc_channel_control, 

25 logger, 

26) 

27from bot.status_runtime import parse_channel_logins 

28from bot.twitch_tokens import TwitchTokenManager, TwitchTokenManagerSettings 

29 

30 

31def get_secret(secret_name: str = "twitch-client-secret") -> str: 

32 # Ler do ENV usando o nome do secret 

33 env_key = secret_name.upper().replace("-", "_") 

34 val = os.environ.get(env_key) 

35 if val: 

36 return val 

37 raise RuntimeError(f"Secret {env_key} not in env.") 

38 

39 

40def require_env(name: str) -> str: 

41 value = (os.environ.get(name) or "").strip() 

42 if not value: 

43 raise RuntimeError(f"Variavel obrigatoria ausente: {name}") 

44 return value 

45 

46 

47def resolve_irc_channel_logins() -> list[str]: 

48 explicit_channels = parse_channel_logins(TWITCH_CHANNEL_LOGINS_RAW) 

49 if explicit_channels: 

50 return explicit_channels 

51 fallback_channels = parse_channel_logins(TWITCH_CHANNEL_LOGIN) 

52 if fallback_channels: 

53 return fallback_channels 

54 

55 required_single_channel = require_env("TWITCH_CHANNEL_LOGIN") 

56 required_channels = parse_channel_logins(required_single_channel) 

57 if not required_channels: 

58 raise RuntimeError( 

59 "TWITCH_CHANNEL_LOGIN invalido. Use login Twitch sem # e com 3-25 caracteres." 

60 ) 

61 return required_channels 

62 

63 

64def resolve_client_secret_for_irc_refresh() -> str: 

65 if TWITCH_CLIENT_SECRET_INLINE: 

66 return TWITCH_CLIENT_SECRET_INLINE 

67 

68 secret_name = TWITCH_CLIENT_SECRET_NAME or "twitch-client-secret" 

69 try: 

70 return get_secret(secret_name=secret_name) 

71 except Exception as error: 

72 logger.warning( 

73 "Nao foi possivel ler segredo '%s' para refresh automatico: %s", 

74 secret_name, 

75 error, 

76 ) 

77 return "" 

78 

79 

80def build_irc_token_manager() -> TwitchTokenManager: 

81 user_token = TWITCH_USER_TOKEN or require_env("TWITCH_USER_TOKEN") 

82 refresh_token = TWITCH_REFRESH_TOKEN.strip() 

83 settings = TwitchTokenManagerSettings( 

84 validate_timeout_seconds=TWITCH_TOKEN_VALIDATE_TIMEOUT_SECONDS, 

85 refresh_timeout_seconds=TWITCH_TOKEN_REFRESH_TIMEOUT_SECONDS, 

86 ) 

87 if not refresh_token: 

88 return TwitchTokenManager( 

89 access_token=user_token, 

90 settings=settings, 

91 observability=observability, 

92 logger=logger, 

93 ) 

94 

95 client_id = CLIENT_ID or require_env("TWITCH_CLIENT_ID") 

96 client_secret = resolve_client_secret_for_irc_refresh() 

97 if not client_secret: 

98 logger.warning( 

99 "TWITCH_REFRESH_TOKEN definido, mas TWITCH_CLIENT_SECRET nao encontrado. " 

100 "Refresh automatico do token nao funcionara. Configure TWITCH_CLIENT_SECRET na Host." 

101 ) 

102 

103 return TwitchTokenManager( 

104 access_token=user_token, 

105 refresh_token=refresh_token, 

106 client_id=client_id, 

107 client_secret=client_secret or "", 

108 refresh_margin_seconds=TWITCH_TOKEN_REFRESH_MARGIN_SECONDS, 

109 settings=settings, 

110 observability=observability, 

111 logger=logger, 

112 ) 

113 

114 

115def run_irc_mode() -> None: 

116 try: 

117 token_manager = build_irc_token_manager() 

118 channel_logins = resolve_irc_channel_logins() 

119 bot = IrcByteBot( 

120 host=TWITCH_IRC_HOST, 

121 port=TWITCH_IRC_PORT, 

122 use_tls=TWITCH_IRC_TLS, 

123 bot_login=TWITCH_BOT_LOGIN or require_env("TWITCH_BOT_LOGIN"), 

124 channel_logins=channel_logins, 

125 token_manager=token_manager, 

126 ) 

127 

128 async def run_with_channel_control() -> None: 

129 running_loop = asyncio.get_running_loop() 

130 irc_channel_control.bind(loop=running_loop, bot=bot) 

131 

132 async def _check_clips_auth() -> None: 

133 # Avoid circular import 

134 from bot.control_plane import control_plane 

135 

136 try: 

137 token_valid, has_clips_edit = await token_manager.validate_clips_auth() 

138 

139 config = control_plane.get_config() 

140 if config.get("clip_pipeline_enabled") and not has_clips_edit: 

141 logger.warning( 

142 "Pipeline de clips habilitado mas scope 'clips:edit' ausente no token." 

143 ) 

144 

145 except Exception as error: 

146 logger.error("Erro ao validar auth de clips: %s", error) 

147 observability.record_error(category="clips_auth", details=str(error)) 

148 

149 async def verify_clips_auth_loop() -> None: 

150 # Initial check 

151 await asyncio.sleep(5) # Wait for boot 

152 while True: 

153 await _check_clips_auth() 

154 await asyncio.sleep(3600) 

155 

156 async def send_autonomy_chat(text: str) -> None: 

157 await bot.send_reply(text) 

158 

159 autonomy_runtime.bind( 

160 loop=running_loop, 

161 mode="irc", 

162 auto_chat_dispatcher=send_autonomy_chat, 

163 ) 

164 

165 # Clip Jobs Runtime 

166 clip_jobs.bind_token_provider(token_manager.ensure_token_for_connection) 

167 clip_jobs.start(running_loop) 

168 

169 # Start background task 

170 asyncio.create_task(verify_clips_auth_loop()) 

171 

172 try: 

173 await bot.run_forever() 

174 finally: 

175 clip_jobs.stop() 

176 autonomy_runtime.unbind() 

177 irc_channel_control.unbind() 

178 

179 asyncio.run(run_with_channel_control()) 

180 except Exception as error: 

181 logger.error("Erro fatal no modo IRC: %s", error) 

182 observability.record_error(category="bootstrap_irc", details=str(error)) 

183 

184 

185def run_eventsub_mode() -> None: 

186 require_env("TWITCH_CLIENT_ID") 

187 require_env("TWITCH_BOT_ID") 

188 require_env("TWITCH_CHANNEL_ID") 

189 bot = ByteBot(client_secret=get_secret()) 

190 try: 

191 bot.run() 

192 finally: 

193 autonomy_runtime.unbind()