Coverage for integrations / channels / plugins / registry.py: 0.0%

1""" 

2Plugin Registry for HevolveBot Integration. 

3 

4Provides a registry for discovering, installing, and managing plugins 

5from various sources. 

6""" 

7 

8import logging 

9import json 

10import hashlib 

11from dataclasses import dataclass, field 

12from typing import Any, Dict, List, Optional 

13from datetime import datetime 

14from enum import Enum 

15import os 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20class PluginSource(Enum): 

21 """Source types for plugins.""" 

22 LOCAL = "local" 

23 REMOTE = "remote" 

24 GIT = "git" 

25 

26 

27@dataclass 

28class PluginInfo: 

29 """Information about an available plugin.""" 

30 name: str 

31 version: str 

32 description: str 

33 author: str 

34 source: PluginSource 

35 source_url: str = "" 

36 dependencies: List[str] = field(default_factory=list) 

37 tags: List[str] = field(default_factory=list) 

38 downloads: int = 0 

39 rating: float = 0.0 

40 checksum: str = "" 

41 created_at: Optional[datetime] = None 

42 updated_at: Optional[datetime] = None 

43 

44 def to_dict(self) -> Dict[str, Any]: 

45 """Convert to dictionary.""" 

46 return { 

47 "name": self.name, 

48 "version": self.version, 

49 "description": self.description, 

50 "author": self.author, 

51 "source": self.source.value, 

52 "source_url": self.source_url, 

53 "dependencies": self.dependencies, 

54 "tags": self.tags, 

55 "downloads": self.downloads, 

56 "rating": self.rating, 

57 "checksum": self.checksum, 

58 "created_at": self.created_at.isoformat() if self.created_at else None, 

59 "updated_at": self.updated_at.isoformat() if self.updated_at else None 

60 } 

61 

62 @classmethod 

63 def from_dict(cls, data: Dict[str, Any]) -> "PluginInfo": 

64 """Create from dictionary.""" 

65 return cls( 

66 name=data["name"], 

67 version=data["version"], 

68 description=data.get("description", ""), 

69 author=data.get("author", ""), 

70 source=PluginSource(data.get("source", "local")), 

71 source_url=data.get("source_url", ""), 

72 dependencies=data.get("dependencies", []), 

73 tags=data.get("tags", []), 

74 downloads=data.get("downloads", 0), 

75 rating=data.get("rating", 0.0), 

76 checksum=data.get("checksum", ""), 

77 created_at=datetime.fromisoformat(data["created_at"]) if data.get("created_at") else None, 

78 updated_at=datetime.fromisoformat(data["updated_at"]) if data.get("updated_at") else None 

79 ) 

80 

81 

82@dataclass 

83class InstalledPlugin: 

84 """Information about an installed plugin.""" 

85 info: PluginInfo 

86 install_path: str 

87 installed_at: datetime 

88 enabled: bool = False 

89 config: Dict[str, Any] = field(default_factory=dict) 

90 

91 

92class PluginRegistry: 

93 """ 

94 Registry for plugin discovery and management. 

95 

96 Handles searching for plugins, installing, uninstalling, 

97 and checking for updates. 

98 """ 

99 

100 def __init__(self, install_dir: str = "./plugins", 

101 registry_url: str = ""): 

102 self._install_dir = install_dir 

103 self._registry_url = registry_url 

104 self._available_plugins: Dict[str, PluginInfo] = {} 

105 self._installed_plugins: Dict[str, InstalledPlugin] = {} 

106 self._cache_file = os.path.join(install_dir, ".registry_cache.json") 

107 

108 # Ensure install directory exists 

109 os.makedirs(install_dir, exist_ok=True) 

110 

111 @property 

112 def install_dir(self) -> str: 

113 """Return the plugin installation directory.""" 

114 return self._install_dir 

115 

116 @property 

117 def available_plugins(self) -> Dict[str, PluginInfo]: 

