Coverage for integrations / channels / commands / registry.py: 48.9%

1""" 

2Command Registry 

3 

4Provides the command registration and lookup system. 

5Ported from HevolveBot's commands-registry pattern. 

6""" 

7 

8from dataclasses import dataclass, field 

9from enum import Enum 

10from typing import ( 

11 Any, 

12 Callable, 

13 Dict, 

14 List, 

15 Optional, 

16 Set, 

17 Union, 

18) 

19import re 

20import logging 

21 

22logger = logging.getLogger(__name__) 

23 

24 

25class CommandScope(Enum): 

26 """Scope where command is available.""" 

27 TEXT = "text" # Only available as text command (e.g., /help) 

28 NATIVE = "native" # Only available as native command (platform-specific) 

29 BOTH = "both" # Available in both text and native forms 

30 

31 

32class CommandCategory(Enum): 

33 """Category of command for grouping in help.""" 

34 SESSION = "session" 

35 OPTIONS = "options" 

36 STATUS = "status" 

37 MANAGEMENT = "management" 

38 MEDIA = "media" 

39 TOOLS = "tools" 

40 CUSTOM = "custom" 

41 

42 

43@dataclass 

44class CommandDefinition: 

45 """ 

46 Definition of a command. 

47 

48 Attributes: 

49 key: Unique command identifier (e.g., "help", "model") 

50 description: Human-readable description 

51 handler: Async callable that handles the command 

52 aliases: List of text aliases (e.g., ["/help", "/h", "/?"]) 

53 native_name: Name for native platform commands 

54 scope: Where the command is available 

55 category: Category for help grouping 

56 accepts_args: Whether command accepts arguments 

57 hidden: Whether to hide from help listings 

58 enabled: Whether command is currently enabled 

59 metadata: Additional command metadata 

60 """ 

61 key: str 

62 description: str 

63 handler: Optional[Callable] = None 

64 aliases: List[str] = field(default_factory=list) 

65 native_name: Optional[str] = None 

66 scope: CommandScope = CommandScope.BOTH 

67 category: CommandCategory = CommandCategory.CUSTOM 

68 accepts_args: bool = False 

69 hidden: bool = False 

70 enabled: bool = True 

71 metadata: Dict[str, Any] = field(default_factory=dict) 

72 

73 def __post_init__(self): 

74 # Ensure aliases are normalized 

75 normalized = [] 

76 for alias in self.aliases: 

77 alias = alias.strip().lower() 

78 if alias and alias not in normalized: 

79 normalized.append(alias) 

80 self.aliases = normalized 

81 

82 # Auto-add /key as alias if no aliases provided 

83 if not self.aliases and self.scope != CommandScope.NATIVE: 

84 self.aliases = [f"/{self.key}"] 

85 

86 @property 

87 def primary_alias(self) -> str: 

88 """Get the primary text alias.""" 

89 if self.aliases: 

90 return self.aliases[0] 

91 return f"/{self.key}" 

92 

93 

94class CommandRegistry: 

95 """ 

96 Registry for managing commands. 

97 

98 Provides: 

99 - Command registration and unregistration 

100 - Alias resolution 

101 - Command lookup by key or alias 

102 - Category-based listing 

103 """ 

104 

105 def __init__(self): 

106 self._commands: Dict[str, CommandDefinition] = {} 

107 self._alias_map: Dict[str, str] = {} # alias -> key 

108 self._native_map: Dict[str, str] = {} # native_name -> key 

109 

110 def register(self, command: CommandDefinition) -> None: 

111 """ 

112 Register a command. 

113 

114 Args: 

115 command: Command definition to register 

116 

117 Raises: 

118 ValueError: If command key or alias conflicts with existing 

119 """ 

120 # Check for key conflict 

121 if command.key in self._commands: 

122 raise ValueError(f"Command key already registered: {command.key}") 

123 

124 # Check for alias conflicts 

125 for alias in command.aliases: 

126 normalized = alias.strip().lower() 

127 if normalized in self._alias_map: 

128 existing_key = self._alias_map[normalized] 

