Coverage for integrations / channels / commands / mention_gating.py: 46.5%

127 statements  

« prev     ^ index     » next       coverage.py v7.14.0, created at 2026-05-12 04:49 +0000

1""" 

2Mention Gating 

3 

4Provides mention-based command gating for group contexts. 

5Ported from HevolveBot's activation mode pattern. 

6""" 

7 

8from dataclasses import dataclass 

9from enum import Enum 

10from typing import Optional, List, Set 

11import re 

12import logging 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17class MentionMode(Enum): 

18 """Mode for mention gating.""" 

19 ALWAYS = "always" # Always respond, no mention required 

20 MENTION = "mention" # Require @mention or reply to bot 

21 COMMANDS_ONLY = "commands_only" # Only respond to commands 

22 QUIET = "quiet" # Silenced, don't respond unless resumed 

23 

24 

25@dataclass 

26class MentionGateConfig: 

27 """ 

28 Configuration for mention gating. 

29 

30 Attributes: 

31 default_mode: Default mode for groups 

32 dm_mode: Mode for direct messages (usually ALWAYS) 

33 bot_username: Bot's username for @mention detection 

34 bot_user_id: Bot's user ID for reply detection 

35 command_prefixes: Prefixes that count as commands 

36 trigger_words: Additional trigger words (optional) 

37 """ 

38 default_mode: MentionMode = MentionMode.MENTION 

39 dm_mode: MentionMode = MentionMode.ALWAYS 

40 bot_username: Optional[str] = None 

41 bot_user_id: Optional[str] = None 

42 command_prefixes: List[str] = None 

43 trigger_words: List[str] = None 

44 

45 def __post_init__(self): 

46 if self.command_prefixes is None: 

47 self.command_prefixes = ["/", "!"] 

48 if self.trigger_words is None: 

49 self.trigger_words = [] 

50 

51 

52@dataclass 

53class MentionCheckResult: 

54 """Result of checking mention gating.""" 

55 should_respond: bool 

56 reason: str 

57 mode: MentionMode 

58 was_mentioned: bool = False 

59 was_replied_to: bool = False 

60 is_command: bool = False 

61 is_dm: bool = False 

62 

63 

64class MentionGate: 

65 """ 

66 Controls when the bot should respond based on mention gating. 

67 

68 Features: 

69 - Different modes for groups vs DMs 

70 - @mention detection 

71 - Reply-to-bot detection 

72 - Command prefix detection 

73 - Trigger word detection 

74 - Per-chat mode overrides 

75 """ 

76 

77 def __init__(self, config: Optional[MentionGateConfig] = None): 

78 self.config = config or MentionGateConfig() 

79 self._chat_modes: dict[str, MentionMode] = {} 

80 

81 def check( 

82 self, 

83 text: str, 

84 chat_id: str, 

85 is_group: bool = False, 

86 is_bot_mentioned: bool = False, 

87 reply_to_bot: bool = False, 

88 sender_id: Optional[str] = None, 

89 ) -> MentionCheckResult: 

90 """ 

91 Check if the bot should respond to a message. 

92 

93 Args: 

94 text: Message text 

95 chat_id: Chat/channel ID 

96 is_group: Whether this is a group chat 

97 is_bot_mentioned: Whether bot was @mentioned 

98 reply_to_bot: Whether this is a reply to the bot 

99 sender_id: Sender's user ID 

100 

101 Returns: 

102 MentionCheckResult with decision 

103 """ 

104 # Determine active mode 

105 mode = self._get_mode(chat_id, is_group) 

106 

107 # DMs always respond (unless quiet) 

108 if not is_group: 

109 if mode == MentionMode.QUIET: 

110 return MentionCheckResult( 

111 should_respond=False, 

112 reason="Quiet mode active", 

113 mode=mode, 

114 is_dm=True, 

115 ) 

116 return MentionCheckResult( 

117 should_respond=True, 

118 reason="Direct message", 

119 mode=mode, 

120 is_dm=True, 

121 ) 

122 

123 # Check if this is a command 

124 is_command = self._is_command(text) 

125 

126 # Check for mention 

127 was_mentioned = is_bot_mentioned or self._check_mention(text) 

128 

129 # Check for reply to bot 

130 was_replied_to = reply_to_bot 

131 

132 # Check for trigger words 

133 has_trigger = self._check_trigger_words(text) 

134 

135 # Apply mode logic 

136 if mode == MentionMode.ALWAYS: 

137 return MentionCheckResult( 

138 should_respond=True, 

139 reason="Always mode active", 

140 mode=mode, 

141 was_mentioned=was_mentioned, 

142 was_replied_to=was_replied_to, 

143 is_command=is_command, 

144 ) 

145 

146 elif mode == MentionMode.QUIET: 

147 # Only respond to /resume command 

148 if is_command and text.strip().lower().startswith(("/resume", "!resume")): 

149 return MentionCheckResult( 

150 should_respond=True, 

151 reason="Resume command in quiet mode", 

152 mode=mode, 

153 is_command=True, 

154 ) 

155 return MentionCheckResult( 

156 should_respond=False, 

157 reason="Quiet mode active", 

158 mode=mode, 

159 was_mentioned=was_mentioned, 

160 was_replied_to=was_replied_to, 

161 is_command=is_command, 

162 ) 

163 

164 elif mode == MentionMode.COMMANDS_ONLY: 

165 if is_command: 

166 return MentionCheckResult( 

167 should_respond=True, 

168 reason="Command detected", 

169 mode=mode, 

170 is_command=True, 

171 ) 

172 return MentionCheckResult( 

173 should_respond=False, 

174 reason="Commands only mode - not a command", 

175 mode=mode, 

176 was_mentioned=was_mentioned, 

177 was_replied_to=was_replied_to, 

178 ) 

