Merge pull request #5 from khakers/feature/forced-plugins

Author: khakers

CHANGELOG.md

@@ -18,7 +18,7 @@ however, insignificant breaking changes do not guarantee a major version bump, s
 ### Added
 - Added `content_type` to attachments stored in the database.
 - `?log key <key>` to retrieve the log link and view a preview using a log key. ([PR #3196](https://github.com/modmail-dev/Modmail/pull/3196))
-
+- Add Forced plugins. Allows auto installing un-removable plugins via `FORCED_PLUGINS` environment variable contain a comma separate list of plugins. (GH#5)
 
 ### Changed
 - Changing a threads title or NSFW status immediately updates the status in the database.

cogs/plugins.py

@@ -117,6 +117,7 @@ class Plugins(commands.Cog):
     """
 
     def __init__(self, bot):
+        self.forced_plugins: list[str] = []
         self.bot = bot
         self.registry = {}
         self.loaded_plugins = set()
@@ -134,33 +135,56 @@ async def populate_registry(self):
         async with self.bot.session.get(url) as resp:
             self.registry = json.loads(await resp.text())
 
-    async def initial_load_plugins(self):
-        for plugin_name in list(self.bot.config["plugins"]):
+    def _get_forced_plugins(self) -> list[str]:
+        env_list = os.getenv("FORCED_PLUGINS")
+        if env_list is None:
+            return []
+        plugins = env_list.split(",")
+        return plugins
+
+    async def _init_load_plugin(self, plugin_name: str):
+        try:
+            # Strict seems to only affect whether specifying the branch is required
+            plugin = Plugin.from_string(plugin_name, strict=False)
+        except InvalidPluginError:
+            self.bot.config["plugins"].remove(plugin_name)
             try:
-                plugin = Plugin.from_string(plugin_name, strict=True)
+                # For backwards compat
+                plugin = Plugin.from_string(plugin_name)
             except InvalidPluginError:
-                self.bot.config["plugins"].remove(plugin_name)
-                try:
-                    # For backwards compat
-                    plugin = Plugin.from_string(plugin_name)
-                except InvalidPluginError:
-                    logger.error("Failed to parse plugin name: %s.", plugin_name, exc_info=True)
-                    continue
+                logger.error("Failed to parse plugin name: %s.", plugin_name, exc_info=True)
+                return
 
-                logger.info("Migrated legacy plugin name: %s, now %s.", plugin_name, str(plugin))
-                self.bot.config["plugins"].append(str(plugin))
+            logger.info("Migrated legacy plugin name: %s, now %s.", plugin_name, str(plugin))
+            self.bot.config["plugins"].append(str(plugin))
 
-            try:
-                await self.download_plugin(plugin)
-                await self.load_plugin(plugin)
-            except Exception:
-                self.bot.config["plugins"].remove(plugin_name)
-                logger.error(
-                    "Error when loading plugin %s. Plugin removed from config.",
-                    plugin,
-                    exc_info=True,
-                )
+        try:
+            await self.download_plugin(plugin)
+            await self.load_plugin(plugin)
+        except Exception:
+            self.bot.config["plugins"].remove(plugin_name)
+            logger.error(
+                "Error when loading plugin %s. Plugin removed from config.",
+                plugin,
+                exc_info=True,
+            )
+
+    async def initial_load_plugins(self):
+        self.forced_plugins = self._get_forced_plugins()
+        logger.debug(f"loading {len(self.forced_plugins)} forced plugins")
+        for plugin_name in self.forced_plugins:
+            logger.debug(f"loading forced plugin {plugin_name}")
+            await self._init_load_plugin(plugin_name)
+
+        user_plugins = list(self.bot.config["plugins"])
+        logger.debug(f"loading {len(user_plugins)} config plugins")
+
+        for plugin_name in user_plugins:
+            # Skip loading this plugin if it is in the force load list and thus already loaded
+            if plugin_name in self.forced_plugins:
+                logger.debug(f"Skipped loading user plugin {plugin_name} because it is forced installed.")
                 continue
+            await self._init_load_plugin(plugin_name)
 
         logger.debug("Finished loading all plugins.")
 
@@ -440,6 +464,15 @@ async def plugins_remove(self, ctx, *, plugin_name: str):
         if plugin is None:
             return
 
+        if str(plugin) in self.forced_plugins:
+            await ctx.send(
+                embed=discord.Embed(
+                    description="This plugin cannot be removed. Contact your admin for more information.",
+                    color=self.bot.error_color,
+                )
+            )
+            return
+
         if str(plugin) not in self.bot.config["plugins"]:
             embed = discord.Embed(description="Plugin is not installed.", color=self.bot.error_color)
             return await ctx.send(embed=embed)