addonmanager_workers_startup.py

# SPDX-License-Identifier: LGPL-2.1-or-later
# ***************************************************************************
# *                                                                         *
# *   Copyright (c) 2022-2023 FreeCAD Project Association                   *
# *   Copyright (c) 2019 Yorik van Havre <yorik@uncreated.net>              *
# *                                                                         *
# *   This file is part of FreeCAD.                                         *
# *                                                                         *
# *   FreeCAD is free software: you can redistribute it and/or modify it    *
# *   under the terms of the GNU Lesser General Public License as           *
# *   published by the Free Software Foundation, either version 2.1 of the  *
# *   License, or (at your option) any later version.                       *
# *                                                                         *
# *   FreeCAD is distributed in the hope that it will be useful, but        *
# *   WITHOUT ANY WARRANTY; without even the implied warranty of            *
# *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
# *   Lesser General Public License for more details.                       *
# *                                                                         *
# *   You should have received a copy of the GNU Lesser General Public      *
# *   License along with FreeCAD. If not, see                               *
# *   <https://www.gnu.org/licenses/>.                                      *
# *                                                                         *
# ***************************************************************************

"""Worker thread classes for Addon Manager startup"""
import hashlib
import json
import os
import queue
import re
import shutil
import stat
import threading
import time
from typing import List, Optional
import xml.etree.ElementTree

try:
    from PySide import QtCore
except ImportError:
    try:
        from PySide6 import QtCore
    except ImportError:
        from PySide2 import QtCore

import addonmanager_utilities as utils
from addonmanager_macro import Macro
from Addon import Addon
from AddonStats import AddonStats
import NetworkManager
from addonmanager_git import initialize_git, GitFailed
from addonmanager_metadata import MetadataReader, get_branch_from_metadata
import addonmanager_freecad_interface as fci

translate = fci.translate

# Workers only have one public method by design
# pylint: disable=c-extension-no-member,too-few-public-methods,too-many-instance-attributes


