121 lines
4.6 KiB
Python
121 lines
4.6 KiB
Python
from core.Constants import Constants
|
|
from core.Errors import UnknownClientPathError, UnknownClientVersionError, CommandNotFoundError
|
|
from core.controllers.ApplicationController import ApplicationController
|
|
from core.controllers.ApplicationVersionController import ApplicationVersionController
|
|
from core.controllers.ClientVersionController import ClientVersionController
|
|
from core.controllers.ConfigurationController import ConfigurationController
|
|
from core.controllers.LocationController import LocationController
|
|
from core.controllers.SubscriptionPlanController import SubscriptionPlanController
|
|
from core.observers import ClientObserver
|
|
from core.observers.ConnectionObserver import ConnectionObserver
|
|
from typing import Optional
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
|
|
|
|
class ClientController:
|
|
|
|
@staticmethod
|
|
def get_working_directory():
|
|
return str(pathlib.Path(__file__).resolve().parent.parent.parent)
|
|
|
|
@staticmethod
|
|
def get_version():
|
|
|
|
if (version_number := Constants.HV_CLIENT_VERSION_NUMBER) is None:
|
|
raise UnknownClientVersionError('The client version could not be determined.')
|
|
|
|
return ClientVersionController.get_or_new(version_number)
|
|
|
|
@staticmethod
|
|
def can_be_updated():
|
|
|
|
try:
|
|
version = ClientController.get_version()
|
|
except UnknownClientVersionError:
|
|
return False
|
|
|
|
return not ClientVersionController.is_latest(version)
|
|
|
|
@staticmethod
|
|
def sync(client_observer: ClientObserver = None, connection_observer: ConnectionObserver = None):
|
|
|
|
from core.controllers.ConnectionController import ConnectionController
|
|
ConnectionController.with_preferred_connection(task=ClientController.__sync, client_observer=client_observer, connection_observer=connection_observer)
|
|
|
|
@staticmethod
|
|
def update(client_observer: ClientObserver = None, connection_observer: ConnectionObserver = None):
|
|
|
|
from core.controllers.ConnectionController import ConnectionController
|
|
ConnectionController.with_preferred_connection(task=ClientController.__update, client_observer=client_observer, connection_observer=connection_observer)
|
|
|
|
@staticmethod
|
|
def __get_path():
|
|
|
|
if (path := Constants.HV_CLIENT_PATH) is None:
|
|
raise UnknownClientPathError('The client path could not be determined.')
|
|
|
|
return path
|
|
|
|
@staticmethod
|
|
def __sync(client_observer: Optional[ClientObserver] = None, proxies: Optional[dict] = None):
|
|
|
|
if client_observer is not None:
|
|
client_observer.notify('synchronizing')
|
|
|
|
# noinspection PyProtectedMember
|
|
ApplicationController._sync(proxies=proxies)
|
|
# noinspection PyProtectedMember
|
|
ApplicationVersionController._sync(proxies=proxies)
|
|
# noinspection PyProtectedMember
|
|
ClientVersionController._sync(proxies=proxies)
|
|
# noinspection PyProtectedMember
|
|
LocationController._sync(proxies=proxies)
|
|
# noinspection PyProtectedMember
|
|
SubscriptionPlanController._sync(proxies=proxies)
|
|
|
|
ConfigurationController.update_last_synced_at()
|
|
|
|
if client_observer is not None:
|
|
client_observer.notify('synchronized')
|
|
|
|
@staticmethod
|
|
|
|
def __update(client_observer: Optional[ClientObserver] = None, proxies: Optional[dict] = None):
|
|
|
|
if ClientController.can_be_updated():
|
|
|
|
if client_observer is not None:
|
|
client_observer.notify('updating')
|
|
|
|
client_path = ClientController.__get_path()
|
|
update_environment = os.environ.copy()
|
|
|
|
if proxies is not None:
|
|
update_environment | dict(http_proxy=proxies['http'], https_proxy=proxies['https'])
|
|
|
|
if shutil.which('hv-updater') is None:
|
|
raise CommandNotFoundError('hv-updater')
|
|
|
|
process = subprocess.Popen(('hv-updater', '--overwrite', '--remove-old', client_path), env=update_environment, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
|
|
highest_reported_progress = 0
|
|
|
|
for line in iter(process.stdout.readline, ''):
|
|
|
|
match = re.search(r'(\d+\.\d+)%', line)
|
|
|
|
if match and (progress := float(match.group(1))) > highest_reported_progress:
|
|
|
|
highest_reported_progress = progress
|
|
|
|
if client_observer is not None:
|
|
|
|
client_observer.notify('update_progressing', None, dict(
|
|
progress=progress
|
|
))
|
|
|
|
if client_observer is not None:
|
|
client_observer.notify('updated')
|