tyler/flares
2
0
Fork 0
Code Issues 24 Pull Requests Actions Packages Projects Releases 25 Wiki Activity

functional connectivity, pylance, and other improvements

Browse Source
This commit is contained in:
tyler
2026-08-22 23:38:13 -07:00
parent 19bd3f1279
commit e37275a1bb
14 changed files with 1455 additions and 841 deletions
Download Patch File Download Diff File Expand all files Collapse all files
updater.py
+97 -56
View File
@@ -1,6 +1,7 @@
"""
Filename: updater.py
Description: Generic updater file
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
License: GPL-3.0
@@ -17,13 +18,14 @@ import zipfile
import traceback
import subprocess
import configparser
from typing import List
# External library imports
import psutil
import requests
from PySide6.QtWidgets import QMessageBox
from PySide6.QtCore import QThread, Signal, QObject
from PySide6.QtWidgets import QMainWindow, QMessageBox
class UpdateDownloadThread(QThread):
@@ -38,7 +40,14 @@ class UpdateDownloadThread(QThread):
update_ready = Signal(str, str)
error_occurred = Signal(str)
def __init__(self, download_url, latest_version, platform_name, app_name):
def __init__(
self,
download_url: str,
latest_version: str,
platform_name: str,
app_name: str,
) -> None:
super().__init__()
self.download_url = download_url
self.latest_version = latest_version
@@ -54,6 +63,7 @@ class UpdateDownloadThread(QThread):
os.makedirs(tmp_dir, exist_ok=True)
local_path = os.path.join(tmp_dir, local_filename)
else:
tmp_dir = os.getcwd()
local_path = os.path.join(os.getcwd(), local_filename)
# Download the file
@@ -92,7 +102,6 @@ class UpdateDownloadThread(QThread):
self.error_occurred.emit(str(e))
class UpdateCheckThread(QThread):
"""
Thread that checks for updates by querying the API and emits a signal based on the result.
@@ -107,7 +116,15 @@ class UpdateCheckThread(QThread):
no_update_available = Signal()
error_occurred = Signal(str)
def __init__(self, api_url, api_url_sec, current_version, platform_name, app_name):
def __init__(
self,
api_url: str,
api_url_sec: str,
current_version: str,
platform_name: str,
app_name: str,
) -> None:
super().__init__()
self.api_url = api_url
self.api_url_sec = api_url_sec
@@ -137,8 +154,9 @@ class UpdateCheckThread(QThread):
except Exception as e:
self.error_occurred.emit(f"Update check failed: {e}")
def version_compare(self, v1, v2):
def normalize(v): return [int(x) for x in v.split(".")]
def version_compare(self, v1: str, v2: str) -> int:
def normalize(v: str) -> List[int]:
return [int(x) for x in v.split(".")]
return (normalize(v1) > normalize(v2)) - (normalize(v1) < normalize(v2))
def get_latest_release_for_platform(self):
@@ -165,7 +183,7 @@ class UpdateCheckThread(QThread):
return tag, asset["browser_download_url"]
return tag, None
except (requests.RequestException, ValueError) as e:
except (requests.RequestException, ValueError, KeyError):
continue
return None, None
@@ -182,15 +200,23 @@ class LocalPendingUpdateCheckThread(QThread):
pending_update_found = Signal(str, str)
no_pending_update = Signal()
def __init__(self, current_version, platform_suffix, platform_name, app_name):
def __init__(
self,
current_version: str,
platform_suffix: str,
platform_name: str,
app_name: str,
) -> None:
super().__init__()
self.current_version = current_version
self.platform_suffix = platform_suffix
self.platform_name = platform_name
self.app_name = app_name
def version_compare(self, v1, v2):
def normalize(v): return [int(x) for x in v.split(".")]
def version_compare(self, v1: str, v2: str) -> int:
def normalize(v: str) -> List[int]:
return [int(x) for x in v.split(".")]
return (normalize(v1) > normalize(v2)) - (normalize(v1) < normalize(v2))
def run(self):
@@ -220,18 +246,25 @@ class LocalPendingUpdateCheckThread(QThread):
self.no_pending_update.emit()
class UpdateManager(QObject):
"""
Orchestrates the update process.
Main apps should instantiate this and call check_for_updates().
"""
def __init__(self, main_window, api_url, api_url_sec, current_version, platform_name, platform_suffix, app_name):
super().__init__()
self.parent = main_window
def __init__(
self,
main_window: QMainWindow,
api_url: str,
api_url_sec: str,
current_version: str,
platform_name: str,
platform_suffix: str,
app_name: str,
) -> None:
super().__init__(main_window)
self.main_window: QMainWindow = main_window
self.api_url = api_url
self.api_url_sec = api_url_sec
self.current_version = current_version
@@ -243,59 +276,64 @@ class UpdateManager(QObject):
self.pending_update_path = None
def manual_check_for_updates(self):
def manual_check_for_updates(self) -> None:
self.local_check_thread = LocalPendingUpdateCheckThread(self.current_version, self.platform_suffix, self.platform_name, self.app_name)
self.local_check_thread.pending_update_found.connect(self.on_pending_update_found)
self.local_check_thread.no_pending_update.connect(self.on_no_pending_update)
self.local_check_thread.start()
def on_pending_update_found(self, version, folder_path):
self.parent.statusBar().showMessage(f"Pending update found: version {version}")
def on_pending_update_found(self, version: str, folder_path: str) -> None:
self.main_window.statusBar().showMessage(f"Pending update found: version {version}")
self.pending_update_version = version
self.pending_update_path = folder_path
self.show_pending_update_popup()
def on_no_pending_update(self):
def on_no_pending_update(self) -> None:
# No pending update found locally, start server check directly
self.parent.statusBar().showMessage("No pending local update found. Checking server...")
self.main_window.statusBar().showMessage("No pending local update found. Checking server...")
self.start_update_check_thread()
def show_pending_update_popup(self):
msg_box = QMessageBox(self.parent)
def show_pending_update_popup(self) -> None:
msg_box = QMessageBox(self.main_window)
msg_box.setWindowTitle("Pending Update Found")
msg_box.setText(f"A previously downloaded update for {self.app_name.upper()} (version {self.pending_update_version}) is available at:\n{self.pending_update_path}\nWould you like to install it now?")
install_now_button = msg_box.addButton("Install Now", QMessageBox.ButtonRole.AcceptRole)
install_later_button = msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole)
msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole)
msg_box.exec()
if msg_box.clickedButton() == install_now_button:
if msg_box.clickedButton() == install_now_button and self.pending_update_path:
self.install_update(self.pending_update_path)
else:
self.parent.statusBar().showMessage("Pending update available. Install later.")
if self.main_window.statusBar():
self.main_window.statusBar().showMessage("Pending update available. Install later.")
# After user dismisses, still check the server for new updates
self.start_update_check_thread()
def start_update_check_thread(self):
def start_update_check_thread(self) -> None:
self.check_thread = UpdateCheckThread(self.api_url, self.api_url_sec, self.current_version, self.platform_name, self.app_name)
self.check_thread.download_requested.connect(self.on_server_update_requested)
self.check_thread.no_update_available.connect(self.on_server_no_update)
self.check_thread.error_occurred.connect(self.on_error)
self.check_thread.start()
def on_server_no_update(self):
self.parent.statusBar().showMessage("No new updates found on server.", 5000)
def on_server_no_update(self) -> None:
if self.main_window.statusBar():
self.main_window.statusBar().showMessage("No new updates found on server.", 5000)
def on_server_update_requested(self, download_url, latest_version):
if self.pending_update_version:
cmp = self.version_compare(latest_version, self.pending_update_version)
def on_server_update_requested(self, download_url: str, latest_version: str) -> None:
pending_path = self.pending_update_path
pending_version = self.pending_update_version
if pending_version and pending_path:
cmp = self.version_compare(latest_version, pending_version)
if cmp > 0:
# Server version is newer than pending update
self.parent.statusBar().showMessage(f"Newer version {latest_version} available on server. Removing old pending update...")
self.main_window.statusBar().showMessage(f"Newer version {latest_version} available on server. Removing old pending update...")
try:
shutil.rmtree(self.pending_update_path)
self.parent.statusBar().showMessage(f"Deleted old update folder: {self.pending_update_path}")
shutil.rmtree(pending_path)
self.main_window.statusBar().showMessage(f"Deleted old update folder: {pending_path}")
except Exception as e:
self.parent.statusBar().showMessage(f"Failed to delete old update folder: {e}")
self.main_window.statusBar().showMessage(f"Failed to delete old update folder: {e}")
# Clear pending update info so new download proceeds
self.pending_update_version = None
@@ -305,39 +343,41 @@ class UpdateManager(QObject):
self.download_update(download_url, latest_version)
elif cmp == 0:
# Versions equal, no download needed
self.parent.statusBar().showMessage(f"Pending update version {self.pending_update_version} is already latest. No download needed.")
self.main_window.statusBar().showMessage(f"Pending update version {self.pending_update_version} is already latest. No download needed.")
else:
# Server version older than pending? Unlikely but just keep pending update
self.parent.statusBar().showMessage(f"Pending update version {self.pending_update_version} is newer than server version. No action.")
self.main_window.statusBar().showMessage(f"Pending update version {self.pending_update_version} is newer than server version. No action.")
else:
# No pending update, just download
self.download_update(download_url, latest_version)
def download_update(self, download_url, latest_version):
self.parent.statusBar().showMessage("Downloading update...")
def download_update(self, download_url: str, latest_version: str) -> None:
if self.main_window.statusBar():
self.main_window.statusBar().showMessage("Downloading update...")
self.download_thread = UpdateDownloadThread(download_url, latest_version, self.platform_name, self.app_name)
self.download_thread.update_ready.connect(self.on_update_ready)
self.download_thread.error_occurred.connect(self.on_error)
self.download_thread.start()
def on_update_ready(self, latest_version, extract_folder):
self.parent.statusBar().showMessage("Update downloaded and extracted.")
def on_update_ready(self, latest_version: str, extract_folder: str) -> None:
if self.main_window.statusBar():
self.main_window.statusBar().showMessage("Update downloaded and extracted.")
msg_box = QMessageBox(self.parent)
msg_box = QMessageBox(self.main_window)
msg_box.setWindowTitle("Update Ready")
msg_box.setText(f"Version {latest_version} has been downloaded and extracted to:\n{extract_folder}\nWould you like to install it now?")
install_now_button = msg_box.addButton("Install Now", QMessageBox.ButtonRole.AcceptRole)
install_later_button = msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole)
msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole)
msg_box.exec()
if msg_box.clickedButton() == install_now_button:
self.install_update(extract_folder)
else:
self.parent.statusBar().showMessage("Update ready. Install later.")
self.main_window.statusBar().showMessage("Update ready. Install later.")
def install_update(self, extract_folder):
def install_update(self, extract_folder: str) -> None:
# Path to updater executable
if self.platform_name == 'windows':
@@ -354,7 +394,7 @@ class UpdateManager(QObject):
updater_path = os.getcwd()
if not os.path.exists(updater_path):
QMessageBox.critical(self.parent, "Error", f"Updater not found at:\n{updater_path}. The absolute path was {os.path.abspath(updater_path)}")
QMessageBox.critical(self.main_window, "Error", f"Updater not found at:\n{updater_path}. The absolute path was {os.path.abspath(updater_path)}")
return
# Launch updater with extracted folder path as argument
@@ -373,18 +413,19 @@ class UpdateManager(QObject):
sys.exit(0)
except Exception as e:
QMessageBox.critical(self.parent, "Error", f"[Updater Launch Failed]\n{str(e)}\n{traceback.format_exc()}")
QMessageBox.critical(self.main_window, "Error", f"[Updater Launch Failed]\n{str(e)}\n{traceback.format_exc()}")
def on_error(self, message):
# print(f"Error: {message}")
self.parent.statusBar().showMessage(f"Error occurred during update process. {message}")
def on_error(self, message: str) -> None:
if self.main_window.statusBar():
self.main_window.statusBar().showMessage(f"Error occurred during update process. {message}")
def version_compare(self, v1, v2):
def normalize(v): return [int(x) for x in v.split(".")]
def version_compare(self, v1: str, v2: str) -> int:
def normalize(v: str) -> List[int]:
return [int(x) for x in v.split(".")]
return (normalize(v1) > normalize(v2)) - (normalize(v1) < normalize(v2))
def wait_for_process_to_exit(process_name, timeout=10):
def wait_for_process_to_exit(process_name: str, timeout: int = 10) -> bool:
"""
Waits for a process with the specified name to exit within a timeout period.
@@ -416,7 +457,7 @@ def wait_for_process_to_exit(process_name, timeout=10):
return False
def finish_update_if_needed(platform_name, app_name, cfg_path, finish_update):
def finish_update_if_needed(platform_name: str, app_name: str, cfg_path: str, finish_update: bool) -> None:
"""
Completes a pending application update if '--finish-update' is present in the command-line arguments.
"""
@@ -534,7 +575,7 @@ def finish_update_if_needed(platform_name, app_name, cfg_path, finish_update):
sys.argv.remove("--finish-update")
def remove_quarantine(app_path, app_name):
def remove_quarantine(app_path: str, app_name: str) -> None:
"""
Removes the macOS quarantine attribute from the specified application path.
"""