118 """Return available plugins.""" 

119 return self._available_plugins.copy() 

120 

121 @property 

122 def installed_plugins(self) -> Dict[str, InstalledPlugin]: 

123 """Return installed plugins.""" 

124 return self._installed_plugins.copy() 

125 

126 def register_plugin(self, info: PluginInfo) -> bool: 

127 """ 

128 Register a plugin in the registry. 

129 

130 Args: 

131 info: The plugin information. 

132 

133 Returns: 

134 True if registration was successful. 

135 """ 

136 if info.name in self._available_plugins: 

137 existing = self._available_plugins[info.name] 

138 # Update if newer version 

139 if self._compare_versions(info.version, existing.version) > 0: 

140 self._available_plugins[info.name] = info 

141 logger.info(f"Updated plugin {info.name} to v{info.version}") 

142 else: 

143 logger.debug(f"Plugin {info.name} already registered with same or newer version") 

144 return True 

145 

146 self._available_plugins[info.name] = info 

147 logger.info(f"Registered plugin {info.name} v{info.version}") 

148 return True 

149 

150 def unregister_plugin(self, name: str) -> bool: 

151 """ 

152 Unregister a plugin from the registry. 

153 

154 Args: 

155 name: The plugin name. 

156 

157 Returns: 

158 True if unregistration was successful. 

159 """ 

160 if name not in self._available_plugins: 

161 logger.warning(f"Plugin {name} not found in registry") 

162 return False 

163 

164 del self._available_plugins[name] 

165 logger.info(f"Unregistered plugin {name}") 

166 return True 

167 

168 def search(self, query: str = "", tags: Optional[List[str]] = None, 

169 author: str = "", min_rating: float = 0.0) -> List[PluginInfo]: 

170 """ 

171 Search for plugins in the registry. 

172 

173 Args: 

174 query: Search query string (searches name and description). 

175 tags: Filter by tags. 

176 author: Filter by author. 

177 min_rating: Minimum rating filter. 

178 

179 Returns: 

180 List of matching plugins. 

181 """ 

182 results = [] 

183 query_lower = query.lower() 

184 

185 for info in self._available_plugins.values(): 

186 # Check query match 

187 if query and query_lower not in info.name.lower() and \ 

188 query_lower not in info.description.lower(): 

189 continue 

190 

191 # Check tags 

192 if tags and not any(t in info.tags for t in tags): 

193 continue 

194 

195 # Check author 

196 if author and author.lower() != info.author.lower(): 

197 continue 

198 

199 # Check rating 

200 if info.rating < min_rating: 

201 continue 

202 

203 results.append(info) 

204 

205 # Sort by downloads (popularity) 

206 results.sort(key=lambda x: x.downloads, reverse=True) 

207 return results 

208 

209 def install(self, plugin_name: str, version: str = "", 

210 config: Optional[Dict[str, Any]] = None) -> Optional[InstalledPlugin]: 

211 """ 

212 Install a plugin from the registry. 

213 

214 Args: 

215 plugin_name: The name of the plugin to install. 

216 version: Specific version to install (latest if empty). 

217 config: Optional configuration for the plugin. 

218 

219 Returns: 

220 InstalledPlugin if successful, None otherwise. 

221 """ 

222 if plugin_name not in self._available_plugins: 

223 logger.error(f"Plugin {plugin_name} not found in registry") 

224 return None 

225 

226 info = self._available_plugins[plugin_name] 

227 

228 # Check version 

229 if version and version != info.version: 

230 logger.error(f"Version {version} not available for {plugin_name}") 

231 return None 

232 

233 # Check if already installed 

234 if plugin_name in self._installed_plugins: 

235 installed = self._installed_plugins[plugin_name] 

236 if installed.info.version == info.version: 

237 logger.warning(f"Plugin {plugin_name} v{info.version} already installed") 

238 return installed 