class CreateAddonListWorker(QtCore.QThread):
    """This worker updates the list of available workbenches, emitting an "addon_repo"
    signal for each Addon as they are processed."""

    addon_repo = QtCore.Signal(object)
    progress_made = QtCore.Signal(str, int, int)

    def __init__(self):
        QtCore.QThread.__init__(self)

        # reject_listed addons
        self.macros_reject_list = []
        self.mod_reject_list = []

        # These addons will print an additional message informing the user
        self.obsolete = []

        # These addons will print an additional message informing the user Python2 only
        self.py2only = []

        self.package_names = []
        self.moddir = fci.DataPaths().mod_dir
        self.current_thread = None

        self.git_manager = initialize_git()

    def run(self):
        "populates the list of addons"

        self.current_thread = QtCore.QThread.currentThread()
        try:
            self._get_freecad_addon_repo_data()
        except ConnectionError:
            return
        self._get_custom_addons()
        self._get_official_addons()
        self._retrieve_macros_from_git()
        self._retrieve_macros_from_wiki()

    def _get_freecad_addon_repo_data(self):
        # update info lists
        p = NetworkManager.AM_NETWORK_MANAGER.blocking_get(
            "https://raw.githubusercontent.com/FreeCAD/FreeCAD-addons/master/addonflags.json", 5000
        )
        if p:
            p = p.data().decode("utf8")
            j = json.loads(p)
            if "obsolete" in j and "Mod" in j["obsolete"]:
                self.obsolete = j["obsolete"]["Mod"]

            if "blacklisted" in j and "Macro" in j["blacklisted"]:
                self.macros_reject_list = j["blacklisted"]["Macro"]

            if "blacklisted" in j and "Mod" in j["blacklisted"]:
                self.mod_reject_list = j["blacklisted"]["Mod"]

            if "py2only" in j and "Mod" in j["py2only"]:
                self.py2only = j["py2only"]["Mod"]

            if "deprecated" in j:
                self._process_deprecated(j["deprecated"])

        else:
            message = translate(
                "AddonsInstaller",
                "Failed to connect to GitHub. Check your connection and proxy settings.",
            )
            fci.Console.PrintError(message + "\n")
            raise ConnectionError

    def _process_deprecated(self, deprecated_addons):
        """Parse the section on deprecated addons"""

        fc_major = int(fci.Version()[0])
        fc_minor = int(fci.Version()[1])
        for item in deprecated_addons:
            if "as_of" in item and "name" in item:
                try:
                    version_components = item["as_of"].split(".")
                    major = int(version_components[0])
                    if len(version_components) > 1:
                        minor = int(version_components[1])
                    else:
                        minor = 0
                    if major < fc_major or (major == fc_major and minor <= fc_minor):
                        if "kind" not in item or item["kind"] == "mod":
                            self.obsolete.append(item["name"])
                        elif item["kind"] == "macro":
                            self.macros_reject_list.append(item["name"])
                        else:
                            fci.Console.PrintMessage(
                                f'Unrecognized Addon kind {item["kind"]} in deprecation list.'
                            )
                except ValueError:
                    fci.Console.PrintMessage(
                        f"Failed to parse version from {item['name']}, version {item['as_of']}"
                    )

    def _get_custom_addons(self):

        # querying custom addons first
        addon_list = fci.Preferences().get("CustomRepositories").split("\n")
        custom_addons = []
        for addon in addon_list:
            if " " in addon:
                addon_and_branch = addon.split(" ")
                custom_addons.append({"url": addon_and_branch[0], "branch": addon_and_branch[1]})
            else:
                custom_addons.append({"url": addon, "branch": "master"})
        for addon in custom_addons:
            if self.current_thread.isInterruptionRequested():
                return
            if addon and addon["url"]:
                if addon["url"][-1] == "/":
                    addon["url"] = addon["url"][0:-1]  # Strip trailing slash
                addon["url"] = addon["url"].split(".git")[0]  # Remove .git
                name = addon["url"].split("/")[-1]
                if name in self.package_names:
                    # We already have something with this name, skip this one
                    fci.Console.PrintWarning(
                        translate("AddonsInstaller", "WARNING: Duplicate addon {} ignored").format(
                            name
                        )
                    )
                    continue
                fci.Console.PrintLog(
                    f"Adding custom location {addon['url']} with branch {addon['branch']}\n"
                )
                self.package_names.append(name)
                addondir = os.path.join(self.moddir, name)
                if os.path.exists(addondir) and os.listdir(addondir):
                    state = Addon.Status.UNCHECKED
                else:
                    state = Addon.Status.NOT_INSTALLED
                repo = Addon(name, addon["url"], state, addon["branch"])
                md_file = os.path.join(addondir, "package.xml")
                if os.path.isfile(md_file):
                    try:
                        repo.installed_metadata = MetadataReader.from_file(md_file)
                        repo.installed_version = repo.installed_metadata.version
                        repo.updated_timestamp = os.path.getmtime(md_file)
                        repo.verify_url_and_branch(addon["url"], addon["branch"])
                    except xml.etree.ElementTree.ParseError:
                        fci.Console.PrintWarning(
                            "An invalid or corrupted package.xml file was installed for"
                        )
                        fci.Console.PrintWarning(
                            f" custom addon {self.name}... ignoring the bad data.\n"
                        )

                self.addon_repo.emit(repo)

    def _get_official_addons(self):
        # querying official addons
        p = NetworkManager.AM_NETWORK_MANAGER.blocking_get(
            "https://raw.githubusercontent.com/FreeCAD/FreeCAD-addons/master/.gitmodules", 5000
        )
        if not p:
            return
        p = p.data().decode("utf8")
        p = re.findall(
            (
                r'(?m)\[submodule\s*"(?P<name>.*)"\]\s*'
                r"path\s*=\s*(?P<path>.+)\s*"
                r"url\s*=\s*(?P<url>https?://.*)\s*"
                r"(branch\s*=\s*(?P<branch>[^\s]*)\s*)?"
            ),
            p,
        )
        for name, _, url, _, branch in p:
            if self.current_thread.isInterruptionRequested():
                return
            if name in self.package_names:
                # We already have something with this name, skip this one
                continue
            self.package_names.append(name)
            if branch is None or len(branch) == 0:
                branch = "master"
            url = url.split(".git")[0]
            addondir = os.path.join(self.moddir, name)
            if os.path.exists(addondir) and os.listdir(addondir):
                # make sure the folder exists and it contains files!
                state = Addon.Status.UNCHECKED
            else:
                state = Addon.Status.NOT_INSTALLED
            repo = Addon(name, url, state, branch)
            md_file = os.path.join(addondir, "package.xml")
            if os.path.isfile(md_file):
                try:
                    repo.installed_metadata = MetadataReader.from_file(md_file)
                    repo.installed_version = repo.installed_metadata.version
                    repo.updated_timestamp = os.path.getmtime(md_file)
                    repo.verify_url_and_branch(url, branch)
                except xml.etree.ElementTree.ParseError:
                    fci.Console.PrintWarning(
                        "An invalid or corrupted package.xml file was installed for"
                    )
                    fci.Console.PrintWarning(f" addon {self.name}... ignoring the bad data.\n")

            if name in self.py2only:
                repo.python2 = True
            if name in self.mod_reject_list:
                repo.rejected = True
            if name in self.obsolete:
                repo.obsolete = True
            self.addon_repo.emit(repo)

    def _retrieve_macros_from_git(self):
        """Retrieve macros from FreeCAD-macros.git

        Emits a signal for each macro in
        https://github.com/FreeCAD/FreeCAD-macros.git
        """

        macro_cache_location = utils.get_cache_file_name("Macros")

        if not self.git_manager:
            message = translate(
                "AddonsInstaller",
                "Git is disabled, skipping Git macros",
            )
            fci.Console.PrintWarning(message + "\n")
            return

        update_succeeded = self._update_local_git_repo()
        if not update_succeeded:
            return

        n_files = 0
        for _, _, filenames in os.walk(macro_cache_location):
            n_files += len(filenames)
        counter = 0
        for dirpath, _, filenames in os.walk(macro_cache_location):
            counter += 1
            if self.current_thread.isInterruptionRequested():
                return
            if ".git" in dirpath:
                continue
            for filename in filenames:
                if self.current_thread.isInterruptionRequested():
                    return
                if filename.lower().endswith(".fcmacro"):
                    macro = Macro(filename[:-8])  # Remove ".FCMacro".
                    if macro.name in self.package_names:
                        fci.Console.PrintLog(
                            f"Ignoring second macro named {macro.name} (found on git)\n"
                        )
                        continue  # We already have a macro with this name
                    self.package_names.append(macro.name)
                    macro.on_git = True
                    macro.src_filename = os.path.join(dirpath, filename)
                    macro.fill_details_from_file(macro.src_filename)
                    repo = Addon.from_macro(macro)
                    fci.Console.PrintLog(f"Found macro {repo.name}\n")
                    repo.url = "https://github.com/FreeCAD/FreeCAD-macros.git"
                    utils.update_macro_installation_details(repo)
                    self.addon_repo.emit(repo)

    def _update_local_git_repo(self) -> bool:
        macro_cache_location = utils.get_cache_file_name("Macros")
        try:
            if os.path.exists(macro_cache_location):
                if not os.path.exists(os.path.join(macro_cache_location, ".git")):
                    fci.Console.PrintWarning(
                        translate(
                            "AddonsInstaller",
                            "Attempting to change non-Git Macro setup to use Git\n",
                        )
                    )
                    self.git_manager.repair(
                        "https://github.com/FreeCAD/FreeCAD-macros.git",
                        macro_cache_location,
                    )
                self.git_manager.update(macro_cache_location)
            else:
                self.git_manager.clone(
                    "https://github.com/FreeCAD/FreeCAD-macros.git",
                    macro_cache_location,
                )
        except GitFailed as e:
            fci.Console.PrintMessage(
                translate(
                    "AddonsInstaller",
                    "An error occurred updating macros from GitHub, trying clean checkout...",
                )
                + f":\n{e}\n"
            )
            fci.Console.PrintMessage(f"{macro_cache_location}\n")
            fci.Console.PrintMessage(
                translate("AddonsInstaller", "Attempting to do a clean checkout...") + "\n"
            )
            try:
                os.chdir(
                    os.path.join(macro_cache_location, "..")
                )  # Make sure we are not IN this directory
                shutil.rmtree(macro_cache_location, onerror=self._remove_readonly)
                self.git_manager.clone(
                    "https://github.com/FreeCAD/FreeCAD-macros.git",
                    macro_cache_location,
                )
                fci.Console.PrintMessage(
                    translate("AddonsInstaller", "Clean checkout succeeded") + "\n"
                )
            except GitFailed as e2:
                # The Qt Python translation extractor doesn't support splitting this string (yet)
                # pylint: disable=line-too-long
                fci.Console.PrintWarning(
                    translate(
                        "AddonsInstaller",
                        "Failed to update macros from GitHub -- try clearing the Addon Manager's cache.",
                    )
                    + f":\n{str(e2)}\n"
                )
                return False
        return True

    def _retrieve_macros_from_wiki(self):
        """Retrieve macros from the wiki

        Read the wiki and emit a signal for each found macro.
        Reads only the page https://wiki.freecad.org/Macros_recipes
        """

        p = NetworkManager.AM_NETWORK_MANAGER.blocking_get(
            "https://wiki.freecad.org/Macros_recipes", 5000
        )
        if not p:
            # The Qt Python translation extractor doesn't support splitting this string (yet)
            # pylint: disable=line-too-long
            fci.Console.PrintWarning(
                translate(
                    "AddonsInstaller",
                    "Error connecting to the Wiki, FreeCAD cannot retrieve the Wiki macro list at this time",
                )
                + "\n"
            )
            return
        p = p.data().decode("utf8")
        macros = re.findall(r'title="(Macro.*?)"', p)
        macros = [mac for mac in macros if "translated" not in mac]
        macro_names = []
        for _, mac in enumerate(macros):
            if self.current_thread.isInterruptionRequested():
                return
            macname = mac[6:]  # Remove "Macro ".
            macname = macname.replace("&amp;", "&")
            if not macname:
                continue
            if (
                (macname not in self.macros_reject_list)
                and ("recipes" not in macname.lower())
                and (macname not in macro_names)
            ):
                macro_names.append(macname)
                macro = Macro(macname)
                if macro.name in self.package_names:
                    fci.Console.PrintLog(
                        f"Ignoring second macro named {macro.name} (found on wiki)\n"
                    )
                    continue  # We already have a macro with this name
                self.package_names.append(macro.name)
                macro.on_wiki = True
                macro.parsed = False
                repo = Addon.from_macro(macro)
                repo.url = "https://wiki.freecad.org/Macros_recipes"
                utils.update_macro_installation_details(repo)
                self.addon_repo.emit(repo)

    def _remove_readonly(self, func, path, _) -> None:
        """Remove a read-only file."""

        os.chmod(path, stat.S_IWRITE)
        func(path)


