1017 lines
42 KiB
Python
1017 lines
42 KiB
Python
# 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
|
|
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<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)
|
|
|
|
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 = (
|
|
"<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:
|
|
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}"
|
|
)
|