Source code for dak.import_repository

#! /usr/bin/env python3
#
# Copyright (C) 2015, Ansgar Burchardt <ansgar@debian.org>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

import sys
from collections import defaultdict

import apt_pkg

import daklib.archive
import daklib.config
import daklib.dbconn
import daklib.import_repository
import daklib.utils

# This is still work-in-progress and far too incomplete.
# ruff: noqa


[docs]def usage(status=0): print( """ dak import-repository --keyring=/usr/share/keyrings/debian-archive-keyring.gpg [--key=${fingerprint}] [--architectures=a,b,c (default: architectures in origin suite)] [--components=main,contrib (default: components in origin suite)] [--target-suite=${suite} (default: origin suite name)] [--add-overrides] [--max-packages=${n} (import at maximum ${n} packages, default: no limit)] http://httpredir.debian.org/debian unstable Things to think about: - Import Built-Using sources - all / only referenced - Remove old packages: - by-source: remove source X_v, if no X exists upstream - by-version: remove source X_v, if no X_v exists upstream (X denotes package name, v version, X_v package at a specific version) - Import all or only newest? - Expire binary packages? """ ) sys.exit(status)
[docs]def entry_is_newer(entry, packages): version = entry["Version"] for p in packages[entry["Package"]]: if apt_pkg.version_compare(version, p.version) <= 0: return False return True
[docs]def entry_in_packages(entry, packages): return entry["Package"] in packages
[docs]def get_packages_in_suite(suite): sources = defaultdict(list) for s in suite.sources: sources[s.source].append(s) packages = defaultdict(list) for b in suite.binaries: packages[b.package].append(b) return sources, packages
[docs]def import_sources( base, sources, transaction, target_suite, component, target_sources, extra_sources, extra_sources_comp, max_packages=None, ): n = 0 for entry in sources: if max_packages is not None and n > max_packages: break if entry.get("Extra-Source-Only", "no") == "yes": # Remember package, we might need to import it later. key = (entry["Package"], entry["Version"]) extra_sources[key] = entry extra_sources_comp[key].add(c) continue if not entry_in_packages(entry, target_sources) or entry_is_newer( entry, target_sources ): print("Importing {0}={1}".format(entry["Package"], entry["Version"])) daklib.import_repository.import_source_to_suite( base, entry, transaction, target_suite, component ) n += 1 return n
[docs]def import_built_using( base, source, version, transaction, target_suite, component, extra_sources, extra_sources_comp, ): if not daklib.import_repository.source_in_archive( bu_source, bu_version, target_suite.archive ): print("Importing extra source {0}={1}".format(bu_source, bu_version)) key = (bu_source, bu_version) extra_entry = extra_sources.get(key) if extra_entry is None: raise Exception( "Extra source {0}={1} referenced by {2}={3} ({4}) not found in source suite.".format( bu_source, bu_version, entry["Package"], entry["Version"], architecture, ) ) # extra_components = extra_sources_comp[key] if c in components: extra_component = component else: # TODO: Take preferred components from those listed... raise Exception("Not implemented.") daklib.import_repository.import_source_to_suite( base, extra_entry, transaction, target_suite, extra_component )
[docs]def import_packages( base, packages, transaction, target_suite, component, architecture, target_binaries, extra_sources, extra_sources_comp, max_packages=None, ): n = 0 for entry in packages: if max_packages is not None and n > max_packages: break if not entry_in_packages(entry, target_binaries) or entry_is_newer( entry, target_binaries ): print( "Importing {0}={1} ({2})".format( entry["Package"], entry["Version"], architecture ) ) # Import Built-Using sources: for bu_source, bu_version in daklib.utils.parse_built_using(entry): import_built_using( base, bu_source, bu_version, transaction, target_suite, component, extra_sources, extra_sources_comp, ) # Import binary: daklib.import_repository.import_package_to_suite( base, entry, transaction, target_suite, component ) n += 1 return n
[docs]def main(argv=None): if argv is None: argv = sys.argv arguments = [ ("h", "help", "Import-Repository::Help"), ("k", "keyring", "Import-Repository::Keyring", "HasArg"), ("K", "key", "Import-Repository::Key", "HasArg"), ("a", "architectures", "Import-Repository::Architectures", "HasArg"), ("c", "components", "Import-Repository::Components", "HasArg"), ("t", "target-suite", "Import-Repository::Target-Suite", "HasArg"), ("A", "add-overrides", "Import-Repository::AddOverrides"), ("n", "max-packages", "Import-Repository::MaxPackages", "HasArg"), ] cnf = daklib.config.Config() argv = apt_pkg.parse_commandline(cnf.Cnf, arguments, argv) options = cnf.subtree("Import-Repository") if "Help" in options or len(argv) < 2: usage(0) keyring = options.find("Keyring") or None if keyring is None: print("Error: No keyring specified") print() if "Key" in options: raise Exception("Not implemented.") if "AddOverrides" in options: raise Exception("Not implemented.") if "MaxPackages" in options: max_packages = int(options["MaxPackages"]) else: max_packages = None base, suite = argv[0:2] target_suite_name = options.find("Target-Suite") or suite print( "Importing packages from {0}/dists/{1} to {2}".format( base, suite, target_suite_name ) ) with daklib.archive.ArchiveTransaction() as transaction: target_suite = daklib.dbconn.get_suite(target_suite_name, transaction.session) if target_suite is None: daklib.utils.fubar( "Target suite '{0}' is unknown.".format(target_suite_name) ) release = daklib.import_repository.obtain_release(base, suite, keyring) target_sources, target_binaries = get_packages_in_suite(target_suite) if "Architectures" in options: architectures = options["Architectures"].split(",") else: architectures = ["all"] + release.architectures() if "Components" in options: components = options["Components"].split(",") else: components = release.components() # TODO: Clean this up... n = 0 # For Extra-Source-Only sources packages, keep a dict # (name, version) -> entry and (name, version) -> set of components # to allow importing needed packages at a later stage extra_sources = dict() extra_sources_comp = defaultdict(set) for c in components: component = daklib.dbconn.get_component(c, transaction.session) print("Processing {0}/source...".format(c)) sources = release.sources(c) imported = import_sources( base, sources, transaction, target_suite, component, target_sources, extra_sources, extra_sources_comp, max_packages, ) print(" imported {0} source packages".format(imported)) n += imported if max_packages is not None: max_packages -= n for c in components: component = daklib.dbconn.get_component(c, transaction.session) for architecture in architectures: print("Processing {0}/{1}...".format(c, architecture)) packages = release.packages(c, architecture) imported = import_packages( base, packages, transaction, target_suite, component, architecture, target_binaries, extra_sources, extra_sources_comp, max_packages, ) print(" imported {0} binary packages".format(imported)) n += imported if max_packages is not None: max_packages -= n transaction.rollback()
if __name__ == "__main__": main()