class LoadPackagesFromCacheWorker(QtCore.QThread):
    """A subthread worker that loads package information from its cache file."""

    addon_repo = QtCore.Signal(object)

    def __init__(self, cache_file: str):
        QtCore.QThread.__init__(self)
        self.cache_file = cache_file
        self.metadata_cache_path = os.path.join(
            fci.DataPaths().cache_dir, "AddonManager", "PackageMetadata"
        )

    def override_metadata_cache_path(self, path):
        """For testing purposes, override the location to fetch the package metadata from."""
        self.metadata_cache_path = path

    def run(self):
        """Rarely called directly: create an instance and call start() on it instead to
        launch in a new thread"""
        with open(self.cache_file, encoding="utf-8") as f:
            data = f.read()
            if data:
                dict_data = json.loads(data)
                for item in dict_data.values():
                    if QtCore.QThread.currentThread().isInterruptionRequested():
                        return
                    repo = Addon.from_cache(item)
                    repo_metadata_cache_path = os.path.join(
                        self.metadata_cache_path, repo.name, "package.xml"
                    )
                    if os.path.isfile(repo_metadata_cache_path):
                        try:
                            repo.load_metadata_file(repo_metadata_cache_path)
                        except RuntimeError as e:
                            fci.Console.PrintLog(f"Failed loading {repo_metadata_cache_path}\n")
                            fci.Console.PrintLog(str(e) + "\n")
                    self.addon_repo.emit(repo)


