diff --git a/bin/updater/README.ES.md b/bin/updater/README.ES.md index 4b5068a..c9efa9b 100644 --- a/bin/updater/README.ES.md +++ b/bin/updater/README.ES.md @@ -42,7 +42,7 @@ Los valores utilizados para la configuración son: |----------------------|-------------|--------------------------------------------------------------------------------------------------------| | `folder` | `SI` | Carpeta donde se guardara la herramienta. Si no existe la misma se creara. | | `url` | `SI` | Web que se usara para realizar los chequeos con las regex. | -| `from` | `NO` | Indica la estrategia usada para el update. Los valores admitidos por el momento son: `web` o `github`. | +| `from` | `NO` | Indica la estrategia usada para el update. Los valores admitidos por el momento son: `web`, `github` o `http`. | | `local_version` | `NO` | Versión descargada actualmente. Este valor se actualizara con cada update. | | `re_version` | `NO` | Regex usado para comprobar si hay nuevas versiones en la web de `url`. | | `re_download` | `NO` | Regex usado para obtener el link de descarga en la web de `url`. | @@ -63,6 +63,7 @@ Combinando el uso de `update_url` y `re_download` se consiguen las siguientes es 2. Usando solo `re_download` se obtiene el link de descarga en la web de `url`. 3. Cuando se usan ambos parámetros se concatena el resultado de `re_download` con `update_url`. Esto es útil para arreglar los links de descarga de GitHub o Sourceforge. +4. También se dispone de un método de detección de nuevas versiones que en lugar de regex usa las cabeceras http con las que responde el servidor. ## Ejemplos @@ -121,5 +122,5 @@ SCHTASKS /DELETE /TN "ToolkitUpdater" ```bash pip install pyinstaller -pyinstaller --onefile updater.py --icon=appicon.ico +pyinstaller --onefile updater.py --icon=assets/appicon.ico ``` diff --git a/bin/updater/README.md b/bin/updater/README.md index 3d044e3..b633708 100644 --- a/bin/updater/README.md +++ b/bin/updater/README.md @@ -42,7 +42,7 @@ The values used for configuration are: |--------------------|-----------|------------------------------------------------------------------------------------------------| | `folder` | `YES` | Folder where the tool will be saved. If it does not exist, it will be created. | | `url` | `YES` | Web that will be used to perform the checks with the regex. | -| `from` | `NO` | Indicates the strategy used for the update. Currently supported values are: `web` or `github`. | +| `from` | `NO` | Indicates the strategy used for the update. Currently supported values are: `web`, `github` or `http`. | | `local_version` | `NO` | Currently downloaded version. This value will be updated with each update. | | `re_version` | `NO` | Regex used to check for new versions on the web used in `url`. | | `re_download` | `NO` | Regex used to get the download link on the web used in `url`. | @@ -61,6 +61,7 @@ Combining the use of `update_url` and `re_download` the following download strat 2. Using only `re_download` you get the download link on the web from `url`. 3. When using both parameters, the result of `re_download` is concatenated with `update_url`. This is useful for fixing GitHub or Sourceforge download links. +4. A new version detection method is also available that instead of regex uses the http headers with which the server responds. ## Examples @@ -119,5 +120,5 @@ SCHTASKS /DELETE /TN "ToolkitUpdater" ```bash pip install pyinstaller -pyinstaller --onefile updater.py --icon=appicon.ico +pyinstaller --onefile updater.py --icon=assets/appicon.ico ``` diff --git a/bin/updater/add-task.bat b/bin/updater/extras/add-task.bat similarity index 64% rename from bin/updater/add-task.bat rename to bin/updater/extras/add-task.bat index 9a14f68..ff6edda 100644 --- a/bin/updater/add-task.bat +++ b/bin/updater/extras/add-task.bat @@ -8,9 +8,6 @@ echo ------------------------------------------------- echo. cd /D "%~dp0" - -:: for hide window use cmdow.exe or hstart.exe -::SCHTASKS /CREATE /SC DAILY /TN "IndetectablesToolkit_Updater" /TR "%cd%\..\hstart\hstart.exe /NOCONSOLE %cd%\updater.exe" /ST 14:00 -SCHTASKS /CREATE /SC WEEKLY /TN "IndetectablesToolkit_Updater" /TR "%cd%\..\hstart\hstart.exe /NOCONSOLE %cd%\updater.exe" - +::SCHTASKS /CREATE /SC DAILY /TN "IndetectablesToolkit_Updater" /TR "%cd%\custom-task-loader.bat" /ST 14:00 +SCHTASKS /CREATE /SC WEEKLY /TN "IndetectablesToolkit_Updater" /TR "%cd%\custom-task-loader.bat" pause diff --git a/bin/updater/extras/custom-task-loader.bat b/bin/updater/extras/custom-task-loader.bat new file mode 100644 index 0000000..b4df3a4 --- /dev/null +++ b/bin/updater/extras/custom-task-loader.bat @@ -0,0 +1,8 @@ +@echo off + +::cd /D D:\code\indetectables\toolkit\Updater\ +cd /D "%~dp0" + +echo Executing Indetectables Toolkit Updater +echo %cd% +updater.exe > last-run.log diff --git a/bin/updater/delete-task.bat b/bin/updater/extras/delete-task.bat similarity index 100% rename from bin/updater/delete-task.bat rename to bin/updater/extras/delete-task.bat diff --git a/bin/updater/requirements.txt b/bin/updater/requirements.txt new file mode 100644 index 0000000..9976414 --- /dev/null +++ b/bin/updater/requirements.txt @@ -0,0 +1,5 @@ +requests>=2.26.0 +colorama>=0.4.4 +tqdm>=4.62.3 +py7zr>=0.16.1 +rarfile>=4.0 diff --git a/bin/updater/src/ColoredFormatter.py b/bin/updater/src/ColoredFormatter.py new file mode 100644 index 0000000..1c6021e --- /dev/null +++ b/bin/updater/src/ColoredFormatter.py @@ -0,0 +1,16 @@ +import colorama +import logging + + +class ColoredFormatter(logging.Formatter): + COLORS = { + 'DEBUG': colorama.Fore.CYAN, + # 'INFO': colorama.Fore.GREEN, + 'WARNING': colorama.Fore.YELLOW, + 'ERROR': colorama.Fore.RED, + 'CRITICAL': colorama.Fore.RED + colorama.Back.YELLOW, + } + + def format(self, record): + log_message = super().format(record) + return self.COLORS.get(record.levelname, '') + log_message + colorama.Style.RESET_ALL diff --git a/bin/updater/src/ConfigManager.py b/bin/updater/src/ConfigManager.py new file mode 100644 index 0000000..7a9615b --- /dev/null +++ b/bin/updater/src/ConfigManager.py @@ -0,0 +1,91 @@ +import configparser +import colorama + + +class ConfigManager: + """ + Manages configuration settings, including getting and setting values, + as well as updating the local version of a tool. + """ + + def __init__(self, config_file_name): + """ + Initialize with the name of the config file. + + :param config_file_name: Name of the configuration file + """ + self.config = configparser.ConfigParser() + self.config_file_name = config_file_name + self.config.read(self.config_file_name) + + def get_config(self, section, key, fallback=None): + """ + Get a configuration value. + + :param section: Section in the config file + :param key: Key in the section + :param fallback: Fallback value if key is not found + :return: Value of the configuration key + """ + return self.config.get(section, key, fallback=fallback) + + def set_config(self, section, key, value): + """ + Set a configuration value. + + :param section: Section in the config file + :param key: Key in the section + :param value: Value to set + """ + if not self.config.has_section(section): + self.config.add_section(section) + + self.config.set(section, key, value) + self.save_config() + + def get_tool_config(self, name): + """ + Retrieve the configuration for a specific tool. + + :param name: Name of the tool + :return: Dictionary containing the tool's configuration + """ + if name in self.config.sections(): + return dict(self.config.items(name)) + + raise Exception(colorama.Fore.RED + f'No entries were found for {name}') + + def get_boolean(self, section, option, fallback=None): + """ + Get a boolean configuration value. + + :param section: Section in the config file + :param option: Option in the section + :param fallback: Fallback value if option is not found + :return: Boolean value of the configuration option + """ + return self.config.getboolean(section, option, fallback=fallback) + + def get_sections(self): + """ + Get all sections in the configuration file. + + :return: List of section names + """ + return self.config.sections() + + def update_local_version(self, name, version): + """ + Update the local version of a tool in the configuration file. + + :param name: Name of the tool + :param version: New version to update + """ + self.set_config(name, 'local_version', version) + + def save_config(self): + """ + Save the current configuration to file. + """ + with open(self.config_file_name, 'w') as config_file: + self.config.write(config_file) diff --git a/bin/updater/src/Downloader.py b/bin/updater/src/Downloader.py new file mode 100644 index 0000000..6fad764 --- /dev/null +++ b/bin/updater/src/Downloader.py @@ -0,0 +1,91 @@ +import requests +from tqdm import tqdm +import pathlib +import colorama +import logging + + +class Downloader: + """Handles file downloads.""" + + def __init__(self, user_agent, disable_progress, update_folder_path): + """ + Initialize with optional user_agent, disable_progress flag, and update_folder_path. + + :param user_agent: User agent string for HTTP requests + :param disable_progress: Flag to disable progress bar + :param update_folder_path: Path to the folder where updates will be saved + """ + self.user_agent = user_agent + self.disable_progress = disable_progress + self.update_folder_path = update_folder_path + + def get_filename_from_url(self, url): + """ + Extract the filename from a given URL. + + :param url: URL string + :return: Extracted filename as a string + """ + fragment_removed = url.split('#')[0] # keep to left of first "#" + query_string_removed = fragment_removed.split('?')[0] + scheme_removed = query_string_removed.split('://')[-1].split(':')[-1] + + if scheme_removed.find('/') == -1: + return '' + + return pathlib.Path(scheme_removed).name + + def download_file(self, url, dest_path, progress_bar_desc): + """ + Download a file from a given URL. + + :param url: URL of the file to download + :param dest_path: Destination path to save the file + :param progress_bar_desc: Description for the progress bar + """ + headers = {'User-Agent': self.user_agent} + response = requests.get(url, headers=headers, stream=True) + response.raise_for_status() + + total_length = int(response.headers.get('content-length', 0)) + with open(dest_path, 'wb') as handle, tqdm( + disable=self.disable_progress, + colour='green', + ncols=100, + desc=progress_bar_desc, + total=total_length, + unit='iB', + unit_scale=True, + unit_divisor=1024, + ) as bar: + for block in response.iter_content(1024): + size = handle.write(block) + bar.update(size) + + bar.close() + + def download_from_web(self, tool_name, download_url): + """ + Perform a download step for a given tool. + + :param tool_name: Name of the tool + :param download_url: URL from which to download the tool + :return: Path where the file has been saved + """ + file_name = self.get_filename_from_url(download_url) + file_path = pathlib.Path(self.update_folder_path).joinpath(file_name) + + logging.info(f'{tool_name}: downloading update "{file_name}"') + + try: + self.download_file( + url=download_url, + dest_path=file_path, + progress_bar_desc=tool_name, + ) + except Exception as exception: + logging.error(f'{tool_name}: download url {download_url}') + raise Exception(colorama.Fore.RED + f'{tool_name}: Error {exception}') + + return file_path diff --git a/bin/updater/src/FileManager.py b/bin/updater/src/FileManager.py new file mode 100644 index 0000000..8fa6a08 --- /dev/null +++ b/bin/updater/src/FileManager.py @@ -0,0 +1,90 @@ +import os +import pathlib +import shutil +import logging + +from src.Helpers import Helpers + + +class FileManager: + """Handles file and folder operations like cleanup and copy.""" + + def __init__(self, script_path, disable_clean): + """ + Initialize FileManager with script path and clean option. + + :param script_path: Path to the script + :param disable_clean: Flag to disable folder cleanup + """ + self.script_path = script_path + self.disable_clean = disable_clean + self.tool_name = "" + self.tool_config = {} + + def tool_setup(self, tool_name, tool_config): + """ + Initialize tool-specific settings. + + :param tool_name: Name of the tool + :param tool_config: Configuration object for the specific tool + """ + self.tool_name = tool_name + self.tool_config = tool_config + + def get_tool_install_path(self): + """ + Retrieve the installation path for the tool. + + :return: Absolute path to the tool's installation folder + """ + tool_folder_path = self.tool_config['folder'] + if not pathlib.Path(tool_folder_path).is_absolute(): + tool_folder_path = pathlib.Path.resolve( + pathlib.Path(self.script_path).joinpath(tool_folder_path) + ) + + return tool_folder_path + + def processing_tool_path(self, tool_unpack_path): + """ + Process and return the tool's folder and unpack paths. + + :param tool_unpack_path: Path where the tool is unpacked + :return: Dictionary containing 'folder_path' and 'unpack_path' + """ + # dirty hack for correct folders structure + folder_list = os.listdir(tool_unpack_path) + folder_sample = pathlib.Path(tool_unpack_path).joinpath(folder_list[0]) + if len(folder_list) == 1 & os.path.isdir(folder_sample): + tool_unpack_path = folder_sample + + # tool folder + tool_folder_path = self.get_tool_install_path() + pathlib.Path(tool_folder_path).mkdir(parents=True, exist_ok=True) + + return { + 'folder_path': tool_folder_path, + 'unpack_path': tool_unpack_path, + } + + def save(self, tool_folder_path, tool_unpack_path): + """ + Save the tool to the specified folder path. + + :param tool_folder_path: Path to the folder where the tool will be saved + :param tool_unpack_path: Path where the tool is unpacked + :return: Dictionary containing 'tool_name', 'tool_folder', and 'save_compress_name' + """ + logging.info(f'{self.tool_name}: saving to folder {tool_folder_path}') + + use_merge = self.tool_config.get('merge', None) + if not self.disable_clean and not use_merge: + Helpers.cleanup_folder(tool_folder_path) + + shutil.copytree(tool_unpack_path, tool_folder_path, copy_function=shutil.copy, dirs_exist_ok=True) + + return { + 'tool_name': self.tool_name, + 'tool_folder': str(tool_folder_path), + 'save_compress_name': '', + } diff --git a/bin/updater/src/Helpers.py b/bin/updater/src/Helpers.py new file mode 100644 index 0000000..f4ff349 --- /dev/null +++ b/bin/updater/src/Helpers.py @@ -0,0 +1,18 @@ +import pathlib +import shutil + + +class Helpers: + + @staticmethod + def cleanup_folder(path): + """ + Clean up a folder by deleting all its contents. + + :param path: Path to the folder to be cleaned + """ + for file in pathlib.Path(path).iterdir(): + if file.is_dir(): + shutil.rmtree(file) + else: + file.unlink() diff --git a/bin/updater/src/Packer.py b/bin/updater/src/Packer.py new file mode 100644 index 0000000..ba23b24 --- /dev/null +++ b/bin/updater/src/Packer.py @@ -0,0 +1,202 @@ +import zipfile +import rarfile +import py7zr +import pathlib +import os +import shutil +import colorama +import logging + +from src.Helpers import Helpers + + +class Packer: + """Handles unpacking and repacking of compressed files.""" + + def __init__(self, update_folder_path, save_format_type, disable_clean): + """ + Initialize the Packer with configurations. + + :param update_folder_path: Path to the update folder + :param save_format_type: Format type for saving compressed files + :param disable_clean: Flag to disable cleaning + """ + self.update_folder_path = update_folder_path + self.save_format_type = save_format_type + self.disable_clean = disable_clean + self.tool_name = "" + self.tool_config = {} + self.valid_extensions = ['.zip', '.rar', '.7z'] + + def tool_setup(self, tool_name, tool_config): + """ + Initialize tool-specific settings. + + :param tool_name: Name of the tool + :param tool_config: Configuration object for the specific tool + """ + self.tool_name = tool_name + self.tool_config = tool_config + + def unpack_zip(self, file_path, unpack_path, file_pass=None): + """ + Unpack a ZIP file. + + :param file_path: Path to the ZIP file + :param unpack_path: Destination path to unpack + :param file_pass: Password for the ZIP file, default is None + """ + if file_pass: + file_pass = bytes(file_pass, 'utf-8') + + with zipfile.ZipFile(file_path, 'r') as compressed: + compressed.extractall(unpack_path, pwd=file_pass) + + def unpack_rar(self, file_path, unpack_path, file_pass=None): + """ + Unpack a RAR file. + + :param file_path: Path to the RAR file + :param unpack_path: Destination path to unpack + :param file_pass: Password for the RAR file, default is None + """ + # download first "UnRAR for Windows" from https://www.rarlab.com/rar_add.htm + # direct link: https://www.rarlab.com/rar/unrarw32.exe + with rarfile.RarFile(file_path, 'r') as compressed: + compressed.extractall(unpack_path, pwd=file_pass) + + def unpack_7z(self, file_path, unpack_path, file_pass=None): + """ + Unpack a 7z file. + + :param file_path: Path to the 7z file + :param unpack_path: Destination path to unpack + :param file_pass: Password for the 7z file, default is None + """ + with py7zr.SevenZipFile(file_path, 'r', password=file_pass) as compressed: + compressed.extractall(unpack_path) + + def unpack(self, file_path, file_ext, unpack_path, file_pass): + """ + Unpack a compressed file based on its extension. + + :param file_path: Path to the compressed file + :param file_ext: File extension (.zip, .rar, .7z) + :param unpack_path: Destination path to unpack + :param file_pass: Password for the compressed file, default is None + :return: None + :raises: Exception if file extension is not supported + """ + if file_ext == '.zip': + return self.unpack_zip(file_path, unpack_path, file_pass) + elif file_ext == '.rar': + return self.unpack_rar(file_path, unpack_path, file_pass) + elif file_ext == '.7z': + return self.unpack_7z(file_path, unpack_path, file_pass) + + raise Exception(colorama.Fore.RED + f'{file_ext} extension not accepted by internal unpacker') + + def unpack_step(self, file_path): + """ + Perform the unpack step for a specific tool. + + :param file_path: Path to the compressed file + :return: Path to the unpacked folder + """ + file_ext = pathlib.Path(file_path).suffix + update_folder = str(file_path).replace(file_ext, '') + unpack_path = pathlib.Path(self.update_folder_path).joinpath(update_folder) + + if file_ext in self.valid_extensions: + update_file_pass = self.tool_config.get('update_file_pass', None) + self.unpack(file_path, file_ext, unpack_path, update_file_pass) + else: + # this case is because there are times when the downloaded program is not compressed and is an exe + pathlib.Path(unpack_path).mkdir(exist_ok=True) + shutil.copy(file_path, unpack_path) + + # dirty hack for correct zip inside a zip... + folder_list = os.listdir(unpack_path) + folder_sample = pathlib.Path(unpack_path).joinpath(folder_list[0]) + file_ext = pathlib.Path(folder_sample).suffix + if len(folder_list) == 1 & (file_ext in self.valid_extensions): + self.unpack(folder_sample, file_ext, unpack_path, None) + os.remove(folder_sample) + + return unpack_path + + def repack_save_compress_name(self, name, version): + """ + Generate the compressed file name based on the save format type. + + :param name: Name of the tool + :param version: Version of the tool + :return: Compressed file name + """ + pack_name = f'{name} - {version}.7z' + if self.save_format_type == 'version': + pack_name = f'{version}.7z' + elif self.save_format_type == 'name': + pack_name = f'{name}.7z' + + return pack_name + + def repack_merge(self, tool_folder_path, tool_unpack_path): + """ + Merge the new unpacked version with the old one. + + :param tool_folder_path: Path to the tool folder + :param tool_unpack_path: Path to the unpacked folder + :return: None + """ + # checks and preparation + old_version = self.tool_config.get('local_version', '0') + old_compress_name = self.repack_save_compress_name(self.tool_name, old_version) + old_tool_compress_path = tool_folder_path.joinpath(old_compress_name) + if not old_tool_compress_path.exists(): + return False + + logging.info(f'{self.tool_name}: merging with "{old_compress_name}"') + + # unpack old version + old_tool_unpack_folder = str(old_compress_name).replace('.7z', '') + old_tool_unpack_path = pathlib.Path(self.update_folder_path).joinpath(old_tool_unpack_folder) + self.unpack(old_tool_compress_path, '.7z', old_tool_unpack_path, None) + + # merge + shutil.copytree(tool_unpack_path, old_tool_unpack_path, copy_function=shutil.copy, dirs_exist_ok=True) + shutil.rmtree(tool_unpack_path) + shutil.move(old_tool_unpack_path, tool_unpack_path, copy_function=shutil.copy) + + def repack_step(self, tool_folder_path, tool_unpack_path, unpack_folder_path, version): + """ + Perform the repack step for a specific tool. + + :param tool_folder_path: Path to the tool folder + :param tool_unpack_path: Path to the unpacked folder + :param unpack_folder_path: Path to the folder containing unpacked files + :param version: Version of the tool + :return: Dictionary containing tool name, tool folder, and compressed file name + """ + use_merge = self.tool_config.get('merge', None) + if use_merge: + self.repack_merge(tool_folder_path, tool_unpack_path) + + logging.info(f'{self.tool_name}: saving to folder {tool_folder_path}') + + if not self.disable_clean: + Helpers.cleanup_folder(tool_folder_path) + + save_compress_name = self.repack_save_compress_name(self.tool_name, version) + tool_repack_path = pathlib.Path(pathlib.Path(unpack_folder_path).parent).joinpath(save_compress_name) + + with py7zr.SevenZipFile(tool_repack_path, 'w') as archive: + archive.writeall(tool_unpack_path, arcname='') + + shutil.copy(tool_repack_path, tool_folder_path) + + return { + 'tool_name': self.tool_name, + 'tool_folder': str(tool_folder_path), + 'save_compress_name': save_compress_name, + } diff --git a/bin/updater/src/Scraper.py b/bin/updater/src/Scraper.py new file mode 100644 index 0000000..69a325e --- /dev/null +++ b/bin/updater/src/Scraper.py @@ -0,0 +1,322 @@ +import re +import requests +import urllib.parse +import binascii +import colorama +import logging + + +class Scraper: + """ + Handles all scraping tasks for the Updater. + """ + + def __init__(self, force_download, use_github_api, user_agent): + """ + Initialize the Scraper with necessary configurations. + + :param use_github_api: Boolean to determine if GitHub API should be used + :param user_agent: User agent string for HTTP requests + """ + self.user_agent = user_agent + self.force_download = force_download + self.use_github_api = use_github_api + self.tool_name = "" + self.tool_config = {} + self.github_version_check = 'https://github.com/{0}/releases.atom' + self.github_files = 'https://github.com/{0}/releases/expanded_assets/{1}' + self.github_api_files = 'https://api.github.com/repos/{0}/releases/latest' + self.re_github_version = '\/releases\/tag\/(\S+)"' + self.re_github_download = '"(.*?/{0})"' + + def tool_setup(self, tool_name, tool_config): + """ + Initialize tool-specific settings. + + :param tool_name: Name of the tool + :param tool_config: Configuration object for the specific tool + """ + self.tool_name = tool_name + self.tool_config = tool_config + + def web_get_request(self, url, headers=None): + """ + Performs a GET request to a given URL. + + :param url: The URL to perform the GET request to + :param headers: Optional dictionary containing HTTP headers. If not provided, the default User-Agent is used. + :return: Response object from the GET request + :raises Exception: If an error occurs during the request + """ + if headers is None: + headers = {'User-Agent': self.user_agent} + + try: + response = requests.get(url, headers=headers) + response.raise_for_status() + return response + except Exception as exception: + raise Exception(colorama.Fore.RED + f'{self.tool_name}: Error {exception}') + + ################# + # Scraper methods + ################# + def scrape_web(self): + """ + Scrape web for version and download URL based on tool_config. + + :return: Dictionary containing 'download_version' and 'download_url' + """ + update_url = self.tool_config.get('update_url', None) + re_version = self.tool_config.get('re_version', None) + re_download = self.tool_config.get('re_download', None) + + # load html + url = self.tool_config.get('url', None) + html_response = self.web_get_request(url) + logging.debug(f'{self.tool_name}: HTML content fetched, starting regex matching.') + + # regex shit + download_version = self.check_version_from_web(html_response.text, re_version) + download_url = self.get_download_url_from_web(html_response.text, url, update_url, re_download) + logging.debug(f'{self.tool_name}: Regex matching done.') + + return { + 'download_version': download_version, + 'download_url': download_url, + } + + def scrape_github(self): + """ + Scrape GitHub for version and download URL based on tool_config. + + :return: Dictionary containing 'download_version' and 'download_url' + """ + if self.use_github_api: + return self.scrape_github_api() + + github_repo = self.tool_config.get('url', None) + + # load html + version_url = self.github_version_check.format(github_repo) + version_html_response = self.web_get_request(version_url) + logging.debug(f'{self.tool_name}: Version HTML fetched, starting regex matching for version.') + + download_version = self.check_version_from_web(version_html_response.text, self.re_github_version) + logging.debug(f'{self.tool_name}: Regex matching for version done.') + + # load second html + update_url = self.tool_config.get('update_url', None) + if not update_url: + logging.debug(f'{self.tool_name}: update_url not set. I try to generate it.') + download_url = self.github_files.format(github_repo, download_version) + download_html_response = self.web_get_request(download_url) + + re_download = self.re_github_download.format(self.tool_config.get('re_download', None)) + update_url = self.get_download_url_from_web(download_html_response.text, version_url, '', re_download) + + return { + 'download_version': download_version, + 'download_url': update_url, + } + + def scrape_github_api(self): + """ + Scrape GitHub API for version and download URL based on tool_config. + + :return: Dictionary containing 'download_version' and 'download_url' + """ + logging.debug(f'{self.tool_name}: Consuming GitHub via Api') + github_repo = self.tool_config.get('url', None) + repo_url = self.github_api_files.format(github_repo) + + # load json + headers = {'Authorization': f'token {self.use_github_api}'} + api_response = self.web_get_request(repo_url, headers) + json_response = api_response.json() + logging.debug(f'{self.tool_name}: JSON fetched, extracting version and download URL.') + + update_url = self.tool_config.get('update_url', None) + if not update_url: + logging.debug(f'{self.tool_name}: update_url not set. I try to generate it.') + update_url = self.get_download_url_from_github(json_response) + + download_version = self.check_version_from_github(json_response) + logging.debug(f'{self.tool_name}: Version and download URL extracted.') + + # regex shit + return { + 'download_version': download_version, + 'download_url': update_url, + } + + def scrape_http(self): + """ + Scrape HTTP headers for version based on tool_config. + + :return: Dictionary containing 'download_version' and 'download_url' + """ + # get http response + update_url = self.tool_config.get('update_url', None) + headers = {'User-Agent': self.user_agent} + + try: + http_response = requests.head(update_url, headers=headers) + http_response.raise_for_status() + logging.debug(f'{self.tool_name}: HTTP headers fetched, extracting version.') + except Exception as exception: + raise Exception(colorama.Fore.RED + f'{self.tool_name}: Error {exception}') + + download_version = self.check_version_from_http(http_response.headers) + logging.debug(f'{self.tool_name}: Version extracted.') + + return { + 'download_version': download_version, + 'download_url': update_url, + } + + ################# + # Check methods + ################# + def check_version_from_web(self, html, re_version): + """ + Check version from web HTML content. + + :param html: HTML content + :param re_version: Regex pattern for version + :return: Version string + """ + local_version = self.tool_config.get('local_version', '0') + html_regex_version = re.findall(re_version, html) + + if not html_regex_version: + raise Exception(colorama.Fore.RED + f'{self.tool_name}: re_version regex not match ({re_version})') + + if not self.force_download and local_version == html_regex_version[0]: + raise Exception(f'{self.tool_name}: {local_version} is the latest version') + + logging.info(f'{self.tool_name}: updated from {local_version} --> {html_regex_version[0]}') + + return html_regex_version[0] + + def check_version_from_http(self, headers): + """ + Check version from HTTP headers. + + :param headers: HTTP headers + :return: Version string + """ + local_version = self.tool_config.get('local_version', '0') + + remote_version = None + if 'last-modified' in headers: + logging.info(f'{self.tool_name}: using "last-modified" as version number...') + input_bytes = headers['last-modified'].encode() + remote_version = str(binascii.crc32(input_bytes)) + elif 'content-length' in headers: + logging.info(f'{self.tool_name}: using "content-length" as version number...') + input_bytes = headers['content-length'].encode() + remote_version = str(binascii.crc32(input_bytes)) + else: + raise Exception(colorama.Fore.RED + + f'{self.tool_name}: no header is found with which to determine if there is an update') + + if not self.force_download and local_version == remote_version: + raise Exception(f'{self.tool_name}: {local_version} is the latest version') + + logging.info(f'{self.tool_name}: updated from {local_version} --> {remote_version}') + + return remote_version + + def check_version_from_github(self, json): + """ + Check version from GitHub API JSON response. + + :param json: JSON response from GitHub API + :return: Version string + """ + local_version = self.tool_config.get('local_version', '0') + + if not self.force_download and local_version == json['tag_name']: + raise Exception(f'{self.tool_name}: {local_version} is the latest version') + + logging.info(f'{self.tool_name}: updated from {local_version} --> {json["tag_name"]}') + + return json['tag_name'] + + ################# + # Check methods + ################# + def get_download_url_from_web(self, html, html_url, update_url, re_download): + """ + Get download URL from a web page using regex. + + :param html: HTML content of the web page + :param html_url: Original URL of the web page + :param update_url: Base URL for download + :param re_download: Regex pattern to find the download URL + :return: Download URL found or None + """ + # case 2: if update_url is not set, scrape the link from html + if re_download: + html_regex_download = re.findall(re_download, html) + if not html_regex_download: + raise Exception(colorama.Fore.RED + f'{self.tool_name}: re_download regex not match ({re_download})') + + # check if valid url + download_url_parse = urllib.parse.urlparse(html_regex_download[0]) + + # case 3: if update_url and re_download is set.... generate download link + if update_url: + update_url = f'{update_url}{html_regex_download[0]}' + elif download_url_parse.scheme == '': + html_url_parse_fix = urllib.parse.urlparse(html_url) + update_url = f'{html_url_parse_fix.scheme}://{html_url_parse_fix.netloc}/{html_regex_download[0]}' + else: + update_url = html_regex_download[0] + + # case 1: if update_url is set... download it! + if not update_url: + raise Exception(colorama.Fore.RED + f'{self.tool_name}: update_url not generated!') + + return update_url + + def get_download_url_from_github(self, json): + """ + Get download URL from GitHub API JSON response. + + :param json: JSON response from GitHub API + :return: Download URL found or None + """ + re_download = self.tool_config.get('re_download', None) + if not re_download: + raise Exception(colorama.Fore.RED + f'{self.tool_name}: re_download regex not set') + + update_url = None + for attachment in json['assets']: + html_regex_download = re.findall(re_download, attachment['browser_download_url']) + if html_regex_download: + update_url = attachment['browser_download_url'] + break + + if not update_url: + raise Exception(colorama.Fore.RED + f'{self.tool_name}: re_download regex not match ({re_download})') + + return update_url + + ################# + # Scrape step + ################# + def scrape_step(self): + """ + Execute a specific script for a given tool based on tool_config. + + :return: Dictionary containing 'download_version' and 'download_url' + """ + from_url = self.tool_config.get('from', 'web') + if from_url == 'github': + return self.scrape_github() + elif from_url == 'http': + return self.scrape_http() + + return self.scrape_web() diff --git a/bin/updater/src/ScriptExecutor.py b/bin/updater/src/ScriptExecutor.py new file mode 100644 index 0000000..dabec84 --- /dev/null +++ b/bin/updater/src/ScriptExecutor.py @@ -0,0 +1,56 @@ +import subprocess +import colorama +import logging + + +class ScriptExecutor: + """ + Class responsible for executing scripts during the update process. + """ + + def __init__(self): + """ + Initialize ScriptExecutor. + """ + self.tool_name = "" + self.tool_config = {} + self.valid_types = ['post_unpack', 'pre_update', 'post_update'] + + def tool_setup(self, tool_name, tool_config): + """ + Initialize tool-specific settings. + + :param tool_name: Name of the tool + :param tool_config: Configuration object for the specific tool + """ + self.tool_name = tool_name + self.tool_config = tool_config + + def execute_script(self, script_type): + """ + Execute a specific script for a given tool. + + :param script_type: Type of script to execute ('pre_update', 'post_update', 'post_unpack') + """ + if script_type in self.valid_types and script_type in self.tool_config: + script = self.tool_config[script_type] + logging.info(f'{self.tool_name}: exec {script_type} "{script}"') + logging.info(colorama.Fore.BLUE + '------------------------------') + subprocess.run(script) + logging.info(colorama.Fore.BLUE + '------------------------------') + + def execute_global_script(self, processing_info): + """ + Execute a global script. + + :param processing_info: Dictionary containing processing information + """ + if 'global_post_update' in self.tool_config: + script = self.tool_config['global_post_update'] + logging.info(f'{self.tool_name}: exec global script "{script}"') + subprocess.run([ + script, + processing_info['tool_name'], + processing_info['tool_folder'], + processing_info['save_compress_name'], + ]) diff --git a/bin/updater/src/UpdateManager.py b/bin/updater/src/UpdateManager.py new file mode 100644 index 0000000..f686340 --- /dev/null +++ b/bin/updater/src/UpdateManager.py @@ -0,0 +1,291 @@ +import argparse +import signal +import sys +import os +import colorama +import logging + +from src.Updater import Updater +from src.ConfigManager import ConfigManager +from src.ColoredFormatter import ColoredFormatter + + +class UpdateManager: + """ + Manages the application setup, including argument parsing and update handling. + """ + + def __init__(self): + """ + Initialize the UpdateManager with a ConfigManager instance and command-line arguments. + """ + self.version = '2.0.0' + self.config_file_name = 'tools.ini' + self.config_section_defaults = 'UpdaterConfig' + self.config_section_self_update = 'UpdaterAutoUpdater' + self.arguments = {} + self.config_manager = ConfigManager(self.config_file_name) + colorama.init(autoreset=True) + + def print_banner(self): + """ + Prints the banner for the update process. + """ + print(colorama.Fore.YELLOW + f""" + ____ __ __ __ __ __ + / _/___ ____/ /__ / /____ _____/ /_____ _/ /_ / /__ _____ + / // __ \/ __ / _ \/ __/ _ \/ ___/ __/ __ `/ __ \/ / _ \/ ___/ + _/ // / / / /_/ / __/ /_/ __/ /__/ /_/ /_/ / /_/ / / __(__ ) + /___/_/ /_/\__,_/\___/\__/\___/\___/\__/\__,_/_.___/_/\___/____/ + + Universal Tool Updater - by DSR! + Web: https://github.com/xchwarze/universal-tool-updater + Version: {self.version} + """) + + def exit_handler(self, signal, frame): + """ + Handles signals like SIGINT for graceful exit. + + :param signal: Signal type + :param frame: Current stack frame + """ + print(colorama.Fore.YELLOW + 'SIGINT or CTRL-C detected. Exiting gracefully') + sys.exit(0) + + def get_argparse_default(self, option, default, is_bool=True): + """ + Retrieves the default value for a given argparse option from the configuration. + + :param option: The name of the argparse option + :param default: The default value to return if the option is not found in the configuration + :param is_bool: Flag indicating if the option is a boolean + :return: The default value for the argparse option + """ + if is_bool: + # return value.lower() in ('true', '1', 'yes', 'on') + return self.config_manager.get_boolean(self.config_section_defaults, option, fallback=default) + + return self.config_manager.get_config(self.config_section_defaults, option, fallback=default) + + def parse_arguments(self): + """ + Parses command-line arguments and updates the internal state. + """ + parser = argparse.ArgumentParser(usage='%(prog)s [ARGUMENTS]') + parser.add_argument( + '-v', + '--version', + action='version', + version=f'Version: {self.version}' + ) + parser.add_argument( + '-u', + '--update', + dest='update', + help='list of tools (default: all)', + nargs='*' + ) + parser.add_argument( + '-dsu', + '--disable-self-update', + dest='disable_self_update', + help='disable self update of this script', + action=argparse.BooleanOptionalAction, + default=False + ) + parser.add_argument( + '-dfc', + '--disable-folder-clean', + dest='disable_clean', + help='disable tool folder clean', + action=argparse.BooleanOptionalAction, + default=self.get_argparse_default('disable_clean', False) + ) + parser.add_argument( + '-dr', + '--disable-repack', + dest='disable_repack', + help='disable tool repack', + action=argparse.BooleanOptionalAction, + default=self.get_argparse_default('disable_repack', False) + ) + parser.add_argument( + '-dic', + '--disable-install-check', + dest='disable_install_check', + help='disable tool install check', + action=argparse.BooleanOptionalAction, + default=self.get_argparse_default('disable_install_check', False) + ) + parser.add_argument( + '-dpb', + '--disable-progress-bar', + dest='disable_progress', + help='disable download progress bar', + action=argparse.BooleanOptionalAction, + default=self.get_argparse_default('disable_progress', False) + ) + parser.add_argument( + '-sft', + '--save-format-type', + dest='save_format_type', + help='compress save format name', + choices=['full', 'version', 'name'], + default=self.get_argparse_default('save_format_type', 'full', False) + ) + parser.add_argument( + '-f', + '--force', + dest='force_download', + help='force download', + action=argparse.BooleanOptionalAction, + default=False + ) + parser.add_argument( + '-uga', + '--use-github-api', + dest='use_github_api', + help='use github api with this token', + default=self.get_argparse_default('use_github_api', '', False) + ) + parser.add_argument( + '-udp', + '--update-default-params', + dest='update_default_params', + help='update default params', + action=argparse.BooleanOptionalAction, + default=False + ) + parser.add_argument( + '-d', + '--debug', + dest='debug', + help='enable debug output', + action=argparse.BooleanOptionalAction, + default=False + ) + + self.arguments = parser.parse_args() + + def update_default_params(self): + """ + Updates default parameters in the configuration based on command-line arguments. + """ + if not self.arguments.update_default_params: + return False + + self.config_manager.set_config(self.config_section_defaults, 'disable_clean', str(self.arguments.disable_clean)) + self.config_manager.set_config(self.config_section_defaults, 'disable_repack', str(self.arguments.disable_repack)) + self.config_manager.set_config(self.config_section_defaults, 'disable_install_check', + str(self.arguments.disable_install_check)) + self.config_manager.set_config(self.config_section_defaults, 'disable_progress', str(self.arguments.disable_progress)) + self.config_manager.set_config(self.config_section_defaults, 'save_format_type', self.arguments.save_format_type) + self.config_manager.set_config(self.config_section_defaults, 'use_github_api', self.arguments.use_github_api) + + logging.info(colorama.Fore.GREEN + 'Update default params successful') + + def set_logging_level(self): + """ + Sets the logging level based on the 'debug' argument. + If 'debug' is True, sets the logging level to DEBUG. + Otherwise, sets the logging level to INFO. + """ + logger = logging.getLogger() + if self.arguments.debug: + logger.setLevel(logging.DEBUG) + else: + logger.setLevel(logging.INFO) + + for old_handler in logger.handlers: + logger.removeHandler(old_handler) + + #formatter = ColoredFormatter('%(asctime)s - %(levelname)s - %(message)s') + formatter = ColoredFormatter('%(message)s') + handler = logging.StreamHandler() + handler.setFormatter(formatter) + logger.addHandler(handler) + + def change_current_directory(self): + """ + Changes the current directory to the directory where the script is located. + This fixes "FileNotFoundError" in exe builds. + """ + current_dir = os.path.dirname(sys.argv[0]) + if current_dir: + os.chdir(current_dir) + + def generate_update_list(self): + """ + Generates the list of tools to be updated. + + :return: List of tools to update + """ + update_list = self.arguments.update + if not update_list: + update_list = self.config_manager.get_sections() + + # delete config data of this script + if self.config_section_defaults in update_list: + update_list.remove(self.config_section_defaults) + + # delete self updater config + if self.config_section_self_update in update_list: + update_list.remove(self.config_section_self_update) + + return update_list + + def handle_auto_update(self): + """ + Handles the auto-update logic for the script itself. + """ + updater = Updater( + config_manager=self.config_manager, + ) + if self.config_section_self_update in self.config_manager.get_sections() and \ + not self.arguments.disable_self_update: + try: + updater.update(self.config_section_self_update) + except Exception as exception: + logging.info(exception) + + def handle_tool_updates(self, updater, update_list): + """ + Handles the update process for each tool in the update list. + + :param updater: An instance of the Updater class responsible for updating tools + :param update_list: List of tools to update + """ + for tool in update_list: + try: + updater.update(tool) + except Exception as exception: + logging.info(exception) + + def handle_updates(self): + """ + Orchestrates the update process based on the configuration. + """ + self.handle_auto_update() + + updater = Updater( + config_manager=self.config_manager, + + # To convert from Namespace to Dict I have to use vars() + updater_setup=vars(self.arguments), + ) + update_list = self.generate_update_list() + self.handle_tool_updates(updater, update_list) + updater.cleanup_update_folder() + + def main(self): + """ + Main entry point for the UpdateManager. + """ + signal.signal(signal.SIGINT, self.exit_handler) + self.change_current_directory() + self.print_banner() + self.parse_arguments() + self.set_logging_level() + self.update_default_params() + self.handle_updates() diff --git a/bin/updater/src/Updater.py b/bin/updater/src/Updater.py new file mode 100644 index 0000000..94a622e --- /dev/null +++ b/bin/updater/src/Updater.py @@ -0,0 +1,167 @@ +import os +import pathlib +import colorama +import logging + +from src.Scraper import Scraper +from src.Downloader import Downloader +from src.Packer import Packer +from src.FileManager import FileManager +from src.ScriptExecutor import ScriptExecutor +from src.Helpers import Helpers + + +class Updater: + """ + Manages the update process for a tool. The class is responsible for + scraping the latest version, downloading updates, unpacking, and saving or + repacking the tool. It also handles pre-update and post-update scripts. + """ + + def __init__(self, config_manager, updater_setup = {}): + """ + Initialize the Updater class with various configurations. + + :param config_manager: Configuration manager instance + :param updater_setup: Dictionary containing various flags and settings for the updater. + Possible keys are: + - force_download: Flag to force download + - disable_repack: Flag to disable repacking + - disable_clean: Flag to disable cleaning + - disable_install_check: Flag to disable installation check + - disable_progress: Flag to disable progress bar + - save_format_type: Format type for saving (default "full") + - use_github_api: Flag to use GitHub API. The value is the token to use the api. + """ + self.tool_name = "" + self.tool_config = {} + self.script_path = os.fsdecode(os.getcwdb()) + self.update_folder_path = pathlib.Path(self.script_path).joinpath('updates') + self.request_user_agent = 'curl/7.84.0' + self.config_manager = config_manager + self.disable_install_check = updater_setup.get('disable_install_check', False) + self.disable_repack = updater_setup.get('disable_repack', False) + self.scraper = Scraper( + force_download=updater_setup.get('force_download', False), + use_github_api=updater_setup.get('use_github_api', ''), + user_agent=self.request_user_agent, + ) + self.downloader = Downloader( + disable_progress=updater_setup.get('disable_progress', False), + user_agent=self.request_user_agent, + update_folder_path=self.update_folder_path, + ) + self.packer = Packer( + save_format_type=updater_setup.get('save_format_type', 'full'), + disable_clean=updater_setup.get('disable_clean', False), + update_folder_path=self.update_folder_path, + ) + self.file_manager = FileManager( + disable_clean=updater_setup.get('disable_clean', False), + script_path=self.script_path, + ) + self.script_executor = ScriptExecutor() + + def check_tool_installed(self): + """ + Check if the tool is installed. Raise an exception if not found. + """ + tool_folder_path = self.file_manager.get_tool_install_path() + if not tool_folder_path.exists() and not self.disable_install_check: + raise Exception(colorama.Fore.YELLOW + f'{self.tool_name}: The program was not found') + + def pre_update(self): + """ + Execute pre-update checks and scripts. + """ + self.check_tool_installed() + self.script_executor.execute_script('pre_update') + + def post_update(self, processing_info): + """ + Execute post-update scripts. + + :param processing_info: Information about the processing step + """ + self.script_executor.execute_script('post_update') + self.script_executor.execute_global_script(processing_info) + + def download_step(self, download_url): + """ + Download the tool from the given URL. + + :param download_url: URL to download the tool from + :return: Path to the downloaded file + """ + # create updates folder if don't exist + if not pathlib.Path.exists(self.update_folder_path): + pathlib.Path.mkdir(self.update_folder_path) + + # download + self.cleanup_update_folder() + return self.downloader.download_from_web(self.tool_name, download_url) + + def processing_tool_step(self, file_path, download_version): + """ + Process the downloaded tool. + + :param file_path: Path to the downloaded file + :param download_version: Version of the downloaded tool + :return: Dictionary containing processing information + """ + # unpack logic + unpack_folder_path = self.packer.unpack_step(file_path) + self.script_executor.execute_script('post_unpack') + + # save or repack logic + disable_repack = self.tool_config.get('disable_repack', None) + tool_path = self.file_manager.processing_tool_path(unpack_folder_path) + if self.disable_repack or disable_repack: + return self.file_manager.save( + tool_folder_path=tool_path['folder_path'], + tool_unpack_path=tool_path['unpack_path'], + ) + + return self.packer.repack_step( + tool_folder_path=tool_path['folder_path'], + tool_unpack_path=tool_path['unpack_path'], + unpack_folder_path=unpack_folder_path, + version=download_version, + ) + + def cleanup_update_folder(self): + """ + Clean up the update folder. + """ + if pathlib.Path.exists(self.update_folder_path): + Helpers.cleanup_folder(self.update_folder_path) + + def update(self, tool_name): + """ + Perform the update process for a given tool. + + :param tool_name: Name of the tool to update + """ + # tool data setup + self.tool_name = tool_name + self.tool_config = self.config_manager.get_tool_config(tool_name) + self.scraper.tool_setup(self.tool_name, self.tool_config) + self.packer.tool_setup(self.tool_name, self.tool_config) + self.file_manager.tool_setup(self.tool_name, self.tool_config) + self.script_executor.tool_setup(self.tool_name, self.tool_config) + + # execute checks and scripts + self.pre_update() + + # generate version and download data + scrape_data = self.scraper.scrape_step() + + # download and process file + update_file_path = self.download_step(scrape_data['download_url']) + processing_info = self.processing_tool_step(update_file_path, scrape_data['download_version']) + + # update complete + self.config_manager.update_local_version(self.tool_name, scrape_data['download_version']) + self.post_update(processing_info) + + logging.info(f'{self.tool_name}: update complete') diff --git a/bin/updater/unrar.exe b/bin/updater/unrar.exe index 16c7094..5c51398 100644 Binary files a/bin/updater/unrar.exe and b/bin/updater/unrar.exe differ diff --git a/bin/updater/updater.exe b/bin/updater/updater.exe index c1caca7..fdd31e9 100644 Binary files a/bin/updater/updater.exe and b/bin/updater/updater.exe differ diff --git a/bin/updater/updater.py b/bin/updater/updater.py index 2779999..7afb445 100644 --- a/bin/updater/updater.py +++ b/bin/updater/updater.py @@ -1,703 +1,7 @@ -# -*- coding: utf-8 -*- -# -# Copyright (C) 2021 - 2023 DSR! -# Released under the terms of the MIT License -# Developed for Python 3.6+ -# pip install requests py7zr rarfile colorama tqdm +# Import the UpdateManager class from the src package +from src.UpdateManager import UpdateManager -import argparse -import configparser -import requests -import re -import os -import shutil -import pathlib -import zipfile -import py7zr -import rarfile -import subprocess -import signal -import sys -import colorama -import tqdm -import urllib -import binascii - - -# Helpers functions -def get_filename_from_url(url): - fragment_removed = url.split('#')[0] # keep to left of first # - query_string_removed = fragment_removed.split('?')[0] - scheme_removed = query_string_removed.split('://')[-1].split(':')[-1] - - if scheme_removed.find('/') == -1: - return '' - - return pathlib.Path(scheme_removed).name - - -def cleanup_folder(path): - for file in pathlib.Path(path).iterdir(): - if file.is_dir(): - shutil.rmtree(file) - else: - file.unlink() - - -def download(url, file_path, progress_bar_desc, disable_progress_bar=False): - file_response = requests.get(url, allow_redirects=True, stream=True) - file_response.raise_for_status() - - # for debug redirects - # print('DEBUG: download url "{0}"'.format(file_response.url)) - - total_length = int(file_response.headers.get('content-length', 0)) - with open(file_path, 'wb') as handle, tqdm.tqdm( - disable=disable_progress_bar, - colour='green', - ncols=100, - desc=progress_bar_desc, - total=total_length, - unit='iB', - unit_scale=True, - unit_divisor=1024, - ) as bar: - for block in file_response.iter_content(1024): - size = handle.write(block) - bar.update(size) - - bar.close() - - -def unpack(file_path, file_ext, unpack_path, file_pass): - if file_ext == '.zip': - if file_pass: - file_pass = bytes(file_pass, 'utf-8') - - with zipfile.ZipFile(file_path, 'r') as compressed: - compressed.extractall(unpack_path, pwd=file_pass) - - elif file_ext == '.rar': - # downlad first "UnRAR for Windows" from https://www.rarlab.com/rar_add.htm - # https://www.rarlab.com/rar/unrarw32.exe - with rarfile.RarFile(file_path, 'r') as compressed: - compressed.extractall(unpack_path, pwd=file_pass) - - elif file_ext == '.7z': - with py7zr.SevenZipFile(file_path, 'r', password=file_pass) as compressed: - compressed.extractall(unpack_path) - - else: - pathlib.Path(unpack_path).mkdir(exist_ok=True) - shutil.copy(file_path, unpack_path) - - -def save_config_to_file(config, config_file_name): - with open(config_file_name, 'w') as config_file: - config.write(config_file) - - -# Main Updater class -class Updater: - def __init__( - self, config, config_file_name, force_download, disable_repack, disable_clean, - disable_install_check, disable_progress, save_format_type, use_github_api, - ): - self.name = '' - self.config = config - self.config_file_name = config_file_name - self.force_download = force_download - self.disable_repack = disable_repack - self.disable_clean = disable_clean - self.disable_install_check = disable_install_check - self.disable_progress = disable_progress - self.save_format_type = save_format_type - self.script_path = os.fsdecode(os.getcwdb()) - self.update_folder_path = pathlib.Path(self.script_path).joinpath('updates') - self.use_github_api = use_github_api - self.github_version_check = 'https://github.com/{0}/releases.atom' - self.github_files = 'https://github.com/{0}/releases/expanded_assets/{1}' - self.github_api_files = 'https://api.github.com/repos/{0}/releases/latest' - self.re_github_version = '\/releases\/tag\/(\S+)"' - self.re_github_download = '"(.*?/{0})"' - self.request_user_agent = 'curl/7.84.0' - - def _check_version_from_web(self, html, re_version): - local_version = self.config.get(self.name, 'local_version', fallback='0') - html_regex_version = re.findall(re_version, html) - - if not html_regex_version: - raise Exception(colorama.Fore.RED + '{0}: re_version regex not match ({1})'.format(self.name, re_version)) - - if not self.force_download and local_version == html_regex_version[0]: - raise Exception('{0}: {1} is the latest version'.format(self.name, local_version)) - - print('{0}: updated from {1} --> {2}'.format(self.name, local_version, html_regex_version[0])) - - return html_regex_version[0] - - def _get_download_url_from_web(self, html, html_url, update_url, re_download): - # case 2: if update_url is not set, scrape the link from html - if re_download: - html_regex_download = re.findall(re_download, html) - if not html_regex_download: - raise Exception(colorama.Fore.RED + '{0}: re_download regex not match ({1})'.format(self.name, re_download)) - - # check if valid url - download_url_parse = urllib.parse.urlparse(html_regex_download[0]) - - # case 3: if update_url and re_download is set.... generate download link - if update_url: - update_url = '{0}{1}'.format(update_url, html_regex_download[0]) - elif download_url_parse.scheme == '': - html_url_parse_fix = urllib.parse.urlparse(html_url) - update_url = '{0}://{1}/{2}'.format(html_url_parse_fix.scheme, html_url_parse_fix.netloc, html_regex_download[0]) - else: - update_url = html_regex_download[0] - - # case 1: if update_url is set... download it! - if not update_url: - raise Exception(colorama.Fore.RED + '{0}: update_url not generated!'.format(self.name)) - - return update_url - - def _scrape_web(self, url, update_url, re_version, re_download): - # load html - headers={'User-Agent': self.request_user_agent} - html_response = requests.get(url, headers=headers) - html_response.raise_for_status() - - # regex shit - return { - 'download_version': self._check_version_from_web(html_response.text, re_version), - 'download_url': self._get_download_url_from_web(html_response.text, url, update_url, re_download), - } - - def _check_version_from_github(self, json): - local_version = self.config.get(self.name, 'local_version', fallback='0') - - if not self.force_download and local_version == json['tag_name']: - raise Exception('{0}: {1} is the latest version'.format(self.name, local_version)) - - print('{0}: updated from {1} --> {2}'.format(self.name, local_version, json['tag_name'])) - - return json['tag_name'] - - def _get_download_url_from_github(self, json, re_download): - if not re_download: - raise Exception(colorama.Fore.RED + '{0}: re_download regex not set'.format(self.name)) - - update_url = None - for attachment in json['assets']: - html_regex_download = re.findall(re_download, attachment['browser_download_url']) - if html_regex_download: - update_url = attachment['browser_download_url'] - break - - if not update_url: - raise Exception(colorama.Fore.RED + '{0}: re_download regex not match ({1})'.format(self.name, re_download)) - - return update_url - - def _scrape_github(self, github_repo, update_url, re_download): - if self.use_github_api: - return self._scrape_github_api(github_repo, re_download) - - headers = {'User-Agent': self.request_user_agent} - - # load html - version_url = self.github_version_check.format(github_repo) - version_html_response = requests.get(version_url, headers=headers) - version_html_response.raise_for_status() - - download_version = self._check_version_from_web(version_html_response.text, self.re_github_version) - - # load second html - if not update_url: - download_url = self.github_files.format(github_repo, download_version) - download_html_response = requests.get(download_url, headers=headers) - download_html_response.raise_for_status() - - re_download = self.re_github_download.format(re_download) - update_url = self._get_download_url_from_web(download_html_response.text, version_url, '', re_download) - - return { - 'download_version': download_version, - 'download_url': update_url, - } - - def _scrape_github_api(self, github_repo, update_url, re_download): - repo_url = self.github_api_files.format(github_repo) - - # load json - headers = {'Authorization': f'token {self.use_github_api}'} - html_response = requests.get(repo_url, headers=headers) - html_response.raise_for_status() - json_response = html_response.json() - - if not update_url: - update_url = self._get_download_url_from_github(json_response, re_download) - - # regex shit - return { - 'download_version': self._check_version_from_github(json_response), - 'download_url': update_url, - } - - def _check_version_from_http(self, headers): - local_version = self.config.get(self.name, 'local_version', fallback='0') - - remote_version = None - if 'last-modified' in headers: - input_bytes = headers['last-modified'].encode() - remote_version = str(binascii.crc32(input_bytes)) - elif 'content-length' in headers: - print('{0}: using "content-length" as version number...'.format(self.name)) - input_bytes = headers['content-length'].encode() - remote_version = str(binascii.crc32(input_bytes)) - else: - raise Exception(colorama.Fore.RED + '{0}: no header is found with which to determine if there is an update'.format(self.name)) - - if not self.force_download and local_version == remote_version: - raise Exception('{0}: {1} is the latest version'.format(self.name, local_version)) - - print('{0}: updated from {1} --> {2}'.format(self.name, local_version, remote_version)) - - return remote_version - - def _scrape_http(self, update_url): - # get http response - headers={'User-Agent': self.request_user_agent} - http_response = requests.head(update_url, headers=headers) - http_response.raise_for_status() - - return { - 'download_version': self._check_version_from_http(http_response.headers), - 'download_url': update_url, - } - - def _unpack(self, file_path): - file_ext = pathlib.Path(file_path).suffix - update_folder = str(file_path).replace(file_ext, '') - - update_file_pass = self.config.get(self.name, 'update_file_pass', fallback=None) - unpack_path = pathlib.Path(self.update_folder_path).joinpath(update_folder) - unpack(file_path, file_ext, unpack_path, update_file_pass) - - # dirty hack for correct compress inside compress... - folder_list = os.listdir(unpack_path) - folder_sample = pathlib.Path(unpack_path).joinpath(folder_list[0]) - file_ext = pathlib.Path(folder_sample).suffix - if len(folder_list) == 1 & (file_ext in ['.zip', '.rar', '.7z']): - unpack(folder_sample, file_ext, unpack_path, None) - os.remove(folder_sample) - - return unpack_path - - def _save(self, use_merge, tool_folder_path, tool_unpack_path): - print('{0}: saving to folder {1}'.format(self.name, tool_folder_path)) - - if not self.disable_clean and not use_merge: - cleanup_folder(tool_folder_path) - - shutil.copytree(tool_unpack_path, tool_folder_path, copy_function=shutil.copy, dirs_exist_ok=True) - - return { - 'tool_name': self.name, - 'tool_folder': str(tool_folder_path), - 'save_compress_name': '', - } - - def _repack_save_compress_name(self, name, version): - pack_name = '{0} - {1}.7z'.format(name, version) - if self.save_format_type == 'version': - pack_name = '{0}.7z'.format(version) - elif self.save_format_type == 'name': - pack_name = '{0}.7z'.format(name) - - return pack_name - - def _repack_merge(self, tool_folder_path, tool_unpack_path): - # checks and preparation - old_version = self.config.get(self.name, 'local_version', fallback='0') - old_compress_name = self._repack_save_compress_name(self.name, old_version) - old_tool_compress_path = tool_folder_path.joinpath(old_compress_name) - if not old_tool_compress_path.exists(): - return False - - print('{0}: merging with "{1}"'.format(self.name, old_compress_name)) - - # unpack old version - old_tool_unpack_folder = str(old_compress_name).replace('.7z', '') - old_tool_unpack_path = pathlib.Path(self.update_folder_path).joinpath(old_tool_unpack_folder) - unpack(old_tool_compress_path, '.7z', old_tool_unpack_path, None) - - # merge - shutil.copytree(tool_unpack_path, old_tool_unpack_path, copy_function=shutil.copy, dirs_exist_ok=True) - shutil.rmtree(tool_unpack_path) - shutil.move(old_tool_unpack_path, tool_unpack_path, copy_function=shutil.copy) - - def _repack(self, use_merge, tool_folder_path, tool_unpack_path, unpack_folder_path, version): - if use_merge: - self._repack_merge(tool_folder_path, tool_unpack_path) - - print('{0}: saving to folder {1}'.format(self.name, tool_folder_path)) - - if not self.disable_clean: - cleanup_folder(tool_folder_path) - - save_compress_name = self._repack_save_compress_name(self.name, version) - tool_repack_path = pathlib.Path(pathlib.Path(unpack_folder_path).parent).joinpath(save_compress_name) - - with py7zr.SevenZipFile(tool_repack_path, 'w') as archive: - archive.writeall(tool_unpack_path, arcname='') - - shutil.copy(tool_repack_path, tool_folder_path) - - return { - 'tool_name': self.name, - 'tool_folder': str(tool_folder_path), - 'save_compress_name': save_compress_name, - } - - def _tool_install_path(self): - tool_folder_path = self.config.get(self.name, 'folder') - if not pathlib.Path(tool_folder_path).is_absolute(): - tool_folder_path = pathlib.Path.resolve( - pathlib.Path(self.script_path).joinpath(tool_folder_path) - ) - - return tool_folder_path - - def _processing_tool(self, tool_unpack_path): - # dirty hack for correct folders structure - folder_list = os.listdir(tool_unpack_path) - folder_sample = pathlib.Path(tool_unpack_path).joinpath(folder_list[0]) - if len(folder_list) == 1 & os.path.isdir(folder_sample): - tool_unpack_path = folder_sample - - # tool folder - tool_folder_path = self._tool_install_path() - pathlib.Path(tool_folder_path).mkdir(parents=True, exist_ok=True) - - return { - 'folder_path': tool_folder_path, - 'unpack_path': tool_unpack_path, - } - - def _bump_version(self, latest_version): - self.config.set(self.name, 'local_version', latest_version) - save_config_to_file(self.config, self.config_file_name) - - def _exec_update_script(self, script_type): - script = self.config.get(self.name, script_type, fallback=None) - if script: - print('{0}: exec {1} "{2}"'.format(self.name, script_type, script)) - print(colorama.Fore.BLUE + '------------------------------') - subprocess.run(script) - print(colorama.Fore.BLUE + '------------------------------') - - def _exec_global_update_script(self, processing_info): - script = self.config.get('UpdaterConfig', 'global_post_update', fallback=None) - if script: - print('{0}: exec global script "{1}"'.format(self.name, script)) - subprocess.run([ - script, - processing_info['tool_name'], - processing_info['tool_folder'], - processing_info['save_compress_name'], - ]) - - def _scrape_step(self): - tool_url = self.config.get(self.name, 'url', fallback=None) - update_url = self.config.get(self.name, 'update_url', fallback=None) - re_download = self.config.get(self.name, 're_download', fallback=None) - re_version = self.config.get(self.name, 're_version', fallback=None) - from_url = self.config.get(self.name, 'from', fallback='web') - - if from_url == 'github': - return self._scrape_github(tool_url, update_url, re_download) - elif from_url == 'http': - return self._scrape_http(update_url) - - return self._scrape_web(tool_url, update_url, re_version, re_download) - - def _download_step(self, download_url): - # create updates folder if don't exist - if not pathlib.Path.exists(self.update_folder_path): - pathlib.Path.mkdir(self.update_folder_path) - - # download - self.cleanup_update_folder() - file_name = get_filename_from_url(download_url) - file_path = pathlib.Path(self.update_folder_path).joinpath(file_name) - - print('{0}: downloading update "{1}"'.format(self.name, file_name)) - - try: - download( - url=download_url, - file_path=file_path, - progress_bar_desc=self.name, - disable_progress_bar=self.disable_progress, - ) - except Exception as exception: - print(colorama.Fore.RED + '{0}: download url {1}'.format(self.name, download_url)) - raise Exception(colorama.Fore.RED + '{0}: Error {1}'.format(self.name, exception)) - - return file_path - - def _processing_step(self, file_path, download_version): - unpack_folder_path = self._unpack(file_path) - self._exec_update_script('post_unpack') - - use_merge = self.config.get(self.name, 'merge', fallback=None) - disable_repack = self.config.get(self.name, 'disable_repack', fallback=None) - tool = self._processing_tool(unpack_folder_path) - if self.disable_repack or disable_repack: - return self._save( - use_merge=use_merge, - tool_folder_path=tool['folder_path'], - tool_unpack_path=tool['unpack_path'], - ) - else: - return self._repack( - use_merge=use_merge, - tool_folder_path=tool['folder_path'], - tool_unpack_path=tool['unpack_path'], - unpack_folder_path=unpack_folder_path, - version=download_version, - ) - - def _check_tool_installed(self): - tool_folder_path = self._tool_install_path() - if not tool_folder_path.exists() and not self.disable_install_check: - raise Exception('{0}: The program was not found'.format(self.name)) - - def update(self, name): - self.name = name - - # check if installed - self._check_tool_installed() - - # execute custom pre update script - self._exec_update_script('pre_update') - - # generate download url - scrape_data = self._scrape_step() - - # download - update_file_path = self._download_step(scrape_data['download_url']) - - # processing file - processing_info = self._processing_step(update_file_path, scrape_data['download_version']) - - # update complete - self._bump_version(scrape_data['download_version']) - self._exec_update_script('post_update') - self._exec_global_update_script(processing_info) - print('{0}: update complete'.format(self.name)) - - def cleanup_update_folder(self): - cleanup_folder(self.update_folder_path) - - -# Implementation -class Setup: - def __init__(self): - self.version = '1.8.1' - self.arguments = {} - self.config = configparser.ConfigParser() - self.default_config = {} - self.config_file_name = 'tools.ini' - - def print_banner(self): - print(colorama.Fore.YELLOW + """ - ____ __ __ __ __ __ - / _/___ ____/ /__ / /____ _____/ /_____ _/ /_ / /__ _____ - / // __ \/ __ / _ \/ __/ _ \/ ___/ __/ __ `/ __ \/ / _ \/ ___/ - _/ // / / / /_/ / __/ /_/ __/ /__/ /_/ /_/ / /_/ / / __(__ ) - /___/_/ /_/\__,_/\___/\__/\___/\___/\__/\__,_/_.___/_/\___/____/ - - Universal Tool Updater v{0} - by DSR! - https://github.com/xchwarze/universal-tool-updater - """.format(self.version)) - - def exit_handler(self, signal, frame): - print(colorama.Fore.YELLOW + 'SIGINT or CTRL-C detected. Exiting gracefully') - sys.exit(0) - - def get_argparse_default(self, option, default, is_bool=True): - if is_bool: - #return value.lower() in ('true', '1', 'yes', 'on') - return self.config.getboolean('UpdaterConfig', option, fallback=default) - - return self.config.get('UpdaterConfig', option, fallback=default) - - def init_argparse(self): - parser = argparse.ArgumentParser( - usage='%(prog)s [ARGUMENTS]', - ) - parser.add_argument( - '-v', - '--version', - action='version', - version='Version: {0}'.format(self.version) - ) - parser.add_argument( - '-u', - '--update', - dest='update', - help='list of tools (default: all)', - nargs='*' - ) - parser.add_argument( - '-dsu', - '--disable-self-update', - dest='disable_self_update', - help='disable self update of this script', - action=argparse.BooleanOptionalAction, - default=False - ) - parser.add_argument( - '-dfc', - '--disable-folder-clean', - dest='disable_clean', - help='disable tool folder clean', - action=argparse.BooleanOptionalAction, - default=self.get_argparse_default('disable_clean', False) - ) - parser.add_argument( - '-dr', - '--disable-repack', - dest='disable_repack', - help='disable tool repack', - action=argparse.BooleanOptionalAction, - default=self.get_argparse_default('disable_repack', False) - ) - parser.add_argument( - '-dic', - '--disable-install-check', - dest='disable_install_check', - help='disable tool install check', - action=argparse.BooleanOptionalAction, - default=self.get_argparse_default('disable_install_check', False) - ) - parser.add_argument( - '-dpb', - '--disable-progress-bar', - dest='disable_progress', - help='disable download progress bar', - action=argparse.BooleanOptionalAction, - default=self.get_argparse_default('disable_progress', False) - ) - parser.add_argument( - '-sft', - '--save-format-type', - dest='save_format_type', - help='compress save format name', - choices=['full', 'version', 'name'], - default=self.get_argparse_default('save_format_type', 'full', False) - ) - parser.add_argument( - '-f', - '--force', - dest='force_download', - help='force download', - action=argparse.BooleanOptionalAction, - default=False - ) - parser.add_argument( - '-uga', - '--use-github-api', - dest='use_github_api', - help='use github api with this token', - default=self.get_argparse_default('use_github_api', '', False) - ) - parser.add_argument( - '-udp', - '--update-default-params', - dest='update_default_params', - help='update default params', - action=argparse.BooleanOptionalAction, - default=False - ) - - self.arguments = parser.parse_args() - - def update_default_params(self): - if not self.arguments.update_default_params: - return False - - self.config.set('UpdaterConfig', 'disable_clean', str(self.arguments.disable_clean)) - self.config.set('UpdaterConfig', 'disable_repack', str(self.arguments.disable_repack)) - self.config.set('UpdaterConfig', 'disable_install_check', str(self.arguments.disable_install_check)) - self.config.set('UpdaterConfig', 'disable_progress', str(self.arguments.disable_progress)) - self.config.set('UpdaterConfig', 'save_format_type', self.arguments.save_format_type) - self.config.set('UpdaterConfig', 'use_github_api', self.arguments.use_github_api) - - save_config_to_file(self.config, self.config_file_name) - print(colorama.Fore.GREEN + 'Update default params successful') - - def handle_updates(self): - updater = Updater( - config=self.config, - config_file_name=self.config_file_name, - force_download=self.arguments.force_download, - disable_repack=self.arguments.disable_repack, - disable_clean=self.arguments.disable_clean, - disable_install_check=self.arguments.disable_install_check, - disable_progress=self.arguments.disable_progress, - save_format_type=self.arguments.save_format_type, - use_github_api=self.arguments.use_github_api, - ) - - # generate update list - update_list = self.arguments.update - if not update_list: - update_list = self.config.sections() - - # config data of this script - if 'UpdaterConfig' in update_list: - update_list.remove('UpdaterConfig') - - # auto updater logic - if 'UpdaterAutoUpdater' in update_list: - update_list.remove('UpdaterAutoUpdater') - - if not self.arguments.disable_self_update: - try: - updater.update('UpdaterAutoUpdater') - except Exception as exception: - print(exception) - - # normal execution - for ini_name in update_list: - try: - updater.update(ini_name) - except Exception as exception: - print(exception) - - updater.cleanup_update_folder() - - def main(self): - colorama.init(autoreset=True) - signal.signal(signal.SIGINT, self.exit_handler) - - # Fix updates "FileNotFoundError" in exe build - current_dir = os.path.dirname(sys.argv[0]) - if current_dir: - os.chdir(current_dir) - - self.print_banner() - - self.config.read(self.config_file_name) - self.init_argparse() - self.update_default_params() - - self.handle_updates() - - -# se fini +# Entry point for the script if __name__ == '__main__': - Setup().main() + # Initialize and run the main method of UpdateManager + UpdateManager().main()