599 lines
23 KiB
Python
599 lines
23 KiB
Python
"""
|
|
Filename: updater.py
|
|
Description: Generic updater file
|
|
Note: Compliant with pylance strict type checking
|
|
|
|
Author: Tyler de Zeeuw
|
|
License: GPL-3.0
|
|
"""
|
|
|
|
# Built-in imports
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import shlex
|
|
import shutil
|
|
import zipfile
|
|
import traceback
|
|
import subprocess
|
|
import configparser
|
|
from typing import List, Tuple, Optional, Callable, Any
|
|
|
|
|
|
# External library imports
|
|
import psutil
|
|
import requests
|
|
|
|
from PySide6.QtCore import QThread, Signal, QObject
|
|
from PySide6.QtWidgets import QMainWindow, QMessageBox
|
|
from src.shared.shareddata import get_app_dir
|
|
|
|
class UpdateDownloadThread(QThread):
|
|
"""
|
|
Thread that downloads and extracts an update package and emits a signal on completion or error.
|
|
|
|
Args:
|
|
download_url (str): URL of the update zip file to download.
|
|
latest_version (str): Version string of the latest update.
|
|
"""
|
|
|
|
update_ready = Signal(str, str)
|
|
error_occurred = Signal(str)
|
|
|
|
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
|
|
self.platform_name = platform_name
|
|
self.app_name = app_name
|
|
|
|
def run(self) -> None:
|
|
try:
|
|
local_filename = os.path.basename(self.download_url)
|
|
|
|
if self.platform_name == 'darwin':
|
|
tmp_dir = f'/tmp/{self.app_name}tempupdate'
|
|
os.makedirs(tmp_dir, exist_ok=True)
|
|
local_path = os.path.join(tmp_dir, local_filename)
|
|
else:
|
|
tmp_dir = get_app_dir()
|
|
local_path = os.path.join(tmp_dir, local_filename)
|
|
|
|
# Download the file
|
|
with requests.get(self.download_url, stream=True, timeout=15) as r:
|
|
r.raise_for_status()
|
|
with open(local_path, 'wb') as f:
|
|
for chunk in r.iter_content(chunk_size=8192):
|
|
if chunk:
|
|
f.write(chunk)
|
|
|
|
# Extract folder name (remove .zip)
|
|
if self.platform_name == 'darwin':
|
|
extract_folder = os.path.splitext(local_filename)[0]
|
|
extract_path = os.path.join(tmp_dir, extract_folder)
|
|
else:
|
|
extract_folder = os.path.splitext(local_filename)[0]
|
|
extract_path = os.path.join(get_app_dir(), extract_folder)
|
|
|
|
# Create the folder if not exists
|
|
os.makedirs(extract_path, exist_ok=True)
|
|
|
|
# Extract the zip file contents
|
|
if self.platform_name == 'darwin':
|
|
subprocess.run(['ditto', '-xk', local_path, extract_path], check=True)
|
|
else:
|
|
with zipfile.ZipFile(local_path, 'r') as zip_ref:
|
|
zip_ref.extractall(extract_path)
|
|
|
|
# Remove the zip once extracted and emit a signal
|
|
os.remove(local_path)
|
|
self.update_ready.emit(self.latest_version, extract_path)
|
|
|
|
except Exception as e:
|
|
# Emit a signal signifying failure
|
|
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.
|
|
|
|
Signals:
|
|
download_requested(str, str): Emitted with (download_url, latest_version) when an update is available.
|
|
no_update_available(): Emitted when no update is found or current version is up to date.
|
|
error_occurred(str): Emitted with an error message if the update check fails.
|
|
"""
|
|
|
|
download_requested = Signal(str, str)
|
|
no_update_available = Signal()
|
|
error_occurred = Signal(str)
|
|
|
|
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
|
|
self.current_version = current_version
|
|
self.platform_name = platform_name
|
|
self.app_name = app_name
|
|
|
|
def run(self) -> None:
|
|
try:
|
|
latest_version, download_url = self.get_latest_release_for_platform()
|
|
if not latest_version:
|
|
self.no_update_available.emit()
|
|
return
|
|
|
|
if not download_url:
|
|
self.error_occurred.emit(f"No download available for platform '{self.platform_name}'")
|
|
return
|
|
|
|
if self.version_compare(latest_version, self.current_version) > 0:
|
|
self.download_requested.emit(download_url, latest_version)
|
|
else:
|
|
self.no_update_available.emit()
|
|
|
|
except Exception as e:
|
|
self.error_occurred.emit(f"Update check failed: {e}")
|
|
|
|
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) -> Tuple[Optional[str], Optional[str]]:
|
|
urls = [self.api_url, self.api_url_sec]
|
|
for url in urls:
|
|
try:
|
|
response = requests.get(url, timeout=5)
|
|
response.raise_for_status()
|
|
releases = response.json()
|
|
|
|
if not releases:
|
|
continue
|
|
|
|
latest = next((r for r in releases if not r.get("prerelease") and not r.get("draft")), None)
|
|
|
|
if not latest:
|
|
continue
|
|
|
|
tag = latest["tag_name"].lstrip("v")
|
|
|
|
for asset in latest.get("assets", []):
|
|
if self.platform_name in asset["name"].lower():
|
|
return tag, asset["browser_download_url"]
|
|
|
|
return tag, None
|
|
except (requests.RequestException, ValueError, KeyError):
|
|
continue
|
|
return None, None
|
|
|
|
|
|
class LocalPendingUpdateCheckThread(QThread):
|
|
"""
|
|
Thread that checks for locally pending updates by scanning the download directory and emits a signal accordingly.
|
|
|
|
Args:
|
|
current_version (str): Current application version.
|
|
platform_suffix (str): Platform-specific suffix to identify update folders.
|
|
"""
|
|
|
|
pending_update_found = Signal(str, str)
|
|
no_pending_update = Signal()
|
|
|
|
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: 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) -> None:
|
|
if self.platform_name == 'darwin':
|
|
cwd = f'/tmp/{self.app_name}tempupdate'
|
|
else:
|
|
cwd = get_app_dir()
|
|
|
|
pattern = re.compile(r".*-(\d+\.\d+\.\d+)" + re.escape(self.platform_suffix) + r"$")
|
|
found = False
|
|
|
|
try:
|
|
for item in os.listdir(cwd):
|
|
folder_path = os.path.join(cwd, item)
|
|
if os.path.isdir(folder_path) and item.endswith(self.platform_suffix):
|
|
match = pattern.match(item)
|
|
if match:
|
|
folder_version = match.group(1)
|
|
if self.version_compare(folder_version, self.current_version) > 0:
|
|
self.pending_update_found.emit(folder_version, folder_path)
|
|
found = True
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
if not found:
|
|
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: 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
|
|
self.platform_name = platform_name
|
|
self.platform_suffix = platform_suffix
|
|
self.app_name = app_name
|
|
|
|
self.pending_update_version: Optional[str] = None
|
|
self.pending_update_path: Optional[str] = None
|
|
|
|
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: str, folder_path: str) -> None:
|
|
if self.main_window.statusBar():
|
|
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) -> None:
|
|
if self.main_window.statusBar():
|
|
self.main_window.statusBar().showMessage("No pending local update found. Checking server...")
|
|
self.start_update_check_thread()
|
|
|
|
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)
|
|
msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole)
|
|
msg_box.exec()
|
|
|
|
if msg_box.clickedButton() == install_now_button and self.pending_update_path:
|
|
self.install_update(self.pending_update_path)
|
|
else:
|
|
if self.main_window.statusBar():
|
|
self.main_window.statusBar().showMessage("Pending update available. Install later.")
|
|
self.start_update_check_thread()
|
|
|
|
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) -> 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: 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:
|
|
if self.main_window.statusBar():
|
|
self.main_window.statusBar().showMessage(f"Newer version {latest_version} available on server. Removing old pending update...")
|
|
try:
|
|
shutil.rmtree(pending_path)
|
|
if self.main_window.statusBar():
|
|
self.main_window.statusBar().showMessage(f"Deleted old update folder: {pending_path}")
|
|
except Exception as e:
|
|
if self.main_window.statusBar():
|
|
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
|
|
self.pending_update_path = None
|
|
|
|
# Download the new update
|
|
self.download_update(download_url, latest_version)
|
|
elif cmp == 0:
|
|
if self.main_window.statusBar():
|
|
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.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: 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: str, extract_folder: str) -> None:
|
|
if self.main_window.statusBar():
|
|
self.main_window.statusBar().showMessage("Update downloaded and extracted.")
|
|
|
|
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)
|
|
msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole)
|
|
|
|
msg_box.exec()
|
|
|
|
if msg_box.clickedButton() == install_now_button:
|
|
self.install_update(extract_folder)
|
|
else:
|
|
if self.main_window.statusBar():
|
|
self.main_window.statusBar().showMessage("Update ready. Install later.")
|
|
|
|
def install_update(self, extract_folder: str) -> None:
|
|
base_dir = get_app_dir()
|
|
|
|
# Path to updater executable
|
|
if self.platform_name == 'windows':
|
|
updater_path = os.path.join(base_dir, f"{self.app_name}_updater.exe")
|
|
elif self.platform_name == 'darwin':
|
|
if getattr(sys, 'frozen', False):
|
|
updater_path = os.path.join(base_dir, f"../../../{self.app_name}_updater.app")
|
|
else:
|
|
updater_path = os.path.join(base_dir, f"../{self.app_name}_updater.app")
|
|
elif self.platform_name == 'linux':
|
|
updater_path = os.path.join(base_dir, f"{self.app_name}_updater")
|
|
else:
|
|
updater_path = base_dir
|
|
|
|
updater_path = os.path.abspath(updater_path)
|
|
|
|
if not os.path.exists(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
|
|
try:
|
|
main_app_executable = sys.executable if getattr(sys, 'frozen', False) else os.path.abspath(sys.argv[0])
|
|
|
|
print(f'Launching updater with: "{updater_path}" "{extract_folder}" "{main_app_executable}"')
|
|
|
|
if self.platform_name == 'darwin':
|
|
subprocess.Popen(['open', updater_path, '--args', extract_folder, main_app_executable])
|
|
else:
|
|
subprocess.Popen([updater_path, f'{extract_folder}', f'{main_app_executable}'], cwd=os.path.dirname(updater_path))
|
|
|
|
# Close the current app so updater can replace files
|
|
sys.exit(0)
|
|
|
|
except Exception as e:
|
|
QMessageBox.critical(self.main_window, "Error", f"[Updater Launch Failed]\n{str(e)}\n{traceback.format_exc()}")
|
|
|
|
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: 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: str, timeout: int = 10) -> bool:
|
|
"""
|
|
Waits for a process with the specified name to exit within a timeout period.
|
|
|
|
Args:
|
|
process_name (str): Name (or part of the name) of the process to wait for.
|
|
timeout (int, optional): Maximum time to wait in seconds. Defaults to 10.
|
|
|
|
Returns:
|
|
bool: True if the process exited before the timeout, False otherwise.
|
|
"""
|
|
|
|
print(f"Waiting for {process_name} to exit...")
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
still_running = False
|
|
for proc in psutil.process_iter(['name']):
|
|
try:
|
|
if proc.info['name'] and process_name.lower() in proc.info['name'].lower():
|
|
still_running = True
|
|
print(f"Still running: {proc.info['name']} (PID: {proc.pid})")
|
|
break
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
continue
|
|
if not still_running:
|
|
print(f"{process_name} has exited.")
|
|
return True
|
|
time.sleep(0.5)
|
|
print(f"{process_name} did not exit in time.")
|
|
return False
|
|
|
|
|
|
def _remove_readonly(func: Callable[[str], Any], path: str, exc: Any) -> None:
|
|
"""Error handler for shutil.rmtree to clear read-only files on Windows."""
|
|
import stat
|
|
os.chmod(path, stat.S_IWRITE)
|
|
func(path)
|
|
|
|
|
|
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.
|
|
"""
|
|
|
|
if not finish_update:
|
|
return
|
|
|
|
print("Finishing update...")
|
|
|
|
# 1. Reset welcome dialog config flag
|
|
update_cfg = configparser.ConfigParser()
|
|
try:
|
|
if os.path.exists(cfg_path):
|
|
update_cfg.read(cfg_path)
|
|
if not update_cfg.has_section("Options"):
|
|
update_cfg.add_section("Options")
|
|
update_cfg.set("Options", "show_welcome_dialog", "true")
|
|
|
|
with open(cfg_path, "w") as f:
|
|
update_cfg.write(f)
|
|
print("Welcome dialog flag successfully reset to 'true' for next run.")
|
|
except Exception as e:
|
|
print(f"Warning: Could not update welcome dialog preference flag: {e}")
|
|
|
|
app_dir = f'/tmp/{app_name}tempupdate' if platform_name == 'darwin' else get_app_dir()
|
|
|
|
if not os.path.exists(app_dir):
|
|
print(f"App directory does not exist: {app_dir}. Skipping cleanup.")
|
|
return
|
|
|
|
# 2. Find update folder(s)
|
|
update_folders: List[str] = []
|
|
try:
|
|
for entry in os.listdir(app_dir):
|
|
entry_path = os.path.join(app_dir, entry)
|
|
if os.path.isdir(entry_path) and entry.startswith(f"{app_name}-") and entry.endswith("-" + platform_name):
|
|
update_folders.append(entry_path)
|
|
except Exception as e:
|
|
print(f"Error scanning app directory for update folders: {e}")
|
|
|
|
if not update_folders:
|
|
print("No update folder found. Skipping update steps.")
|
|
return
|
|
|
|
primary_update_folder = update_folders[0]
|
|
target_updater_folder = os.path.join(primary_update_folder, f"{app_name}-darwin") if platform_name == 'darwin' else primary_update_folder
|
|
|
|
# 3. Wait for updater process to exit
|
|
print(f"Waiting for {app_name}_updater to exit...")
|
|
updater_bin_name = f"{app_name}_updater"
|
|
for proc in psutil.process_iter(['pid', 'name']):
|
|
try:
|
|
if proc.info['name'] and updater_bin_name.lower() in proc.info['name'].lower():
|
|
proc.wait(timeout=5)
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
pass
|
|
except psutil.TimeoutExpired:
|
|
print(f"Force killing lingering {app_name}_updater")
|
|
try:
|
|
proc.kill()
|
|
except Exception:
|
|
pass
|
|
|
|
# 4. Replace the updater executable
|
|
target_base = get_app_dir()
|
|
if platform_name == 'windows':
|
|
new_updater = os.path.join(target_updater_folder, f"{app_name}_updater.exe")
|
|
dest_updater = os.path.join(target_base, f"{app_name}_updater.exe")
|
|
elif platform_name == 'darwin':
|
|
new_updater = os.path.join(target_updater_folder, f"{app_name}_updater.app")
|
|
dest_updater = os.path.abspath(os.path.join(sys.executable, f"../../../../{app_name}_updater.app"))
|
|
elif platform_name == 'linux':
|
|
new_updater = os.path.join(target_updater_folder, f"{app_name}_updater")
|
|
dest_updater = os.path.join(target_base, f"{app_name}_updater")
|
|
else:
|
|
new_updater = target_base
|
|
dest_updater = target_base
|
|
|
|
if os.path.exists(new_updater):
|
|
try:
|
|
if os.path.exists(dest_updater):
|
|
if platform_name == 'darwin' and os.path.isdir(dest_updater):
|
|
shutil.rmtree(dest_updater, onexc=_remove_readonly)
|
|
else:
|
|
os.remove(dest_updater)
|
|
|
|
if platform_name == 'darwin':
|
|
wait_for_process_to_exit(f"{app_name}_updater", timeout=10)
|
|
subprocess.check_call(["ditto", new_updater, dest_updater])
|
|
else:
|
|
shutil.copy2(new_updater, dest_updater)
|
|
|
|
if platform_name in ('linux', 'darwin'):
|
|
os.chmod(dest_updater, 0o755)
|
|
|
|
if platform_name == 'darwin':
|
|
remove_quarantine(dest_updater, app_name)
|
|
|
|
print(f"{app_name}_updater replaced successfully.")
|
|
except Exception as e:
|
|
print(f"Failed to replace {app_name}_updater: {e}")
|
|
|
|
# 5. Clean up all temporary update folders (with retries for Windows file locks)
|
|
for folder in update_folders:
|
|
for attempt in range(3):
|
|
try:
|
|
if os.path.exists(folder):
|
|
shutil.rmtree(folder, onexc=_remove_readonly)
|
|
print(f"Successfully deleted update folder: {folder}")
|
|
break
|
|
except Exception as e:
|
|
if attempt < 2:
|
|
time.sleep(1.0)
|
|
else:
|
|
print(f"Failed to delete update folder '{folder}' after 3 attempts: {e}")
|
|
|
|
if "--finish-update" in sys.argv:
|
|
sys.argv.remove("--finish-update")
|
|
|
|
|
|
def remove_quarantine(app_path: str, app_name: str) -> None:
|
|
"""
|
|
Removes the macOS quarantine attribute from the specified application path.
|
|
"""
|
|
|
|
script = f'''
|
|
do shell script "xattr -d -r com.apple.quarantine {shlex.quote(app_path)}" with administrator privileges with prompt "{app_name.upper()} needs privileges to finish the update. (2/2)"
|
|
'''
|
|
try:
|
|
subprocess.run(['osascript', '-e', script], check=True)
|
|
print("✅ Quarantine attribute removed.")
|
|
except subprocess.CalledProcessError as e:
|
|
print("❌ Failed to remove quarantine attribute.")
|
|
print(e) |