239 logger.info(f"Upgrading {plugin_name} from v{installed.info.version} to v{info.version}") 

240 

241 # Check dependencies 

242 for dep in info.dependencies: 

243 if dep not in self._installed_plugins: 

244 logger.warning(f"Missing dependency: {dep}") 

245 # Attempt to install dependency 

246 if not self.install(dep): 

247 logger.error(f"Failed to install dependency {dep}") 

248 return None 

249 

250 # Create install path 

251 install_path = os.path.join(self._install_dir, plugin_name) 

252 os.makedirs(install_path, exist_ok=True) 

253 

254 # Create installed plugin record 

255 installed = InstalledPlugin( 

256 info=info, 

257 install_path=install_path, 

258 installed_at=datetime.utcnow(), 

259 config=config or {} 

260 ) 

261 

262 self._installed_plugins[plugin_name] = installed 

263 logger.info(f"Installed plugin {plugin_name} v{info.version}") 

264 

265 # Save cache 

266 self._save_cache() 

267 

268 return installed 

269 

270 def uninstall(self, plugin_name: str, remove_config: bool = False) -> bool: 

271 """ 

272 Uninstall a plugin. 

273 

274 Args: 

275 plugin_name: The name of the plugin to uninstall. 

276 remove_config: Whether to remove configuration files. 

277 

278 Returns: 

279 True if uninstallation was successful. 

280 """ 

281 if plugin_name not in self._installed_plugins: 

282 logger.warning(f"Plugin {plugin_name} is not installed") 

283 return False 

284 

285 installed = self._installed_plugins[plugin_name] 

286 

287 # Check for dependents 

288 dependents = self._find_dependents(plugin_name) 

289 if dependents: 

290 logger.error(f"Cannot uninstall {plugin_name}: required by {dependents}") 

291 return False 

292 

293 # Remove installation 

294 if remove_config and os.path.exists(installed.install_path): 

295 try: 

296 import shutil 

297 shutil.rmtree(installed.install_path) 

298 except Exception as e: 

299 logger.warning(f"Could not remove install directory: {e}") 

300 

301 del self._installed_plugins[plugin_name] 

302 logger.info(f"Uninstalled plugin {plugin_name}") 

303 

304 # Save cache 

305 self._save_cache() 

306 

307 return True 

308 

309 def update(self, plugin_name: str) -> Optional[InstalledPlugin]: 

310 """ 

311 Update a plugin to the latest version. 

312 

313 Args: 

314 plugin_name: The name of the plugin to update. 

315 

316 Returns: 

317 Updated InstalledPlugin if successful, None otherwise. 

318 """ 

319 if plugin_name not in self._installed_plugins: 

320 logger.warning(f"Plugin {plugin_name} is not installed") 

321 return None 

322 

323 installed = self._installed_plugins[plugin_name] 

324 

325 if plugin_name not in self._available_plugins: 

326 logger.warning(f"Plugin {plugin_name} not found in registry") 

327 return None 

328 

329 available = self._available_plugins[plugin_name] 

330 

331 # Check if update is needed 

332 if self._compare_versions(available.version, installed.info.version) <= 0: 

333 logger.info(f"Plugin {plugin_name} is already up to date") 

334 return installed 

335 

336 # Preserve config 

337 config = installed.config.copy() 

338 

339 # Reinstall with new version 

340 return self.install(plugin_name, config=config) 

341 

342 def check_updates(self) -> List[Dict[str, Any]]: 

343 """ 

344 Check for available updates for installed plugins. 

345 

346 Returns: 

347 List of plugins with available updates. 

348 """ 

349 updates = [] 

350 

351 for name, installed in self._installed_plugins.items(): 

352 if name in self._available_plugins: 

353 available = self._available_plugins[name] 

354 if self._compare_versions(available.version, installed.info.version) > 0: 

355 updates.append({ 

356 "name": name, 

357 "current_version": installed.info.version, 

358 "available_version": available.version, 

359 "description": available.description 

360 }) 