class LoadMacrosFromCacheWorker(QtCore.QThread):
    """A worker object to load macros from a cache file"""

    add_macro_signal = QtCore.Signal(object)

    def __init__(self, cache_file: str):
        QtCore.QThread.__init__(self)
        self.cache_file = cache_file

    def run(self):
        """Rarely called directly: create an instance and call start() on it instead to
        launch in a new thread"""

        with open(self.cache_file, encoding="utf-8") as f:
            data = f.read()
            dict_data = json.loads(data)
            for item in dict_data:
                if QtCore.QThread.currentThread().isInterruptionRequested():
                    return
                new_macro = Macro.from_cache(item)
                repo = Addon.from_macro(new_macro)
                utils.update_macro_installation_details(repo)
                self.add_macro_signal.emit(repo)


class CheckSingleUpdateWorker(QtCore.QObject):
    """This worker is a little different from the others: the actual recommended way of
    running in a QThread is to make a worker object that gets moved into the thread."""

    update_status = QtCore.Signal(int)

    def __init__(self, repo: Addon, parent: QtCore.QObject = None):
        super().__init__(parent)
        self.repo = repo

    def do_work(self):
        """Use the UpdateChecker class to do the work of this function, depending on the
        type of Addon"""

        checker = UpdateChecker()
        if self.repo.repo_type == Addon.Kind.WORKBENCH:
            checker.check_workbench(self.repo)
        elif self.repo.repo_type == Addon.Kind.MACRO:
            checker.check_macro(self.repo)
        elif self.repo.repo_type == Addon.Kind.PACKAGE:
            checker.check_package(self.repo)

        self.update_status.emit(self.repo.update_status)


