### ===================================================================================================================
### 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
### ===================================================================================================================