# SPDX-License-Identifier: LGPL-2.1-or-later # *************************************************************************** # * * # * Copyright (c) 2022-2023 FreeCAD Project Association * # * Copyright (c) 2019 Yorik van Havre * # * * # * 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 * # * . * # * * # *************************************************************************** """ 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 import xml.etree.ElementTree from PySide import QtCore import FreeCAD 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 = FreeCAD.Qt.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.""" status_message = QtCore.Signal(str) addon_repo = QtCore.Signal(object) 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 = os.path.join(FreeCAD.getUserAppDataDir(), "Mod") 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.", ) FreeCAD.Console.PrintError(message + "\n") self.status_message.emit(message) raise ConnectionError def _process_deprecated(self, deprecated_addons): """Parse the section on deprecated addons""" fc_major = int(FreeCAD.Version()[0]) fc_minor = int(FreeCAD.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: FreeCAD.Console.PrintMessage( f'Unrecognized Addon kind {item["kind"]} in deprecation list.' ) except ValueError: FreeCAD.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 = ( FreeCAD.ParamGet("User parameter:BaseApp/Preferences/Addons") .GetString("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 FreeCAD.Console.PrintWarning( translate("AddonsInstaller", "WARNING: Duplicate addon {} ignored").format( name ) ) continue FreeCAD.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.*)"\]\s*' r"path\s*=\s*(?P.+)\s*" r"url\s*=\s*(?Phttps?://.*)\s*" r"(branch\s*=\s*(?P[^\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) self.status_message.emit(translate("AddonsInstaller", "Workbenches list was updated.")) 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", ) self.status_message.emit(message) FreeCAD.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: FreeCAD.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) FreeCAD.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")): FreeCAD.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: FreeCAD.Console.PrintMessage( translate( "AddonsInstaller", "An error occurred updating macros from GitHub, trying clean checkout...", ) + f":\n{e}\n" ) FreeCAD.Console.PrintMessage(f"{macro_cache_location}\n") FreeCAD.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, ) FreeCAD.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 FreeCAD.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 FreeCAD.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("&", "&") 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: FreeCAD.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( FreeCAD.getUserCachePath(), "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: FreeCAD.Console.PrintLog(f"Failed loading {repo_metadata_cache_path}\n") FreeCAD.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(int, int) def __init__(self, repos: List[Addon]): QtCore.QThread.__init__(self) self.repos = repos self.current_thread = None self.basedir = FreeCAD.getUserAppDataDir() self.moddir = os.path.join(self.basedir, "Mod") 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 self.progress_made.emit(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.basedir = FreeCAD.getUserAppDataDir() self.moddir = os.path.join(self.basedir, "Mod") 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: FreeCAD.Console.PrintWarning( "AddonManager: " + translate( "AddonsInstaller", "Unable to fetch git updates for workbench {}", ).format(wb.name) + "\n" ) FreeCAD.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: FreeCAD.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: FreeCAD.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: FreeCAD.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(FreeCAD.getUserMacroDir(True), macro_wrapper.macro.filename) test_file_two = os.path.join( FreeCAD.getUserMacroDir(True), "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""" status_message = QtCore.Signal(str) update_macro = QtCore.Signal(Addon) progress_made = QtCore.Signal(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.status_message.emit(translate("AddonsInstaller", "Caching macro code...")) 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 FreeCAD.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) FreeCAD.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): FreeCAD.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: 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 self.progress_made.emit(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)) ) self.status_message.emit( translate( "AddonsInstaller", "Getting metadata from macro {}", ).format(next_repo.macro.name) ) 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 FreeCAD.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(): FreeCAD.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""" status_message = QtCore.Signal(str) 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""" self.status_message.emit(translate("AddonsInstaller", "Retrieving macro description...")) if not self.macro.parsed and self.macro.on_git: self.status_message.emit(translate("AddonsInstaller", "Retrieving info from git")) self.macro.fill_details_from_file(self.macro.src_filename) if not self.macro.parsed and self.macro.on_wiki: self.status_message.emit(translate("AddonsInstaller", "Retrieving info from 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 = ( "

" + self.macro.name + "

" + self.macro.desc + '

Macro location: ' + self.macro.url + "" ) 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: FreeCAD.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: FreeCAD.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: FreeCAD.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): FreeCAD.Console.PrintLog( f"Failed to convert score value '{score}' to an integer for {addon.name}" )