124 lines
5.0 KiB
Python
124 lines
5.0 KiB
Python
import sys
|
|
import subprocess
|
|
import urllib.parse
|
|
import urllib.request
|
|
import os
|
|
from pathlib import Path
|
|
import pathlib
|
|
import time
|
|
import json
|
|
import logging
|
|
import zipfile
|
|
from threading import Thread
|
|
|
|
from libs.app.utils import Utils, Package, get_logger, root_dir
|
|
|
|
logger = get_logger("updater")
|
|
|
|
class Updater():
|
|
def __init__(self, utils:Utils):
|
|
self.utils = utils
|
|
self.libs_path = os.path.join(root_dir, 'libs')
|
|
self.start()
|
|
|
|
def start(self):
|
|
logger.info(f"---------------- Updater Started ----------------")
|
|
# Create better validation code
|
|
if("updateApp" in self.utils.kargs):
|
|
if(self.utils.kargs["updateApp"]):
|
|
self.update_app()
|
|
elif(self.utils.configs.data["updater"]["autoUpdate"]):
|
|
self.update_app()
|
|
|
|
if("restart" in self.utils.flags and self.utils.flags["restart"]):
|
|
self.utils.kargs["updateApp"] = False
|
|
self.utils.restart_app()
|
|
|
|
if("updateLibs" in self.utils.kargs):
|
|
if(self.utils.kargs["updateLibs"]):
|
|
self.update_libs()
|
|
elif(self.utils.configs.data["updater"]["checkUpdates"]):
|
|
self.update_libs()
|
|
|
|
logger.info(f"---------------- Updater Ended ----------------")
|
|
|
|
def update_app(self):
|
|
latest_version_repository = (self.utils.configs.data["app"]["version"],None)
|
|
logger.info(f"App current version: {latest_version_repository[0]}")
|
|
|
|
logger.info(f"Updating app files")
|
|
# TODO FIX THIS
|
|
self.check_package_update(Package(config=self.utils.configs.data["app"]))
|
|
|
|
logger.info(f"App updated to version {latest_version_repository[0]}")
|
|
self.utils.flags["restart"] = True
|
|
|
|
def update_libs(self):
|
|
logger.info(f"Checking packages update")
|
|
threads:list[Thread] = []
|
|
for package in self.utils.configs.packages.values():
|
|
# Default is check updates
|
|
if(not "checkUpdates" in package.config or package.config["checkUpdates"]):
|
|
thread = Thread(target=self.check_package_update, args=(package,))
|
|
thread.start()
|
|
threads.append(thread)
|
|
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
def check_package_update(self, package:Package):
|
|
logger.info(f"Updating package {package.config['id']}")
|
|
package_id = package.config['id']
|
|
|
|
package_path = os.path.join(self.libs_path, package_id)
|
|
if(package.info):
|
|
latest_version_repository = (package.info["version"],None)
|
|
else:
|
|
logger.info(f"Creating package directory: {package_path}")
|
|
Path(package_path).mkdir(parents=True, exist_ok=True)
|
|
latest_version_repository = (-1,None)
|
|
|
|
repositories = self.utils.configs.data["updater"]["repositories"][:]
|
|
if "repositories" in package.config:
|
|
for repository in package.config["repositories"]:
|
|
repositories.append(repository)
|
|
|
|
for repository in repositories:
|
|
try:
|
|
package_url = f"{repository}/{package_id}"
|
|
info_url = f'{package_url}/info.json'
|
|
logger.info(f"Reading remote info of {package_id} from repository {repository}")
|
|
with urllib.request.urlopen(info_url, timeout=2) as data:
|
|
remote_info = json.load(data)
|
|
logger.debug(f"{package_id} remote info: {remote_info}")
|
|
|
|
# TODO Remove 'or' condition -> "or remote_info["version"]==0"
|
|
if remote_info["version"] > latest_version_repository[0] or remote_info["version"]==0:
|
|
latest_version_repository = (remote_info["version"], package_url)
|
|
except Exception as e:
|
|
logger.error(f"Error reading remote info of {package_id} from repository {repository}: {e}")
|
|
|
|
if latest_version_repository[1]:
|
|
package_zip = f"{latest_version_repository[1]}/{package_id}.zip"
|
|
package_local_file = os.path.join(self.libs_path, f"{package_id}.zip")
|
|
logger.debug(f"Downloading package from {package_zip} to local file {package_local_file}")
|
|
self.download_package(package_zip, package_local_file)
|
|
self.pip_install_requirements(package_path)
|
|
|
|
|
|
def download_package(self, url, path):
|
|
urllib.request.urlretrieve(url, path)
|
|
logger.info(f"Extracting all from {path}")
|
|
with zipfile.ZipFile(path, 'r') as zip_ref:
|
|
zip_ref.extractall(Path(path).parent.absolute())
|
|
|
|
def pip_install_requirements(self, package_path):
|
|
path = os.path.join(package_path, "requirements.txt")
|
|
if(os.path.exists(path)):
|
|
logger.info(f"Pip installing requirements in {path}")
|
|
subprocess.check_call([sys.executable, "-m", "pip", "install", "-r", path])
|
|
else:
|
|
logger.info(f"{package_path} no requirements.txt")
|
|
|
|
def check_module_requirements(self, package_path):
|
|
pass |