361 

362 return updates 

363 

364 def refresh(self) -> bool: 

365 """ 

366 Refresh the registry from remote sources. 

367 

368 Returns: 

369 True if refresh was successful. 

370 """ 

371 # In a real implementation, this would fetch from remote registry 

372 logger.info("Refreshing plugin registry") 

373 return True 

374 

375 def _find_dependents(self, plugin_name: str) -> List[str]: 

376 """Find plugins that depend on the given plugin.""" 

377 dependents = [] 

378 for name, installed in self._installed_plugins.items(): 

379 if plugin_name in installed.info.dependencies: 

380 dependents.append(name) 

381 return dependents 

382 

383 def _compare_versions(self, v1: str, v2: str) -> int: 

384 """ 

385 Compare two version strings. 

386 

387 Returns: 

388 1 if v1 > v2, -1 if v1 < v2, 0 if equal. 

389 """ 

390 def parse_version(v): 

391 return [int(x) for x in v.split('.')] 

392 

393 try: 

394 parts1 = parse_version(v1) 

395 parts2 = parse_version(v2) 

396 

397 for p1, p2 in zip(parts1, parts2): 

398 if p1 > p2: 

399 return 1 

400 if p1 < p2: 

401 return -1 

402 

403 if len(parts1) > len(parts2): 

404 return 1 

405 if len(parts1) < len(parts2): 

406 return -1 

407 

408 return 0 

409 except Exception: 

410 # Fall back to string comparison 

411 if v1 > v2: 

412 return 1 

413 if v1 < v2: 

414 return -1 

415 return 0 

416 

417 def _save_cache(self) -> None: 

418 """Save the registry cache to disk.""" 

419 try: 

420 cache_data = { 

421 "available": {k: v.to_dict() for k, v in self._available_plugins.items()}, 

422 "installed": { 

423 k: { 

424 "info": v.info.to_dict(), 

425 "install_path": v.install_path, 

426 "installed_at": v.installed_at.isoformat(), 

427 "enabled": v.enabled, 

428 "config": v.config 

429 } 

430 for k, v in self._installed_plugins.items() 

431 } 

432 } 

433 with open(self._cache_file, 'w') as f: 

434 json.dump(cache_data, f, indent=2) 

435 except Exception as e: 

436 logger.warning(f"Failed to save registry cache: {e}") 

437 

438 def _load_cache(self) -> None: 

439 """Load the registry cache from disk.""" 

440 if not os.path.exists(self._cache_file): 

441 return 

442 

443 try: 

444 with open(self._cache_file, 'r') as f: 

445 cache_data = json.load(f) 

446 

447 for name, data in cache_data.get("available", {}).items(): 

448 self._available_plugins[name] = PluginInfo.from_dict(data) 

449 

450 for name, data in cache_data.get("installed", {}).items(): 

451 self._installed_plugins[name] = InstalledPlugin( 

452 info=PluginInfo.from_dict(data["info"]), 

453 install_path=data["install_path"], 

454 installed_at=datetime.fromisoformat(data["installed_at"]), 

455 enabled=data.get("enabled", False), 

456 config=data.get("config", {}) 

457 ) 

458 except Exception as e: 

459 logger.warning(f"Failed to load registry cache: {e}") 

460 

461 def get_plugin_info(self, plugin_name: str) -> Optional[PluginInfo]: 

462 """Get information about a plugin.""" 

463 return self._available_plugins.get(plugin_name) 

464 

465 def is_installed(self, plugin_name: str) -> bool: 

466 """Check if a plugin is installed.""" 

467 return plugin_name in self._installed_plugins 

468 

469 def get_installed_version(self, plugin_name: str) -> Optional[str]: 

470 """Get the installed version of a plugin.""" 

471 if plugin_name in self._installed_plugins: 

472 return self._installed_plugins[plugin_name].info.version 

473 return None