### ===================================================================================================================
### Update server functions
### ===================================================================================================================
# Copyright ©2026 Haskoning Nederland B.V.
# For use by VIIA
### ===================================================================================================================
### 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 typing import List, Union
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 haskoning_datafusr_py_base package
from haskoning_datafusr_py_base._email import send_email
# References for functions and classes in the haskoning_structural package
from haskoning_structural.fem_tools import fem_copy_file
from haskoning_structural.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[Union[str, List[str]]] = None):
""" Gets the current and latest version of a package """
input = [interpreter, "-m", "pip", "index", "versions", package]
if extra_index_url:
if isinstance(extra_index_url, str):
extra_index_url = [extra_index_url]
for eiu in extra_index_url:
input.append('--extra-index-url=' + eiu)
_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 = "\n\nInstalled software:\n"
message += f"\t{'SOFTWARE'.ljust(90, ' ')}VERSION"
message += f"\n\t{'-'*100}"
programs = get_installed_software()
programs = sorted(programs, key=lambda x: x['name'])
for program in programs:
if not program['name']:
continue
message += f"\n\t{program['name'].ljust(90, ' ')}{program['version']}"
return message
return "\n\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):
r"""
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@haskoning.com,jurriaan.floor@haskoning.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/"
extra_index_url_2 = rf"https://{azure_token}corporateroot.pkgs.visualstudio.com/_packaging/haskoning-py/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, extra_index_url_2], 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
input_pip = [interpreter, "-m", "pip", "install", "--upgrade", "pip"]
_var_pip = run(input_pip, capture_output=True, timeout=1200)
input_artifacts = [interpreter, "-m", "pip", "install", "--upgrade", 'artifacts-keyring']
_var_artifacts = run(input_artifacts, capture_output=True, timeout=1200)
# Update the server venv
input = [
interpreter, "-m", "pip", "install", "--upgrade", package,
'--extra-index-url=' + extra_index_url, '--extra-index-url=' + extra_index_url_2]
_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, extra_index_url_2], 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_pip)
email_message += add_pip_message(run_output=_var_artifacts)
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]
email_message += viia_get_installed_software()
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("C:/Program Files/Diana 10.9/python/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
### ===================================================================================================================