# # Copyright (c) 2023 Wind River Systems, Inc. # # SPDX-License-Identifier: Apache-2.0 # ''' Fetch deb and subdebs from the build system ''' import os import sys import logging import shutil sys.path.append('..') import debsentry import repo_manage import utils BUILD_ROOT = os.environ.get('MY_BUILD_PKG_DIR') DEB_CONFIG_DIR = 'stx-tools/debian-mirror-tools/config/' PKG_LIST_DIR = os.path.join(os.environ.get('MY_REPO_ROOT_DIR'), DEB_CONFIG_DIR, 'debian/distro') logger = logging.getLogger('fetch_debs') utils.set_logger(logger) class FetchDebs(object): def __init__(self): self.need_dl_stx_pkgs = [] self.need_dl_binary_pkgs = [] self.output_dir = os.path.join(BUILD_ROOT, 'dl_debs') self.apt_src_file = os.path.join(BUILD_ROOT, 'aptsrc') self.setup_apt_source() self.debs_fetcher = repo_manage.AptFetch(logger, self.apt_src_file, self.output_dir) def get_debs_clue(self, btype): if btype != 'rt': btype = 'std' return os.path.join(BUILD_ROOT, 'caches', btype + '_debsentry.pkl') def get_all_debs(self): all_debs = set() debs_clue_std = self.get_debs_clue('std') debs_clue_rt = self.get_debs_clue('rt') for pkg in self.need_dl_stx_pkgs: subdebs_std = debsentry.get_subdebs(debs_clue_std, pkg, logger) subdebs_rt = debsentry.get_subdebs(debs_clue_rt, pkg, logger) if not subdebs_std and not subdebs_rt: logger.warning('Failed to get subdebs for %s with local debsentry cache', pkg) continue else: if subdebs_std: all_debs.update(set(subdebs_std)) if subdebs_rt: all_debs.update(set(subdebs_rt)) return all_debs def setup_apt_source(self): # clean up the output dir if os.path.exists(self.output_dir): shutil.rmtree(self.output_dir) os.makedirs(self.output_dir, exist_ok=True) try: with open(self.apt_src_file, 'w') as f: repo_url = os.environ.get('REPOMGR_DEPLOY_URL') apt_item = ' '.join(['deb [trusted=yes]', repo_url + 'deb-local-build', 'bullseye', 'main\n']) f.write(apt_item) apt_item = ' '.join(['deb [trusted=yes]', repo_url + 'deb-local-binary', 'bullseye', 'main\n']) f.write(apt_item) logger.debug(f'Created apt source file {self.apt_src_file} to download debs') except Exception as e: logger.error(str(e)) logger.error('Failed to create the apt source file') sys.exit(1) def fetch_stx_packages(self): ''' Download all debs and subdebs from the build system Save the files to $BUILD_ROOT/dl_debs ''' dl_debs = self.get_all_debs() if not dl_debs: logger.warn('No STX packages were found') return else: dl_debs_dict = {} for deb in dl_debs: # dl_debs_with_ver.append(deb.replace('_', ' ')) name, version = deb.split('_') if name not in dl_debs_dict: dl_debs_dict[name] = version logger.debug('##dldebs:%s', dl_debs_dict) # filter list based on stx-std.lst - Depecrated on master, replaced by debian_iso_image.inc on each repo stx_pkg_list_file = self.get_debian_pkg_iso_list() debs_to_remove = [] for deb in dl_debs_dict.keys(): # try to find the deb in the package list if deb not in stx_pkg_list_file: # remove if not found in all lines debs_to_remove.append(deb) for deb in debs_to_remove: dl_debs_dict.pop(deb) logger.debug(f'Package list after filtering:{dl_debs_dict}') logger.info(f'Total debs need to be downloaded: {len(dl_debs_dict)}') dl_debs_with_ver = [f'{k} {v}' for k, v in dl_debs_dict.items()] fetch_ret = self.download(dl_debs_with_ver) dl_bin_debs_dir = os.path.join(self.output_dir, 'downloads/binary') if len(fetch_ret['deb-failed']) == 0: logger.info(f'Successfully downloaded STX debs to {dl_bin_debs_dir}') else: logger.error(f'Failed to downloaded STX debs to {dl_bin_debs_dir}') def get_debian_pkg_iso_list(self): pkgs = [] cgcs_root_dir = os.environ.get('MY_REPO') package_file_name = 'debian_iso_image.inc' print(cgcs_root_dir) for root, dirs, files in os.walk(cgcs_root_dir): for file in files: if file == package_file_name: with open(os.path.join(root, package_file_name), 'r') as f: pkgs.extend(line.strip() for line in f if line.strip() and not line.startswith('#')) return pkgs def fetch_external_binaries(self): ''' Download all binaries from the build system apt_item = apt_item + ' '.join(['deb [trusted=yes]', repo_url + 'deb-local-binary', 'bullseye', 'main\n']) ''' # Get debs from base-bullseye.lst # https://opendev.org/starlingx/tools/src/branch/master/debian-mirror-tools/config/debian/common/base-bullseye.lst if not self.need_dl_binary_pkgs: logger.debug("No binary packages to download") return all_debs = set() package_list = os.path.join(os.environ.get('MY_REPO_ROOT_DIR'), 'stx-tools/debian-mirror-tools/config/debian/common/base-bullseye.lst') # find pkgs in the list file logger.debug(f'Packages to find {self.need_dl_binary_pkgs}') for pkg in self.need_dl_binary_pkgs: logger.debug(f'checking {pkg}') with open(package_list, 'r') as f: for line in f.readlines(): if pkg == line.split()[0]: logger.debug(f'Line for package {pkg} found') pkg_entry = ' '.join(line.split()[:2]) logger.debug(f'Adding "{pkg_entry}" to be downloaded') all_debs.add(pkg_entry) break else: logger.warning(f"Package '{pkg}' not found in the package list") logger.debug('Binary packages to download:%s', all_debs) fetch_ret = self.download(all_debs) dl_bin_debs_dir = os.path.join(self.output_dir, 'downloads/binary') if len(fetch_ret['deb-failed']) == 0: logger.info(f'Successfully downloaded external debs to {dl_bin_debs_dir} \n') else: logger.info(f'Failed to downloaded external debs to {dl_bin_debs_dir} \n') def download(self, all_debs): try: logger.debug('Downloading debs...') fetch_ret = self.debs_fetcher.fetch_pkg_list(all_debs) except Exception as e: logger.error(str(e)) logger.error('Exception has when fetching debs with repo_manage') sys.exit(1) return fetch_ret if __name__ == '__main__': fetch_debs = FetchDebs() # set the packages you want to download fetch_debs.need_dl_std_pkgs = ['sysinv'] fetch_debs.need_dl_rt_pkgs = [''] fetch_debs.need_dl_binary_pkgs = ['tzdata', 'curl', 'apache2'] fetch_debs.fetch_stx_packages()