Source code for viiapackage.tools.viia_update_server

### ===================================================================================================================
###   Update server functions
### ===================================================================================================================
# Copyright ©VIIA 2025

### ===================================================================================================================
###   1. Import modules
### ===================================================================================================================

# General imports
from pathlib import Path
from sys import executable
from subprocess import run
from os import getenv
from typing import Optional
from shutil import rmtree

from viiapackage import installed_modules
if 'windows_tools' in installed_modules:
    from windows_tools.installed_software import get_installed_software
else:
    get_installed_software = None

# References for functions and classes in the datafusr_py_base package
from datafusr_py_base._email import send_email

# References for functions and classes in the rhdhv_fem
from rhdhv_fem.fem_tools import fem_copy_file
from rhdhv_fem.tools.fem_email import fem_get_email_credentials, fem_get_azure_token


### ===================================================================================================================
###   2. Utility functions
### ===================================================================================================================

[docs]def get_package_version(package: str, interpreter: str, extra_index_url: Optional[str] = None): """ Gets the current and latest version of a package """ input = [interpreter, "-m", "pip", "index", "versions", package] if extra_index_url: input.append('--extra-index-url=' + extra_index_url) _var = run(input, capture_output=True, timeout=60) _installed_version = None _latest_version = None for line in _var.stdout.splitlines(): line = str(line.decode("utf-8")) if 'INSTALLED' in line: _installed_version = line.replace('INSTALLED', '').replace(':', '').strip() if 'LATEST' in line: _latest_version = line.replace('LATEST', '').replace(':', '').strip() return _installed_version, _latest_version
[docs]def add_pip_message(run_output): """ Collects the pip info and return it as a string """ pip_message = '\n' pip_message += f"\n\nDuring installing the following return code is generated:" pip_message += f"\n\t {str(run_output.returncode)}" pip_message += f"\n\nDuring installing the following code is output is generated:" for line in run_output.stdout.splitlines(): line = str(line.decode("utf-8")).strip() pip_message += "\n\t" + line pip_message += f"\n\nDuring installing the following errorcode is generated:" for line in run_output.stderr.splitlines(): line = str(line.decode("utf-8")).strip() pip_message += "\n\t" + line return pip_message
def viia_get_installed_software(): if get_installed_software: message = "\nInstalled software:\n" message += "\tSOFTWARE".ljust(70) + "VERSION" message += f"\n\t{'-'*80}" programms = get_installed_software() programms = sorted(programms, key=lambda x: x['name']) for programm in programms: if not programm['name']: continue message += f"\n\t{programm['name'].ljust(70)}{programm['version']}" return message return "\nInstalled software could not be retrieved because 'windows_tools' is not installed.\n" ### =================================================================================================================== ### 3. Function to update the venv on the server ### ===================================================================================================================
[docs]def viia_update_server_venv(recipient_email: str): """ Updates the venv on the server and will send an email when needed. Can be executed with a batch file with the following lines: set emailaddress=reinier.ringers@rhdhv.com,jurriaan.floor@rhdhv.com "E:\VIIA\server_venv\Scripts\python.exe" -c "from viiapackage.tools import viia_update_server_venv ; viia_update_server_venv(recipient_email=r'%emailaddress%')" TIMEOUT 600 Input: - recipient_email (str): The email address to where the email should be sent. Can be multiple divided by a comma. Output: - Pip install will be updated if possible. Email will be sent if needed. """ package = 'viiapackage' interpreter = Path(executable).as_posix() server_name = getenv('COMPUTERNAME') azure_token = fem_get_azure_token(key_token='SERVER_AZURE_TOKEN', key_key='SERVER_AZURE_KEY') extra_index_url = rf"https://{azure_token}@corporateroot.pkgs.visualstudio.com/VIIA/_packaging/viiapackage@Release/pypi/simple/" email_message = 'Hello server admins,' email_message += '\n' email_message += '\nThis is an auto generated email.' email_message += '\n' # Get the current versions current_installed_version, current_latest_function = \ get_package_version(package=package, extra_index_url=extra_index_url, interpreter=interpreter) if current_installed_version == current_latest_function: # latest version is installed print(f"No update required {current_installed_version=} and {current_latest_function=}") return # Update the server pip install input = [interpreter, "-m", "pip", "install", "--upgrade", package, '--extra-index-url=' + extra_index_url] _var = run(input, capture_output=True, timeout=1200) # Get the new current versions new_installed_version, new_latest_function = \ get_package_version(package=package, extra_index_url=extra_index_url, interpreter=interpreter) diana_text = '' if "C:/" in interpreter and 'Diana 10.6' in interpreter: diana_text = 'Diana 10.6 ' if "C:/" in interpreter and 'Diana 10.9' in interpreter: diana_text = 'Diana 10.9 ' if new_installed_version == new_latest_function: email_message += f"\nThe {diana_text}venv on server {server_name} is updated." email_message += f"\nThe {package} is updated from {current_installed_version} to {new_installed_version}." subject = f"Successful {diana_text}update of {package} on server {server_name} to version {new_installed_version}" elif current_installed_version != new_installed_version: email_message += f"\nThe {diana_text}venv on server {server_name} is updated." email_message += f"\nThe {package} is updated from {current_installed_version} to {new_installed_version}." email_message += f"\nLatest version is {new_latest_function} which could not be installed." subject = f"Unsuccessful {diana_text}update of {package} on server {server_name} to version {new_latest_function} ({new_installed_version})" else: email_message += f"\nThe {diana_text}venv on server {server_name} is not updated." email_message += f"\nThe current version of {package} is {new_installed_version}." email_message += f"\nLatest version is {new_latest_function} which could not be installed." subject = f"Unsuccessful {diana_text}update of {package} on server {server_name} to version {new_latest_function} ({new_installed_version})" email_message += '\n\nKind regards' email_message += add_pip_message(run_output=_var) credentials = fem_get_email_credentials(key_credentials='SERVER_MAIL_ADDRESS', key_key='SERVER_KEY') sender_email, sender_password = credentials[0], credentials[1] send_email( recipient_email=recipient_email, sender_email=sender_email, sender_password=sender_password, subject=subject, message=[{'subtype': 'plain', 'text': email_message}]) print(email_message) # Purge cache to reduce disk size input = [interpreter, "-m", "pip", "cache", "purge"] _var = run(input, capture_output=True, timeout=60)
### =================================================================================================================== ### 4. Install server requirements ### ===================================================================================================================
[docs]def viia_install_server_requirements(recipient_email: str): """ Installs the requirements-servers.txt """ def clone_package(clone_folder, azure_link): print(f'clone_folder {clone_folder}') # make new folder for clone clone_folder.mkdir() # Get azure token and link print('azure_token') # Clone viiapackage print('clone') input = ["git", "clone", azure_link, clone_folder.as_posix()] _var_clone = run(input, capture_output=True, timeout=3000) print('checkout master') input = ["git", "checkout", "master"] _var_checkout = run(input, capture_output=True, timeout=3000, cwd=clone_folder.as_posix()) return _var_clone, _var_checkout # Get default locations server_venv_interpreter = Path("E:/VIIA/server_venv/Scripts/python.exe") diana_interpreter = Path("E:/VIIA/server_venv/Scripts/python.exe") server_admin = Path("E:/VIIA/server_admin") clone_folder = server_admin / 'viiapackage' server_name = getenv('COMPUTERNAME') requirements_file = clone_folder / "requirements-server.txt" cloned_server_admin_folder = clone_folder / 'developers_tools' / 'server_update' / 'server_admin' packages = [] email_message = "" _var_clone = None _var_checkout = None _var_fetch = None _var_pull = None package_versions = dict() output = dict() package_updated = False # Get azure token and link print("Azure link") azure_token = fem_get_azure_token(key_token='SERVER_AZURE_TOKEN', key_key='SERVER_AZURE_KEY') azure_link = f"https://{azure_token}@corporateroot.visualstudio.com/VIIA/_git/viiaPackage" # Create clone or update clone if clone_folder.exists(): if requirements_file.exists(): print("checkout") input = ["git", "checkout", "master"] _var_checkout = run(input, capture_output=True, timeout=3000, cwd=clone_folder.as_posix()) print("fetch") input = ["git", "fetch"] _var_fetch = run(input, capture_output=True, timeout=3000, cwd=clone_folder.as_posix()) print("pull") input = ["git", "pull", clone_folder.as_posix()] _var_pull = run(input, capture_output=True, timeout=3000, cwd=clone_folder.as_posix()) else: print("remove folder") rmtree(clone_folder) _var_clone, _var_checkout = clone_package(clone_folder=clone_folder, azure_link=azure_link) else: _var_clone, _var_checkout = clone_package(clone_folder=clone_folder, azure_link=azure_link) # Install requirements and get old and new version if requirements_file.exists(): lines = open(requirements_file, mode='r').readlines() for line in lines: packages.append(line.strip().split("=")[0].split(">")[0].split("<")[0].split("~")[0]) if packages: for interpreter in [server_venv_interpreter, diana_interpreter]: if interpreter.exists(): package_versions[interpreter] = {} # get old versions for package in packages: if package not in package_versions[interpreter]: package_versions[interpreter][package] = {} installed, new = get_package_version(package=package, interpreter=interpreter.as_posix()) package_versions[interpreter][package]['old_installed'] = installed # install requirements input = [interpreter.as_posix(), "-m", "pip", "install", "-r", requirements_file.as_posix()] output[interpreter] = run(input, capture_output=True, timeout=1200) # get new versions for package in packages: installed, new = get_package_version(package=package, interpreter=interpreter.as_posix()) package_versions[interpreter][package]['new_installed'] = installed else: print(f"requirements_file not found: {requirements_file.as_posix()}.") for interpreter in package_versions: if package_versions[interpreter]: email_message += "\n\n" for package in package_versions[interpreter]: if package_versions[interpreter][package].get('old_installed') is None and package_versions[interpreter][package].get('new_installed') is None: continue if package_versions[interpreter][package]['old_installed'] != package_versions[interpreter][package]['new_installed']: package_updated = True email_message += f"\nPackage {package} for {interpreter} updated from {package_versions[interpreter][package]['old_installed']} to {package_versions[interpreter][package]['new_installed']}." email_message += "\n\n" for interpreter in package_versions: if package_versions[interpreter]: email_message += f"\nFor the server {interpreter} the next error code is generated:\n" add_pip_message(output[interpreter]) email_message += f"\n\n" subject = f"Optional requirements and/or clone installed for {server_name}" credentials = fem_get_email_credentials(key_credentials='SERVER_MAIL_ADDRESS', key_key='SERVER_KEY') sender_email, sender_password = credentials[0], credentials[1] clone_updated = False if _var_clone and _var_checkout: email_message += "\n\nA new clone has been made. The next feedback is given:" email_message += add_pip_message(_var_clone) email_message += "\n\nThe next feedback is given for checkout:" email_message += add_pip_message(_var_checkout) clone_updated = True if _var_checkout and _var_fetch and _var_pull: email_message += "\n\nThe clone has been updated. The next feedback is given:" email_message += "\n\nThe next feedback is given for checkout:" email_message += add_pip_message(_var_checkout) email_message += "\n\nThe next feedback is given for fetch:" email_message += add_pip_message(_var_fetch) email_message += "\n\nThe next feedback is given for pull:" pull_msg = add_pip_message(_var_pull) email_message += pull_msg if "Already up to date." not in pull_msg: clone_updated = True print("Copy files") if cloned_server_admin_folder.exists() and cloned_server_admin_folder.is_dir(): for file in cloned_server_admin_folder.iterdir(): if not file.is_file(): continue file_copied = fem_copy_file(file_to_copy=file, destination_folder=server_admin) if file_copied.is_file() and file_copied.exists(): email_message += f"\n\nThe {file_copied.as_posix()} is updated or added." else: email_message += f"\n\nError with {file_copied.as_posix()}, the file is not updated or added." email_message += viia_get_installed_software() if not package_updated and not clone_updated: # no optional packages are installed and clone not made or updated print(f"No package is updated.") print(email_message) return send_email( recipient_email=recipient_email, sender_email=sender_email, sender_password=sender_password, subject=subject, message=[{'subtype': 'plain', 'text': email_message}]) print(email_message)
### =================================================================================================================== ### 5. End of script ### ===================================================================================================================