129 raise ValueError( 

130 f"Alias '{alias}' already registered for command '{existing_key}'" 

131 ) 

132 

133 # Check native name conflict 

134 if command.native_name: 

135 normalized = command.native_name.strip().lower() 

136 if normalized in self._native_map: 

137 existing_key = self._native_map[normalized] 

138 raise ValueError( 

139 f"Native name '{command.native_name}' already registered for command '{existing_key}'" 

140 ) 

141 

142 # Register command 

143 self._commands[command.key] = command 

144 

145 # Register aliases 

146 for alias in command.aliases: 

147 normalized = alias.strip().lower() 

148 self._alias_map[normalized] = command.key 

149 

150 # Register native name 

151 if command.native_name: 

152 normalized = command.native_name.strip().lower() 

153 self._native_map[normalized] = command.key 

154 

155 logger.debug(f"Registered command: {command.key}") 

156 

157 def unregister(self, key: str) -> bool: 

158 """ 

159 Unregister a command. 

160 

161 Args: 

162 key: Command key to unregister 

163 

164 Returns: 

165 True if command was unregistered, False if not found 

166 """ 

167 if key not in self._commands: 

168 return False 

169 

170 command = self._commands[key] 

171 

172 # Remove aliases 

173 for alias in command.aliases: 

174 normalized = alias.strip().lower() 

175 self._alias_map.pop(normalized, None) 

176 

177 # Remove native name 

178 if command.native_name: 

179 normalized = command.native_name.strip().lower() 

180 self._native_map.pop(normalized, None) 

181 

182 # Remove command 

183 del self._commands[key] 

184 

185 logger.debug(f"Unregistered command: {key}") 

186 return True 

187 

188 def get(self, key: str) -> Optional[CommandDefinition]: 

189 """Get command by key.""" 

190 return self._commands.get(key) 

191 

192 def get_by_alias(self, alias: str) -> Optional[CommandDefinition]: 

193 """ 

194 Get command by text alias. 

195 

196 Args: 

197 alias: Text alias (e.g., "/help" or "help") 

198 

199 Returns: 

200 CommandDefinition if found, None otherwise 

201 """ 

202 normalized = alias.strip().lower() 

203 

204 # Try direct lookup 

205 key = self._alias_map.get(normalized) 

206 if key: 

207 return self._commands.get(key) 

208 

209 # Try with leading slash 

210 if not normalized.startswith("/"): 

211 key = self._alias_map.get(f"/{normalized}") 

212 if key: 

213 return self._commands.get(key) 

214 

215 return None 

216 

217 def get_by_native_name(self, name: str) -> Optional[CommandDefinition]: 

218 """ 

219 Get command by native name. 

220 

221 Args: 

222 name: Native command name 

223 

224 Returns: 

225 CommandDefinition if found, None otherwise 

226 """ 

227 normalized = name.strip().lower() 

228 key = self._native_map.get(normalized) 

229 if key: 

230 return self._commands.get(key) 

231 return None 

232 

233 def resolve_alias(self, alias: str) -> Optional[str]: 

234 """ 

235 Resolve an alias to its command key. 

236 

237 Args: 

238 alias: Text alias 

239 

240 Returns: 

241 Command key if found, None otherwise 

242 """ 

243 normalized = alias.strip().lower() 

244 

245 # Try direct lookup 

246 if normalized in self._alias_map: 

247 return self._alias_map[normalized] 

248 

249 # Try with leading slash 

250 if not normalized.startswith("/"): 

251 prefixed = f"/{normalized}" 

252 if prefixed in self._alias_map: 

253 return self._alias_map[prefixed] 

254 

255 return None 

256 

257 def add_alias(self, key: str, alias: str) -> bool: 

258 """ 

259 Add an alias to an existing command. 

260 

261 Args: 

262 key: Command key 

263 alias: New alias to add 

264 

265 Returns: 

266 True if alias added, False if command not found or alias conflicts 

267 """ 

268 if key not in self._commands: 

269 return False 

270 

271 normalized = alias.strip().lower() 

272 

273 # Check for conflict 

274 if normalized in self._alias_map: 

