from lib import logging as logging_lib from typing import List from lib import utils import time log = logging_lib.log async def ensure_packages_installed( packages: List[str] = [], total_timeout: float = 300 ) -> bool: non_interactive_env = { 'DEBIAN_FRONTEND': 'noninteractive', 'PATH': '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin', } start_time = time.time() packages_to_install = [] for package in packages: check_cmd = f"dpkg -s {package} > /dev/null 2>&1" return_code, _, _ = await utils.async_run_command(check_cmd, env=non_interactive_env) if return_code != 0: packages_to_install.append(package) if not packages_to_install: log.debug("All packages are already installed.") return True update_cmd = ( "apt-get update " "-y " "--no-install-recommends" ) return_code, stdout, stderr = await utils.async_run_command( update_cmd, timeout=None if total_timeout == None else 180, env=non_interactive_env ) if return_code != 0: log.error(f"Failed to update package lists: {stderr}") return False install_cmd = ( "apt-get install " "-y " "--no-install-recommends " "--assume-yes " "-o Dpkg::Options::='--force-confdef' " # Default to existing config "-o Dpkg::Options::='--force-confold' " # Keep existing config f"{' '.join(packages_to_install)}" ) # Calculate remaining timeout remaining_timeout = None if total_timeout == None else max(0, total_timeout - (time.time() - start_time)) # Install packages return_code, stdout, stderr = await utils.async_run_command( install_cmd, timeout=remaining_timeout, env=non_interactive_env ) if return_code == 0: log.debug(f"Successfully installed packages: {packages_to_install}") return True else: log.error(f"Failed to install packages: {stderr}") return False