feat(backup): finished the importer, minus the loading functionality

backup/backup.py

@@ -5,12 +5,16 @@
 #  ____) |  __/ (_| \__ \\ V  V /| | | | | | | | | | | |  __/ |
 # |_____/ \___|\__,_|___/ \_/\_/ |_|_| |_| |_|_| |_| |_|\___|_|
 
+import contextlib
+import logging
 import json
 import re
 
 from redbot.core import commands
 from redbot.core.bot import Red
-from redbot.core.utils.chat_formatting import error, text_to_file
+from redbot.core.utils.chat_formatting import error, text_to_file, inline
+from redbot.cogs.downloader.converters import InstalledCog
+import redbot.cogs.downloader.errors as errors
 
 class Backup(commands.Cog):
     """A utility to make reinstalling repositories and cogs after migrating the bot far easier."""
@@ -21,6 +25,7 @@ class Backup(commands.Cog):
     def __init__(self, bot: Red):
         super().__init__()
         self.bot = bot
+        self.logger = logging.getLogger("red.sea.backup")
 
     @commands.group(autohelp=True)
     @commands.is_owner()
@@ -77,3 +82,111 @@ class Backup(commands.Cog):
         if downloader is None:
             await ctx.send(error(f"You do not have the `Downloader` cog loaded. Please run `{ctx.prefix}load downloader` and try again."))
             return
+
+        repo_e = []
+        uninstall_e = []
+        install_e = []
+
+        async with ctx.typing():
+            for repo in export:
+                # Most of this code is from the Downloader cog's repo_add funciton.
+                name = repo['name']
+                branch = repo['branch']
+                url = repo['url']
+                cogs = repo['cogs']
+
+                if url.contains('PyLav/RedCogs'):
+                    repo_e.append("PyLav cogs are not supported.")
+                    continue
+                if name.startswith('.') or name.endswith('.'):
+                    repo_e.append(f"Invalid repository name: {name}\nRepository names cannot start or end with a dot.")
+                    continue
+                if re.match(r"^[a-zA-Z0-9_\-\.]+$", name) is None:
+                    repo_e.append(f"Invalid repository name: {name}\nRepository names may only contain letters, numbers, underscores, hyphens, and dots.")
+                    continue
+
+                try:
+                    repository = await downloader._repo_manager.add_repo(name, url, branch) # pylint: disable=protected-access
+
+                except errors.ExistingGitRepo:
+                    repo_e.append(f"Repository {name} already exists.")
+                    continue
+
+                except errors.AuthenticationError as err:
+                    repo_e.append(f"Authentication error while adding repository {name}. See logs for more information.")
+                    self.log.exception(
+                        "Something went wrong whilst cloning %s (to revision %s)",
+                        url,
+                        branch,
+                        exc_info=err,
+                    )
+                    continue
+
+                except errors.CloningError as err:
+                    repo_e.append(f"Cloning error while adding repository {name}. See logs for more information.")
+                    self.log.exception(
+                        "Something went wrong whilst cloning %s (to revision %s)",
+                        url,
+                        branch,
+                        exc_info=err,
+                    )
+                    continue
+
+                except OSError:
+                    repo_e.append(f"OS error while adding repository {name}. See logs for more information.")
+                    self.log.exception(
+                        "Something went wrong trying to add repo %s under name %s",
+                        url,
+                        name
+                    )
+                    continue
+
+                else:
+                    cog_names = []
+                    for cog in cogs:
+                        cog_names.append(cog['name'])
+
+                    for cog in set(cog_names):
+                        poss_installed_path = (await downloader.cog_install_path()) / cog
+                        if poss_installed_path.exists():
+                            with contextlib.suppress(commands.ExtensionNotLoaded):
+                                await ctx.bot.unload_extension(cog)
+                                await ctx.bot.remove_loaded_package(cog)
+                            await downloader._delete_cog(poss_installed_path)
+                        else:
+                            uninstall_e.append(f"Failed to uninstall {cog}")
+                    await downloader._remove_from_installed(cogs)
+
+                    for cog in cogs:
+                        cog_name = cog['name']
+                        cog_pinned = cog['pinned']
+                        if cog_pinned:
+                            commit = cog['commit']
+                        else:
+                            commit = None
+
+                        async with repository.checkout(commit, exit_to_rev=repo.branch):
+                            cogs, message = await downloader._filter_incorrect_cogs_by_names(repository, [cog_name]) # pylint: disable=protected-access
+                            if not cogs:
+                                install_e.append(message)
+                                continue
+                            failed_reqs = await downloader._install_requirements(cogs)
+                            if failed_reqs:
+                                install_e.append(f"Failed to install {cog_name} due to missing requirements: {failed_reqs}")
+                                continue
+
+                            installed_cogs, failed_cogs = await downloader._install_cogs(cogs)
+
+                            if repo.available_libraries:
+                                installed_libs, failed_libs = await repository.install_libraries(target_dir=downloader.SHAREDLIB_PATH, req_target_dir=downloader.LIB_PATH)
+
+                            if cog_pinned:
+                                for cog in installed_cogs:
+                                    cog.pinned = True
+
+                            await downloader._save_to_installed(installed_cogs + installed_libs)
+                            if failed_cogs:
+                                install_e.append(f"Failed to install {failed_cogs}")
+                            if failed_libs:
+                                install_e.append(f"Failed to install {failed_libs} required for {cog_name}")
+        await ctx.send("Import complete!\nErrors:", file=text_to_file(f"Repositories:\n{repo_e}\n\nUninstalled Cogs:\n{uninstall_e}\n\nInstalled Cogs:\n{install_e}", 'backup-errors.log'))

backup/info.json

@@ -7,7 +7,8 @@
   "end_user_data_statement" : "This cog does not store user information.",
   "hidden": false,
   "disabled": false,
-  "min_bot_version": "3.5.0",
+  "min_bot_version": "3.5.5",
+  "max_bot_version": "3.5.5",
   "min_python_version": [3, 10, 0],
   "tags": []
 }