class CheckWorkbenchesForUpdatesWorker(QtCore.QThread):
    """This worker checks for available updates for all workbenches"""

    update_status = QtCore.Signal(Addon)
    progress_made = QtCore.Signal(str, int, int)

    def __init__(self, repos: List[Addon]):

        QtCore.QThread.__init__(self)
        self.repos = repos
        self.current_thread = None
        self.moddir = fci.DataPaths().mod_dir

    def run(self):
        """Rarely called directly: create an instance and call start() on it instead to
        launch in a new thread"""

        self.current_thread = QtCore.QThread.currentThread()
        checker = UpdateChecker()
        count = 1
        for repo in self.repos:
            if self.current_thread.isInterruptionRequested():
                return
            message = translate("AddonsInstaller", "Checking {} for update").format(
                repo.display_name
            )
            self.progress_made.emit(message, count, len(self.repos))
            count += 1
            if repo.status() == Addon.Status.UNCHECKED:
                if repo.repo_type == Addon.Kind.WORKBENCH:
                    checker.check_workbench(repo)
                    self.update_status.emit(repo)
                elif repo.repo_type == Addon.Kind.MACRO:
                    checker.check_macro(repo)
                    self.update_status.emit(repo)
                elif repo.repo_type == Addon.Kind.PACKAGE:
                    checker.check_package(repo)
                    self.update_status.emit(repo)


