#!/usr/bin/python3
# codeing = utf-8
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Copyright (C) 2021 WindRiver Corporation
import apt
import apt_pkg
import aptly_deb_usage
import argparse
from concurrent.futures import as_completed
from concurrent.futures import ThreadPoolExecutor
import debian.deb822
import debian.debfile
import logging
import os
import requests
from requests.compat import urljoin
import shutil
from threading import Lock
import urllib.request
import utils
REPOMGR_URL = os.environ.get('REPOMGR_URL')
REPOMGR_ORIGIN = os.environ.get('REPOMGR_ORIGIN')
REPOMGR_DEPLOY_URL = os.environ.get('REPOMGR_DEPLOY_URL')
APTFETCH_JOBS = 20
class AptFetch():
'''
Fetch Debian packages from a set of repositories.
'''
def __init__(self, logger, sources_list='', workdir='/tmp/apt-fetch'):
self.logger = logger
self.aptcache = None
self.aptlock = Lock()
self.workdir = workdir
self.sources_list = sources_list
self.__construct_workdir(sources_list)
self.__init_apt_cache()
def __construct_workdir(self, sources_list):
'''construct some directories for repo and temporary files'''
# In case the sources_list is specified:
#
# ├── apt-root # For apt cache
# │ └── etc
# │ └── apt
# │ └── sources.list
# └── downloads # Sub directory to store downloaded packages
# ├── binary
# └── source
#
# In case the sources_list is ''
# |
# └── downloads # Sub directory to store downloaded packages
# ├── binary
# └── source
basedir = self.workdir
try:
if os.path.exists(basedir):
shutil.rmtree(basedir)
os.makedirs(basedir)
if self.sources_list:
# check to see if meta file exist
if not os.path.exists(sources_list):
raise Exception('Specified sources list file %s does not exist' % sources_list)
aptdir = os.path.join(basedir, 'apt-root/etc/apt')
os.makedirs(aptdir)
shutil.copyfile(sources_list, os.path.join(aptdir, 'sources.list'))
os.makedirs(os.path.join(basedir, 'downloads', 'binary'))
os.makedirs(os.path.join(basedir, 'downloads', 'source'))
except Exception as e:
self.logger.error(str(e))
self.logger.error('Failed to construct workdir %s' % basedir)
raise Exception('Failed to construct workdir %s' % basedir)
def __init_apt_cache(self):
'''Construct APT cache based on specified rootpath. Just like `apt update` on host'''
# In case we use host's apt settings, no need to apt-upgrade.
if not self.sources_list:
self.aptcache = apt.Cache()
return None
try:
self.aptcache = apt.Cache(rootdir=os.path.join(self.workdir, 'apt-root'))
ret = self.aptcache.update()
if not ret:
raise Exception('APT cache update failed')
self.aptcache.open()
except Exception as e:
self.logger.error(str(e))
self.logger.error('apt cache init failed.')
raise Exception('apt cache init failed.')
# Download a binary package into downloaded folder
def fetch_deb(self, pkg_name, pkg_version):
'''Download a binary package'''
ret = ''
if not pkg_name or not pkg_version:
return ret
self.logger.info("Current downloading:%s:%s", pkg_name, pkg_version)
destdir = os.path.join(self.workdir, 'downloads', 'binary')
self.aptlock.acquire()
try:
pkg = self.aptcache[pkg_name]
if not pkg:
self.aptlock.release()
self.logger.error("Failed to find binary package %s", pkg_name)
return ret
default_candidate = pkg.candidate
self.logger.debug("The default candidate is %s", default_candidate.version)
candidate = pkg.versions.get(pkg_version)
if not candidate:
if ':' in default_candidate.version:
epoch, ver = default_candidate.version.split(':')
if epoch.isdigit() and ver == pkg_version:
self.logger.debug('epoch %s will be skipped for %s_%s', epoch, pkg_name, ver)
candidate = default_candidate
if not candidate:
self.aptlock.release()
self.logger.error("Failed to found the matched version %s for %s", pkg_version, pkg_name)
return ret
except Exception as e:
self.aptlock.release()
self.logger.error("Exception during candidate searching:%s", str(e))
return ret
uri = candidate.uri
filename = candidate.filename
self.aptlock.release()
try:
res = requests.get(uri, stream=True)
self.logger.debug('Fetching package file %s' % uri)
with open(os.path.join(destdir, os.path.basename(filename)), 'wb') as download_file:
for chunk in res.iter_content(chunk_size=1024 * 1024):
if chunk:
download_file.write(chunk)
except Exception as e:
self.logger.error(str(e))
self.logger.error('Binary package %s %s download error' % (pkg_name, pkg_version))
return ' '.join(['DEB-F', pkg_name, pkg_version]).strip()
else:
self.logger.debug('Binary package %s %s downloaded.' % (pkg_name, pkg_version))
return ' '.join(['DEB', pkg_name, pkg_version]).strip()
# Download a source package into downloaded folder
def fetch_dsc(self, pkg_name, pkg_version=''):
'''Download a source package'''
if not pkg_name:
raise Exception('Source package name empty')
destdir = os.path.join(self.workdir, 'downloads', 'source')
self.aptlock.acquire()
src = apt_pkg.SourceRecords()
source_find = False
source_lookup = src.lookup(pkg_name)
while source_lookup:
if pkg_name == src.package and pkg_version in ['', src.version]:
source_find = True
break
source_lookup = src.lookup(pkg_name)
if not source_find:
self.aptlock.release()
raise ValueError("Source package %s %s was not found" % (pkg_name, pkg_version))
dict_files = dict()
for src_file in src.files:
dict_files[src_file.path] = src.index.archive_uri(src_file.path)
self.aptlock.release()
# Here the src.files is a list, each one points to a source file
# Download those source files one by one with requests
try:
for file_path, uri in dict_files.items():
res = requests.get(uri, stream=True)
self.logger.debug('Fetch package file %s' % uri)
with open(os.path.join(destdir, os.path.basename(file_path)), 'wb') as download_file:
for chunk in res.iter_content(chunk_size=1024 * 1024):
if chunk:
download_file.write(chunk)
except Exception as e:
self.logger.error(str(e))
self.logger.error('Source package %s %s download error' % (pkg_name, pkg_version))
return ' '.join(['DSC-F', pkg_name, pkg_version]).strip()
else:
self.logger.debug('Source package %s %s downloaded.' % (pkg_name, pkg_version))
return ' '.join(['DSC', pkg_name, pkg_version]).strip()
# Download a bundle of packages into downloaded folder
# deb_list: binary package list file
# dsc_list: source package list file
def fetch_pkg_list(self, deb_set=set(), dsc_set=set()):
'''Download a bundle of packages specified through deb_list and dsc_list.'''
if not deb_set and not dsc_set:
raise Exception('deb_list and dsc_list, at least one is required.')
fetch_result = dict()
fetch_result['deb'] = list()
fetch_result['deb-failed'] = list()
fetch_result['dsc'] = list()
fetch_result['dsc-failed'] = list()
with ThreadPoolExecutor(max_workers=APTFETCH_JOBS) as threads:
obj_list = []
# Download binary packages
for pkg_ver in deb_set:
pkg_name = pkg_ver.split()[0]
if len(pkg_ver.split()) == 1:
pkg_version = ''
else:
pkg_version = pkg_ver.split()[1]
obj = threads.submit(self.fetch_deb, pkg_name, pkg_version)
if not obj:
self.logger.error("Failed to download %s:%s", pkg_name, pkg_version)
else:
obj_list.append(obj)
# Download source packages
for pkg_ver in dsc_set:
pkg_name = pkg_ver.split()[0]
if len(pkg_ver.split()) == 1:
pkg_version = ''
else:
pkg_version = pkg_ver.split()[1]
obj = threads.submit(self.fetch_dsc, pkg_name, pkg_version)
obj_list.append(obj)
# Wait...
for future in as_completed(obj_list):
ret = future.result()
if ret:
if ret.startswith('DEB'):
fetch_result['deb'].append(ret.lstrip('DEB '))
elif ret.startswith('DEB-F'):
fetch_result['deb-failed'].append(ret.lstrip('DEB-F '))
elif ret.startswith('DSC'):
fetch_result['dsc'].append(ret.lstrip('DSC '))
elif ret.startswith('DSC-F'):
fetch_result['dsc-failed'].append(ret.lstrip('DSC-F '))
return fetch_result
class RepoMgr():
'''
Repository management, based on pulp or aptly, mainly used for OBS and LAT
Two kind of repositories: local repo and remote one.
remote repo: mirror of another repository. shouldn't insert or remove packages from it..
local repo: a local repository, we can insert/remove packages into/from them.
'''
def __init__(self, repoType, repoURL, workdir, origin, logger):
if repoType == 'aptly':
self.repo = aptly_deb_usage.Deb_aptly(repoURL, origin, logger)
else:
raise Exception('Currently, only aptly repository supported')
self.aptcache = None
self.logger = logger
self.workdir = workdir
def __sync_pkg_list(self, check_list, pkg_list_file, pkg_type='binary'):
'''Sync packages, return packages need to be downloaded'''
if pkg_type not in ['binary', 'source']:
self.logger.error('Parameter pkg_tye:%s error. Can only be binary or source' % pkg_type)
raise Exception('Parameter pkg_tye:%s error. Can only be binary or source' % pkg_type)
pkg_req = set()
# scan the package list file
pkg_set = self.__scan_pkg_list(pkg_list_file)
for pkg_ver in pkg_set:
pkg_name = pkg_ver.split()[0]
if len(pkg_ver.split()) == 1:
pkg_version = ''
else:
pkg_version = pkg_ver.split()[1]
# Search package in check_list
if self.repo.pkg_exist(check_list, pkg_name, pkg_type, pkg_version):
continue
pkg_req.add(pkg_ver)
self.logger.info('%d %s packages need to be downloaded.' % (len(pkg_req), pkg_type))
return pkg_req
# Scan package list file, return required package_version in a set
def __scan_pkg_list(self, pkg_list_file):
pkg_set = set()
try:
with open(pkg_list_file, 'r') as f_list:
lines = f_list.readlines()
except Exception as e:
self.logger.error(str(e))
self.logger.error('Package list file %s read error.' % pkg_list_file)
raise Exception('Package list file %s read error.' % pkg_list_file)
else:
for line in lines:
pkg_info = line.split('#')[0].split()
if not pkg_info:
continue
if len(pkg_info) == 1:
pkg_set.add(pkg_info[0])
else:
pkg_set.add(pkg_info[0] + ' ' + pkg_info[1])
# If xxx_1.2.3 exist, remove xxx.
tmp_set = pkg_set.copy()
for pkg_ver in tmp_set:
if pkg_ver.split()[0] in pkg_set:
pkg_set.remove(pkg_ver.split()[0])
return pkg_set
# Download a bundle of packages and deployed them through repository
# repo_name: the repository used to deploy these packages
# no_clear: Not delete downloaded package files, for debug
# kwarge:sources_list: Where we should fetch packages from.
# kwarge:deb_list: file contains binary package list
# kwarge:dsc_ist: file contains source package list
# Output: None
def download(self, repo_name, no_clear=False, **kwargs):
'''Download specified packages and deploy them through a specified local repo.'''
if 'deb_list' not in kwargs.keys():
deb_list_file = ''
else:
deb_list_file = kwargs['deb_list']
if 'dsc_list' not in kwargs.keys():
dsc_list_file = ''
else:
dsc_list_file = kwargs['dsc_list']
if not deb_list_file and not dsc_list_file:
raise Exception('deb_list and dsc_list, at least one is required.')
if not repo_name == '':
# No matter if any packages can be download, create related repository firstly.
if not self.repo.create_local(repo_name):
raise Exception('Local repo created failed, Please double check if it already exists.')
# Scan deb/dsc_list_file, get required packages
deb_set = set()
dsc_set = set()
if deb_list_file:
deb_set = self.__scan_pkg_list(deb_list_file)
self.logger.info('%d binary packages will be downloaded.' % len(deb_set))
if dsc_list_file:
dsc_set = self.__scan_pkg_list(dsc_list_file)
self.logger.info('%d source packages will be downloaded.' % len(dsc_set))
# Download packages
apt_fetch = AptFetch(self.logger, kwargs['sources_list'], self.workdir)
fetch_result = apt_fetch.fetch_pkg_list(deb_set=deb_set, dsc_set=dsc_set)
if fetch_result['deb'] or fetch_result['deb-failed']:
self.logger.info('Download %d binary packages' % len(fetch_result['deb']))
self.logger.info('%d binary packages download failed' % len(fetch_result['deb-failed']))
for pkg_ver in fetch_result['deb']:
self.logger.debug('Binary package %s' % pkg_ver)
for pkg_ver in fetch_result['deb-failed']:
self.logger.info('Failed to download binary package %s' % pkg_ver)
if fetch_result['dsc'] or fetch_result['dsc-failed']:
self.logger.info('Download %d source packages' % len(fetch_result['dsc']))
self.logger.info('%d source packages download failed' % len(fetch_result['dsc-failed']))
for pkg_ver in fetch_result['dsc']:
self.logger.debug('Source package %s' % pkg_ver)
for pkg_ver in fetch_result['dsc-failed']:
self.logger.info('Failed to download source package %s' % pkg_ver)
if repo_name == '':
self.logger.info('All packages are downloaded to %s', self.workdir)
return
# Add packages into local repo
pkg_folder = os.path.join(self.workdir, 'downloads', 'binary')
package_files = set()
for filename in os.listdir(pkg_folder):
package_files.add(os.path.join(pkg_folder, filename))
if package_files:
self.repo.upload_pkg_local(package_files, repo_name)
pkg_folder = os.path.join(self.workdir, 'downloads', 'source')
for filename in os.listdir(pkg_folder):
if filename.endswith('.dsc'):
self.upload_pkg(repo_name, os.path.join(pkg_folder, filename))
# Deploy local repo
repo_str = self.repo.deploy_local(repo_name)
if not no_clear:
shutil.rmtree(self.workdir)
self.logger.info('New local repo can be accessed through: %s' % repo_str)
# We need a bundle packages been deployed through a serials of repositories
# We have:
# -) a serials of atply/pulp repositories already contains some packages;
# -) an aptly/pulp repository can be used to deploy downloaded packages
# -) a serials of upstream Debian repository
# -) Two text files list binary and source packages we need
#
# SYNC will download all packages we haven't and deployed them through
# a local repository.
# reponame: name of a local repository used to deploy downloaded packages
# repo_list: String separated with space, contains serials of aptly/pulp
# repos can be used for OBS/LAT.
# no_clear: do not delete downloaded packages. For debug
# kwargs:sources_list: file contains trusted upstream repositories
# kwargs:deb_list: file lists all needed binary packages
# kwargs:dsc_list: file lists all needed source packages
# Output: None
def sync(self, repo_name, repo_list, no_clear=False, **kwargs):
'''
Sync a set of repositories with specified package lists, if any packages
missed, download and deploy them through a specified local repo
'''
if 'deb_list' not in kwargs.keys():
deb_list = ''
else:
deb_list = kwargs['deb_list']
if 'dsc_list' not in kwargs.keys():
dsc_list = ''
else:
dsc_list = kwargs['dsc_list']
if not deb_list and not dsc_list:
raise Exception('deb_list and dsc_list, at least one is required.')
# construct repo list will be checked
local_list = self.repo.list_local(quiet=True)
remote_list = self.repo.list_remotes(quiet=True)
# Specified local repo must exist, or failed
if repo_name not in local_list:
self.logger.info('Sync, local repo %s does not exist, create it.' % repo_name)
if not self.repo.create_local(repo_name):
raise Exception('Local repo %s created failed.' % repo_name)
# Make sure all repos in check_list exist in aptly/pulp database.
check_list = [repo_name]
for repo in repo_list.split():
if repo in local_list or repo in remote_list:
if repo not in check_list:
check_list.append(repo)
else:
self.logger.warning('%s in the list but does not exists.' % repo)
# Download missing packages from remote repo
deb_set = set()
dsc_set = set()
if deb_list:
deb_set = self.__sync_pkg_list(check_list, deb_list, 'binary')
if dsc_list:
dsc_set = self.__sync_pkg_list(check_list, dsc_list, 'source')
# Download packages
if deb_set or dsc_set:
apt_fetch = AptFetch(self.logger, kwargs['sources_list'], self.workdir)
fetch_result = apt_fetch.fetch_pkg_list(deb_set=deb_set, dsc_set=dsc_set)
if fetch_result['deb'] or fetch_result['deb-failed']:
self.logger.info('Download %d binary packages' % len(fetch_result['deb']))
self.logger.info('%d binary packages download failed' % len(fetch_result['deb-failed']))
for pkg_ver in fetch_result['deb']:
self.logger.debug('Binary package %s' % pkg_ver)
for pkg_ver in fetch_result['deb-failed']:
self.logger.info('Failed to download binary package %s' % pkg_ver)
if fetch_result['dsc'] or fetch_result['dsc-failed']:
self.logger.info('Download %d source packages' % len(fetch_result['dsc']))
self.logger.info('%d source packages download failed' % len(fetch_result['dsc-failed']))
for pkg_ver in fetch_result['dsc']:
self.logger.debug('Source package %s' % pkg_ver)
for pkg_ver in fetch_result['dsc-failed']:
self.logger.info('Failed to download source package %s' % pkg_ver)
# Add packages into local repo
pkg_folder = os.path.join(self.workdir, 'downloads', 'binary')
package_files = set()
for filename in os.listdir(pkg_folder):
package_files.add(os.path.join(pkg_folder, filename))
self.repo.upload_pkg_local(package_files, repo_name)
pkg_folder = os.path.join(self.workdir, 'downloads', 'source')
for filename in os.listdir(pkg_folder):
if filename.endswith('.dsc'):
self.upload_pkg(repo_name, os.path.join(pkg_folder, filename))
self.repo.upload_pkg_local(package_files, repo_name)
# Deploy local repo
repo_str = self.repo.deploy_local(repo_name)
if not no_clear:
try:
if os.path.exists(self.workdir):
shutil.rmtree(self.workdir)
except Exception as e:
self.logger.error(str(e))
self.logger.error('Clear work folder %s failed.' % self.workdir)
raise Exception('Clear work folder %s failed.' % self.workdir)
self.logger.info('local repo can be accessed through: %s ' % repo_str)
# Merge all packages of several repositories into a new publication(aptly)
# NOTE: aptly only. Not find similar feature in pulp...
def merge(self, name, source_snapshots):
'''Merge several repositories into a new aptly publication.'''
return self.repo.merge_repos(name, source_snapshots.split(','))
# Construct a repository mirror to an upstream Debian repository
# kwargs:url: URL of the upstream repo (http://deb.debian.org/debian)
# kwargs:distribution: the distribution of the repo (bullseye)
# kwargs:component: component of the repo (main)
# kwargs:architecture: architecture of the repo, "all" is always enabled. (amd64)
# kwargs:with_sources: include source packages, default is False.
# Output: None
def mirror(self, repo_name, **kwargs):
'''Construct a mirror based on a debian repository.'''
url = kwargs['url']
distribution = kwargs['distribution']
component = kwargs['component']
architectures = kwargs['architectures']
if 'with_sources' not in kwargs.keys():
with_sources = False
else:
with_sources = kwargs['with_sources']
self.repo.remove_remote(repo_name)
if with_sources:
self.repo.create_remote(repo_name, url, distribution,
components=[component],
architectures=[architectures],
with_sources=True)
else:
self.repo.create_remote(repo_name, url, distribution,
components=[component],
architectures=[architectures])
repo_str = self.repo.deploy_remote(repo_name)
if repo_str != None:
self.logger.info('New mirror can be accessed through: %s' % repo_str)
return True
else:
self.logger.error('Create mirror failed %s : %s' % (repo_name, url))
return False
# List all repositories
# Output: None
def list(self):
'''List all repos.'''
self.repo.list_remotes()
self.repo.list_local()
# Clean all packages and repositories
def clean(self):
'''Clear all meta files. Construct a clean environment'''
self.repo.clean_all()
# Copy specified packages from one repository to another
def copy_pkgs(self, source, dest, pkg_names, pkg_type='binary', deploy=True, overwrite=True):
'''Copy specified packages from source repository to destination repository.'''
if pkg_type != 'binary' and pkg_type != 'source':
self.logger.error('The package type must be binary or source')
return False
if source == dest:
self.logger.error('%s and %s are the same repository.' % (source, dest))
return False
if pkg_type != 'binary' and pkg_type != 'source':
self.logger.error('The package type must be binary or source')
return False
pkg_list = list()
for pkg in pkg_names.split(','):
pkg_list.append(pkg.strip())
ret = self.repo.copy_pkgs(source, dest, pkg_list, pkg_type)
if not ret:
self.logger.error('Copy %s packages from %s to %s failed' % (pkg_type, source, dest))
return False
elif deploy:
self.repo.deploy_local(dest)
return True
# list a repository
# repo_name: the name of the repo been listed.
# Output: True is all works in order
def list_pkgs(self, repo_name, quiet=False):
'''List a specified repository.'''
local_list = self.repo.list_local(quiet=True)
remote_list = self.repo.list_remotes(quiet=True)
pkg_list = []
for repo in local_list:
if repo == repo_name:
self.logger.info('List a local repo')
pkgs = self.repo.pkg_list([repo])
pkg_list.extend(pkgs)
if not quiet:
self.logger.info("Local repo %s:" % repo_name)
for pkg in sorted(pkgs):
self.logger.info(" %s" % pkg)
for repo in remote_list:
if repo == repo_name:
self.logger.info('List a remote mirror')
pkgs = self.repo.pkg_list([repo])
pkg_list.extend(pkgs)
if not quiet:
self.logger.info("Remote repo %s:" % repo_name)
for pkg in sorted(pkgs):
self.logger.info(" %s" % pkg)
return pkg_list
# delete a repository
# repo_name: the name of the repo been deleted.
# Output: True is all works in order
def remove_repo(self, repo_name):
'''Remove a specified repository.'''
local_list = self.repo.list_local(quiet=True)
remote_list = self.repo.list_remotes(quiet=True)
for repo in local_list:
if repo == repo_name:
self.logger.info('Remove a local repo')
self.repo.remove_local(repo)
return True
for repo in remote_list:
if repo == repo_name:
self.logger.info('Remove a remote mirror')
self.repo.remove_remote(repo)
return True
self.logger.warning("Remove repo failed: repo '%s' not found" % repo_name)
return False
# Before uploading a source package into a local repo, scan all repos,
# find all duplicate files with different size.
# dsc: Dsc data of the source package. <class 'debian.deb822.Dsc'>
# Return a dictionary: {repo_1: {file_a, ...}, ...}
# Input dsc contains file_a, while repo_1 also contains such a file
# with different size.
def __check_orig_files(self, dsc):
different_files = {}
repo_list = self.repo.list_local(quiet=True)
repo_list += self.repo.list_remotes(quiet=True)
for repo in repo_list:
for meta_file in dsc['Files']:
if meta_file['name'].find('.orig.'):
new_file_size = meta_file['size']
if dsc['Source'].startswith('lib'):
prefix_dir = dsc['Source'][:4] + '/' + dsc['Source']
else:
prefix_dir = dsc['Source'][0] + '/' + dsc['Source']
target_path = repo + '/pool/main/' + prefix_dir + '/' + meta_file['name']
target_url = urljoin(REPOMGR_DEPLOY_URL, target_path)
try:
orig_file = urllib.request.urlopen(target_url)
except Exception:
# no such file in repo, that is good
self.logger.debug('%s does not contain %s' % (repo, meta_file['name']))
continue
if orig_file.length != int(new_file_size):
self.logger.debug('File %s is not same as the one in %s.' %
(meta_file['name'], repo))
if repo not in different_files.keys():
different_files[repo] = {meta_file['name']}
else:
different_files[repo].add(meta_file['name'])
return different_files
# upload a Debian package and deploy it
# repo_name: the name of the repository used to contain and deploy the package
# package: pathname of the package(xxx.deb or xxx.dsc) to be uploaded
# deploy: If True, deploy the repository after "binary" package been uploaded.
# Output: True if all works.
def upload_pkg(self, repo_name, package, deploy=True):
'''Upload a Debian package into a specified repository.'''
self.logger.info("upload_pkg: %s to %s", package, repo_name)
local_list = self.repo.list_local(quiet=True)
if repo_name not in local_list:
self.logger.info('upload_pkg: repository %s does not exist, creating it.' % repo_name)
self.repo.create_local(repo_name)
if not package and deploy:
# No repo found, no package specified, just create & deploy the repo and return
self.repo.deploy_local(repo_name)
return True
self.logger.debug('upload_pkg: upload package %s into %s' % (package, repo_name))
if '.deb' == os.path.splitext(package)[-1]:
pkg_type = 'binary'
try:
deb = debian.debfile.DebFile(package, 'r').debcontrol()
except Exception as e:
self.logger.error('Error: %s' % e)
self.logger.error('Binary package %s read error.' % package)
raise Exception('Binary package error.')
pkg_version = deb['Version']
pkg_name = deb['Package']
self.repo.upload_pkg_local({package}, repo_name)
elif '.dsc' == os.path.splitext(package)[-1]:
pkg_type = 'source'
try:
dsc = debian.deb822.Dsc(open(package, 'r'))
except Exception as e:
self.logger.error('Error: %s' % e)
self.logger.error('Source package %s read error.' % package)
raise Exception('Source package error.')
pkg_name = dsc['Source']
pkg_version = dsc['Version']
# In case there is already an *.orig.* file with different size in any repos,
# refuse to upload it.
different_files = self.__check_orig_files(dsc)
if different_files:
for repo, meta_files in different_files.items():
self.logger.error('%s contains different file: %s' % (repo, str(meta_files)))
self.logger.error('Package %s upload failed. Repo %s already contains '
'file %s with different content.' %
(package, repo, str(meta_files)))
return False
pkg_files = set()
pkg_files.add(package)
for meta_file in dsc['Files']:
pkg_files.add(os.path.join(os.path.dirname(package), meta_file['name']))
self.repo.upload_pkg_local(pkg_files, repo_name)
else:
self.logger.warning('Only Debian style files, like deb and dsc, are supported.')
return False
self.logger.debug('upload_pkg: package %s been uploaded into %s' % (package, repo_name))
# deploy = False only effect on binary package.
if 'binary' == pkg_type and not deploy:
return True
self.repo.deploy_local(repo_name)
# Double check if the package been uploaded successfully
if not self.search_pkg(repo_name, pkg_name, pkg_version, pkg_type == 'binary'):
self.logger.error('upload_pkg: verify failed, no %s package %s %s in %s'
% (pkg_type, pkg_name, pkg_version, repo_name))
return False
return True
# search a package from a repository
# repo_name: name of the repository to search the package in
# pkg_name: name of the Debian package to be searched
# pkg_version: version number of the package to be searched
# binary: binary package or source one?
# Output: True if find, or False
def search_pkg(self, repo_name, pkg_name, pkg_version=None, binary=True):
'''Find a package from a specified repo.'''
repo_find = False
repo = None
r_list = self.repo.list_local(quiet=True)
for repo in r_list:
if repo == repo_name:
repo_find = True
r_list = self.repo.list_remotes(quiet=True)
for repo in r_list:
if repo == repo_name:
repo_find = True
if not repo_find:
self.logger.error('Search package, repository does not exist.')
return False
if binary:
pkg_type = 'binary'
else:
pkg_type = 'source'
if not self.repo.pkg_exist([repo_name], pkg_name, pkg_type, pkg_version):
self.logger.info('Search %s package %s, not found.' % (pkg_type, pkg_name))
return False
return True
# Deploy a local repository
def deploy_repo(self, repo_name, suffix=''):
'''Deploy a local repository manually.'''
local_list = self.repo.list_local(quiet=True)
if repo_name not in local_list:
self.logger.error('Local repository deploy failed, %s does not exist.' % repo_name)
return
return self.repo.deploy_local(repo_name, suffix)
# Delete a Debian package from a local repository
# repo_name: name of the LOCAL repository to delete the package from
# pkg_name: name of the binary package to be deleted
# pkg_type: 'source' or 'binary'
# pkg_version: version number of the package to be deleted
# Output: True if find and delete, or False
def delete_pkg(self, repo_name, pkg_name, pkg_type, pkg_version='', deploy=True):
'''Find and delete a binary package from a specified local repo.'''
repo_find = False
repo = None
if pkg_type not in {'binary', 'source'}:
self.logger.error('Delete package, pkg_type must be one of '
'either "binary" or "source".')
return False
if not repo_name.startswith(aptly_deb_usage.PREFIX_LOCAL):
self.logger.error('Delete package, only local repositories support this operation.')
return False
local_list = self.repo.list_local(quiet=True)
for repo in local_list:
if repo == repo_name:
repo_find = True
if not repo_find:
self.logger.error('Delete package, repository does not exist.')
return False
if not self.repo.pkg_exist([repo_name], pkg_name, pkg_type, pkg_version):
self.logger.info('Delete package, package not found.')
return False
self.repo.delete_pkg_local(repo_name, pkg_name, pkg_type, pkg_version)
# deploy = False only effect on binary packages
if 'binary' == pkg_type and not deploy:
return True
self.repo.deploy_local(repo_name)
return True
# Simple example on using this class.
applogger = logging.getLogger('repomgr')
utils.set_logger(applogger)
def _handleDownload(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, args.basedir, REPOMGR_ORIGIN, applogger)
kwargs = {'sources_list': args.sources_list, 'deb_list': args.deb_list,
'dsc_list': args.dsc_list}
repomgr.download(args.repository, **kwargs, no_clear=args.no_clear)
def _handleSync(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, args.basedir, REPOMGR_ORIGIN, applogger)
kwargs = {'sources_list': args.sources_list, 'deb_list': args.deb_list,
'dsc_list': args.dsc_list}
repomgr.sync(args.repository, args.repo_list, **kwargs, no_clear=args.no_clear)
def _handleMirror(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
kwargs = {'url': args.url, 'distribution': args.distribution, 'component': args.component,
'architectures': args.architectures, 'with_sources': args.with_sources}
repomgr.mirror(args.repository, **kwargs)
def _handleMerge(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
repomgr.merge(args.name, args.repo_list)
def _handleCopyPkg(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
repomgr.copy_pkgs(args.source, args.destination, args.package_name, deploy=True, pkg_type=args.package_type)
def _handleUploadPkg(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
repomgr.upload_pkg(args.repository, args.package)
def _handleDeletePkg(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
repomgr.delete_pkg(args.repository, args.package_name, args.package_type,
pkg_version=args.package_version)
def _handleSearchPkg(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
if args.package_type == 'binary':
repomgr.search_pkg(args.repository, args.package_name, pkg_version=args.package_version,
binary=True)
else:
repomgr.search_pkg(args.repository, args.package_name, pkg_version=args.package_version,
binary=False)
def _handleRemoveRope(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
repomgr.remove_repo(args.repository)
def _handleList(_args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
repomgr.list()
def _handleListPkgs(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
repomgr.list_pkgs(args.repository)
def _handleClean(_args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', REPOMGR_ORIGIN, applogger)
repomgr.clean()
def subcmd_download(subparsers):
download_parser = subparsers.add_parser('download',
help='Download specified packages and deploy them '
'through a new build repository.\n\n')
download_parser.add_argument('--repository', '-r', help='Name of the local repository', required=False,
default='deb-local-down')
download_parser.add_argument('--deb_list', help='Binary package list file', required=False, default='')
download_parser.add_argument('--dsc_list', help='Source package list file', required=False, default='')
download_parser.add_argument('--basedir', help='Temporary folder to store packages', required=False,
default='/tmp/repomgr')
download_parser.add_argument('--sources_list', help='Upstream sources list file.', required=False,
default='')
download_parser.add_argument('--no-clear', help='Not remove temporary files', action='store_true')
download_parser.set_defaults(handle=_handleDownload)
def subcmd_sync(subparsers):
sync_parser = subparsers.add_parser('sync',
help='Sync a set of repositories with specified package'
'lists.\n\n')
sync_parser.add_argument('--repository', '-r', help='Name of the local repository', required=False,
default='deb-local-sync')
sync_parser.add_argument('--repo_list', '-l', help='a set of local repos', required=False,
default='')
sync_parser.add_argument('--deb_list', help='Binary package list file', required=False,
default='')
sync_parser.add_argument('--dsc_list', help='Source package list file', required=False,
default='')
sync_parser.add_argument('--basedir', help='Temporary folder to store packages', required=False,
default='/tmp/repomgr')
sync_parser.add_argument('--sources_list', help='Upstream sources list file', required=False,
default='')
sync_parser.add_argument('--no-clear', help='Not remove temporary files', action='store_true')
sync_parser.set_defaults(handle=_handleSync)
def subcmd_mirror(subparsers):
mirror_parser = subparsers.add_parser('mirror',
help='Construct a mirror based on a remote'
'repository.\n\n')
mirror_parser.add_argument('--repository', '-r', help='Name of the remote repository', required=False,
default='deb-remote-tmp')
mirror_parser.add_argument('--url', help='URL of remote repository', required=False,
default='http://nginx.org/packages/debian/')
mirror_parser.add_argument('--distribution', '-d', help='distribution name', required=False,
default='buster')
mirror_parser.add_argument('--component', '-c', help='component name', required=False,
default='nginx')
mirror_parser.add_argument('--architectures', '-a', help='architectures', required=False,
default='amd64')
mirror_parser.add_argument('--with-sources', '-s', help='include source packages',
action='store_true')
mirror_parser.set_defaults(handle=_handleMirror)
def subcmd_merge(subparsers):
merge_parser = subparsers.add_parser('merge',
help='Merge several repositories into a new publication.\n\n')
merge_parser.add_argument('--name', '-n', help='Name of the new merged publication', required=True)
merge_parser.add_argument('--repo_list', '-l', help='a set of repos, seperate by comma', required=True)
merge_parser.set_defaults(handle=_handleMerge)
def subcmd_copy_pkg(subparsers):
copy_pkg_parser = subparsers.add_parser('copy_pkg',
help='Copy specified packages from source repository into destination one.\n\n')
copy_pkg_parser.add_argument('--source', '-s', help='Name of the source repository', required=True)
copy_pkg_parser.add_argument('--destination', '-d', help='Name of the destination repository, must be a local repo', required=True)
copy_pkg_parser.add_argument('--package_name', '-p', help='Name of the package(s) to be copied, separate by comma', required=True)
copy_pkg_parser.add_argument('--package_type', '-t', help='Package type, "binary" or "source"', required=False, default='binary')
copy_pkg_parser.set_defaults(handle=_handleCopyPkg)
def main():
# command line arguments
parser = argparse.ArgumentParser(add_help=True,
description='Repository management Tool',
epilog='''Tips: Use %(prog)s --help to get help for all of '
'parameters\n\n''')
subparsers = parser.add_subparsers(title='Repo control Commands:',
help='sub-command for repo-ctl\n\n')
# Three functions for three sub commands: pylint checking(too-many-statements)
subcmd_download(subparsers)
subcmd_sync(subparsers)
subcmd_mirror(subparsers)
subcmd_merge(subparsers)
subcmd_copy_pkg(subparsers)
clean_parser = subparsers.add_parser('clean', help='Clear all aptly repos.\n\n')
clean_parser.set_defaults(handle=_handleClean)
remove_repo_parser = subparsers.add_parser('remove_repo',
help='Remove a specific repository.\n\n')
remove_repo_parser.add_argument('--repository', '-r',
help='Name of the repo to be removed')
remove_repo_parser.set_defaults(handle=_handleRemoveRope)
upload_pkg_parser = subparsers.add_parser('upload_pkg',
help='Upload a Debian package into a specific '
'repository.\n\n')
upload_pkg_parser.add_argument('--package', '-p',
help='Debian package to be uploaded.', required=False)
upload_pkg_parser.add_argument('--repository', '-r',
help='Repository used to deploy this package.')
upload_pkg_parser.set_defaults(handle=_handleUploadPkg)
delete_pkg_parser = subparsers.add_parser('delete_pkg',
help='Delete a specified Debian package from a '
'specified repository.\n\n')
delete_pkg_parser.add_argument('--package_name', '-p', help='Package name to be deleted.')
delete_pkg_parser.add_argument('--package_type', '-t',
help='Package type to be deleted, "binary" or "source".')
delete_pkg_parser.add_argument('--package_version', '-v',
help='Version number of the package.', required=False)
delete_pkg_parser.add_argument('--repository', '-r', help='Local Repository to delete the '
'package from.')
delete_pkg_parser.set_defaults(handle=_handleDeletePkg)
search_pkg_parser = subparsers.add_parser('search_pkg',
help='Search a specified package from a specified '
'repository.\n\n')
search_pkg_parser.add_argument('--package_name', '-p', help='package to be looking for.')
search_pkg_parser.add_argument('--package_version', '-v', help='The version of the package.',
required=False)
search_pkg_parser.add_argument('--repository', '-r',
help='The Repository to search the package in.')
search_pkg_parser.add_argument('--package_type', '-t', help='binary or source',
required=False)
search_pkg_parser.set_defaults(handle=_handleSearchPkg)
list_parser = subparsers.add_parser('list',
help='List all aptly repos, including local repo and '
'remote mirror.\n\n')
list_parser.set_defaults(handle=_handleList)
list_pkgs_parser = subparsers.add_parser('list_pkgs',
help='List contents of a specific repo.\n\n')
list_pkgs_parser.add_argument('--repository', '-r',
help='Name of the repo to be listed')
list_pkgs_parser.set_defaults(handle=_handleListPkgs)
args = parser.parse_args()
if hasattr(args, 'handle'):
args.handle(args)
else:
parser.print_help()
if __name__ == '__main__':
main()