275 return False 

276 

277 # Add alias 

278 self._alias_map[normalized] = key 

279 self._commands[key].aliases.append(normalized) 

280 

281 return True 

282 

283 def remove_alias(self, alias: str) -> bool: 

284 """ 

285 Remove an alias. 

286 

287 Args: 

288 alias: Alias to remove 

289 

290 Returns: 

291 True if removed, False if not found 

292 """ 

293 normalized = alias.strip().lower() 

294 

295 if normalized not in self._alias_map: 

296 return False 

297 

298 key = self._alias_map[normalized] 

299 

300 # Don't remove if it's the only alias 

301 command = self._commands.get(key) 

302 if command and len(command.aliases) <= 1: 

303 return False 

304 

305 # Remove alias 

306 del self._alias_map[normalized] 

307 if command: 

308 command.aliases = [a for a in command.aliases if a != normalized] 

309 

310 return True 

311 

312 def list_commands( 

313 self, 

314 category: Optional[CommandCategory] = None, 

315 scope: Optional[CommandScope] = None, 

316 include_hidden: bool = False, 

317 include_disabled: bool = False, 

318 ) -> List[CommandDefinition]: 

319 """ 

320 List registered commands. 

321 

322 Args: 

323 category: Filter by category 

324 scope: Filter by scope 

325 include_hidden: Include hidden commands 

326 include_disabled: Include disabled commands 

327 

328 Returns: 

329 List of matching command definitions 

330 """ 

331 commands = [] 

332 

333 for command in self._commands.values(): 

334 # Filter by enabled 

335 if not include_disabled and not command.enabled: 

336 continue 

337 

338 # Filter by hidden 

339 if not include_hidden and command.hidden: 

340 continue 

341 

342 # Filter by category 

343 if category is not None and command.category != category: 

344 continue 

345 

346 # Filter by scope 

347 if scope is not None and command.scope != scope: 

348 continue 

349 

350 commands.append(command) 

351 

352 # Sort by key 

353 commands.sort(key=lambda c: c.key) 

354 

355 return commands 

356 

357 def list_aliases(self) -> Dict[str, str]: 

358 """Get all aliases mapped to command keys.""" 

359 return dict(self._alias_map) 

360 

361 def list_native_names(self) -> Dict[str, str]: 

362 """Get all native names mapped to command keys.""" 

363 return dict(self._native_map) 

364 

365 def has_command(self, key: str) -> bool: 

366 """Check if a command is registered.""" 

367 return key in self._commands 

368 

369 def has_alias(self, alias: str) -> bool: 

370 """Check if an alias is registered.""" 

371 normalized = alias.strip().lower() 

372 return normalized in self._alias_map 

373 

374 def enable_command(self, key: str) -> bool: 

375 """Enable a command.""" 

376 if key in self._commands: 

377 self._commands[key].enabled = True 

378 return True 

379 return False 

380 

381 def disable_command(self, key: str) -> bool: 

382 """Disable a command.""" 

383 if key in self._commands: 

384 self._commands[key].enabled = False 

385 return True 

386 return False 

387 

388 def clear(self) -> None: 

389 """Clear all registered commands.""" 

390 self._commands.clear() 

391 self._alias_map.clear() 

392 self._native_map.clear() 

393 

394 def __len__(self) -> int: 

395 """Get number of registered commands.""" 

396 return len(self._commands) 

397 

398 def __contains__(self, key: str) -> bool: 

399 """Check if command key is registered.""" 

400 return key in self._commands 

401 

402 def __iter__(self): 

403 """Iterate over command definitions.""" 

404 return iter(self._commands.values()) 

405 

406 

407# Global registry instance 

408_global_registry: Optional[CommandRegistry] = None 

409 

410 

411def get_command_registry() -> CommandRegistry: 

412 """Get the global command registry.""" 

413 global _global_registry 

414 if _global_registry is None: 

415 _global_registry = CommandRegistry() 

416 return _global_registry 

417 

418 

419def reset_command_registry() -> None: 

420 """Reset the global command registry.""" 

421 global _global_registry 

422 _global_registry = None