class UpdateChecker:
    """A utility class used by the CheckWorkbenchesForUpdatesWorker class. Each function is
    designed for a specific Addon type, and modifies the passed-in Addon with the determined
    update status."""

    def __init__(self):
        self.moddir = fci.DataPaths().mod_dir
        self.git_manager = initialize_git()

    def override_mod_directory(self, moddir):
        """Primarily for use when testing, sets an alternate directory to use for mods"""
        self.moddir = moddir

    def check_workbench(self, wb):
        """Given a workbench Addon wb, check it for updates using git. If git is not
        available, does nothing."""
        if not self.git_manager:
            wb.set_status(Addon.Status.CANNOT_CHECK)
            return
        clonedir = os.path.join(self.moddir, wb.name)
        if os.path.exists(clonedir):
            # mark as already installed AND already checked for updates
            if not os.path.exists(os.path.join(clonedir, ".git")):
                with wb.git_lock:
                    self.git_manager.repair(wb.url, clonedir)
            with wb.git_lock:
                try:
                    status = self.git_manager.status(clonedir)
                    if "(no branch)" in status:
                        # By definition, in a detached-head state we cannot
                        # update, so don't even bother checking.
                        wb.set_status(Addon.Status.NO_UPDATE_AVAILABLE)
                        wb.branch = self.git_manager.current_branch(clonedir)
                        return
                except GitFailed as e:
                    fci.Console.PrintWarning(
                        "AddonManager: "
                        + translate(
                            "AddonsInstaller",
                            "Unable to fetch Git updates for workbench {}",
                        ).format(wb.name)
                        + "\n"
                    )
                    fci.Console.PrintWarning(str(e) + "\n")
                    wb.set_status(Addon.Status.CANNOT_CHECK)
                else:
                    try:
                        if self.git_manager.update_available(clonedir):
                            wb.set_status(Addon.Status.UPDATE_AVAILABLE)
                        else:
                            wb.set_status(Addon.Status.NO_UPDATE_AVAILABLE)
                    except GitFailed:
                        fci.Console.PrintWarning(
                            translate("AddonsInstaller", "Git status failed for {}").format(wb.name)
                            + "\n"
                        )
                        wb.set_status(Addon.Status.CANNOT_CHECK)

    def _branch_name_changed(self, package: Addon) -> bool:
        clone_dir = os.path.join(self.moddir, package.name)
        installed_metadata_file = os.path.join(clone_dir, "package.xml")
        if not os.path.isfile(installed_metadata_file):
            return False
        if not hasattr(package, "metadata") or package.metadata is None:
            return False
        try:
            installed_metadata = MetadataReader.from_file(installed_metadata_file)
            installed_default_branch = get_branch_from_metadata(installed_metadata)
            remote_default_branch = get_branch_from_metadata(package.metadata)
            if installed_default_branch != remote_default_branch:
                return True
        except RuntimeError:
            return False
        return False

    def check_package(self, package: Addon) -> None:
        """Given a packaged Addon package, check it for updates. If git is available that is
        used. If not, the package's metadata is examined, and if the metadata file has changed
        compared to the installed copy, an update is flagged. In addition, a change to the
        default branch name triggers an update."""

        clone_dir = self.moddir + os.sep + package.name
        if os.path.exists(clone_dir):

            # First, see if the branch name changed, which automatically triggers an update
            if self._branch_name_changed(package):
                package.set_status(Addon.Status.UPDATE_AVAILABLE)
                return

            # Next, try to just do a git-based update, which will give the most accurate results:
            if self.git_manager:
                self.check_workbench(package)
                if package.status() != Addon.Status.CANNOT_CHECK:
                    # It worked, just exit now
                    return

            # If we were unable to do a git-based update, try using the package.xml file instead:
            installed_metadata_file = os.path.join(clone_dir, "package.xml")
            if not os.path.isfile(installed_metadata_file):
                # If there is no package.xml file, then it's because the package author added it
                # after the last time the local installation was updated. By definition, then,
                # there is an update available, if only to download the new XML file.
                package.set_status(Addon.Status.UPDATE_AVAILABLE)
                package.installed_version = None
                return
            package.updated_timestamp = os.path.getmtime(installed_metadata_file)
            try:
                installed_metadata = MetadataReader.from_file(installed_metadata_file)
                package.installed_version = installed_metadata.version
                # Packages are considered up-to-date if the metadata version matches.
                # Authors should update their version string when they want the addon
                # manager to alert users of a new version.
                if package.metadata.version != installed_metadata.version:
                    package.set_status(Addon.Status.UPDATE_AVAILABLE)
                else:
                    package.set_status(Addon.Status.NO_UPDATE_AVAILABLE)
            except RuntimeError:
                fci.Console.PrintWarning(
                    translate(
                        "AddonsInstaller",
                        "Failed to read metadata from {name}",
                    ).format(name=installed_metadata_file)
                    + "\n"
                )
                package.set_status(Addon.Status.CANNOT_CHECK)

    def check_macro(self, macro_wrapper: Addon) -> None:
        """Check to see if the online copy of the macro's code differs from the local copy."""

        # Make sure this macro has its code downloaded:
        try:
            if not macro_wrapper.macro.parsed and macro_wrapper.macro.on_git:
                macro_wrapper.macro.fill_details_from_file(macro_wrapper.macro.src_filename)
            elif not macro_wrapper.macro.parsed and macro_wrapper.macro.on_wiki:
                mac = macro_wrapper.macro.name.replace(" ", "_")
                mac = mac.replace("&", "%26")
                mac = mac.replace("+", "%2B")
                url = "https://wiki.freecad.org/Macro_" + mac
                macro_wrapper.macro.fill_details_from_wiki(url)
        except RuntimeError:
            fci.Console.PrintWarning(
                translate(
                    "AddonsInstaller",
                    "Failed to fetch code for macro '{name}'",
                ).format(name=macro_wrapper.macro.name)
                + "\n"
            )
            macro_wrapper.set_status(Addon.Status.CANNOT_CHECK)
            return

        hasher1 = hashlib.sha1()
        hasher2 = hashlib.sha1()
        hasher1.update(macro_wrapper.macro.code.encode("utf-8"))
        new_sha1 = hasher1.hexdigest()
        test_file_one = os.path.join(fci.DataPaths().macro_dir, macro_wrapper.macro.filename)
        test_file_two = os.path.join(
            fci.DataPaths().macro_dir, "Macro_" + macro_wrapper.macro.filename
        )
        if os.path.exists(test_file_one):
            with open(test_file_one, "rb") as f:
                contents = f.read()
                hasher2.update(contents)
                old_sha1 = hasher2.hexdigest()
        elif os.path.exists(test_file_two):
            with open(test_file_two, "rb") as f:
                contents = f.read()
                hasher2.update(contents)
                old_sha1 = hasher2.hexdigest()
        else:
            return
        if new_sha1 == old_sha1:
            macro_wrapper.set_status(Addon.Status.NO_UPDATE_AVAILABLE)
        else:
            macro_wrapper.set_status(Addon.Status.UPDATE_AVAILABLE)


