mirror of
https://github.com/aclist/dztui.git
synced 2026-08-26 17:57:06 +02:00
chore: move update logic to manager
This commit is contained in:
parent
e715e6da1c
commit
86ba8d9187
60
dzgui/managers/update.py
Normal file
60
dzgui/managers/update.py
Normal file
@ -0,0 +1,60 @@
|
||||
import logging
|
||||
import requests
|
||||
import shutil
|
||||
|
||||
from dzgui.const.constants import (
|
||||
TMP_EXE,
|
||||
TMP_PATH,
|
||||
TMP_TARBALL,
|
||||
)
|
||||
from dzgui.strings import dialogs
|
||||
from dzgui.managers.threading import call_on_thread, StoredFunc, ThreadingManager
|
||||
|
||||
import gi
|
||||
|
||||
gi.require_version("Gtk", "3.0")
|
||||
from gi.repository import Gtk, GLib # noqa E402
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from dzgui.controllers.mc import Controller
|
||||
|
||||
logger = logging.getLogger(APP_NAME)
|
||||
|
||||
class UpdateManager:
|
||||
def __init__(self) -> None:
|
||||
|
||||
self.thread_man = ThreadingManager()
|
||||
|
||||
@call_on_thread(dialogs.fetching_update)
|
||||
def _update_version(self, exe_path: str, url: str) -> None:
|
||||
try:
|
||||
res = requests.get(url)
|
||||
if res.status_code == 200:
|
||||
with open(TMP_TARBALL, "wb") as file:
|
||||
file.write(res.content)
|
||||
with tarfile.open(TMP_TARBALL) as tar:
|
||||
tar.extractall(TMP_PATH)
|
||||
|
||||
shutil.move(TMP_EXE, exe_path)
|
||||
|
||||
proc = subprocess.run([exe_path, "self", "restore"])
|
||||
if proc.returncode == 0:
|
||||
func = StoredFunc(self._on_update_success)
|
||||
self.thread_man.set_cleanup_func(func, destroy_first=True)
|
||||
else:
|
||||
msg = dialogs.failed_to_update
|
||||
func = StoredFunc(self._on_update_failure, msg)
|
||||
self.thread_man.set_cleanup_func(func, destroy_first=True)
|
||||
except Exception as e:
|
||||
func = StoredFunc(self._on_update_failure, e)
|
||||
self.thread_man.set_cleanup_func(func, destroy_first=True)
|
||||
logger.warning(e)
|
||||
|
||||
def _on_update_success(self) -> None:
|
||||
msg = dialogs.update_success
|
||||
dialog = QuitDialog(self.controller, msg)
|
||||
dialog.run()
|
||||
|
||||
def _on_update_failure(self, msg: str) -> None:
|
||||
dialog = ExceptionDialog(self.controller, msg)
|
||||
dialog.run()
|
||||
@ -19,7 +19,7 @@ from dzgui.const.constants import (
|
||||
)
|
||||
from dzgui.const.update import ALLOW_UPDATES
|
||||
from dzgui.const.enum import ServerTab
|
||||
from dzgui.managers.threading import call_on_thread, StoredFunc, ThreadingManager
|
||||
from dzgui.managers.update import UpdateManager
|
||||
from dzgui.strings import dialogs
|
||||
from dzgui.util.clip import copy_clipboard
|
||||
from dzgui.views.components.buttonbox import ButtonBox
|
||||
@ -45,7 +45,6 @@ class RightPanel(Gtk.Box):
|
||||
def __init__(self, controller: "Controller"):
|
||||
super().__init__(spacing=6, orientation=Gtk.Orientation.VERTICAL)
|
||||
|
||||
self.thread_man = ThreadingManager(controller)
|
||||
self.controller = controller
|
||||
self.controller.register_widget("right_panel", self)
|
||||
|
||||
@ -96,54 +95,18 @@ class RightPanel(Gtk.Box):
|
||||
valign=Gtk.Align.END,
|
||||
spacing=10,
|
||||
)
|
||||
if ALLOW_UPDATES and update is not None:
|
||||
exe_path = os.getenv("PYAPP")
|
||||
if exe_path and update is not None:
|
||||
self.update_button.set_halign(Gtk.Align.END)
|
||||
self.gutter_box.add(self.update_button)
|
||||
self.update_button.connect(
|
||||
"clicked", self._on_update_button_clicked, update
|
||||
"clicked", self._on_update_button_clicked, exe_path, update
|
||||
)
|
||||
self.gutter_box.add(eb)
|
||||
self.pack_start(self.gutter_box, NO_EXPAND, FILL, NO_PADDING)
|
||||
|
||||
def _on_update_success(self) -> None:
|
||||
msg = dialogs.update_success
|
||||
dialog = QuitDialog(self.controller, msg)
|
||||
dialog.run()
|
||||
|
||||
def _on_update_failure(self, msg: str) -> None:
|
||||
dialog = ExceptionDialog(self.controller, msg)
|
||||
dialog.run()
|
||||
|
||||
@call_on_thread(dialogs.fetching_update)
|
||||
def _on_update_button_clicked(self, button: Gtk.Button, url: str) -> None:
|
||||
try:
|
||||
res = requests.get(url)
|
||||
if res.status_code == 200:
|
||||
with open(TMP_TARBALL, "wb") as file:
|
||||
file.write(res.content)
|
||||
with tarfile.open(TMP_TARBALL) as tar:
|
||||
tar.extractall(TMP_PATH)
|
||||
|
||||
exe_path = os.getenv("PYAPP")
|
||||
if exe_path is None:
|
||||
msg = dialogs.failed_to_update
|
||||
func = StoredFunc(self._on_update_failure, msg)
|
||||
self.thread_man.set_cleanup_func(func, destroy_first=True)
|
||||
return
|
||||
shutil.move(TMP_EXE, exe_path)
|
||||
|
||||
proc = subprocess.run([exe_path, "self", "restore"])
|
||||
if proc.returncode == 0:
|
||||
func = StoredFunc(self._on_update_success)
|
||||
self.thread_man.set_cleanup_func(func, destroy_first=True)
|
||||
else:
|
||||
msg = dialogs.failed_to_update
|
||||
func = StoredFunc(self._on_update_failure, msg)
|
||||
self.thread_man.set_cleanup_func(func, destroy_first=True)
|
||||
except Exception as e:
|
||||
func = StoredFunc(self._on_update_failure, e)
|
||||
self.thread_man.set_cleanup_func(func, destroy_first=True)
|
||||
logger.warning(e)
|
||||
def _on_update_button_clicked(self, button: Gtk.Button, exe_path: str, url: str) -> None:
|
||||
UpdateManager().update_version()
|
||||
|
||||
def _on_server_page_changed(
|
||||
self, emitter: "Emitter", page: "ServerTreeView"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user