179 

180 elif mode == MentionMode.MENTION: 

181 # Respond if mentioned, replied to, command, or trigger word 

182 if was_mentioned: 

183 return MentionCheckResult( 

184 should_respond=True, 

185 reason="Bot was mentioned", 

186 mode=mode, 

187 was_mentioned=True, 

188 is_command=is_command, 

189 ) 

190 if was_replied_to: 

191 return MentionCheckResult( 

192 should_respond=True, 

193 reason="Reply to bot", 

194 mode=mode, 

195 was_replied_to=True, 

196 is_command=is_command, 

197 ) 

198 if is_command: 

199 return MentionCheckResult( 

200 should_respond=True, 

201 reason="Command detected", 

202 mode=mode, 

203 is_command=True, 

204 ) 

205 if has_trigger: 

206 return MentionCheckResult( 

207 should_respond=True, 

208 reason="Trigger word detected", 

209 mode=mode, 

210 was_mentioned=was_mentioned, 

211 ) 

212 return MentionCheckResult( 

213 should_respond=False, 

214 reason="Mention required - not mentioned", 

215 mode=mode, 

216 ) 

217 

218 # Default: don't respond 

219 return MentionCheckResult( 

220 should_respond=False, 

221 reason="Unknown mode", 

222 mode=mode, 

223 ) 

224 

225 def _get_mode(self, chat_id: str, is_group: bool) -> MentionMode: 

226 """Get the active mode for a chat.""" 

227 # Check for chat-specific override 

228 if chat_id in self._chat_modes: 

229 return self._chat_modes[chat_id] 

230 

231 # Use default based on chat type 

232 if is_group: 

233 return self.config.default_mode 

234 return self.config.dm_mode 

235 

236 def _is_command(self, text: str) -> bool: 

237 """Check if text starts with a command prefix.""" 

238 text = text.strip() 

239 return any(text.startswith(p) for p in self.config.command_prefixes) 

240 

241 def _check_mention(self, text: str) -> bool: 

242 """Check if bot is @mentioned in text.""" 

243 if not self.config.bot_username: 

244 return False 

245 

246 # Case-insensitive mention check 

247 pattern = rf"@{re.escape(self.config.bot_username)}\b" 

248 return bool(re.search(pattern, text, re.IGNORECASE)) 

249 

250 def _check_trigger_words(self, text: str) -> bool: 

251 """Check for trigger words in text.""" 

252 if not self.config.trigger_words: 

253 return False 

254 

255 text_lower = text.lower() 

256 return any(word.lower() in text_lower for word in self.config.trigger_words) 

257 

258 def set_mode(self, chat_id: str, mode: MentionMode) -> None: 

259 """Set mode for a specific chat.""" 

260 self._chat_modes[chat_id] = mode 

261 logger.info(f"Set mention mode for {chat_id}: {mode.value}") 

262 

263 def get_mode(self, chat_id: str, is_group: bool = True) -> MentionMode: 

264 """Get mode for a specific chat.""" 

265 return self._get_mode(chat_id, is_group) 

266 

267 def clear_mode(self, chat_id: str) -> bool: 

268 """Clear mode override for a chat.""" 

269 if chat_id in self._chat_modes: 

270 del self._chat_modes[chat_id] 

271 return True 

272 return False 

273 

274 def quiet(self, chat_id: str) -> None: 

275 """Set chat to quiet mode.""" 

276 self.set_mode(chat_id, MentionMode.QUIET) 

277 

278 def resume(self, chat_id: str) -> None: 

279 """Resume from quiet mode (restore default).""" 

280 self.clear_mode(chat_id) 

281 

282 def set_always(self, chat_id: str) -> None: 

283 """Set chat to always respond mode.""" 

284 self.set_mode(chat_id, MentionMode.ALWAYS) 

285 

286 def set_mention_only(self, chat_id: str) -> None: 

287 """Set chat to mention-only mode.""" 

288 self.set_mode(chat_id, MentionMode.MENTION) 

289 

290 def set_commands_only(self, chat_id: str) -> None: 

291 """Set chat to commands-only mode.""" 

292 self.set_mode(chat_id, MentionMode.COMMANDS_ONLY) 

293 

294 def list_overrides(self) -> dict[str, MentionMode]: 

295 """List all chat mode overrides.""" 

296 return dict(self._chat_modes) 

297 

298 def update_config( 

299 self, 

300 bot_username: Optional[str] = None, 

301 bot_user_id: Optional[str] = None, 

302 default_mode: Optional[MentionMode] = None, 

303 dm_mode: Optional[MentionMode] = None, 

304 trigger_words: Optional[List[str]] = None, 

305 ) -> None: 

306 """Update configuration.""" 

307 if bot_username is not None: 

308 self.config.bot_username = bot_username 

309 if bot_user_id is not None: 

310 self.config.bot_user_id = bot_user_id 

311 if default_mode is not None: 

312 self.config.default_mode = default_mode 

313 if dm_mode is not None: 

314 self.config.dm_mode = dm_mode 

315 if trigger_words is not None: 

316 self.config.trigger_words = trigger_words 

317 

318 

319# Global instance 

320_global_gate: Optional[MentionGate] = None 

321 

322 

323def get_mention_gate() -> MentionGate: 

324 """Get the global mention gate.""" 

325 global _global_gate 

326 if _global_gate is None: 

327 _global_gate = MentionGate() 

328 return _global_gate 

329 

330 

331def reset_mention_gate() -> None: 

332 """Reset the global mention gate.""" 

333 global _global_gate 

334 _global_gate = None