class CacheMacroCodeWorker(QtCore.QThread):
    """Download and cache the macro code, and parse its internal metadata"""

    update_macro = QtCore.Signal(Addon)
    progress_made = QtCore.Signal(str, int, int)

    def __init__(self, repos: List[Addon]) -> None:
        QtCore.QThread.__init__(self)
        self.repos = repos
        self.workers = []
        self.terminators = []
        self.lock = threading.Lock()
        self.failed = []
        self.counter = 0
        self.repo_queue = None

    def run(self):
        """Rarely called directly: create an instance and call start() on it instead to
        launch in a new thread"""

        self.repo_queue = queue.Queue()
        num_macros = 0
        for repo in self.repos:
            if repo.macro is not None:
                self.repo_queue.put(repo)
                num_macros += 1

        interrupted = self._process_queue(num_macros)
        if interrupted:
            return

        # Make sure all of our child threads have fully exited:
        for worker in self.workers:
            worker.wait(50)
            if not worker.isFinished():
                # The Qt Python translation extractor doesn't support splitting this string (yet)
                # pylint: disable=line-too-long
                fci.Console.PrintError(
                    translate(
                        "AddonsInstaller",
                        "Addon Manager: a worker process failed to complete while fetching {name}",
                    ).format(name=worker.macro.name)
                    + "\n"
                )

        self.repo_queue.join()
        for terminator in self.terminators:
            if terminator and terminator.isActive():
                terminator.stop()

        if len(self.failed) > 0:
            num_failed = len(self.failed)
            fci.Console.PrintWarning(
                translate(
                    "AddonsInstaller",
                    "Out of {num_macros} macros, {num_failed} timed out while processing",
                ).format(num_macros=num_macros, num_failed=num_failed)
                + "\n"
            )

    def _process_queue(self, num_macros) -> bool:
        """Spools up six network connections and downloads the macro code. Returns True if
        it was interrupted by user request, or False if it ran to completion."""

        # Emulate QNetworkAccessManager and spool up six connections:
        for _ in range(6):
            self.update_and_advance(None)

        current_thread = QtCore.QThread.currentThread()
        while True:
            if current_thread.isInterruptionRequested():
                for worker in self.workers:
                    worker.blockSignals(True)
                    worker.requestInterruption()
                    if not worker.wait(100):
                        fci.Console.PrintWarning(
                            translate(
                                "AddonsInstaller",
                                "Addon Manager: a worker process failed to halt ({name})",
                            ).format(name=worker.macro.name)
                            + "\n"
                        )
                return True
            # Ensure our signals propagate out by running an internal thread-local event loop
            QtCore.QCoreApplication.processEvents()
            with self.lock:
                if self.counter >= num_macros:
                    break
            time.sleep(0.1)
        return False

    def update_and_advance(self, repo: Optional[Addon]) -> None:
        """Emit the updated signal and launch the next item from the queue."""
        if repo is not None:
            if repo.macro.name not in self.failed:
                self.update_macro.emit(repo)
            self.repo_queue.task_done()
            with self.lock:
                self.counter += 1

        if QtCore.QThread.currentThread().isInterruptionRequested():
            return

        if repo is not None:
            message = translate("AddonsInstaller", "Caching {} macro").format(repo.display_name)
        else:
            message = translate("AddonsInstaller", "Caching macros")
        self.progress_made.emit(message, len(self.repos) - self.repo_queue.qsize(), len(self.repos))

        try:
            next_repo = self.repo_queue.get_nowait()
            worker = GetMacroDetailsWorker(next_repo)
            worker.finished.connect(lambda: self.update_and_advance(next_repo))
            with self.lock:
                self.workers.append(worker)
                self.terminators.append(
                    QtCore.QTimer.singleShot(10000, lambda: self.terminate(worker))
                )
            worker.start()
        except queue.Empty:
            pass

    def terminate(self, worker) -> None:
        """Shut down all running workers and exit the thread"""
        if not worker.isFinished():
            macro_name = worker.macro.name
            fci.Console.PrintWarning(
                translate(
                    "AddonsInstaller",
                    "Timeout while fetching metadata for macro {}",
                ).format(macro_name)
                + "\n"
            )
            # worker.blockSignals(True)
            worker.requestInterruption()
            worker.wait(100)
            if worker.isRunning():
                fci.Console.PrintError(
                    translate(
                        "AddonsInstaller",
                        "Failed to kill process for macro {}!\n",
                    ).format(macro_name)
                )
            with self.lock:
                self.failed.append(macro_name)


