Bump Universal Updater to 2.0.0

This commit is contained in:
DSR! committed 2023-10-18 11:44:32 -03:00
1 parent 0c8c423a58
commit b139bdfd49
19 files changed
+1370 -710

No files matched your search

+3 -2
View File
@@ -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
```
+3 -2
View File
@@ -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
```
@@ -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
@@ -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
File renamed without changes.
+5
View File
@@ -0,0 +1,5 @@
requests>=2.26.0
colorama>=0.4.4
tqdm>=4.62.3
py7zr>=0.16.1
rarfile>=4.0
+16
View File
@@ -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
+91
View File
@@ -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)
+91
View File
@@ -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
+90
View File
@@ -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': '',
}
+18
View File
@@ -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()
+202
View File
@@ -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,
}
+322
View File
@@ -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()
+56
View File
@@ -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'],
])
+291
View File
@@ -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()
+167
View File
@@ -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')
Binary file not shown.
Binary file not shown.
+5 -701
View File
@@ -1,703 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2021 - 2023 DSR! <[email protected]>
# 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()