class GetMacroDetailsWorker(QtCore.QThread):
    """Retrieve the macro details for a macro"""

    readme_updated = QtCore.Signal(str)

    def __init__(self, repo):

        QtCore.QThread.__init__(self)
        self.macro = repo.macro

    def run(self):
        """Rarely called directly: create an instance and call start() on it instead to
        launch in a new thread"""

        if not self.macro.parsed and self.macro.on_git:
            self.macro.fill_details_from_file(self.macro.src_filename)
        if not self.macro.parsed and self.macro.on_wiki:
            mac = self.macro.name.replace(" ", "_")
            mac = mac.replace("&", "%26")
            mac = mac.replace("+", "%2B")
            url = "https://wiki.freecad.org/Macro_" + mac
            self.macro.fill_details_from_wiki(url)
        message = (
            "<h1>"
            + self.macro.name
            + "</h1>"
            + self.macro.desc
            + '<br/><br/>Macro location: <a href="'
            + self.macro.url
            + '">'
            + self.macro.url
            + "</a>"
        )
        if QtCore.QThread.currentThread().isInterruptionRequested():
            return
        self.readme_updated.emit(message)


class GetBasicAddonStatsWorker(QtCore.QThread):
    """Fetch data from an addon stats repository."""

    update_addon_stats = QtCore.Signal(Addon)

    def __init__(self, url: str, addons: List[Addon], parent: QtCore.QObject = None):
        super().__init__(parent)
        self.url = url
        self.addons = addons

    def run(self):
        """Fetch the remote data and load it into the addons"""

        fetch_result = NetworkManager.AM_NETWORK_MANAGER.blocking_get(self.url, 5000)
        if fetch_result is None:
            fci.Console.PrintError(
                translate(
                    "AddonsInstaller",
                    "Failed to get Addon statistics from {} -- only sorting alphabetically will"
                    " be accurate\n",
                ).format(self.url)
            )
            return
        text_result = fetch_result.data().decode("utf8")
        json_result = json.loads(text_result)

        for addon in self.addons:
            if addon.url in json_result:
                addon.stats = AddonStats.from_json(json_result[addon.url])
                self.update_addon_stats.emit(addon)


class GetAddonScoreWorker(QtCore.QThread):
    """Fetch data from an addon score file."""

    update_addon_score = QtCore.Signal(Addon)

    def __init__(self, url: str, addons: List[Addon], parent: QtCore.QObject = None):
        super().__init__(parent)
        self.url = url
        self.addons = addons

    def run(self):
        """Fetch the remote data and load it into the addons"""

        if self.url != "TEST":
            fetch_result = NetworkManager.AM_NETWORK_MANAGER.blocking_get(self.url, 5000)
            if fetch_result is None:
                fci.Console.PrintError(
                    translate(
                        "AddonsInstaller",
                        "Failed to get Addon score from '{}' -- sorting by score will fail\n",
                    ).format(self.url)
                )
                return
            text_result = fetch_result.data().decode("utf8")
            json_result = json.loads(text_result)
        else:
            fci.Console.PrintWarning("Running score generation in TEST mode...\n")
            json_result = {}
            for addon in self.addons:
                if addon.macro:
                    json_result[addon.name] = len(addon.macro.comment) if addon.macro.comment else 0
                else:
                    json_result[addon.url] = len(addon.description) if addon.description else 0

        for addon in self.addons:
            score = None
            if addon.url in json_result:
                score = json_result[addon.url]
            elif addon.name in json_result:
                score = json_result[addon.name]
            if score is not None:
                try:
                    addon.score = int(score)
                    self.update_addon_score.emit(addon)
                except (ValueError, OverflowError):
                    fci.Console.PrintLog(
                        f"Failed to convert score value '{score}' to an integer for {addon.name}"
                    )