Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:47
📄 __init__.py # __init__.py # DNF cli subpackage. # # Copyright (C) 2012-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import import dnf.exceptions class CliError(dnf.exceptions.Error): """CLI Exception. :api""" pass from dnf.cli.cli import Cli # :api from dnf.cli.commands import Command # :api	1.19 KB	2022-09-09 11:21
📁 __pycache__	-	2026-03-21 15:47
📄 aliases.py # aliases.py # Resolving aliases in CLI arguments. # # Copyright (C) 2018 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import _ import collections import dnf.cli from dnf.conf.config import PRIO_DEFAULT import dnf.exceptions import libdnf.conf import logging import os import os.path logger = logging.getLogger('dnf') ALIASES_DROPIN_DIR = '/etc/dnf/aliases.d/' ALIASES_CONF_PATH = os.path.join(ALIASES_DROPIN_DIR, 'ALIASES.conf') ALIASES_USER_PATH = os.path.join(ALIASES_DROPIN_DIR, 'USER.conf') class AliasesConfig(object): def __init__(self, path): self._path = path self._parser = libdnf.conf.ConfigParser() self._parser.read(self._path) @property def enabled(self): option = libdnf.conf.OptionBool(True) try: option.set(PRIO_DEFAULT, self._parser.getData()["main"]["enabled"]) except IndexError: pass return option.getValue() @property def aliases(self): result = collections.OrderedDict() section = "aliases" if not self._parser.hasSection(section): return result for key in self._parser.options(section): value = self._parser.getValue(section, key) if not value: continue result[key] = value.split() return result class Aliases(object): def __init__(self): self.aliases = collections.OrderedDict() self.conf = None self.enabled = True if self._disabled_by_environ(): self.enabled = False return self._load_main() if not self.enabled: return self._load_aliases() def _disabled_by_environ(self): option = libdnf.conf.OptionBool(True) try: option.set(PRIO_DEFAULT, os.environ['DNF_DISABLE_ALIASES']) return option.getValue() except KeyError: return False except RuntimeError: logger.warning( _('Unexpected value of environment variable: ' 'DNF_DISABLE_ALIASES=%s'), os.environ['DNF_DISABLE_ALIASES']) return True def _load_conf(self, path): try: return AliasesConfig(path) except RuntimeError as e: raise dnf.exceptions.ConfigError( _('Parsing file "%s" failed: %s') % (path, e)) except IOError as e: raise dnf.exceptions.ConfigError( _('Cannot read file "%s": %s') % (path, e)) def _load_main(self): try: self.conf = self._load_conf(ALIASES_CONF_PATH) self.enabled = self.conf.enabled except dnf.exceptions.ConfigError as e: logger.debug(_('Config error: %s'), e) def _load_aliases(self, filenames=None): if filenames is None: try: filenames = self._dropin_dir_filenames() except dnf.exceptions.ConfigError: return for filename in filenames: try: conf = self._load_conf(filename) if conf.enabled: self.aliases.update(conf.aliases) except dnf.exceptions.ConfigError as e: logger.warning(_('Config error: %s'), e) def _dropin_dir_filenames(self): # Get default aliases config filenames: # all files from ALIASES_DROPIN_DIR, # and ALIASES_USER_PATH as the last one (-> override all others) ignored_filenames = [os.path.basename(ALIASES_CONF_PATH), os.path.basename(ALIASES_USER_PATH)] def _ignore_filename(filename): return filename in ignored_filenames or\ filename.startswith('.') or\ not filename.endswith(('.conf', '.CONF')) filenames = [] try: if not os.path.exists(ALIASES_DROPIN_DIR): os.mkdir(ALIASES_DROPIN_DIR) for fn in sorted(os.listdir(ALIASES_DROPIN_DIR)): if _ignore_filename(fn): continue filenames.append(os.path.join(ALIASES_DROPIN_DIR, fn)) except (IOError, OSError) as e: raise dnf.exceptions.ConfigError(e) if os.path.exists(ALIASES_USER_PATH): filenames.append(ALIASES_USER_PATH) return filenames def _resolve(self, args): stack = [] self.prefix_options = [] def store_prefix(args): num = 0 for arg in args: if arg and arg[0] != '-': break num += 1 self.prefix_options += args[:num] return args[num:] def subresolve(args): suffix = store_prefix(args) if (not suffix or # Current alias on stack is resolved suffix[0] not in self.aliases or # End resolving suffix[0].startswith('\\')): # End resolving try: stack.pop() # strip the '\' if it exists if suffix[0].startswith('\\'): suffix[0] = suffix[0][1:] except IndexError: pass return suffix if suffix[0] in stack: # Infinite recursion detected raise dnf.exceptions.Error( _('Aliases contain infinite recursion')) # Next word must be an alias stack.append(suffix[0]) current_alias_result = subresolve(self.aliases[suffix[0]]) if current_alias_result: # We reached non-alias or '\' return current_alias_result + suffix[1:] else: # Need to resolve aliases in the rest return subresolve(suffix[1:]) suffix = subresolve(args) return self.prefix_options + suffix def resolve(self, args): if self.enabled: try: args = self._resolve(args) except dnf.exceptions.Error as e: logger.error(_('%s, using original arguments.'), e) return args	6.97 KB	2022-09-09 11:21
📄 cli.py # Copyright 2005 Duke University # Copyright (C) 2012-2016 Red Hat, Inc. # # 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 Library 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. # # Written by Seth Vidal """ Command line interface yum class and related. """ from __future__ import print_function from __future__ import absolute_import from __future__ import unicode_literals try: from collections.abc import Sequence except ImportError: from collections import Sequence from collections import defaultdict import datetime from fnmatch import fnmatch import logging import operator import os import random import rpm import sys import time import hawkey import libdnf.transaction from . import output from dnf.cli import CliError from dnf.i18n import ucd, _ import dnf import dnf.cli.aliases import dnf.cli.commands import dnf.cli.commands.alias import dnf.cli.commands.autoremove import dnf.cli.commands.check import dnf.cli.commands.clean import dnf.cli.commands.deplist import dnf.cli.commands.distrosync import dnf.cli.commands.downgrade import dnf.cli.commands.group import dnf.cli.commands.history import dnf.cli.commands.install import dnf.cli.commands.makecache import dnf.cli.commands.mark import dnf.cli.commands.module import dnf.cli.commands.reinstall import dnf.cli.commands.remove import dnf.cli.commands.repolist import dnf.cli.commands.repoquery import dnf.cli.commands.search import dnf.cli.commands.shell import dnf.cli.commands.swap import dnf.cli.commands.updateinfo import dnf.cli.commands.upgrade import dnf.cli.commands.upgrademinimal import dnf.cli.demand import dnf.cli.format import dnf.cli.option_parser import dnf.conf import dnf.conf.substitutions import dnf.const import dnf.db.history import dnf.exceptions import dnf.logging import dnf.persistor import dnf.plugin import dnf.rpm import dnf.sack import dnf.transaction import dnf.util import dnf.yum.misc logger = logging.getLogger('dnf') def _add_pkg_simple_list_lens(data, pkg, indent=''): """ Get the length of each pkg's column. Add that to data. This "knows" about simpleList and printVer. """ na = len(pkg.name) + 1 + len(pkg.arch) + len(indent) ver = len(pkg.evr) rid = len(pkg._from_repo) for (d, v) in (('na', na), ('ver', ver), ('rid', rid)): data[d].setdefault(v, 0) data[d][v] += 1 def _list_cmd_calc_columns(output, ypl): """ Work out the dynamic size of the columns to pass to fmtColumns. """ data = {'na' : {}, 'ver' : {}, 'rid' : {}} for lst in (ypl.installed, ypl.available, ypl.extras, ypl.autoremove, ypl.updates, ypl.recent): for pkg in lst: _add_pkg_simple_list_lens(data, pkg) if len(ypl.obsoletes) > 0: for (npkg, opkg) in ypl.obsoletesTuples: _add_pkg_simple_list_lens(data, npkg) _add_pkg_simple_list_lens(data, opkg, indent=" " * 4) data = [data['na'], data['ver'], data['rid']] columns = output.calcColumns(data, remainder_column=1) return (-columns[0], -columns[1], -columns[2]) def print_versions(pkgs, base, output): def sm_ui_time(x): return time.strftime("%c", time.gmtime(x)) rpmdb_sack = dnf.sack.rpmdb_sack(base) done = False for pkg in rpmdb_sack.query().installed().filterm(name=pkgs): if done: print("") done = True if pkg.epoch == '0': ver = '%s-%s.%s' % (pkg.version, pkg.release, pkg.arch) else: ver = '%s:%s-%s.%s' % (pkg.epoch, pkg.version, pkg.release, pkg.arch) name = output.term.bold(pkg.name) print(_(" Installed: %s-%s at %s") %(name, ver, sm_ui_time(pkg.installtime))) print(_(" Built : %s at %s") % (pkg.packager if pkg.packager else "", sm_ui_time(pkg.buildtime))) # :hawkey, no changelist information yet # print(_(" Committed: %s at %s") % (pkg.committer, # sm_ui_date(pkg.committime))) def report_module_switch(switchedModules): msg1 = _("The operation would result in switching of module '{0}' stream '{1}' to " "stream '{2}'") for moduleName, streams in switchedModules.items(): logger.warning(msg1.format(moduleName, streams[0], streams[1])) class BaseCli(dnf.Base): """This is the base class for yum cli.""" def __init__(self, conf=None): conf = conf or dnf.conf.Conf() super(BaseCli, self).__init__(conf=conf) self.output = output.Output(self, self.conf) def do_transaction(self, display=()): """Take care of package downloading, checking, user confirmation and actually running the transaction. :param display: `rpm.callback.TransactionProgress` object(s) :return: history database transaction ID or None """ if dnf.base.WITH_MODULES: if not self.conf.module_stream_switch: switchedModules = dict(self._moduleContainer.getSwitchedStreams()) if switchedModules: report_module_switch(switchedModules) msg = _("It is not possible to switch enabled streams of a module unless explicitly " "enabled via configuration option module_stream_switch.\n" "It is recommended to rather remove all installed content from the module, and " "reset the module using '{prog} module reset <module_name>' command. After " "you reset the module, you can install the other stream.").format( prog=dnf.util.MAIN_PROG) raise dnf.exceptions.Error(msg) trans = self.transaction pkg_str = self.output.list_transaction(trans) if pkg_str: logger.info(pkg_str) if trans: # Check which packages have to be downloaded install_pkgs = [] rmpkgs = [] install_only = True for tsi in trans: if tsi.action in dnf.transaction.FORWARD_ACTIONS: install_pkgs.append(tsi.pkg) elif tsi.action in dnf.transaction.BACKWARD_ACTIONS: install_only = False rmpkgs.append(tsi.pkg) # Close the connection to the rpmdb so that rpm doesn't hold the # SIGINT handler during the downloads. del self._ts # report the total download size to the user if not install_pkgs: self.output.reportRemoveSize(rmpkgs) else: self.output.reportDownloadSize(install_pkgs, install_only) if trans or self._moduleContainer.isChanged() or \ (self._history and (self._history.group or self._history.env)): # confirm with user if self.conf.downloadonly: logger.info(_("{prog} will only download packages for the transaction.").format( prog=dnf.util.MAIN_PROG_UPPER)) elif 'test' in self.conf.tsflags: logger.info(_("{prog} will only download packages, install gpg keys, and check the " "transaction.").format(prog=dnf.util.MAIN_PROG_UPPER)) is_bootc_transaction = dnf.util._BootcSystem.is_bootc_system() and \ os.path.realpath(self.conf.installroot) == "/" and \ not self.conf.downloadonly # Handle bootc transactions. `--transient` must be specified if # /usr is not already writeable. bootc_system = None if is_bootc_transaction: if self.conf.persistence == "persist": logger.info(_("Persistent transactions aren't supported on bootc systems.")) raise CliError(_("Operation aborted.")) assert self.conf.persistence in ("auto", "transient") bootc_system = dnf.util._BootcSystem() if not bootc_system.is_writable(): if self.conf.persistence == "auto": logger.info(_("This bootc system is configured to be read-only. Pass --transient to " "perform this transaction in a transient overlay which will reset when " "the system reboots.")) raise CliError(_("Operation aborted.")) assert self.conf.persistence == "transient" if not bootc_system.is_unlocked_transient(): # Only tell the user about the transient overlay if # it's not already in place logger.info(_("A transient overlay will be created on /usr that will be discarded on reboot. " "Keep in mind that changes to /etc and /var will still persist, and packages " "commonly modify these directories.")) self._persistence = libdnf.transaction.TransactionPersistence_TRANSIENT # Check whether the transaction modifies usr_drift_protected_paths transaction_protected_paths = defaultdict(list) for pkg in trans: for pkg_file_path in sorted(pkg.files): for protected_pattern in self.conf.usr_drift_protected_paths: if fnmatch(pkg_file_path, protected_pattern): transaction_protected_paths[pkg.nevra].append(pkg_file_path) if transaction_protected_paths: logger.info(_('This operation would modify the following paths, possibly introducing ' 'inconsistencies when the transient overlay on /usr is discarded. See the ' 'usr_drift_protected_paths configuration option for more information.')) for nevra, protected_paths in transaction_protected_paths.items(): logger.info(nevra) for protected_path in protected_paths: logger.info(" %s" % protected_path) raise CliError(_("Operation aborted. Pass --setopt=usr_drift_protected_paths= to disable this check and proceed anyway.")) else: # Not a bootc transaction. if self.conf.persistence == "transient": raise CliError(_("Transient transactions are only supported on bootc systems.")) self._persistence = libdnf.transaction.TransactionPersistence_PERSIST if self._promptWanted(): if self.conf.assumeno or not self.output.userconfirm(): raise CliError(_("Operation aborted.")) if bootc_system: bootc_system.make_writable() else: logger.info(_('Nothing to do.')) return if trans: if install_pkgs: logger.info(_('Downloading Packages:')) try: total_cb = self.output.download_callback_total_cb self.download_packages(install_pkgs, self.output.progress, total_cb) except dnf.exceptions.DownloadError as e: specific = dnf.cli.format.indent_block(ucd(e)) errstr = _('Error downloading packages:') + '\n%s' % specific # setting the new line to prevent next chars being eaten up # by carriage returns print() raise dnf.exceptions.Error(errstr) # Check GPG signatures self.gpgsigcheck(install_pkgs) if self.conf.downloadonly: return if not isinstance(display, Sequence): display = [display] display = [output.CliTransactionDisplay()] + list(display) tid = super(BaseCli, self).do_transaction(display) # display last transaction (which was closed during do_transaction()) if tid is not None: trans = self.history.old([tid])[0] trans = dnf.db.group.RPMTransaction(self.history, trans._trans) else: trans = None if trans: # the post transaction summary is already written to log during # Base.do_transaction() so here only print the messages to the # user arranged in columns print() print('\n'.join(self.output.post_transaction_output(trans))) print() for tsi in trans: if tsi.state == libdnf.transaction.TransactionItemState_ERROR: raise dnf.exceptions.Error(_('Transaction failed')) return tid def gpgsigcheck(self, pkgs): """Perform GPG signature verification on the given packages, installing keys if possible. :param pkgs: a list of package objects to verify the GPG signatures of :raises: Will raise :class:`Error` if there's a problem """ error_messages = [] for po in pkgs: result, errmsg = self._sig_check_pkg(po) if result == 0: # Verified ok, or verify not req'd continue elif result == 1: ay = self.conf.assumeyes and not self.conf.assumeno if (not sys.stdin or not sys.stdin.isatty()) and not ay: raise dnf.exceptions.Error(_('Refusing to automatically import keys when running ' \ 'unattended.\nUse "-y" to override.')) # the callback here expects to be able to take options which # userconfirm really doesn't... so fake it fn = lambda x, y, z: self.output.userconfirm() try: self._get_key_for_package(po, fn) except (dnf.exceptions.Error, ValueError) as e: error_messages.append(str(e)) else: # Fatal error error_messages.append(errmsg) if error_messages: for msg in error_messages: logger.critical(msg) raise dnf.exceptions.Error(_("GPG check FAILED")) def latest_changelogs(self, package): """Return list of changelogs for package newer then installed version""" newest = None # find the date of the newest changelog for installed version of package # stored in rpmdb for mi in self._rpmconn.readonly_ts.dbMatch('name', package.name): changelogtimes = mi[rpm.RPMTAG_CHANGELOGTIME] if changelogtimes: newest = datetime.date.fromtimestamp(changelogtimes[0]) break chlogs = [chlog for chlog in package.changelogs if newest is None or chlog['timestamp'] > newest] return chlogs def format_changelog(self, changelog): """Return changelog formatted as in spec file""" chlog_str = '* %s %s\n%s\n' % ( changelog['timestamp'].strftime("%a %b %d %X %Y"), dnf.i18n.ucd(changelog['author']), dnf.i18n.ucd(changelog['text'])) return chlog_str def print_changelogs(self, packages): # group packages by src.rpm to avoid showing duplicate changelogs bysrpm = dict() for p in packages: # there are packages without source_name, use name then. bysrpm.setdefault(p.source_name or p.name, []).append(p) for source_name in sorted(bysrpm.keys()): bin_packages = bysrpm[source_name] print(_("Changelogs for {}").format(', '.join([str(pkg) for pkg in bin_packages]))) for chl in self.latest_changelogs(bin_packages[0]): print(self.format_changelog(chl)) def check_updates(self, patterns=(), reponame=None, print_=True, changelogs=False): """Check updates matching given patterns in selected repository.""" ypl = self.returnPkgLists('upgrades', patterns, reponame=reponame) if self.conf.obsoletes or self.conf.verbose: typl = self.returnPkgLists('obsoletes', patterns, reponame=reponame) ypl.obsoletes = typl.obsoletes ypl.obsoletesTuples = typl.obsoletesTuples if print_: columns = _list_cmd_calc_columns(self.output, ypl) if len(ypl.updates) > 0: local_pkgs = {} highlight = self.output.term.MODE['bold'] if highlight: # Do the local/remote split we get in "yum updates" for po in sorted(ypl.updates): local = po.localPkg() if os.path.exists(local) and po.verifyLocalPkg(): local_pkgs[(po.name, po.arch)] = po cul = self.conf.color_update_local cur = self.conf.color_update_remote self.output.listPkgs(ypl.updates, '', outputType='list', highlight_na=local_pkgs, columns=columns, highlight_modes={'=' : cul, 'not in' : cur}) if changelogs: self.print_changelogs(ypl.updates) if len(ypl.obsoletes) > 0: print(_('Obsoleting Packages')) # The tuple is (newPkg, oldPkg) ... so sort by new for obtup in sorted(ypl.obsoletesTuples, key=operator.itemgetter(0)): self.output.updatesObsoletesList(obtup, 'obsoletes', columns=columns) return ypl.updates or ypl.obsoletes def distro_sync_userlist(self, userlist): """ Upgrade or downgrade packages to match the latest versions available in the enabled repositories. :return: (exit_code, [ errors ]) exit_code is:: 0 = we're done, exit 1 = we've errored, exit with error string 2 = we've got work yet to do, onto the next stage """ oldcount = self._goal.req_length() if len(userlist) == 0: self.distro_sync() else: for pkg_spec in userlist: self.distro_sync(pkg_spec) cnt = self._goal.req_length() - oldcount if cnt <= 0 and not self._goal.req_has_distupgrade_all(): msg = _('No packages marked for distribution synchronization.') raise dnf.exceptions.Error(msg) def downgradePkgs(self, specs=[], file_pkgs=[], strict=False): """Attempt to take the user specified list of packages or wildcards and downgrade them. If a complete version number is specified, attempt to downgrade them to the specified version :param specs: a list of names or wildcards specifying packages to downgrade :param file_pkgs: a list of pkg objects from local files """ result = False for pkg in file_pkgs: try: self.package_downgrade(pkg, strict=strict) result = True except dnf.exceptions.MarkingError as e: logger.info(_('No match for argument: %s'), self.output.term.bold(pkg.location)) for arg in specs: try: self.downgrade_to(arg, strict=strict) result = True except dnf.exceptions.PackageNotFoundError as err: msg = _('No package %s available.') logger.info(msg, self.output.term.bold(arg)) except dnf.exceptions.PackagesNotInstalledError as err: logger.info(_('Packages for argument %s available, but not installed.'), self.output.term.bold(err.pkg_spec)) except dnf.exceptions.MarkingError: assert False if not result: raise dnf.exceptions.Error(_('No packages marked for downgrade.')) def output_packages(self, basecmd, pkgnarrow='all', patterns=(), reponame=None): """Output selection pkgnarrow of packages matching patterns and repoid.""" try: highlight = self.output.term.MODE['bold'] ypl = self.returnPkgLists( pkgnarrow, patterns, installed_available=highlight, reponame=reponame) except dnf.exceptions.Error as e: return 1, [str(e)] else: update_pkgs = {} inst_pkgs = {} local_pkgs = {} columns = None if basecmd == 'list': # Dynamically size the columns columns = _list_cmd_calc_columns(self.output, ypl) if highlight and ypl.installed: # If we have installed and available lists, then do the # highlighting for the installed packages so you can see what's # available to update, an extra, or newer than what we have. for pkg in (ypl.hidden_available + ypl.reinstall_available + ypl.old_available): key = (pkg.name, pkg.arch) if key not in update_pkgs or pkg > update_pkgs[key]: update_pkgs[key] = pkg if highlight and ypl.available: # If we have installed and available lists, then do the # highlighting for the available packages so you can see what's # available to install vs. update vs. old. for pkg in ypl.hidden_installed: key = (pkg.name, pkg.arch) if key not in inst_pkgs or pkg > inst_pkgs[key]: inst_pkgs[key] = pkg if highlight and ypl.updates: # Do the local/remote split we get in "yum updates" for po in sorted(ypl.updates): if po.reponame != hawkey.SYSTEM_REPO_NAME: local_pkgs[(po.name, po.arch)] = po # Output the packages: clio = self.conf.color_list_installed_older clin = self.conf.color_list_installed_newer clir = self.conf.color_list_installed_reinstall clie = self.conf.color_list_installed_extra rip = self.output.listPkgs(ypl.installed, _('Installed Packages'), basecmd, highlight_na=update_pkgs, columns=columns, highlight_modes={'>' : clio, '<' : clin, '=' : clir, 'not in' : clie}) clau = self.conf.color_list_available_upgrade clad = self.conf.color_list_available_downgrade clar = self.conf.color_list_available_reinstall clai = self.conf.color_list_available_install rap = self.output.listPkgs(ypl.available, _('Available Packages'), basecmd, highlight_na=inst_pkgs, columns=columns, highlight_modes={'<' : clau, '>' : clad, '=' : clar, 'not in' : clai}) raep = self.output.listPkgs(ypl.autoremove, _('Autoremove Packages'), basecmd, columns=columns) rep = self.output.listPkgs(ypl.extras, _('Extra Packages'), basecmd, columns=columns) cul = self.conf.color_update_local cur = self.conf.color_update_remote rup = self.output.listPkgs(ypl.updates, _('Available Upgrades'), basecmd, highlight_na=local_pkgs, columns=columns, highlight_modes={'=' : cul, 'not in' : cur}) # XXX put this into the ListCommand at some point if len(ypl.obsoletes) > 0 and basecmd == 'list': # if we've looked up obsolete lists and it's a list request rop = len(ypl.obsoletes) print(_('Obsoleting Packages')) for obtup in sorted(ypl.obsoletesTuples, key=operator.itemgetter(0)): self.output.updatesObsoletesList(obtup, 'obsoletes', columns=columns) else: rop = self.output.listPkgs(ypl.obsoletes, _('Obsoleting Packages'), basecmd, columns=columns) rrap = self.output.listPkgs(ypl.recent, _('Recently Added Packages'), basecmd, columns=columns) if len(patterns) and \ rrap == 0 and rop == 0 and rup == 0 and rep == 0 and rap == 0 and raep == 0 and rip == 0: raise dnf.exceptions.Error(_('No matching Packages to list')) def returnPkgLists(self, pkgnarrow='all', patterns=None, installed_available=False, reponame=None): """Return a :class:`dnf.yum.misc.GenericHolder` object containing lists of package objects that match the given names or wildcards. :param pkgnarrow: a string specifying which types of packages lists to produce, such as updates, installed, available, etc. :param patterns: a list of names or wildcards specifying packages to list :param installed_available: whether the available package list is present as .hidden_available when doing all, available, or installed :param reponame: limit packages list to the given repository :return: a :class:`dnf.yum.misc.GenericHolder` instance with the following lists defined:: available = list of packageObjects installed = list of packageObjects upgrades = tuples of packageObjects (updating, installed) extras = list of packageObjects obsoletes = tuples of packageObjects (obsoleting, installed) recent = list of packageObjects """ done_hidden_available = False done_hidden_installed = False if installed_available and pkgnarrow == 'installed': done_hidden_available = True pkgnarrow = 'all' elif installed_available and pkgnarrow == 'available': done_hidden_installed = True pkgnarrow = 'all' ypl = self._do_package_lists( pkgnarrow, patterns, ignore_case=True, reponame=reponame) if self.conf.showdupesfromrepos: for pkg in ypl.reinstall_available: if not pkg.installed and not done_hidden_available: ypl.available.append(pkg) if installed_available: ypl.hidden_available = ypl.available ypl.hidden_installed = ypl.installed if done_hidden_available: ypl.available = [] if done_hidden_installed: ypl.installed = [] return ypl def provides(self, args): """Print out a list of packages that provide the given file or feature. This a cli wrapper to the provides methods in the rpmdb and pkgsack. :param args: the name of a file or feature to search for :return: (exit_code, [ errors ]) exit_code is:: 0 = we're done, exit 1 = we've errored, exit with error string 2 = we've got work yet to do, onto the next stage """ # always in showdups mode old_sdup = self.conf.showdupesfromrepos self.conf.showdupesfromrepos = True matches = [] used_search_strings = [] for spec in args: query, used_search_string = super(BaseCli, self).provides(spec) matches.extend(query) used_search_strings.extend(used_search_string) for pkg in sorted(matches): self.output.matchcallback_verbose(pkg, used_search_strings, args) self.conf.showdupesfromrepos = old_sdup if not matches: raise dnf.exceptions.Error(_('No matches found. If searching for a file, ' 'try specifying the full path or using a ' 'wildcard prefix ("/") at the beginning.')) def _promptWanted(self): # shortcut for the always-off/always-on options if self.conf.assumeyes and not self.conf.assumeno: return False return True class Cli(object): def __init__(self, base): self.base = base self.cli_commands = {} self.command = None self.demands = dnf.cli.demand.DemandSheet() # :api self.register_command(dnf.cli.commands.alias.AliasCommand) self.register_command(dnf.cli.commands.autoremove.AutoremoveCommand) self.register_command(dnf.cli.commands.check.CheckCommand) self.register_command(dnf.cli.commands.clean.CleanCommand) self.register_command(dnf.cli.commands.distrosync.DistroSyncCommand) self.register_command(dnf.cli.commands.deplist.DeplistCommand) self.register_command(dnf.cli.commands.downgrade.DowngradeCommand) self.register_command(dnf.cli.commands.group.GroupCommand) self.register_command(dnf.cli.commands.history.HistoryCommand) self.register_command(dnf.cli.commands.install.InstallCommand) self.register_command(dnf.cli.commands.makecache.MakeCacheCommand) self.register_command(dnf.cli.commands.mark.MarkCommand) self.register_command(dnf.cli.commands.module.ModuleCommand) self.register_command(dnf.cli.commands.reinstall.ReinstallCommand) self.register_command(dnf.cli.commands.remove.RemoveCommand) self.register_command(dnf.cli.commands.repolist.RepoListCommand) self.register_command(dnf.cli.commands.repoquery.RepoQueryCommand) self.register_command(dnf.cli.commands.search.SearchCommand) self.register_command(dnf.cli.commands.shell.ShellCommand) self.register_command(dnf.cli.commands.swap.SwapCommand) self.register_command(dnf.cli.commands.updateinfo.UpdateInfoCommand) self.register_command(dnf.cli.commands.upgrade.UpgradeCommand) self.register_command(dnf.cli.commands.upgrademinimal.UpgradeMinimalCommand) self.register_command(dnf.cli.commands.InfoCommand) self.register_command(dnf.cli.commands.ListCommand) self.register_command(dnf.cli.commands.ProvidesCommand) self.register_command(dnf.cli.commands.CheckUpdateCommand) self.register_command(dnf.cli.commands.RepoPkgsCommand) self.register_command(dnf.cli.commands.HelpCommand) def _configure_repos(self, opts): self.base.read_all_repos(opts) if opts.repofrompath: for label, path in opts.repofrompath.items(): this_repo = self.base.repos.add_new_repo(label, self.base.conf, baseurl=[path]) this_repo._configure_from_options(opts) # do not let this repo to be disabled opts.repos_ed.append((label, "enable")) if opts.repo: opts.repos_ed.insert(0, ("", "disable")) opts.repos_ed.extend([(r, "enable") for r in opts.repo]) notmatch = set() # Process repo enables and disables in order try: for (repo, operation) in opts.repos_ed: repolist = self.base.repos.get_matching(repo) if not repolist: if self.base.conf.strict and operation == "enable": msg = _("Unknown repo: '%s'") raise dnf.exceptions.RepoError(msg % repo) notmatch.add(repo) if operation == "enable": repolist.enable() else: repolist.disable() except dnf.exceptions.ConfigError as e: logger.critical(e) self.optparser.print_help() sys.exit(1) for repo in notmatch: logger.warning(_("No repository match: %s"), repo) expired_repos = self.base._repo_persistor.get_expired_repos() if expired_repos is None: expired_repos = self.base.repos.keys() for rid in expired_repos: repo = self.base.repos.get(rid) if repo: repo._repo.expire() # setup the progress bars/callbacks (bar, self.base._ds_callback) = self.base.output.setup_progress_callbacks() self.base.repos.all().set_progress_bar(bar) key_import = output.CliKeyImport(self.base, self.base.output) self.base.repos.all()._set_key_import(key_import) def _log_essentials(self): logger.debug('{prog} version: %s'.format(prog=dnf.util.MAIN_PROG_UPPER), dnf.const.VERSION) logger.log(dnf.logging.DDEBUG, 'Command: %s', self.cmdstring) logger.log(dnf.logging.DDEBUG, 'Installroot: %s', self.base.conf.installroot) logger.log(dnf.logging.DDEBUG, 'Releasever: %s', self.base.conf.releasever) logger.debug("cachedir: %s", self.base.conf.cachedir) def _process_demands(self): demands = self.demands repos = self.base.repos if demands.root_user: if not dnf.util.am_i_root(): raise dnf.exceptions.Error( _('This command has to be run with superuser privileges ' '(under the root user on most systems).')) if demands.changelogs: for repo in repos.iter_enabled(): repo.load_metadata_other = True if demands.cacheonly or self.base.conf.cacheonly: self.base.conf.cacheonly = True for repo in repos.values(): repo._repo.setSyncStrategy(dnf.repo.SYNC_ONLY_CACHE) else: if demands.freshest_metadata: for repo in repos.iter_enabled(): repo._repo.expire() elif not demands.fresh_metadata: for repo in repos.values(): repo._repo.setSyncStrategy(dnf.repo.SYNC_LAZY) if demands.sack_activation: self.base.fill_sack( load_system_repo='auto' if self.demands.load_system_repo else False, load_available_repos=self.demands.available_repos) def _parse_commands(self, opts, args): """Check that the requested CLI command exists.""" basecmd = opts.command command_cls = self.cli_commands.get(basecmd) if command_cls is None: logger.critical(_('No such command: %s. Please use %s --help'), basecmd, sys.argv[0]) if self.base.conf.plugins: logger.critical(_("It could be a {PROG} plugin command, " "try: \"{prog} install 'dnf-command(%s)'\"").format( prog=dnf.util.MAIN_PROG, PROG=dnf.util.MAIN_PROG_UPPER), basecmd) else: logger.critical(_("It could be a {prog} plugin command, " "but loading of plugins is currently disabled.").format( prog=dnf.util.MAIN_PROG_UPPER)) raise CliError self.command = command_cls(self) logger.log(dnf.logging.DDEBUG, 'Base command: %s', basecmd) logger.log(dnf.logging.DDEBUG, 'Extra commands: %s', args) def configure(self, args, option_parser=None): """Parse command line arguments, and set up :attr:`self.base.conf` and :attr:`self.cmds`, as well as logger objects in base instance. :param args: a list of command line arguments :param option_parser: a class for parsing cli options """ aliases = dnf.cli.aliases.Aliases() args = aliases.resolve(args) self.optparser = dnf.cli.option_parser.OptionParser() \ if option_parser is None else option_parser opts = self.optparser.parse_main_args(args) # Just print out the version if that's what the user wanted if opts.version: print(dnf.const.VERSION) print_versions(self.base.conf.history_record_packages, self.base, self.base.output) sys.exit(0) if opts.quiet: opts.debuglevel = 0 opts.errorlevel = 2 if opts.verbose: opts.debuglevel = opts.errorlevel = dnf.const.VERBOSE_LEVEL # Read up configuration options and initialize plugins try: if opts.cacheonly: self.base.conf._set_value("cachedir", self.base.conf.system_cachedir, dnf.conf.PRIO_DEFAULT) self.demands.cacheonly = True self.base.conf._configure_from_options(opts) self._read_conf_file(opts.releasever, opts.releasever_major, opts.releasever_minor) if 'arch' in opts: self.base.conf.arch = opts.arch self.base.conf._adjust_conf_options() except (dnf.exceptions.ConfigError, ValueError) as e: logger.critical(_('Config error: %s'), e) sys.exit(1) except IOError as e: e = '%s: %s' % (ucd(str(e)), repr(e.filename)) logger.critical(_('Config error: %s'), e) sys.exit(1) if opts.destdir is not None: self.base.conf.destdir = opts.destdir if not self.base.conf.downloadonly and opts.command not in ( 'download', 'system-upgrade', 'reposync', 'modulesync'): logger.critical(_('--destdir or --downloaddir must be used with --downloadonly ' 'or download or system-upgrade command.') ) sys.exit(1) if (opts.set_enabled or opts.set_disabled) and opts.command != 'config-manager': logger.critical( _('--enable, --set-enabled and --disable, --set-disabled ' 'must be used with config-manager command.')) sys.exit(1) if opts.sleeptime is not None: time.sleep(random.randrange(opts.sleeptime * 60)) # store the main commands & summaries, before plugins are loaded self.optparser.add_commands(self.cli_commands, 'main') # store the plugin commands & summaries self.base.init_plugins(opts.disableplugin, opts.enableplugin, self) self.optparser.add_commands(self.cli_commands,'plugin') # show help if no command specified # this is done here, because we first have the full # usage info after the plugins are loaded. if not opts.command: self.optparser.print_help() sys.exit(0) # save our original args out self.base.args = args # save out as a nice command string self.cmdstring = self.optparser.prog + ' ' for arg in self.base.args: self.cmdstring += '%s ' % arg self._log_essentials() try: self._parse_commands(opts, args) except CliError: sys.exit(1) # show help for dnf <command> --help / --help-cmd if opts.help: self.optparser.print_help(self.command) sys.exit(0) opts = self.optparser.parse_command_args(self.command, args) if opts.allowerasing: self.demands.allow_erasing = opts.allowerasing self.base._allow_erasing = True if opts.freshest_metadata: self.demands.freshest_metadata = opts.freshest_metadata if opts.debugsolver: self.base.conf.debug_solver = True if opts.obsoletes: self.base.conf.obsoletes = True self.command.pre_configure() self.base.pre_configure_plugins() # with cachedir in place we can configure stuff depending on it: self.base._activate_persistor() self._configure_repos(opts) self.base.configure_plugins() self.base.conf._configure_from_options(opts) self.command.configure() if self.base.conf.destdir: dnf.util.ensure_dir(self.base.conf.destdir) self.base.repos.all().pkgdir = self.base.conf.destdir if self.base.conf.color != 'auto': self.base.output.term.reinit(color=self.base.conf.color) if rpm.expandMacro('%_pkgverify_level') in ('signature', 'all'): forcing = False for repo in self.base.repos.iter_enabled(): if repo.gpgcheck: continue repo.gpgcheck = True forcing = True if not self.base.conf.localpkg_gpgcheck: self.base.conf.localpkg_gpgcheck = True forcing = True if forcing: logger.warning( _("Warning: Enforcing GPG signature check globally " "as per active RPM security policy (see 'gpgcheck' in " "dnf.conf(5) for how to squelch this message)" ) ) def _read_conf_file(self, releasever=None, releasever_major=None, releasever_minor=None): timer = dnf.logging.Timer('config') conf = self.base.conf # replace remote config path with downloaded file conf._check_remote_file('config_file_path') # search config file inside the installroot first conf._search_inside_installroot('config_file_path') # check whether a config file is requested from command line and the file exists filename = conf._get_value('config_file_path') if (conf._get_priority('config_file_path') == dnf.conf.PRIO_COMMANDLINE) and \ not os.path.isfile(filename): raise dnf.exceptions.ConfigError(_('Config file "{}" does not exist').format(filename)) # read config conf.read(priority=dnf.conf.PRIO_MAINCONFIG) # search reposdir file inside the installroot first from_root = conf._search_inside_installroot('reposdir') # Update vars from same root like repos were taken if conf._get_priority('varsdir') == dnf.conf.PRIO_COMMANDLINE: from_root = "/" subst = conf.substitutions subst.update_from_etc(from_root, varsdir=conf._get_value('varsdir')) # cachedir, logs, releasever, and gpgkey are taken from or stored in installroot det_major = None det_minor = None if releasever is None and conf.releasever is None: releasever, det_major, det_minor = dnf.rpm.detect_releasevers(conf.installroot) elif releasever == '/': releasever, det_major, det_minor = dnf.rpm.detect_releasevers(releasever) def or_else(*args): for arg in args: if arg is not None: return arg return None conf.releasever = or_else(releasever, conf.releasever) conf.releasever_major = or_else(releasever_major, det_major, conf.releasever_major) conf.releasever_minor = or_else(releasever_minor, det_minor, conf.releasever_minor) if conf.releasever is None: logger.warning(_("Unable to detect release version (use '--releasever' to specify " "release version)")) for opt in ('cachedir', 'logdir', 'persistdir'): conf.prepend_installroot(opt) self.base._logging._setup_from_dnf_conf(conf) timer() return conf def _populate_update_security_filter(self, opts, cmp_type='eq', all=None): """ :param opts: :param cmp_type: string supported "eq", "gte" :param all: :return: """ if (opts is None) and (all is None): return types = [] if opts.bugfix or all: types.append('bugfix') if opts.enhancement or all: types.append('enhancement') if opts.newpackage or all: types.append('newpackage') if opts.security or all: types.append('security') self.base.add_security_filters(cmp_type, types=types, advisory=opts.advisory, bugzilla=opts.bugzilla, cves=opts.cves, severity=opts.severity) def redirect_logger(self, stdout=None, stderr=None): # :api """ Change minimal logger level for terminal output to stdout and stderr according to specific command requirements @param stdout: logging.INFO, logging.WARNING, ... @param stderr:logging.INFO, logging.WARNING, ... """ if stdout is not None: self.base._logging.stdout_handler.setLevel(stdout) if stderr is not None: self.base._logging.stderr_handler.setLevel(stderr) def redirect_repo_progress(self, fo=sys.stderr): progress = dnf.cli.progress.MultiFileProgressMeter(fo) self.base.output.progress = progress self.base.repos.all().set_progress_bar(progress) def _check_running_kernel(self): kernel = self.base.sack.get_running_kernel() if kernel is None: return q = self.base.sack.query().filterm(provides=kernel.name) q = q.installed() q.filterm(advisory_type='security') ikpkg = kernel for pkg in q: if pkg > ikpkg: ikpkg = pkg if ikpkg > kernel: print('Security: %s is an installed security update' % ikpkg) print('Security: %s is the currently running version' % kernel) def _option_conflict(self, option_string_1, option_string_2): print(self.optparser.print_usage()) raise dnf.exceptions.Error(_("argument {}: not allowed with argument {}".format( option_string_1, option_string_2))) def register_command(self, command_cls): """Register a Command. :api""" for name in command_cls.aliases: if name in self.cli_commands: raise dnf.exceptions.ConfigError(_('Command "%s" already defined') % name) self.cli_commands[name] = command_cls def run(self): """Call the base command, and pass it the extended commands or arguments. :return: (exit_code, [ errors ]) exit_code is:: 0 = we're done, exit 1 = we've errored, exit with error string 2 = we've got work yet to do, onto the next stage """ self._process_demands() # Reports about excludes and includes (but not from plugins) if self.base.conf.excludepkgs: logger.debug( _('Excludes in dnf.conf: ') + ", ".join(sorted(set(self.base.conf.excludepkgs)))) if self.base.conf.includepkgs: logger.debug( _('Includes in dnf.conf: ') + ", ".join(sorted(set(self.base.conf.includepkgs)))) for repo in self.base.repos.iter_enabled(): if repo.excludepkgs: logger.debug(_('Excludes in repo ') + repo.id + ": " + ", ".join(sorted(set(repo.excludepkgs)))) if repo.includepkgs: logger.debug(_('Includes in repo ') + repo.id + ": " + ", ".join(sorted(set(repo.includepkgs)))) return self.command.run()	47.33 KB	2025-09-22 11:27
📁 commands	-	2026-03-21 15:47
📄 completion_helper.py #!/usr/bin/python3.9 # # This file is part of dnf. # # Copyright 2015 (C) Igor Gnatenko <i.gnatenko.brain@gmail.com> # Copyright 2016 (C) Red Hat, Inc. # # 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 dnf.exceptions import dnf.cli import dnf.cli.commands.clean import sys def filter_list_by_kw(kw, lst): return filter(lambda k: str(k).startswith(kw), lst) def listpkg_to_setstr(pkgs): return set([str(x) for x in pkgs]) class RemoveCompletionCommand(dnf.cli.commands.remove.RemoveCommand): def __init__(self, args): super(RemoveCompletionCommand, self).__init__(args) def configure(self): self.cli.demands.root_user = False self.cli.demands.sack_activation = True def run(self): for pkg in ListCompletionCommand.installed(self.base, self.opts.pkg_specs): print(str(pkg)) class InstallCompletionCommand(dnf.cli.commands.install.InstallCommand): def __init__(self, args): super(InstallCompletionCommand, self).__init__(args) def configure(self): self.cli.demands.root_user = False self.cli.demands.available_repos = True self.cli.demands.sack_activation = True def run(self): installed = listpkg_to_setstr(ListCompletionCommand.installed(self.base, self.opts.pkg_specs)) available = listpkg_to_setstr(ListCompletionCommand.available(self.base, self.opts.pkg_specs)) for pkg in (available - installed): print(str(pkg)) class ReinstallCompletionCommand(dnf.cli.commands.reinstall.ReinstallCommand): def __init__(self, args): super(ReinstallCompletionCommand, self).__init__(args) def configure(self): self.cli.demands.root_user = False self.cli.demands.available_repos = True self.cli.demands.sack_activation = True def run(self): installed = listpkg_to_setstr(ListCompletionCommand.installed(self.base, self.opts.pkg_specs)) available = listpkg_to_setstr(ListCompletionCommand.available(self.base, self.opts.pkg_specs)) for pkg in (installed & available): print(str(pkg)) class ListCompletionCommand(dnf.cli.commands.ListCommand): def __init__(self, args): super(ListCompletionCommand, self).__init__(args) def run(self): subcmds = self.pkgnarrows args = self.opts.packages action = self.opts.packages_action if len(args) > 1 and args[1] not in subcmds: print("\n".join(filter_list_by_kw(args[1], subcmds))) else: if action == "installed": pkgs = self.installed(self.base, args) elif action == "available": pkgs = self.available(self.base, args) elif action == "updates": pkgs = self.updates(self.base, args) else: available = listpkg_to_setstr(self.available(self.base, args)) installed = listpkg_to_setstr(self.installed(self.base, args)) pkgs = (available \| installed) if not pkgs: print("\n".join(filter_list_by_kw(args[0], subcmds))) return for pkg in pkgs: print(str(pkg)) @staticmethod def installed(base, arg): return base.sack.query().installed().filterm(name__glob="{}".format(arg[0])) @staticmethod def available(base, arg): return base.sack.query().available().filterm(name__glob="{}".format(arg[0])) @staticmethod def updates(base, arg): return base.check_updates(["{}*".format(arg[0])], print_=False) class RepoListCompletionCommand(dnf.cli.commands.repolist.RepoListCommand): def __init__(self, args): super(RepoListCompletionCommand, self).__init__(args) def run(self): args = self.opts if args.repos_action == "enabled": print("\n".join(filter_list_by_kw(args.repos[0], [r.id for r in self.base.repos.iter_enabled()]))) elif args.repos_action == "disabled": print("\n".join(filter_list_by_kw(args.repos[0], [r.id for r in self.base.repos.all() if not r.enabled]))) elif args.repos_action == "all": print("\n".join(filter_list_by_kw(args.repos[0], [r.id for r in self.base.repos.all()]))) class UpgradeCompletionCommand(dnf.cli.commands.upgrade.UpgradeCommand): def __init__(self, args): super(UpgradeCompletionCommand, self).__init__(args) def configure(self): self.cli.demands.root_user = False self.cli.demands.available_repos = True self.cli.demands.sack_activation = True def run(self): for pkg in ListCompletionCommand.updates(self.base, self.opts.pkg_specs): print(str(pkg)) class DowngradeCompletionCommand(dnf.cli.commands.downgrade.DowngradeCommand): def __init__(self, args): super(DowngradeCompletionCommand, self).__init__(args) def configure(self): self.cli.demands.root_user = False self.cli.demands.available_repos = True self.cli.demands.sack_activation = True def run(self): for pkg in ListCompletionCommand.available(self.base, self.opts.pkg_specs).downgrades(): print(str(pkg)) class CleanCompletionCommand(dnf.cli.commands.clean.CleanCommand): def __init__(self, args): super(CleanCompletionCommand, self).__init__(args) def run(self): subcmds = dnf.cli.commands.clean._CACHE_TYPES.keys() print("\n".join(filter_list_by_kw(self.opts.type[1], subcmds))) def main(args): base = dnf.cli.cli.BaseCli() cli = dnf.cli.Cli(base) if args[0] == "_cmds": base.init_plugins([], [], cli) print("\n".join(filter_list_by_kw(args[1], cli.cli_commands))) return cli.cli_commands.clear() cli.register_command(RemoveCompletionCommand) cli.register_command(InstallCompletionCommand) cli.register_command(ReinstallCompletionCommand) cli.register_command(ListCompletionCommand) cli.register_command(RepoListCompletionCommand) cli.register_command(UpgradeCompletionCommand) cli.register_command(DowngradeCompletionCommand) cli.register_command(CleanCompletionCommand) cli.configure(args) try: cli.run() except (OSError, dnf.exceptions.Error): sys.exit(0) if __name__ == "__main__": try: main(sys.argv[1:]) except KeyboardInterrupt: sys.exit(1)	7.29 KB	2025-09-22 11:27
📄 demand.py # demand.py # Demand sheet and related classes. # # Copyright (C) 2014-2015 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import unicode_literals class _BoolDefault(object): def __init__(self, default): self.default = default self._storing_name = '__%s%x' % (self.__class__.__name__, id(self)) def __get__(self, obj, objtype=None): objdict = obj.__dict__ if self._storing_name in objdict: return objdict[self._storing_name] return self.default def __set__(self, obj, val): objdict = obj.__dict__ if self._storing_name in objdict: current_val = objdict[self._storing_name] if current_val != val: raise AttributeError('Demand already set.') objdict[self._storing_name] = val class DemandSheet(object): """Collection of demands that different CLI parts have on other parts. :api""" # :api... allow_erasing = _BoolDefault(False) available_repos = _BoolDefault(False) resolving = _BoolDefault(False) root_user = _BoolDefault(False) sack_activation = _BoolDefault(False) load_system_repo = _BoolDefault(True) success_exit_status = 0 cacheonly = _BoolDefault(False) fresh_metadata = _BoolDefault(True) freshest_metadata = _BoolDefault(False) changelogs = _BoolDefault(False) transaction_display = None # This demand controlls applicability of the plugins that could filter # repositories packages (e.g. versionlock). # If it stays None, the demands.resolving is used as a fallback. plugin_filtering_enabled = _BoolDefault(None)	2.48 KB	2022-09-09 11:21
📄 format.py # Copyright (C) 2013-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. from __future__ import unicode_literals from dnf.pycomp import long def format_number(number, SI=0, space=' '): """Return a human-readable metric-like string representation of a number. :param number: the number to be converted to a human-readable form :param SI: If is 0, this function will use the convention that 1 kilobyte = 1024 bytes, otherwise, the convention that 1 kilobyte = 1000 bytes will be used :param space: string that will be placed between the number and the SI prefix :return: a human-readable metric-like string representation of number """ # copied from from urlgrabber.progress symbols = [ ' ', # (none) 'k', # kilo 'M', # mega 'G', # giga 'T', # tera 'P', # peta 'E', # exa 'Z', # zetta 'Y'] # yotta if SI: step = 1000.0 else: step = 1024.0 thresh = 999 depth = 0 max_depth = len(symbols) - 1 if number is None: number = 0.0 # we want numbers between 0 and thresh, but don't exceed the length # of our list. In that event, the formatting will be screwed up, # but it'll still show the right number. while number > thresh and depth < max_depth: depth = depth + 1 number = number / step if isinstance(number, int) or isinstance(number, long): format = '%i%s%s' elif number < 9.95: # must use 9.95 for proper sizing. For example, 9.99 will be # rounded to 10.0 with the .1f format string (which is too long) format = '%.1f%s%s' else: format = '%.0f%s%s' return(format % (float(number or 0), space, symbols[depth])) def format_time(seconds, use_hours=0): """Return a human-readable string representation of a number of seconds. The string will show seconds, minutes, and optionally hours. :param seconds: the number of seconds to convert to a human-readable form :param use_hours: If use_hours is 0, the representation will be in minutes and seconds. Otherwise, it will be in hours, minutes, and seconds :return: a human-readable string representation of seconds """ # copied from from urlgrabber.progress if seconds is None or seconds < 0: if use_hours: return '--:--:--' else: return '--:--' elif seconds == float('inf'): return 'Infinite' else: seconds = int(seconds) minutes = seconds // 60 seconds = seconds % 60 if use_hours: hours = minutes // 60 minutes = minutes % 60 return '%02i:%02i:%02i' % (hours, minutes, seconds) else: return '%02i:%02i' % (minutes, seconds) def indent_block(s): return '\n'.join(' ' + s for s in s.splitlines())	3.76 KB	2022-09-09 11:21
📄 main.py # Copyright 2005 Duke University # Copyright (C) 2012-2016 Red Hat, Inc. # # 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 Library 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. """ Entrance point for the yum command line interface. """ from __future__ import print_function from __future__ import absolute_import from __future__ import unicode_literals from dnf.conf import Conf from dnf.cli.cli import Cli from dnf.cli.option_parser import OptionParser from dnf.i18n import ucd from dnf.cli.utils import show_lock_owner from dnf.i18n import _ import dnf.cli import dnf.cli.cli import dnf.cli.option_parser import dnf.exceptions import dnf.i18n import dnf.logging import dnf.util import errno import hawkey import libdnf.error import logging import os import os.path import sys logger = logging.getLogger("dnf") def ex_IOError(e): logger.log(dnf.logging.SUBDEBUG, '', exc_info=True) logger.critical(ucd(e)) return 1 def ex_Error(e): logger.log(dnf.logging.SUBDEBUG, '', exc_info=True) if e.value is not None: logger.critical(_('Error: %s'), ucd(e)) return 1 def main(args, conf_class=Conf, cli_class=Cli, option_parser_class=OptionParser): try: dnf.i18n.setup_stdout() with dnf.cli.cli.BaseCli(conf_class()) as base: return _main(base, args, cli_class, option_parser_class) except dnf.exceptions.ProcessLockError as e: logger.critical(e.value) show_lock_owner(e.pid) return 200 except dnf.exceptions.LockError as e: logger.critical(e.value) return 200 except dnf.exceptions.DepsolveError as e: return 1 except dnf.exceptions.Error as e: return ex_Error(e) except hawkey.Exception as e: logger.critical(_('Error: %s'), ucd(e)) return 1 except libdnf.error.Error as e: logger.critical(_('Error: %s'), ucd(e)) return 1 except IOError as e: return ex_IOError(e) except KeyboardInterrupt as e: logger.critical('{}: {}'.format(type(e).__name__, _("Terminated."))) return 1 def _main(base, args, cli_class, option_parser): """Run the dnf program from a command line interface.""" # our core object for the cli base._logging._presetup() cli = cli_class(base) # do our cli parsing and config file setup # also sanity check the things being passed on the cli try: cli.configure(list(map(ucd, args)), option_parser()) except (IOError, OSError) as e: return ex_IOError(e) return cli_run(cli, base) def cli_run(cli, base): # Try to open the current directory to see if we have # read and execute access. If not, chdir to / try: f = open(".") except IOError as e: if e.errno == errno.EACCES: logger.critical(_('No read/execute access in current directory, moving to /')) os.chdir("/") else: f.close() try: cli.run() except dnf.exceptions.LockError: raise except (IOError, OSError) as e: return ex_IOError(e) if cli.demands.resolving: try: ret = resolving(cli, base) except dnf.exceptions.DepsolveError as e: ex_Error(e) msg = "" if not cli.demands.allow_erasing and base._goal.problem_conflicts(available=True): msg += _("try to add '{}' to command line to replace conflicting " "packages").format("--allowerasing") if cli.base.conf.strict: if not msg: msg += _("try to add '{}' to skip uninstallable packages").format( "--skip-broken") else: msg += _(" or '{}' to skip uninstallable packages").format("--skip-broken") if cli.base.conf.best: prio = cli.base.conf._get_priority("best") if prio <= dnf.conf.PRIO_MAINCONFIG: if not msg: msg += _("try to add '{}' to use not only best candidate packages").format( "--nobest") else: msg += _(" or '{}' to use not only best candidate packages").format( "--nobest") if msg: logger.info("({})".format(msg)) raise if ret: return ret cli.command.run_transaction() return cli.demands.success_exit_status def resolving(cli, base): """Perform the depsolve, download and RPM transaction stage.""" if base.transaction is None: base.resolve(cli.demands.allow_erasing) logger.info(_('Dependencies resolved.')) cli.command.run_resolved() # Run the transaction displays = [] if cli.demands.transaction_display is not None: displays.append(cli.demands.transaction_display) try: base.do_transaction(display=displays) except dnf.cli.CliError as exc: logger.error(ucd(exc)) return 1 except dnf.exceptions.TransactionCheckError as err: for msg in cli.command.get_error_output(err): logger.critical(msg) return 1 except IOError as e: return ex_IOError(e) else: logger.info(_('Complete!')) return 0 def user_main(args, exit_code=False): """Call one of the multiple main() functions based on environment variables. :param args: command line arguments passed into yum :param exit_code: if exit_code is True, this function will exit python with its exit code when it has finished executing. Otherwise, it will return its exit code. :return: the exit code from dnf.yum execution """ errcode = main(args) if exit_code: sys.exit(errcode) return errcode if __name__ == "__main__": user_main(sys.argv[1:], exit_code=True)	6.35 KB	2022-09-09 11:21
📄 option_parser.py # optparse.py # CLI options parser. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import unicode_literals from dnf.i18n import _ from dnf.util import _parse_specs import argparse import dnf.exceptions import dnf.util import dnf.rpm import dnf.yum.misc import logging import os.path import re import sys logger = logging.getLogger("dnf") class MultilineHelpFormatter(argparse.HelpFormatter): def _split_lines(self, text, width): if '\n' in text: return text.splitlines() return super(MultilineHelpFormatter, self)._split_lines(text, width) class OptionParser(argparse.ArgumentParser): """ArgumentParser like class to do things the "yum way".""" def __init__(self, reset_usage=True): super(OptionParser, self).__init__(add_help=False, formatter_class=MultilineHelpFormatter) self.command_positional_parser = None self.command_group = None self._add_general_options() if reset_usage: self._cmd_usage = {} # names, summary for dnf commands, to build usage self._cmd_groups = set() # cmd groups added (main, plugin) def error(self, msg): """Output an error message, and exit the program. This method overrides standard argparser's error so that error output goes to the logger. :param msg: the error message to output """ self.print_usage() logger.critical(_("Command line error: %s"), msg) sys.exit(1) class _RepoCallback(argparse.Action): def __call__(self, parser, namespace, values, opt_str): operation = 'disable' if opt_str == '--disablerepo' else 'enable' l = getattr(namespace, self.dest) l.extend((x, operation) for x in re.split(r'\s[,\s]\s', values)) class _RepoCallbackEnable(argparse.Action): def __call__(self, parser, namespace, values, opt_str): namespace.repos_ed.append((values[0], 'enable')) setattr(namespace, 'reponame', values) class _SplitCallback(argparse._AppendAction): """ Split all strings in seq, at "," and whitespace. Returns a new list. """ SPLITTER = r'\s[,\s]\s' def __call__(self, parser, namespace, values, opt_str): first = True for val in re.split(self.SPLITTER, values): if first or val: # Empty values are sometimes used to clear existing content of the option. # Only the first value in the parsed string can be empty. Other empty values # are ignored. super(OptionParser._SplitCallback, self).__call__(parser, namespace, val, opt_str) first = False class _SplitExtendDictCallback(argparse.Action): """ Split string at "," or whitespace to (key, value). Extends dict with {key: value}.""" def __call__(self, parser, namespace, values, opt_str): try: key, val = values.split(',') if not key or not val: raise ValueError except ValueError: msg = _('bad format: %s') % values raise argparse.ArgumentError(self, msg) dct = getattr(namespace, self.dest) dct[key] = val class _SetoptsCallback(argparse.Action): """ Parse setopts arguments and put them into main_<setopts> and repo_<setopts>.""" def __call__(self, parser, namespace, values, opt_str): vals = values.split('=') if len(vals) > 2: logger.warning(_("Setopt argument has multiple values: %s"), values) return if len(vals) < 2: logger.warning(_("Setopt argument has no value: %s"), values) return k, v = vals period = k.rfind('.') if period != -1: repo = k[:period] k = k[period+1:] if hasattr(namespace, 'repo_setopts'): repoopts = namespace.repo_setopts else: repoopts = {} repoopts.setdefault(repo, {}).setdefault(k, []).append(v) setattr(namespace, 'repo_' + self.dest, repoopts) else: if hasattr(namespace, 'main_setopts'): mainopts = namespace.main_setopts else: mainopts = {} mainopts.setdefault(k, []).append(v) setattr(namespace, 'main_' + self.dest, mainopts) class ParseSpecGroupFileCallback(argparse.Action): def __call__(self, parser, namespace, values, opt_str): _parse_specs(namespace, values) class PkgNarrowCallback(argparse.Action): def __init__(self, args, kwargs): self.pkgnarrow = {} try: for k in ['choices', 'default']: self.pkgnarrow[k] = kwargs[k] del kwargs[k] except KeyError as e: raise TypeError("%s() missing mandatory argument %s" % (self.__class__.__name__, e)) kwargs['default'] = [] super(OptionParser.PkgNarrowCallback, self).__init__(args, *kwargs) def __call__(self, parser, namespace, values, opt_str): dest_action = self.dest + '_action' if not values or values[0] not in self.pkgnarrow['choices']: narrow = self.pkgnarrow['default'] else: narrow = values.pop(0) setattr(namespace, dest_action, narrow) setattr(namespace, self.dest, values) class ForceArchAction(argparse.Action): def __call__(self, parser, namespace, values, opt_str): namespace.ignorearch = True namespace.arch = values def _add_general_options(self): """ Standard options known to all dnf subcommands. """ # All defaults need to be a None, so we can always tell whether the user # has set something or whether we are getting a default. general_grp = self.add_argument_group(_('General {prog} options'.format( prog=dnf.util.MAIN_PROG_UPPER))) general_grp.add_argument("-c", "--config", dest="config_file_path", default=None, metavar='CONFIG_FILE', help=_("config file location")) general_grp.add_argument("-q", "--quiet", dest="quiet", action="store_true", default=None, help=_("quiet operation")) general_grp.add_argument("-v", "--verbose", action="store_true", default=None, help=_("verbose operation")) general_grp.add_argument("--version", action="store_true", default=None, help=_("show {prog} version and exit").format( prog=dnf.util.MAIN_PROG_UPPER)) general_grp.add_argument("--installroot", help=_("set install root"), metavar='PATH') general_grp.add_argument("--nodocs", action="store_const", const=['nodocs'], dest='tsflags', help=_("do not install documentations")) general_grp.add_argument("--noplugins", action="store_false", default=None, dest='plugins', help=_("disable all plugins")) general_grp.add_argument("--enableplugin", dest="enableplugin", default=[], action=self._SplitCallback, help=_("enable plugins by name"), metavar='PLUGIN') general_grp.add_argument("--disableplugin", dest="disableplugin", default=[], action=self._SplitCallback, help=_("disable plugins by name"), metavar='PLUGIN') general_grp.add_argument("--releasever", default=None, help=_("override the value of $releasever" " in config and repo files")) general_grp.add_argument("--releasever-major", default=None, help=_("override the value of $releasever_major" " in config and repo files")) general_grp.add_argument("--releasever-minor", default=None, help=_("override the value of $releasever_minor" " in config and repo files")) general_grp.add_argument("--setopt", dest="setopts", default=[], action=self._SetoptsCallback, help=_("set arbitrary config and repo options")) general_grp.add_argument("--skip-broken", dest="skip_broken", action="store_true", default=None, help=_("resolve depsolve problems by skipping packages")) general_grp.add_argument('-h', '--help', '--help-cmd', action="store_true", dest='help', help=_("show command help")) general_grp.add_argument('--allowerasing', action='store_true', default=None, help=_('allow erasing of installed packages to ' 'resolve dependencies')) best_group = general_grp.add_mutually_exclusive_group() best_group.add_argument("-b", "--best", action="store_true", dest='best', default=None, help=_("try the best available package versions in transactions.")) best_group.add_argument("--nobest", action="store_false", dest='best', help=_("do not limit the transaction to the best candidate")) general_grp.add_argument("-C", "--cacheonly", dest="cacheonly", action="store_true", default=None, help=_("run entirely from system cache, " "don't update cache")) general_grp.add_argument("-R", "--randomwait", dest="sleeptime", type=int, default=None, metavar='MINUTES', help=_("maximum command wait time")) general_grp.add_argument("-d", "--debuglevel", dest="debuglevel", metavar='DEBUG_LEVEL', default=None, help=_("debugging output level"), type=int) general_grp.add_argument("--debugsolver", action="store_true", default=None, help=_("dumps detailed solving results into" " files")) general_grp.add_argument("--showduplicates", dest="showdupesfromrepos", action="store_true", default=None, help=_("show duplicates, in repos, " "in list/search commands")) general_grp.add_argument("-e", "--errorlevel", default=None, type=int, help=_("error output level")) general_grp.add_argument("--obsoletes", default=None, dest="obsoletes", action="store_true", help=_("enables {prog}'s obsoletes processing logic " "for upgrade or display capabilities that " "the package obsoletes for info, list and " "repoquery").format(prog=dnf.util.MAIN_PROG)) general_grp.add_argument("--rpmverbosity", default=None, help=_("debugging output level for rpm"), metavar='DEBUG_LEVEL_NAME') general_grp.add_argument("-y", "--assumeyes", action="store_true", default=None, help=_("automatically answer yes" " for all questions")) general_grp.add_argument("--assumeno", action="store_true", default=None, help=_("automatically answer no" " for all questions")) general_grp.add_argument("--enablerepo", action=self._RepoCallback, dest='repos_ed', default=[], metavar='REPO', help=_("Temporarily enable repositories for the purpose " "of the current dnf command. Accepts an id, a " "comma-separated list of ids, or a glob of ids. " "This option can be specified multiple times.")) repo_group = general_grp.add_mutually_exclusive_group() repo_group.add_argument("--disablerepo", action=self._RepoCallback, dest='repos_ed', default=[], metavar='REPO', help=_("Temporarily disable active repositories for the " "purpose of the current dnf command. Accepts an id, " "a comma-separated list of ids, or a glob of ids. " "This option can be specified multiple times, but " "is mutually exclusive with `--repo`.")) repo_group.add_argument('--repo', '--repoid', metavar='REPO', dest='repo', action=self._SplitCallback, default=[], help=_('enable just specific repositories by an id or a glob, ' 'can be specified multiple times')) enable_group = general_grp.add_mutually_exclusive_group() enable_group.add_argument("--enable", default=False, dest="set_enabled", action="store_true", help=_("enable repos with config-manager " "command (automatically saves)")) enable_group.add_argument("--disable", default=False, dest="set_disabled", action="store_true", help=_("disable repos with config-manager " "command (automatically saves)")) general_grp.add_argument("-x", "--exclude", "--excludepkgs", default=[], dest='excludepkgs', action=self._SplitCallback, help=_("exclude packages by name or glob"), metavar='PACKAGE') general_grp.add_argument("--disableexcludes", "--disableexcludepkgs", default=[], dest="disable_excludes", action=self._SplitCallback, help=_("disable excludepkgs"), metavar='{all, main, REPOID}') general_grp.add_argument("--repofrompath", default={}, action=self._SplitExtendDictCallback, metavar='REPO,PATH', help=_("label and path to an additional repository to use (same " "path as in a baseurl), can be specified multiple times.")) general_grp.add_argument("--noautoremove", action="store_false", default=None, dest='clean_requirements_on_remove', help=_("disable removal of dependencies that are no longer used")) general_grp.add_argument("--nogpgcheck", action="store_false", default=None, dest='gpgcheck', help=_("disable gpg signature checking (if RPM policy allows)")) general_grp.add_argument("--color", dest="color", default=None, help=_("control whether color is used")) general_grp.add_argument("--refresh", dest="freshest_metadata", action="store_true", help=_("set metadata as expired before running" " the command")) general_grp.add_argument("-4", dest="ip_resolve", default=None, help=_("resolve to IPv4 addresses only"), action="store_const", const='ipv4') general_grp.add_argument("-6", dest="ip_resolve", default=None, help=_("resolve to IPv6 addresses only"), action="store_const", const='ipv6') general_grp.add_argument("--destdir", "--downloaddir", dest="destdir", default=None, help=_("set directory to copy packages to")) general_grp.add_argument("--downloadonly", dest="downloadonly", action="store_true", default=False, help=_("only download packages")) general_grp.add_argument("--transient", dest="persistence", action="store_const", const="transient", default=None, help=_("Use a transient overlay which will reset on reboot")) general_grp.add_argument("--comment", dest="comment", default=None, help=_("add a comment to transaction")) # Updateinfo options... general_grp.add_argument("--bugfix", action="store_true", help=_("Include bugfix relevant packages, " "in updates")) general_grp.add_argument("--enhancement", action="store_true", help=_("Include enhancement relevant packages," " in updates")) general_grp.add_argument("--newpackage", action="store_true", help=_("Include newpackage relevant packages," " in updates")) general_grp.add_argument("--security", action="store_true", help=_("Include security relevant packages, " "in updates")) general_grp.add_argument("--advisory", "--advisories", dest="advisory", default=[], action=self._SplitCallback, help=_("Include packages needed to fix the " "given advisory, in updates")) general_grp.add_argument("--bz", "--bzs", default=[], dest="bugzilla", action=self._SplitCallback, help=_( "Include packages needed to fix the given BZ, in updates")) general_grp.add_argument("--cve", "--cves", default=[], dest="cves", action=self._SplitCallback, help=_("Include packages needed to fix the given CVE, in updates")) general_grp.add_argument( "--sec-severity", "--secseverity", choices=['Critical', 'Important', 'Moderate', 'Low'], default=[], dest="severity", action=self._SplitCallback, help=_( "Include security relevant packages matching the severity, " "in updates")) general_grp.add_argument("--forcearch", metavar="ARCH", dest=argparse.SUPPRESS, action=self.ForceArchAction, choices=sorted(dnf.rpm._BASEARCH_MAP.keys()), help=_("Force the use of an architecture")) general_grp.add_argument('command', nargs='?', help=argparse.SUPPRESS) def _add_cmd_usage(self, cmd, group): """ store usage info about a single dnf command.""" summary = dnf.i18n.ucd(cmd.summary) name = dnf.i18n.ucd(cmd.aliases[0]) if not name in self._cmd_usage: self._cmd_usage[name] = (group, summary) self._cmd_groups.add(group) def add_commands(self, cli_cmds, group): """ store name & summary for dnf commands The stored information is used build usage information grouped by build-in & plugin commands. """ for cmd in set(cli_cmds.values()): self._add_cmd_usage(cmd, group) def get_usage(self): """ get the usage information to show the user. """ desc = {'main': _('List of Main Commands:'), 'plugin': _('List of Plugin Commands:')} usage = '%s [options] COMMAND\n' % dnf.util.MAIN_PROG for grp in ['main', 'plugin']: if not grp in self._cmd_groups: # dont add plugin usage, if we dont have plugins continue usage += "\n%s\n\n" % desc[grp] for name in sorted(self._cmd_usage.keys()): group, summary = self._cmd_usage[name] if group == grp: usage += "%-25s %s\n" % (name, summary) return usage def _add_command_options(self, command): self.prog = "%s %s" % (dnf.util.MAIN_PROG, command._basecmd) self.description = command.summary self.command_positional_parser = argparse.ArgumentParser(self.prog, add_help=False) self.command_positional_parser.print_usage = self.print_usage self.command_positional_parser._positionals.title = None self.command_group = self.add_argument_group( '{} command-specific options'.format(command._basecmd.capitalize())) self.command_group.add_argument = self.cmd_add_argument self.command_group._command = command._basecmd command.set_argparser(self.command_group) def cmd_add_argument(self, args, *kwargs): if all([(arg[0] in self.prefix_chars) for arg in args]): return type(self.command_group).add_argument(self.command_group, args, *kwargs) else: return self.command_positional_parser.add_argument(args, **kwargs) def _check_encoding(self, args): for arg in args: try: arg.encode('utf-8') except UnicodeEncodeError as e: raise dnf.exceptions.ConfigError( _("Cannot encode argument '%s': %s") % (arg, str(e))) def parse_main_args(self, args): self._check_encoding(args) namespace, _unused_args = self.parse_known_args(args) return namespace def parse_command_args(self, command, args): self._add_command_options(command) namespace, unused_args = self.parse_known_args(args) namespace = self.command_positional_parser.parse_args(unused_args, namespace) command.opts = namespace return command.opts def print_usage(self, file_=None): if self.command_positional_parser: self._actions += self.command_positional_parser._actions super(OptionParser, self).print_usage(file_) def print_help(self, command=None): # pylint: disable=W0212 if command: if not self.command_group or self.command_group._command != command._basecmd: self._add_command_options(command) self._actions += self.command_positional_parser._actions self._action_groups.append(self.command_positional_parser._positionals) else: self.usage = self.get_usage() super(OptionParser, self).print_help()	24.41 KB	2025-09-22 11:27
📄 output.py # Copyright 2005 Duke University # Copyright (C) 2012-2016 Red Hat, Inc. # # 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 Library 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. """Handle actual output from the cli.""" from __future__ import absolute_import from __future__ import print_function from __future__ import unicode_literals import fnmatch import hawkey import itertools import libdnf.transaction import logging import operator import pwd import re import sys import time from dnf.cli.format import format_number, format_time from dnf.i18n import _, C_, P_, ucd, fill_exact_width, textwrap_fill, exact_width, select_short_long from dnf.pycomp import xrange, basestring, long, unicode, sys_maxsize from dnf.yum.rpmtrans import TransactionDisplay from dnf.db.history import MergedTransactionWrapper import dnf.base import dnf.callback import dnf.cli.progress import dnf.cli.term import dnf.conf import dnf.crypto import dnf.i18n import dnf.transaction import dnf.util import dnf.yum.misc logger = logging.getLogger('dnf') def _spread_in_columns(cols_count, label, lst): left = itertools.chain((label,), itertools.repeat('')) lst_length = len(lst) right_count = cols_count - 1 missing_items = -lst_length % right_count if not lst_length: lst = itertools.repeat('', right_count) elif missing_items: lst.extend(('',) * missing_items) lst_iter = iter(lst) return list(zip(left, [lst_iter] right_count)) class Output(object): """Main output class for the yum command line.""" GRP_PACKAGE_INDENT = ' ' * 3 FILE_PROVIDE_RE = re.compile(r'^\{0,2}/') def __init__(self, base, conf): self.conf = conf self.base = base self.term = dnf.cli.term.Term() self.progress = None def _banner(self, col_data, row): term_width = self.term.columns rule = '%s' % '=' term_width header = self.fmtColumns(zip(row, col_data), ' ') return rule, header, rule def _col_widths(self, rows): col_data = [dict() for _ in rows[0]] for row in rows: for (i, val) in enumerate(row): col_dct = col_data[i] length = len(val) col_dct[length] = col_dct.get(length, 0) + 1 cols = self.calcColumns(col_data, None, indent=' ') # align to the left return list(map(operator.neg, cols)) def _highlight(self, highlight): hibeg = '' hiend = '' if not highlight: pass elif not isinstance(highlight, basestring) or highlight == 'bold': hibeg = self.term.MODE['bold'] elif highlight == 'normal': pass # Minor opt. else: # Turn a string into a specific output: colour, bold, etc. for high in highlight.replace(',', ' ').split(): if high == 'normal': hibeg = '' elif high in self.term.MODE: hibeg += self.term.MODE[high] elif high in self.term.FG_COLOR: hibeg += self.term.FG_COLOR[high] elif (high.startswith('fg:') and high[3:] in self.term.FG_COLOR): hibeg += self.term.FG_COLOR[high[3:]] elif (high.startswith('bg:') and high[3:] in self.term.BG_COLOR): hibeg += self.term.BG_COLOR[high[3:]] if hibeg: hiend = self.term.MODE['normal'] return (hibeg, hiend) def _sub_highlight(self, haystack, highlight, needles, kwds): hibeg, hiend = self._highlight(highlight) return self.term.sub(haystack, hibeg, hiend, needles, kwds) @staticmethod def _calc_columns_spaces_helps(current, data_tups, left): """ Spaces left on the current field will help how many pkgs? """ ret = 0 for tup in data_tups: if left < (tup[0] - current): break ret += tup[1] return ret @property def history(self): return self.base.history @property def sack(self): return self.base.sack def calcColumns(self, data, columns=None, remainder_column=0, total_width=None, indent=''): """Dynamically calculate the widths of the columns that the fields in data should be placed into for output. :param data: a list of dictionaries that represent the data to be output. Each dictionary in the list corresponds to a column of output. The keys of the dictionary are the lengths of the items to be output, and the value associated with a key is the number of items of that length. :param columns: a list containing the minimum amount of space that must be allocated for each row. This can be used to ensure that there is space available in a column if, for example, the actual lengths of the items being output cannot be given in data :param remainder_column: number of the column to receive a few extra spaces that may remain after other allocation has taken place :param total_width: the total width of the output. self.term.real_columns is used by default :param indent: string that will be prefixed to a line of output to create e.g. an indent :return: a list of the widths of the columns that the fields in data should be placed into for output """ cols = len(data) # Convert the data to ascending list of tuples, (field_length, pkgs) pdata = data data = [None] * cols # Don't modify the passed in data for d in range(0, cols): data[d] = sorted(pdata[d].items()) if total_width is None: total_width = self.term.real_columns # We start allocating 1 char to everything but the last column, and a # space between each (again, except for the last column). Because # at worst we are better with: # \|one two three\| # \| four \| # ...than: # \|one two three\| # \| f\| # \|our \| # ...the later being what we get if we pre-allocate the last column, and # thus. the space, due to "three" overflowing it's column by 2 chars. if columns is None: columns = [1] * (cols - 1) columns.append(0) # i'm not able to get real terminal width so i'm probably # running in non interactive terminal (pipe to grep, redirect to file...) # avoid splitting lines to enable filtering output if not total_width: full_columns = [] for d in xrange(0, cols): col = data[d] if col: full_columns.append(col[-1][0]) else: full_columns.append(columns[d] + 1) full_columns[0] += len(indent) # if possible, try to keep default width (usually 80 columns) default_width = self.term.columns if sum(full_columns) > default_width: return full_columns total_width = default_width total_width -= (sum(columns) + (cols - 1) + exact_width(indent)) if not columns[-1]: total_width += 1 while total_width > 0: # Find which field all the spaces left will help best helps = 0 val = 0 for d in xrange(0, cols): thelps = self._calc_columns_spaces_helps(columns[d], data[d], total_width) if not thelps: continue # We prefer to overflow: the last column, and then earlier # columns. This is so that in the best case (just overflow the # last) ... grep still "works", and then we make it prettier. if helps and (d == (cols - 1)) and (thelps / 2) < helps: continue if thelps < helps: continue helps = thelps val = d # If we found a column to expand, move up to the next level with # that column and start again with any remaining space. if helps: diff = data[val].pop(0)[0] - columns[val] if not columns[val] and (val == (cols - 1)): # If we are going from 0 => N on the last column, take 1 # for the space before the column. total_width -= 1 columns[val] += diff total_width -= diff continue overflowed_columns = 0 for d in xrange(0, cols): if not data[d]: continue overflowed_columns += 1 if overflowed_columns: # Split the remaining spaces among each overflowed column # equally norm = total_width // overflowed_columns for d in xrange(0, cols): if not data[d]: continue columns[d] += norm total_width -= norm # Split the remaining spaces among each column equally, except the # last one. And put the rest into the remainder column cols -= 1 norm = total_width // cols for d in xrange(0, cols): columns[d] += norm columns[remainder_column] += total_width - (cols * norm) total_width = 0 return columns @staticmethod def _fmt_column_align_width(width): """Returns tuple of (align_left, width)""" if width < 0: return (True, -width) return (False, width) def _col_data(self, col_data): assert len(col_data) == 2 or len(col_data) == 3 if len(col_data) == 2: (val, width) = col_data hibeg = hiend = '' if len(col_data) == 3: (val, width, highlight) = col_data (hibeg, hiend) = self._highlight(highlight) return (ucd(val), width, hibeg, hiend) def fmtColumns(self, columns, msg=u'', end=u''): """Return a row of data formatted into a string for output. Items can overflow their columns. :param columns: a list of tuples containing the data to output. Each tuple contains first the item to be output, then the amount of space allocated for the column, and then optionally a type of highlighting for the item :param msg: a string to begin the line of output with :param end: a string to end the line of output with :return: a row of data formatted into a string for output """ columns = list(columns) total_width = len(msg) data = [] for col_data in columns[:-1]: (val, width, hibeg, hiend) = self._col_data(col_data) if not width: # Don't count this column, invisible text msg += u"%s" data.append(val) continue (align_left, width) = self._fmt_column_align_width(width) val_width = exact_width(val) if val_width <= width: # Don't use fill_exact_width() because it sucks performance # wise for 1,000s of rows. Also allows us to use len(), when # we can. msg += u"%s%s%s%s " if align_left: data.extend([hibeg, val, " " * (width - val_width), hiend]) else: data.extend([hibeg, " " * (width - val_width), val, hiend]) else: msg += u"%s%s%s\n" + " " * (total_width + width + 1) data.extend([hibeg, val, hiend]) total_width += width total_width += 1 (val, width, hibeg, hiend) = self._col_data(columns[-1]) (align_left, width) = self._fmt_column_align_width(width) val = fill_exact_width(val, width, left=align_left, prefix=hibeg, suffix=hiend) msg += u"%%s%s" % end data.append(val) return msg % tuple(data) def simpleList(self, pkg, ui_overflow=False, indent='', highlight=False, columns=None): """Print a package as a line. :param pkg: the package to be printed :param ui_overflow: unused :param indent: string to be prefixed onto the line to provide e.g. an indent :param highlight: highlighting options for the name of the package :param columns: tuple containing the space allocated for each column of output. The columns are the package name, version, and repository """ if columns is None: columns = (-40, -22, -16) # Old default na = '%s%s.%s' % (indent, pkg.name, pkg.arch) hi_cols = [highlight, 'normal', 'normal'] columns = zip((na, pkg.evr, pkg._from_repo), columns, hi_cols) print(self.fmtColumns(columns)) def simpleEnvraList(self, pkg, ui_overflow=False, indent='', highlight=False, columns=None): """Print a package as a line, with the package itself in envra format so it can be passed to list/install/etc. :param pkg: the package to be printed :param ui_overflow: unused :param indent: string to be prefixed onto the line to provide e.g. an indent :param highlight: highlighting options for the name of the package :param columns: tuple containing the space allocated for each column of output. The columns the are the package envra and repository """ if columns is None: columns = (-63, -16) # Old default envra = '%s%s' % (indent, ucd(pkg)) hi_cols = [highlight, 'normal', 'normal'] rid = pkg.ui_from_repo columns = zip((envra, rid), columns, hi_cols) print(self.fmtColumns(columns)) def simple_name_list(self, pkg): """Print a package as a line containing its name.""" print(ucd(pkg.name)) def simple_nevra_list(self, pkg): """Print a package as a line containing its NEVRA.""" print(ucd(pkg)) def fmtKeyValFill(self, key, val): """Return a key value pair in the common two column output format. :param key: the key to be formatted :param val: the value associated with key :return: the key value pair formatted in two columns for output """ keylen = exact_width(key) cols = self.term.real_columns if not cols: cols = sys_maxsize elif cols < 20: cols = 20 nxt = ' ' * (keylen - 2) + ': ' if not val: # textwrap.fill in case of empty val returns empty string return key val = ucd(val) ret = textwrap_fill(val, width=cols, initial_indent=key, subsequent_indent=nxt) if ret.count("\n") > 1 and keylen > (cols // 3): # If it's big, redo it again with a smaller subsequent off ret = textwrap_fill(val, width=cols, initial_indent=key, subsequent_indent=' ...: ') return ret def fmtSection(self, name, fill='='): """Format and return a section header. The format of the header is a line with name centered, and fill repeated on either side to fill an entire line on the terminal. :param name: the name of the section :param fill: the character to repeat on either side of name to fill an entire line. fill must be a single character. :return: a string formatted to be a section header """ name = ucd(name) cols = self.term.columns - 2 name_len = exact_width(name) if name_len >= (cols - 4): beg = end = fill * 2 else: beg = fill * ((cols - name_len) // 2) end = fill * (cols - name_len - len(beg)) return "%s %s %s" % (beg, name, end) def infoOutput(self, pkg, highlight=False): """Print information about the given package. :param pkg: the package to print information about :param highlight: highlighting options for the name of the package """ def format_key_val(key, val): return " ".join([fill_exact_width(key, 12, 12), ":", str(val)]) def format_key_val_fill(key, val): return self.fmtKeyValFill(fill_exact_width(key, 12, 12) + " : ", val or "") output_list = [] (hibeg, hiend) = self._highlight(highlight) # Translators: This is abbreviated 'Name'. Should be no longer # than 12 characters. You can use the full version if it is short # enough in your language. key = select_short_long(12, C_("short", "Name"), C_("long", "Name")) output_list.append(format_key_val(key, "%s%s%s" % (hibeg, pkg.name, hiend))) if pkg.epoch: # Translators: This message should be no longer than 12 characters. output_list.append(format_key_val(_("Epoch"), pkg.epoch)) key = select_short_long(12, C_("short", "Version"), C_("long", "Version")) output_list.append(format_key_val(key, pkg.version)) # Translators: This message should be no longer than 12 characters. output_list.append(format_key_val(_("Release"), pkg.release)) key = select_short_long(12, C_("short", "Arch"), C_("long", "Architecture")) output_list.append(format_key_val(key, pkg.arch)) key = select_short_long(12, C_("short", "Size"), C_("long", "Size")) output_list.append(format_key_val(key, format_number(float(pkg._size)))) # Translators: This message should be no longer than 12 characters. output_list.append(format_key_val(_("Source"), pkg.sourcerpm)) key = select_short_long(12, C_("short", "Repo"), C_("long", "Repository")) output_list.append(format_key_val(key, pkg.repoid)) if pkg._from_system: history_repo = self.history.repo(pkg) if history_repo: # Translators: This message should be no longer than 12 chars. output_list.append(format_key_val(_("From repo"), history_repo)) if self.conf.verbose: # :hawkey does not support changelog information # print(_("Committer : %s") % ucd(pkg.committer)) # print(_("Committime : %s") % time.ctime(pkg.committime)) # Translators: This message should be no longer than 12 characters. output_list.append(format_key_val(_("Packager"), pkg.packager)) # Translators: This message should be no longer than 12 characters. output_list.append(format_key_val(_("Buildtime"), dnf.util.normalize_time(pkg.buildtime))) if pkg.installtime: # Translators: This message should be no longer than 12 characters. output_list.append(format_key_val(_("Install time"), dnf.util.normalize_time(pkg.installtime))) history_pkg = self.history.package_data(pkg) if history_pkg: try: uid = int(history_pkg._item.getInstalledBy()) except ValueError: # In case int() fails uid = None # Translators: This message should be no longer than 12 chars. output_list.append(format_key_val(_("Installed by"), self._pwd_ui_username(uid))) # Translators: This is abbreviated 'Summary'. Should be no longer # than 12 characters. You can use the full version if it is short # enough in your language. key = select_short_long(12, C_("short", "Summary"), C_("long", "Summary")) output_list.append(format_key_val_fill(key, pkg.summary)) if pkg.url: output_list.append(format_key_val(_("URL"), ucd(pkg.url))) # Translators: This message should be no longer than 12 characters. output_list.append(format_key_val_fill(_("License"), pkg.license)) # Translators: This is abbreviated 'Description'. Should be no longer # than 12 characters. You can use the full version if it is short # enough in your language. key = select_short_long(12, C_("short", "Description"), C_("long", "Description")) output_list.append(format_key_val_fill(key, pkg.description)) return "\n".join(output_list) def updatesObsoletesList(self, uotup, changetype, columns=None): """Print a simple string that explains the relationship between the members of an update or obsoletes tuple. :param uotup: an update or obsoletes tuple. The first member is the new package, and the second member is the old package :param changetype: a string indicating what the change between the packages is, e.g. 'updates' or 'obsoletes' :param columns: a tuple containing information about how to format the columns of output. The absolute value of each number in the tuple indicates how much space has been allocated for the corresponding column. If the number is negative, the text in the column will be left justified, and if it is positive, the text will be right justified. The columns of output are the package name, version, and repository """ (changePkg, instPkg) = uotup if columns is not None: # New style, output all info. for both old/new with old indented chi = self.conf.color_update_remote if changePkg.reponame != hawkey.SYSTEM_REPO_NAME: chi = self.conf.color_update_local self.simpleList(changePkg, columns=columns, highlight=chi) self.simpleList(instPkg, columns=columns, indent=' ' * 4, highlight=self.conf.color_update_installed) return # Old style c_compact = changePkg.compactPrint() i_compact = '%s.%s' % (instPkg.name, instPkg.arch) c_repo = changePkg.repoid print('%-35.35s [%.12s] %.10s %-20.20s' % (c_compact, c_repo, changetype, i_compact)) def listPkgs(self, lst, description, outputType, highlight_na={}, columns=None, highlight_modes={}): """Prints information about the given list of packages. :param lst: a list of packages to print information about :param description: string describing what the list of packages contains, e.g. 'Available Packages' :param outputType: The type of information to be printed. Current options:: 'list' - simple pkg list 'info' - similar to rpm -qi output 'name' - simple name list 'nevra' - simple nevra list :param highlight_na: a dictionary containing information about packages that should be highlighted in the output. The dictionary keys are (name, arch) tuples for the package, and the associated values are the package objects themselves. :param columns: a tuple containing information about how to format the columns of output. The absolute value of each number in the tuple indicates how much space has been allocated for the corresponding column. If the number is negative, the text in the column will be left justified, and if it is positive, the text will be right justified. The columns of output are the package name, version, and repository :param highlight_modes: dictionary containing information about to highlight the packages in highlight_na. highlight_modes should contain the following keys:: 'not_in' - highlighting used for packages not in highlight_na '=' - highlighting used when the package versions are equal '<' - highlighting used when the package has a lower version number '>' - highlighting used when the package has a higher version number :return: number of packages listed """ if outputType in ['list', 'info', 'name', 'nevra']: if len(lst) > 0: print('%s' % description) info_set = set() if outputType == 'list': unique_item_dict = {} for pkg in lst: unique_item_dict[str(pkg) + str(pkg._from_repo)] = pkg lst = unique_item_dict.values() for pkg in sorted(lst): key = (pkg.name, pkg.arch) highlight = False if key not in highlight_na: highlight = highlight_modes.get('not in', 'normal') elif pkg.evr_eq(highlight_na[key]): highlight = highlight_modes.get('=', 'normal') elif pkg.evr_lt(highlight_na[key]): highlight = highlight_modes.get('>', 'bold') else: highlight = highlight_modes.get('<', 'normal') if outputType == 'list': self.simpleList(pkg, ui_overflow=True, highlight=highlight, columns=columns) elif outputType == 'info': info_set.add(self.infoOutput(pkg, highlight=highlight) + "\n") elif outputType == 'name': self.simple_name_list(pkg) elif outputType == 'nevra': self.simple_nevra_list(pkg) else: pass if info_set: print("\n".join(sorted(info_set))) return len(lst) def userconfirm(self, msg=None, defaultyes_msg=None): """Get a yes or no from the user, and default to No :msg: String for case with [y/N] :defaultyes_msg: String for case with [Y/n] :return: True if the user selects yes, and False if the user selects no """ yui = (ucd(_('y')), ucd(_('yes'))) nui = (ucd(_('n')), ucd(_('no'))) aui = yui + nui while True: if msg is None: msg = _('Is this ok [y/N]: ') choice = '' if self.conf.defaultyes: if defaultyes_msg is None: msg = _('Is this ok [Y/n]: ') else: msg = defaultyes_msg try: choice = dnf.i18n.ucd_input(msg) except EOFError: pass except KeyboardInterrupt: choice = nui[0] choice = ucd(choice).lower() if len(choice) == 0: choice = yui[0] if self.conf.defaultyes else nui[0] if choice in aui: break # If the English one letter names don't mix with the translated # letters, allow them too: if u'y' == choice and u'y' not in aui: choice = yui[0] break if u'n' == choice and u'n' not in aui: choice = nui[0] break if choice in yui: return True return False def _pkgs2name_dict(self, sections): installed = self.sack.query().installed()._name_dict() available = self.sack.query().available()._name_dict() d = {} for pkg_name in itertools.chain(list(zip(sections))[1]): if pkg_name in installed: d[pkg_name] = installed[pkg_name][0] elif pkg_name in available: d[pkg_name] = available[pkg_name][0] return d def _pkgs2col_lengths(self, sections, name_dict): nevra_lengths = {} repo_lengths = {} for pkg_name in itertools.chain(list(zip(sections))[1]): pkg = name_dict.get(pkg_name) if pkg is None: continue nevra_l = exact_width(ucd(pkg)) + exact_width(self.GRP_PACKAGE_INDENT) repo_l = exact_width(ucd(pkg.reponame)) nevra_lengths[nevra_l] = nevra_lengths.get(nevra_l, 0) + 1 repo_lengths[repo_l] = repo_lengths.get(repo_l, 0) + 1 return (nevra_lengths, repo_lengths) def _display_packages(self, pkg_names): for name in pkg_names: print('%s%s' % (self.GRP_PACKAGE_INDENT, name)) def _display_packages_verbose(self, pkg_names, name_dict, columns): for name in pkg_names: try: pkg = name_dict[name] except KeyError: # package not in any repo -> print only package name print('%s%s' % (self.GRP_PACKAGE_INDENT, name)) continue highlight = False if not pkg._from_system: highlight = self.conf.color_list_available_install self.simpleEnvraList(pkg, ui_overflow=True, indent=self.GRP_PACKAGE_INDENT, highlight=highlight, columns=columns) def display_pkgs_in_groups(self, group): """Output information about the packages in a given group :param group: a Group object to output information about """ def names(packages): return sorted(dnf.util._name_unset_wrapper(pkg.name) for pkg in packages) print(_('Group: %s') % dnf.util._name_unset_wrapper(group.ui_name)) verbose = self.conf.verbose if verbose: print(_(' Group-Id: %s') % ucd(group.id)) if group.ui_description: print(_(' Description: %s') % ucd(group.ui_description) or "") if group.lang_only: print(_(' Language: %s') % group.lang_only) sections = ( (_(' Mandatory Packages:'), names(group.mandatory_packages)), (_(' Default Packages:'), names(group.default_packages)), (_(' Optional Packages:'), names(group.optional_packages)), (_(' Conditional Packages:'), names(group.conditional_packages))) if verbose: name_dict = self._pkgs2name_dict(sections) col_lengths = self._pkgs2col_lengths(sections, name_dict) columns = self.calcColumns(col_lengths) columns = (-columns[0], -columns[1]) for (section_name, packages) in sections: if len(packages) < 1: continue print(section_name) self._display_packages_verbose(packages, name_dict, columns) else: for (section_name, packages) in sections: if len(packages) < 1: continue print(section_name) self._display_packages(packages) def display_groups_in_environment(self, environment): """Output information about the packages in a given environment :param environment: an Environment object to output information about """ def names(groups): return sorted(dnf.util._name_unset_wrapper(group.name) for group in groups) print(_('Environment Group: %s') % dnf.util._name_unset_wrapper(environment.ui_name)) if self.conf.verbose: print(_(' Environment-Id: %s') % ucd(environment.id)) if environment.ui_description: description = ucd(environment.ui_description) or "" print(_(' Description: %s') % description) sections = ( (_(' Mandatory Groups:'), names(environment.mandatory_groups)), (_(' Optional Groups:'), names(environment.optional_groups))) for (section_name, packages) in sections: if len(packages) < 1: continue print(section_name) self._display_packages(packages) def matchcallback(self, po, values, matchfor=None, verbose=None, highlight=None): """Output search/provides type callback matches. :param po: the package object that matched the search :param values: the information associated with po that matched the search :param matchfor: a list of strings to be highlighted in the output :param verbose: whether to output extra verbose information :param highlight: highlighting options for the highlighted matches """ def print_highlighted_key_item(key, item, printed_headline, can_overflow=False): if not printed_headline: print(_('Matched from:')) item = ucd(item) or "" if item == "": return if matchfor: item = self._sub_highlight(item, highlight, matchfor, ignore_case=True) if can_overflow: print(self.fmtKeyValFill(key, item)) else: print(key % item) def print_file_provides(item, printed_match): if not self.FILE_PROVIDE_RE.match(item): return False key = _("Filename : %s") file_match = False for filename in po.files: if fnmatch.fnmatch(filename, item): print_highlighted_key_item( key, filename, file_match or printed_match, can_overflow=False) file_match = True return file_match if self.conf.showdupesfromrepos: msg = '%s : ' % po else: msg = '%s.%s : ' % (po.name, po.arch) msg = self.fmtKeyValFill(msg, po.summary or "") if matchfor: if highlight is None: highlight = self.conf.color_search_match msg = self._sub_highlight(msg, highlight, matchfor, ignore_case=True) print(msg) if verbose is None: verbose = self.conf.verbose if not verbose: return print(_("Repo : %s") % po.ui_from_repo) printed_match = False name_match = False for item in set(values): if po.summary == item: name_match = True continue # Skip double name/summary printing if po.description == item: key = _("Description : ") print_highlighted_key_item(key, item, printed_match, can_overflow=True) printed_match = True elif po.url == item: key = _("URL : %s") print_highlighted_key_item(key, item, printed_match, can_overflow=False) printed_match = True elif po.license == item: key = _("License : %s") print_highlighted_key_item(key, item, printed_match, can_overflow=False) printed_match = True elif print_file_provides(item, printed_match): printed_match = True else: key = _("Provide : %s") for provide in po.provides: provide = str(provide) if fnmatch.fnmatch(provide, item): print_highlighted_key_item(key, provide, printed_match, can_overflow=False) printed_match = True else: first_provide = provide.split()[0] possible = set('=<>') if any((char in possible) for char in item): item_new = item.split()[0] else: item_new = item if fnmatch.fnmatch(first_provide, item_new): print_highlighted_key_item( key, provide, printed_match, can_overflow=False) printed_match = True if not any([printed_match, name_match]): for item in set(values): key = _("Other : %s") print_highlighted_key_item(key, item, printed_match, can_overflow=False) print() def matchcallback_verbose(self, po, values, matchfor=None): """Output search/provides type callback matches. This will output more information than :func:`matchcallback`. :param po: the package object that matched the search :param values: the information associated with po that matched the search :param matchfor: a list of strings to be highlighted in the output """ return self.matchcallback(po, values, matchfor, verbose=True) def reportDownloadSize(self, packages, installonly=False): """Report the total download size for a set of packages :param packages: a list of package objects :param installonly: whether the transaction consists only of installations """ totsize = 0 locsize = 0 insize = 0 error = False for pkg in packages: # Just to be on the safe side, if for some reason getting # the package size fails, log the error and don't report download # size try: size = int(pkg._size) totsize += size try: if pkg.verifyLocalPkg(): locsize += size except Exception: pass if not installonly: continue try: size = int(pkg.installsize) except Exception: pass insize += size except Exception: error = True msg = _('There was an error calculating total download size') logger.error(msg) break if not error: if locsize: logger.info(_("Total size: %s"), format_number(totsize)) if locsize != totsize: logger.info(_("Total download size: %s"), format_number(totsize - locsize)) if installonly: logger.info(_("Installed size: %s"), format_number(insize)) def reportRemoveSize(self, packages): """Report the total size of packages being removed. :param packages: a list of package objects """ totsize = 0 error = False for pkg in packages: # Just to be on the safe side, if for some reason getting # the package size fails, log the error and don't report download # size try: size = pkg._size totsize += size except Exception: error = True msg = _('There was an error calculating installed size') logger.error(msg) break if not error: logger.info(_("Freed space: %s"), format_number(totsize)) def list_group_transaction(self, comps, history, diff): if not diff: return None out = [] rows = [] if diff.new_groups: out.append(_('Marking packages as installed by the group:')) for grp_id in diff.new_groups: pkgs = list(diff.added_packages(grp_id)) group_object = comps._group_by_id(grp_id) grp_name = group_object.ui_name if group_object else grp_id rows.extend(_spread_in_columns(4, "@" + grp_name, pkgs)) if diff.removed_groups: out.append(_('Marking packages as removed by the group:')) for grp_id in diff.removed_groups: pkgs = list(diff.removed_packages(grp_id)) grp_name = history.group.get(grp_id).ui_name rows.extend(_spread_in_columns(4, "@" + grp_name, pkgs)) if rows: col_data = self._col_widths(rows) for row in rows: out.append(self.fmtColumns(zip(row, col_data), ' ')) out[0:0] = self._banner(col_data, (_('Group'), _('Packages'), '', '')) return '\n'.join(out) def list_transaction(self, transaction, total_width=None): """Return a string representation of the transaction in an easy-to-read format. """ forward_actions = hawkey.UPGRADE \| hawkey.UPGRADE_ALL \| hawkey.DISTUPGRADE \| \ hawkey.DISTUPGRADE_ALL \| hawkey.DOWNGRADE \| hawkey.INSTALL skipped_conflicts = set() skipped_broken = set() if transaction is None: # set empty transaction list instead of returning None # in order to display module changes when RPM transaction is empty transaction = [] list_bunch = dnf.util._make_lists(transaction) pkglist_lines = [] data = {'n' : {}, 'v' : {}, 'r' : {}} a_wid = 0 # Arch can't get "that big" ... so always use the max. def _add_line(lines, data, a_wid, po, obsoletes=[]): (n, a, e, v, r) = po.pkgtup evr = po.evr repoid = po._from_repo size = format_number(po._size) if a is None: # gpgkeys are weird a = 'noarch' # none, partial, full? if po._from_system: hi = self.conf.color_update_installed elif po._from_cmdline: hi = self.conf.color_update_local else: hi = self.conf.color_update_remote lines.append((n, a, evr, repoid, size, obsoletes, hi)) # Create a dict of field_length => number of packages, for # each field. for (d, v) in (("n", len(n)), ("v", len(evr)), ("r", len(repoid))): data[d].setdefault(v, 0) data[d][v] += 1 a_wid = max(a_wid, len(a)) return a_wid ins_group_msg = _('Installing group/module packages') if dnf.base.WITH_MODULES \ else _('Installing group packages') for (action, pkglist) in [ # TRANSLATORS: This is for a list of packages to be installed. (C_('summary', 'Installing'), list_bunch.installed), # TRANSLATORS: This is for a list of packages to be upgraded. (C_('summary', 'Upgrading'), list_bunch.upgraded), # TRANSLATORS: This is for a list of packages to be reinstalled. (C_('summary', 'Reinstalling'), list_bunch.reinstalled), (ins_group_msg, list_bunch.installed_group), (_('Installing dependencies'), list_bunch.installed_dep), (_('Installing weak dependencies'), list_bunch.installed_weak), # TRANSLATORS: This is for a list of packages to be removed. (_('Removing'), list_bunch.erased), (_('Removing dependent packages'), list_bunch.erased_dep), (_('Removing unused dependencies'), list_bunch.erased_clean), # TRANSLATORS: This is for a list of packages to be downgraded. (C_('summary', 'Downgrading'), list_bunch.downgraded)]: lines = [] # build a reverse mapping to 'replaced_by' # this is required to achieve reasonable speed replaces = {} for tsi in transaction: if tsi.action != libdnf.transaction.TransactionItemAction_OBSOLETED: continue for i in tsi._item.getReplacedBy(): replaces.setdefault(i, set()).add(tsi) for tsi in sorted(pkglist, key=lambda x: x.pkg): if tsi.action not in dnf.transaction.FORWARD_ACTIONS + [libdnf.transaction.TransactionItemAction_REMOVE]: continue # get TransactionItems obsoleted by tsi obsoleted = sorted(replaces.get(tsi._item, [])) a_wid = _add_line(lines, data, a_wid, tsi.pkg, obsoleted) pkglist_lines.append((action, lines)) installedProfiles = sorted(dict(self.base._moduleContainer.getInstalledProfiles()).items()) if installedProfiles: action = _("Installing module profiles") lines = [] for name, profiles in installedProfiles: for profile in list(profiles): lines.append(("%s/%s" % (name, profile), "", "", "", "", "", "")) pkglist_lines.append((action, lines)) removedProfiles = sorted(dict(self.base._moduleContainer.getRemovedProfiles()).items()) if removedProfiles: action = _("Disabling module profiles") lines = [] for name, profiles in removedProfiles: for profile in list(profiles): lines.append(("%s/%s" % (name, profile), "", "", "", "", "", "")) pkglist_lines.append((action, lines)) enabledStreams = sorted(dict(self.base._moduleContainer.getEnabledStreams()).items()) if enabledStreams: action = _("Enabling module streams") lines = [] for name, stream in enabledStreams: lines.append((name, "", stream, "", "", "", "")) pkglist_lines.append((action, lines)) switchedStreams = sorted(dict(self.base._moduleContainer.getSwitchedStreams()).items()) if switchedStreams: action = _("Switching module streams") lines = [] for name, stream in switchedStreams: lines.append((name, "", "%s -> %s" % (stream[0], stream[1]), "", "", "", "")) pkglist_lines.append((action, lines)) disabledModules = sorted(list(self.base._moduleContainer.getDisabledModules())) if disabledModules: action = _("Disabling modules") lines = [] for name in disabledModules: lines.append((name, "", "", "", "", "", "")) pkglist_lines.append((action, lines)) resetModules = sorted(list(self.base._moduleContainer.getResetModules())) if resetModules: action = _("Resetting modules") lines = [] for name in resetModules: lines.append((name, "", "", "", "", "", "")) pkglist_lines.append((action, lines)) if self.base._history: def format_line(group): return (dnf.util._name_unset_wrapper(group.getName()), "", "", "", "", "", "") install_env_group = self.base._history.env._installed if install_env_group: action = _("Installing Environment Groups") lines = [] for group in install_env_group.values(): lines.append(format_line(group)) pkglist_lines.append((action, lines)) upgrade_env_group = self.base._history.env._upgraded if upgrade_env_group: action = _("Upgrading Environment Groups") lines = [] for group in upgrade_env_group.values(): lines.append(format_line(group)) pkglist_lines.append((action, lines)) remove_env_group = self.base._history.env._removed if remove_env_group: action = _("Removing Environment Groups") lines = [] for group in remove_env_group.values(): lines.append(format_line(group)) pkglist_lines.append((action, lines)) install_group = self.base._history.group._installed if install_group: action = _("Installing Groups") lines = [] for group in install_group.values(): lines.append(format_line(group)) pkglist_lines.append((action, lines)) upgrade_group = self.base._history.group._upgraded if upgrade_group: action = _("Upgrading Groups") lines = [] for group in upgrade_group.values(): lines.append(format_line(group)) pkglist_lines.append((action, lines)) remove_group = self.base._history.group._removed if remove_group: action = _("Removing Groups") lines = [] for group in remove_group.values(): lines.append(format_line(group)) pkglist_lines.append((action, lines)) # show skipped conflicting packages if (not self.conf.best or not self.conf.strict) and self.base._goal.actions & forward_actions: lines = [] skipped_conflicts, skipped_broken = self.base._skipped_packages( report_problems=True, transaction=transaction) skipped_broken = dict((str(pkg), pkg) for pkg in skipped_broken) for pkg in sorted(skipped_conflicts): a_wid = _add_line(lines, data, a_wid, pkg, []) recommendations = ["--best"] if not self.base._allow_erasing: recommendations.append("--allowerasing") skip_str = _("Skipping packages with conflicts:\n" "(add '%s' to command line " "to force their upgrade)") % " ".join(recommendations) # remove misleading green color from the "packages with conflicts" lines lines = [i[:-1] + ("", ) for i in lines] pkglist_lines.append((skip_str, lines)) lines = [] for nevra, pkg in sorted(skipped_broken.items()): a_wid = _add_line(lines, data, a_wid, pkg, []) skip_str = _("Skipping packages with broken dependencies%s") if self.base.conf.upgrade_group_objects_upgrade: skip_str = skip_str % "" else: skip_str = skip_str % _(" or part of a group") # remove misleading green color from the "broken dependencies" lines lines = [i[:-1] + ("", ) for i in lines] pkglist_lines.append((skip_str, lines)) output_width = self.term.columns if not data['n'] and not self.base._moduleContainer.isChanged() and not \ (self.base._history and (self.base._history.group or self.base._history.env)): return u'' else: data = [data['n'], {}, data['v'], data['r'], {}] columns = [1, a_wid, 1, 1, 5] columns = self.calcColumns(data, indent=" ", columns=columns, remainder_column=2, total_width=total_width) (n_wid, a_wid, v_wid, r_wid, s_wid) = columns real_width = sum(columns) + 5 output_width = output_width if output_width >= real_width else real_width # Do not use 'Package' without context. Using context resolves # RhBug 1302935 as a side effect. msg_package = select_short_long(n_wid, # Translators: This is the short version of 'Package'. You can # use the full (unabbreviated) term 'Package' if you think that # the translation to your language is not too long and will # always fit to limited space. C_('short', 'Package'), # Translators: This is the full (unabbreviated) term 'Package'. C_('long', 'Package')) msg_arch = select_short_long(a_wid, # Translators: This is abbreviated 'Architecture', used when # we have not enough space to display the full word. C_('short', 'Arch'), # Translators: This is the full word 'Architecture', used when # we have enough space. C_('long', 'Architecture')) msg_version = select_short_long(v_wid, # Translators: This is the short version of 'Version'. You can # use the full (unabbreviated) term 'Version' if you think that # the translation to your language is not too long and will # always fit to limited space. C_('short', 'Version'), # Translators: This is the full (unabbreviated) term 'Version'. C_('long', 'Version')) msg_repository = select_short_long(r_wid, # Translators: This is abbreviated 'Repository', used when # we have not enough space to display the full word. C_('short', 'Repo'), # Translators: This is the full word 'Repository', used when # we have enough space. C_('long', 'Repository')) msg_size = select_short_long(s_wid, # Translators: This is the short version of 'Size'. It should # not be longer than 5 characters. If the term 'Size' in your # language is not longer than 5 characters then you can use it # unabbreviated. C_('short', 'Size'), # Translators: This is the full (unabbreviated) term 'Size'. C_('long', 'Size')) out = [u"%s\n%s\n%s\n" % ('=' * output_width, self.fmtColumns(((msg_package, -n_wid), (msg_arch, -a_wid), (msg_version, -v_wid), (msg_repository, -r_wid), (msg_size, s_wid)), u" "), '=' * output_width)] for (action, lines) in pkglist_lines: if lines: totalmsg = u"%s:\n" % action for (n, a, evr, repoid, size, obsoletes, hi) in lines: columns = ((n, -n_wid, hi), (a, -a_wid), (evr, -v_wid), (repoid, -r_wid), (size, s_wid)) msg = self.fmtColumns(columns, u" ", u"\n") hibeg, hiend = self._highlight(self.conf.color_update_installed) for obspo in sorted(obsoletes): appended = ' ' + _('replacing') + ' %s%s%s.%s %s\n' appended %= (hibeg, obspo.name, hiend, obspo.arch, obspo.evr) msg += appended totalmsg = totalmsg + msg if lines: out.append(totalmsg) out.append(_(""" Transaction Summary %s """) % ('=' * output_width)) summary_data = ( (_('Install'), len(list_bunch.installed) + len(list_bunch.installed_group) + len(list_bunch.installed_weak) + len(list_bunch.installed_dep), 0), (_('Upgrade'), len(list_bunch.upgraded), 0), (_('Remove'), len(list_bunch.erased) + len(list_bunch.erased_dep) + len(list_bunch.erased_clean), 0), (_('Downgrade'), len(list_bunch.downgraded), 0), (_('Skip'), len(skipped_conflicts) + len(skipped_broken), 0)) max_msg_action = 0 max_msg_count = 0 max_msg_pkgs = 0 max_msg_depcount = 0 for action, count, depcount in summary_data: if not count and not depcount: continue msg_pkgs = P_('Package', 'Packages', count) len_msg_action = exact_width(action) len_msg_count = exact_width(unicode(count)) len_msg_pkgs = exact_width(msg_pkgs) if depcount: len_msg_depcount = exact_width(unicode(depcount)) else: len_msg_depcount = 0 max_msg_action = max(len_msg_action, max_msg_action) max_msg_count = max(len_msg_count, max_msg_count) max_msg_pkgs = max(len_msg_pkgs, max_msg_pkgs) max_msg_depcount = max(len_msg_depcount, max_msg_depcount) for action, count, depcount in summary_data: msg_pkgs = P_('Package', 'Packages', count) if depcount: msg_deppkgs = P_('Dependent package', 'Dependent packages', depcount) action_msg = fill_exact_width(action, max_msg_action) if count: msg = '%s %d %s (+%d %s)\n' out.append(msg % (action_msg, max_msg_count, count, "%-s" % (max_msg_pkgs, msg_pkgs), max_msg_depcount, depcount, msg_deppkgs)) else: msg = '%s %s ( %d %s)\n' out.append(msg % (action_msg, (max_msg_count + max_msg_pkgs) * ' ', max_msg_depcount, depcount, msg_deppkgs)) elif count: msg = '%s %d %s\n' out.append(msg % (fill_exact_width(action, max_msg_action), max_msg_count, count, msg_pkgs)) return ''.join(out) def _pto_callback(self, action, tsis): # Works a bit like calcColumns, but we never overflow a column we just # have a dynamic number of columns. def _fits_in_cols(msgs, num): """ Work out how many columns we can use to display stuff, in the post trans output. """ if len(msgs) < num: return [] left = self.term.columns - ((num - 1) + 2) if left <= 0: return [] col_lens = [0] num col = 0 for msg in msgs: if len(msg) > col_lens[col]: diff = (len(msg) - col_lens[col]) if left <= diff: return [] left -= diff col_lens[col] = len(msg) col += 1 col %= len(col_lens) for col in range(len(col_lens)): col_lens[col] += left // num col_lens[col] = -1 return col_lens if not tsis: return '' out = [] msgs = [] out.append('{}:'.format(action)) for tsi in tsis: msgs.append(str(tsi)) for num in (8, 7, 6, 5, 4, 3, 2): cols = _fits_in_cols(msgs, num) if cols: break if not cols: cols = [-(self.term.columns - 2)] while msgs: current_msgs = msgs[:len(cols)] out.append(' {}'.format(self.fmtColumns(zip(current_msgs, cols)))) msgs = msgs[len(cols):] return out def post_transaction_output(self, transaction): """ Return a human-readable summary of the transaction. Packages in sections are arranged to columns. """ return dnf.util._post_transaction_output(self.base, transaction, self._pto_callback) def setup_progress_callbacks(self): """Set up the progress callbacks and various output bars based on debug level. """ progressbar = None if self.conf.debuglevel >= 2: progressbar = dnf.cli.progress.MultiFileProgressMeter(fo=sys.stdout) self.progress = dnf.cli.progress.MultiFileProgressMeter(fo=sys.stdout) # setup our depsolve progress callback return (progressbar, DepSolveProgressCallBack()) def download_callback_total_cb(self, remote_size, download_start_timestamp): """Outputs summary information about the download process. :param remote_size: the total amount of information that was downloaded, in bytes :param download_start_timestamp: the time when the download process started, in seconds since the epoch """ if remote_size <= 0: return width = self.term.columns logger.info("-" width) dl_time = max(0.01, time.time() - download_start_timestamp) msg = ' %5sB/s \| %5sB %9s ' % ( format_number(remote_size // dl_time), format_number(remote_size), format_time(dl_time)) msg = fill_exact_width(_("Total"), width - len(msg)) + msg logger.info(msg) def _history_uiactions(self, hpkgs): actions = set() actions_short = set() count = 0 for pkg in hpkgs: if pkg.action in (libdnf.transaction.TransactionItemAction_UPGRADED, libdnf.transaction.TransactionItemAction_DOWNGRADED): # skip states we don't want to display in user input continue actions.add(pkg.action_name) actions_short.add(pkg.action_short) count += 1 if len(actions) > 1: return count, ", ".join(sorted(actions_short)) # So empty transactions work, although that "shouldn't" really happen return count, "".join(list(actions)) def _pwd_ui_username(self, uid, limit=None): if isinstance(uid, list): return [self._pwd_ui_username(u, limit) for u in uid] # loginuid is set to -1 (0xFFFF_FFFF) on init, in newer kernels. # loginuid is set to INT_MAX (0x7FFF_FFFF) on init, in older kernels. if uid is None or uid in (0xFFFFFFFF, 0x7FFFFFFF): loginid = _("<unset>") name = _("System") + " " + loginid if limit is not None and len(name) > limit: name = loginid return ucd(name) def _safe_split_0(text, args): """ Split gives us a [0] for everything _but_ '', this function returns '' in that case. """ ret = text.split(args) if not ret: return '' return ret[0] try: user = pwd.getpwuid(int(uid)) fullname = _safe_split_0(ucd(user.pw_gecos), ';', 2) user_name = ucd(user.pw_name) name = "%s <%s>" % (fullname, user_name) if limit is not None and len(name) > limit: name = "%s ... <%s>" % (_safe_split_0(fullname), user_name) if len(name) > limit: name = "<%s>" % user_name return name except KeyError: return ucd(uid) def historyListCmd(self, tids, reverse=False): """Output a list of information about the history of yum transactions. :param tids: transaction Ids; lists all transactions if empty """ transactions = self.history.old(tids) if self.conf.history_list_view == 'users': uids = [1, 2] elif self.conf.history_list_view == 'commands': uids = [1] else: assert self.conf.history_list_view == 'single-user-commands' uids = set() done = 0 blanks = 0 for transaction in transactions: done += 1 if transaction.cmdline is None: blanks += 1 uids.add(transaction.loginuid) fmt = "%s \| %s \| %s \| %s \| %s" if len(uids) == 1: name = _("Command line") real_cols = self.term.real_columns if real_cols is None: # if output is redirected in `less` the columns # detected are None value, to detect terminal size # use stdin file descriptor real_cols = dnf.cli.term._real_term_width(0) if real_cols is None: # if even stdin fd fails use 24 to fit to 80 cols real_cols = 24 name_width = real_cols - 55 if real_cols > 79 else 24 else: # TRANSLATORS: user names who executed transaction in history command output name = _("User name") name_width = 24 print(fmt % (fill_exact_width(_("ID"), 6, 6), fill_exact_width(name, name_width, name_width), fill_exact_width(_("Date and time"), 16, 16), fill_exact_width(_("Action(s)"), 14, 14), fill_exact_width(_("Altered"), 7, 7))) # total table width: each column length +3 (padding and separator between columns) table_width = 6 + 3 + name_width + 3 + 16 + 3 + 14 + 3 + 7 print("-" * table_width) fmt = "%6u \| %s \| %-16.16s \| %s \| %4u" if reverse is True: transactions = reversed(transactions) for transaction in transactions: if len(uids) == 1: name = transaction.cmdline or '' else: name = self._pwd_ui_username(transaction.loginuid, 24) name = ucd(name) tm = time.strftime("%Y-%m-%d %H:%M", time.localtime(transaction.beg_timestamp)) num, uiacts = self._history_uiactions(transaction.data()) name = fill_exact_width(name, name_width, name_width) uiacts = fill_exact_width(uiacts, 14, 14) rmark = lmark = ' ' if transaction.return_code is None: rmark = lmark = '' elif transaction.return_code: rmark = lmark = '#' # We don't check .errors, because return_code will be non-0 elif transaction.is_output: rmark = lmark = 'E' if transaction.altered_lt_rpmdb: rmark = '<' if transaction.altered_gt_rpmdb: lmark = '>' print(fmt % (transaction.tid, name, tm, uiacts, num), "%s%s" % (lmark, rmark)) def historyInfoCmd(self, tids, pats=[], mtids=set()): """Output information about a transaction in history :param tids: transaction Ids; prints info for the last transaction if empty :raises dnf.exceptions.Error in case no transactions were found """ tids = set(tids) last = self.history.last() if last is None: logger.critical(_('No transactions')) raise dnf.exceptions.Error(_('Failed history info')) lasttid = last.tid lastdbv = last.end_rpmdb_version transactions = [] if not tids: last = self.history.last(complete_transactions_only=False) if last is not None: tids.add(last.tid) transactions.append(last) else: transactions = self.history.old(tids) if not tids: logger.critical(_('No transaction ID, or package, given')) raise dnf.exceptions.Error(_('Failed history info')) bmtid, emtid = -1, -1 mobj = None done = False if mtids: mtids = sorted(mtids) bmtid, emtid = mtids.pop() for trans in transactions: if lastdbv is not None and trans.tid == lasttid: # If this is the last transaction, is good and it doesn't # match the current rpmdb ... then mark it as bad. rpmdbv = self.base._ts.dbCookie() trans.compare_rpmdbv(str(rpmdbv)) lastdbv = None merged = False if trans.tid >= bmtid and trans.tid <= emtid: if mobj is None: mobj = MergedTransactionWrapper(trans) else: mobj.merge(trans) merged = True elif mobj is not None: if done: print("-" 79) done = True self._historyInfoCmd(mobj) mobj = None if mtids: bmtid, emtid = mtids.pop() if trans.tid >= bmtid and trans.tid <= emtid: mobj = trans merged = True if not merged: if done: print("-" * 79) done = True self._historyInfoCmd(trans, pats) if mobj is not None: if done: print("-" * 79) self._historyInfoCmd(mobj) def _historyInfoCmd(self, old, pats=[]): loginuid = old.loginuid if isinstance(loginuid, int): loginuid = [loginuid] name = [self._pwd_ui_username(uid) for uid in loginuid] _pkg_states_installed = {'i' : _('Installed'), 'e' : _('Erased'), 'o' : _('Upgraded'), 'n' : _('Downgraded')} _pkg_states_available = {'i' : _('Installed'), 'e' : _('Not installed'), 'o' : _('Older'), 'n' : _('Newer')} maxlen = max([len(x) for x in (list(_pkg_states_installed.values()) + list(_pkg_states_available.values()))]) _pkg_states_installed['maxlen'] = maxlen _pkg_states_available['maxlen'] = maxlen def _simple_pkg(pkg, prefix_len, was_installed=False, highlight=False, pkg_max_len=0, show_repo=True): prefix = " " * prefix_len if was_installed: _pkg_states = _pkg_states_installed else: _pkg_states = _pkg_states_available state = _pkg_states['i'] # get installed packages with name = pkg.name ipkgs = self.sack.query().installed().filterm(name=pkg.name).run() if not ipkgs: state = _pkg_states['e'] else: # get latest installed package from software database inst_pkg = self.history.package(ipkgs[0]) if inst_pkg: res = pkg.compare(inst_pkg) # res is: # 0 if inst_pkg == pkg # > 0 when inst_pkg > pkg # < 0 when inst_pkg < pkg if res == 0: pass # installed elif res > 0: state = _pkg_states['o'] # updated else: state = _pkg_states['n'] # downgraded if highlight: (hibeg, hiend) = self._highlight('bold') else: (hibeg, hiend) = self._highlight('normal') state = fill_exact_width(state, _pkg_states['maxlen']) ui_repo = '' if show_repo: ui_repo = pkg.ui_from_repo() print("%s%s%s%s %-s %s" % (prefix, hibeg, state, hiend, pkg_max_len, str(pkg), ui_repo)) tids = old.tids() if len(tids) > 1: print(_("Transaction ID :"), "%u..%u" % (tids[0], tids[-1])) else: print(_("Transaction ID :"), tids[0]) begt = float(old.beg_timestamp) begtm = time.strftime("%c", time.localtime(begt)) print(_("Begin time :"), begtm) if old.beg_rpmdb_version is not None: if old.altered_lt_rpmdb: print(_("Begin rpmdb :"), old.beg_rpmdb_version, "") else: print(_("Begin rpmdb :"), old.beg_rpmdb_version) if old.end_timestamp is not None: endt = old.end_timestamp endtm = time.strftime("%c", time.localtime(endt)) diff = endt - begt if diff < 5 60: diff = _("(%u seconds)") % diff elif diff < 5 * 60 * 60: diff = _("(%u minutes)") % (diff // 60) elif diff < 5 * 60 * 60 * 24: diff = _("(%u hours)") % (diff // (60 * 60)) else: diff = _("(%u days)") % (diff // (60 * 60 * 24)) print(_("End time :"), endtm, diff) if old.end_rpmdb_version is not None: if old.altered_gt_rpmdb: print(_("End rpmdb :"), old.end_rpmdb_version, "") else: print(_("End rpmdb :"), old.end_rpmdb_version) if isinstance(name, (list, tuple)): seen = set() for i in name: if i in seen: continue seen.add(i) print(_("User :"), i) else: print(_("User :"), name) if isinstance(old.return_code, (list, tuple)): codes = old.return_code if codes[0] is None: print(_("Return-Code :"), "", _("Aborted"), "") codes = codes[1:] elif not all(codes): print(_("Return-Code :"), _("Success")) elif codes: print(_("Return-Code :"), _("Failures:"), ", ".join([str(i) for i in codes])) elif old.return_code is None: print(_("Return-Code :"), "", _("Aborted"), "*") elif old.return_code: print(_("Return-Code :"), _("Failure:"), old.return_code) else: print(_("Return-Code :"), _("Success")) if isinstance(old.releasever, (list, tuple)): seen = set() for i in old.releasever: if i in seen: continue seen.add(i) print(_("Releasever :"), i) else: print(_("Releasever :"), old.releasever) if old.cmdline is not None: if isinstance(old.cmdline, (list, tuple)): for cmdline in old.cmdline: print(_("Command Line :"), cmdline) else: print(_("Command Line :"), old.cmdline) def print_persistence(persistence): if old.persistence == libdnf.transaction.TransactionPersistence_PERSIST: persistence_str = "Persist" elif old.persistence == libdnf.transaction.TransactionPersistence_TRANSIENT: persistence_str = "Transient" else: persistence_str = "Unknown" print(_("Persistence :"), persistence_str) if isinstance(old.persistence, (list, tuple)): for persistence in old.persistence: print_persistence(persistence) else: print_persistence(old.persistence) if old.comment is not None: if isinstance(old.comment, (list, tuple)): for comment in old.comment: print(_("Comment :"), comment) else: print(_("Comment :"), old.comment) perf_with = old.performed_with() if perf_with: print(_("Transaction performed with:")) max_len = 0 for with_pkg in perf_with: str_len = len(str(with_pkg)) if str_len > max_len: max_len = str_len for with_pkg in perf_with: _simple_pkg(with_pkg, 4, was_installed=True, pkg_max_len=max_len) print(_("Packages Altered:")) self.historyInfoCmdPkgsAltered(old, pats) t_out = old.output() if t_out: print(_("Scriptlet output:")) num = 0 for line in t_out: num += 1 print("%4d" % num, line) t_err = old.error() if t_err: print(_("Errors:")) num = 0 for line in t_err: num += 1 print("%4d" % num, line) # TODO: remove _history_state2uistate = {'True-Install' : _('Install'), 'Install' : _('Install'), 'Dep-Install' : _('Dep-Install'), 'Obsoleted' : _('Obsoleted'), 'Obsoleting' : _('Obsoleting'), 'Erase' : _('Erase'), 'Reinstall' : _('Reinstall'), 'Downgrade' : _('Downgrade'), 'Downgraded' : _('Downgraded'), 'Update' : _('Upgrade'), 'Updated' : _('Upgraded'), } def historyInfoCmdPkgsAltered(self, old, pats=[]): """Print information about how packages are altered in a transaction. :param old: the :class:`DnfSwdbTrans` to print information about :param pats: a list of patterns. Packages that match a patten in pats* will be highlighted in the output """ # Note that these don't use _simple_pkg() because we are showing what # happened to them in the transaction ... not the difference between the # version in the transaction and now. all_uistates = self._history_state2uistate maxlen = 0 pkg_max_len = 0 packages = old.packages() for pkg in packages: uistate = all_uistates.get(pkg.action_name, pkg.action_name) if maxlen < len(uistate): maxlen = len(uistate) pkg_len = len(str(pkg)) if pkg_max_len < pkg_len: pkg_max_len = pkg_len for pkg in packages: prefix = " " * 4 if pkg.state != libdnf.transaction.TransactionItemState_DONE: prefix = " ** " highlight = 'normal' if pats: if any([pkg.match(pat) for pat in pats]): highlight = 'bold' (hibeg, hiend) = self._highlight(highlight) uistate = all_uistates.get(pkg.action_name, pkg.action_name) uistate = fill_exact_width(ucd(uistate), maxlen) print("%s%s%s%s %-s %s" % (prefix, hibeg, uistate, hiend, pkg_max_len, str(pkg), pkg.ui_from_repo())) class DepSolveProgressCallBack(dnf.callback.Depsolve): """Provides text output callback functions for Dependency Solver callback.""" def pkg_added(self, pkg, mode): """Print information about a package being added to the transaction set. :param pkgtup: tuple containing the package name, arch, version, and repository :param mode: a short string indicating why the package is being added to the transaction set. Valid current values for mode* are:: i = the package will be installed u = the package will be an update e = the package will be erased r = the package will be reinstalled d = the package will be a downgrade o = the package will be obsoleting another package ud = the package will be updated od = the package will be obsoleted """ output = None if mode == 'i': output = _('---> Package %s.%s %s will be installed') elif mode == 'u': output = _('---> Package %s.%s %s will be an upgrade') elif mode == 'e': output = _('---> Package %s.%s %s will be erased') elif mode == 'r': output = _('---> Package %s.%s %s will be reinstalled') elif mode == 'd': output = _('---> Package %s.%s %s will be a downgrade') elif mode == 'o': output = _('---> Package %s.%s %s will be obsoleting') elif mode == 'ud': output = _('---> Package %s.%s %s will be upgraded') elif mode == 'od': output = _('---> Package %s.%s %s will be obsoleted') if output: logger.debug(output, pkg.name, pkg.arch, pkg.evr) def start(self): """Perform setup at the beginning of the dependency solving process. """ logger.debug(_('--> Starting dependency resolution')) def end(self): """Output a message stating that dependency resolution has finished.""" logger.debug(_('--> Finished dependency resolution')) class CliKeyImport(dnf.callback.KeyImport): def __init__(self, base, output): self.base = base self.output = output def _confirm(self, id, userid, fingerprint, url, timestamp): def short_id(id): rj = '0' if dnf.pycomp.PY3 else b'0' return id[-8:].rjust(8, rj) msg = (_('Importing GPG key 0x%s:\n' ' Userid : "%s"\n' ' Fingerprint: %s\n' ' From : %s') % (short_id(id), userid, dnf.crypto._printable_fingerprint(fingerprint), url.replace("file://", ""))) logger.critical("%s", msg) if self.base.conf.assumeyes: return True if self.base.conf.assumeno: return False return self.output.userconfirm() class CliTransactionDisplay(TransactionDisplay): """A YUM specific callback class for RPM operations.""" width = property(lambda self: dnf.cli.term._term_width()) def __init__(self): super(CliTransactionDisplay, self).__init__() self.lastmsg = "" self.lastpackage = None # name of last package we looked at self.output = True # for a progress bar self.mark = "=" self.marks = 22 def progress(self, package, action, ti_done, ti_total, ts_done, ts_total): """Output information about an rpm operation. This may include a text progress bar. :param package: the package involved in the event :param action: the type of action that is taking place. Valid values are given by :func:`rpmtrans.TransactionDisplay.action.keys()` :param ti_done: a number representing the amount of work already done in the current transaction :param ti_total: a number representing the total amount of work to be done in the current transaction :param ts_done: the number of the current transaction in transaction set :param ts_total: the total number of transactions in the transaction set """ action_str = dnf.transaction.ACTIONS.get(action) if action_str is None: return wid1 = self._max_action_width() pkgname = ucd(package) self.lastpackage = package if ti_total == 0: percent = 0 else: percent = (ti_donelong(100))//ti_total self._out_progress(ti_done, ti_total, ts_done, ts_total, percent, action_str, pkgname, wid1) def _max_action_width(self): if not hasattr(self, '_max_action_wid_cache'): wid1 = 0 for val in dnf.transaction.ACTIONS.values(): wid_val = exact_width(val) if wid1 < wid_val: wid1 = wid_val self._max_action_wid_cache = wid1 wid1 = self._max_action_wid_cache return wid1 def _out_progress(self, ti_done, ti_total, ts_done, ts_total, percent, process, pkgname, wid1): if self.output and (sys.stdout.isatty() or ti_done == ti_total): (fmt, wid1, wid2) = self._makefmt(percent, ts_done, ts_total, progress=sys.stdout.isatty(), pkgname=pkgname, wid1=wid1) pkgname = ucd(pkgname) msg = fmt % (fill_exact_width(process, wid1, wid1), fill_exact_width(pkgname, wid2, wid2)) if msg != self.lastmsg: dnf.util._terminal_messenger('write_flush', msg, sys.stdout) self.lastmsg = msg if ti_done == ti_total: print(" ") def _makefmt(self, percent, ts_done, ts_total, progress=True, pkgname=None, wid1=15): l = len(str(ts_total)) size = "%s.%s" % (l, l) fmt_done = "%" + size + "s/%" + size + "s" done = fmt_done % (ts_done, ts_total) # This should probably use TerminLine, but we don't want to dep. on # that. So we kind do an ok job by hand ... at least it's dynamic now. if pkgname is None: pnl = 22 else: pnl = exact_width(pkgname) overhead = (2 l) + 2 # Length of done, above overhead += 2 + wid1 +2 # Length of beginning (" " action " :") overhead += 1 # Space between pn and done overhead += 2 # Ends for progress overhead += 1 # Space for end width = self.width if width < overhead: width = overhead # Give up width -= overhead if pnl > width // 2: pnl = width // 2 marks = self.width - (overhead + pnl) width = "%s.%s" % (marks, marks) fmt_bar = "[%-" + width + "s]" # pnl = str(28 + marks + 1) full_pnl = pnl + marks + 1 if progress and percent == 100: # Don't chop pkg name on 100% fmt = "\r %s: %s " + done wid2 = full_pnl elif progress: if marks > 5: bar = fmt_bar % (self.mark * int(marks * (percent / 100.0)), ) else: bar = "" fmt = "\r %s: %s " + bar + " " + done wid2 = pnl elif percent == 100: fmt = " %s: %s " + done wid2 = full_pnl else: if marks > 5: bar = fmt_bar % (self.mark * marks, ) else: bar = "" fmt = " %s: %s " + bar + " " + done wid2 = pnl return fmt, wid1, wid2 def progressbar(current, total, name=None): """Output the current status to the terminal using a simple text progress bar consisting of 50 # marks. :param current: a number representing the amount of work already done :param total: a number representing the total amount of work to be done :param name: a name to label the progress bar with """ mark = '#' if not sys.stdout.isatty(): return if current == 0: percent = 0 else: if total != 0: percent = float(current) / total else: percent = 0 width = dnf.cli.term._term_width() if name is None and current == total: name = '-' end = ' %d/%d' % (current, total) width -= len(end) + 1 if width < 0: width = 0 if name is None: width -= 2 if width < 0: width = 0 hashbar = mark * int(width * percent) output = '\r[%-s]%s' % (width, hashbar, end) elif current == total: # Don't chop name on 100% output = '\r%s%s' % (fill_exact_width(name, width, width), end) else: width -= 4 if width < 0: width = 0 nwid = width // 2 if nwid > exact_width(name): nwid = exact_width(name) width -= nwid hashbar = mark int(width * percent) output = '\r%s: [%-*s]%s' % (fill_exact_width(name, nwid, nwid), width, hashbar, end) if current <= total: dnf.util._terminal_messenger('write', output, sys.stdout) if current == total: dnf.util._terminal_messenger('write', '\n', sys.stdout) dnf.util._terminal_messenger('flush', out=sys.stdout)	87.24 KB	2025-09-22 11:27
📄 progress.py # Copyright (C) 2013-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. from __future__ import unicode_literals from dnf.cli.format import format_number, format_time from dnf.cli.term import _term_width from dnf.pycomp import unicode from time import time import sys import dnf.callback import dnf.util class MultiFileProgressMeter(dnf.callback.DownloadProgress): """Multi-file download progress meter""" STATUS_2_STR = { dnf.callback.STATUS_FAILED: 'FAILED', dnf.callback.STATUS_ALREADY_EXISTS: 'SKIPPED', dnf.callback.STATUS_MIRROR: 'MIRROR', dnf.callback.STATUS_DRPM: 'DRPM', } def __init__(self, fo=sys.stderr, update_period=0.3, tick_period=1.0, rate_average=5.0): """Creates a new progress meter instance update_period -- how often to update the progress bar tick_period -- how fast to cycle through concurrent downloads rate_average -- time constant for average speed calculation """ self.fo = fo self.update_period = update_period self.tick_period = tick_period self.rate_average = rate_average self.unknown_progres = 0 self.total_drpm = 0 self.isatty = sys.stdout.isatty() self.done_drpm = 0 self.done_files = 0 self.done_size = 0 self.active = [] self.state = {} self.last_time = 0 self.last_size = 0 self.rate = None self.total_files = 0 self.total_size = 0 def message(self, msg): dnf.util._terminal_messenger('write_flush', msg, self.fo) def start(self, total_files, total_size, total_drpms=0): self.total_files = total_files self.total_size = total_size self.total_drpm = total_drpms # download state self.done_drpm = 0 self.done_files = 0 self.done_size = 0 self.active = [] self.state = {} # rate averaging self.last_time = 0 self.last_size = 0 self.rate = None def progress(self, payload, done): now = time() text = unicode(payload) total = int(payload.download_size) done = int(done) # update done_size if text not in self.state: self.state[text] = now, 0 self.active.append(text) start, old = self.state[text] self.state[text] = start, done self.done_size += done - old # update screen if enough time has elapsed if now - self.last_time > self.update_period: if total > self.total_size: self.total_size = total self._update(now) def _update(self, now): if self.last_time: delta_time = now - self.last_time delta_size = self.done_size - self.last_size if delta_time > 0 and delta_size > 0: # update the average rate rate = delta_size / delta_time if self.rate is not None: weight = min(delta_time/self.rate_average, 1) rate = rateweight + self.rate(1 - weight) self.rate = rate self.last_time = now self.last_size = self.done_size if not self.isatty: return # pick one of the active downloads text = self.active[int(now/self.tick_period) % len(self.active)] if self.total_files > 1: n = '%d' % (self.done_files + 1) if len(self.active) > 1: n += '-%d' % (self.done_files + len(self.active)) text = '(%s/%d): %s' % (n, self.total_files, text) # average rate, total done size, estimated remaining time if self.rate and self.total_size: time_eta = format_time((self.total_size - self.done_size) / self.rate) else: time_eta = '--:--' msg = ' %5sB/s \| %5sB %9s ETA\r' % ( format_number(self.rate) if self.rate else '--- ', format_number(self.done_size), time_eta) left = _term_width() - len(msg) bl = (left - 7)//2 if bl > 8: # use part of the remaining space for progress bar if self.total_size: pct = self.done_size * 100 // self.total_size n, p = divmod(self.done_size * bl * 2 // self.total_size, 2) bar = '=' * n + '-' * p msg = '%3d%% [%-s]%s' % (pct, bl, bar, msg) left -= bl + 7 else: n = self.unknown_progres - 3 p = 3 n = 0 if n < 0 else n bar = ' ' n + '=' * p msg = ' [%-s]%s' % (bl, bar, msg) left -= bl + 7 self.unknown_progres = self.unknown_progres + 3 if self.unknown_progres + 3 < bl \ else 0 self.message('%-.s%s' % (left, left, text, msg)) def end(self, payload, status, err_msg): start = now = time() text = unicode(payload) size = int(payload.download_size) done = 0 # update state if status == dnf.callback.STATUS_MIRROR: pass elif status == dnf.callback.STATUS_DRPM: self.done_drpm += 1 elif text in self.state: start, done = self.state.pop(text) self.active.remove(text) size -= done self.done_files += 1 self.done_size += size elif status == dnf.callback.STATUS_ALREADY_EXISTS: self.done_files += 1 self.done_size += size if status: # the error message, no trimming if status is dnf.callback.STATUS_DRPM and self.total_drpm > 1: msg = '[%s %d/%d] %s: ' % (self.STATUS_2_STR[status], self.done_drpm, self.total_drpm, text) else: msg = '[%s] %s: ' % (self.STATUS_2_STR[status], text) left = _term_width() - len(msg) - 1 msg = '%s%-s\n' % (msg, left, err_msg) else: if self.total_files > 1: text = '(%d/%d): %s' % (self.done_files, self.total_files, text) # average rate, file size, download time tm = max(now - start, 0.001) msg = ' %5sB/s \| %5sB %9s \n' % ( format_number(float(done) / tm), format_number(done), format_time(tm)) left = _term_width() - len(msg) msg = '%-.s%s' % (left, left, text, msg) self.message(msg) # now there's a blank line. fill it if possible. if self.active: self._update(now)	7.43 KB	2022-09-09 11:21
📄 term.py # Copyright (C) 2013-2014 Red Hat, Inc. # Terminal routines. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals import curses import dnf.pycomp import fcntl import re import struct import sys import termios def _real_term_width(fd=1): """ Get the real terminal width """ try: buf = 'abcdefgh' buf = fcntl.ioctl(fd, termios.TIOCGWINSZ, buf) ret = struct.unpack(b'hhhh', buf)[1] return ret except IOError: return None def _term_width(fd=1): """ Compute terminal width falling to default 80 in case of trouble""" tw = _real_term_width(fd=1) if not tw: return 80 elif tw < 20: return 20 else: return tw class Term(object): """A class to provide some terminal "UI" helpers based on curses.""" # From initial search for "terminfo and python" got: # http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/475116 # ...it's probably not copyrightable, but if so ASPN says: # # Except where otherwise noted, recipes in the Python Cookbook are # published under the Python license. __enabled = True real_columns = property(lambda self: _real_term_width()) columns = property(lambda self: _term_width()) __cap_names = { 'underline' : 'smul', 'reverse' : 'rev', 'normal' : 'sgr0', } __colors = { 'black' : 0, 'blue' : 1, 'green' : 2, 'cyan' : 3, 'red' : 4, 'magenta' : 5, 'yellow' : 6, 'white' : 7 } __ansi_colors = { 'black' : 0, 'red' : 1, 'green' : 2, 'yellow' : 3, 'blue' : 4, 'magenta' : 5, 'cyan' : 6, 'white' : 7 } __ansi_forced_MODE = { 'bold' : '\x1b[1m', 'blink' : '\x1b[5m', 'dim' : '', 'reverse' : '\x1b[7m', 'underline' : '\x1b[4m', 'normal' : '\x1b(B\x1b[m' } __ansi_forced_FG_COLOR = { 'black' : '\x1b[30m', 'red' : '\x1b[31m', 'green' : '\x1b[32m', 'yellow' : '\x1b[33m', 'blue' : '\x1b[34m', 'magenta' : '\x1b[35m', 'cyan' : '\x1b[36m', 'white' : '\x1b[37m' } __ansi_forced_BG_COLOR = { 'black' : '\x1b[40m', 'red' : '\x1b[41m', 'green' : '\x1b[42m', 'yellow' : '\x1b[43m', 'blue' : '\x1b[44m', 'magenta' : '\x1b[45m', 'cyan' : '\x1b[46m', 'white' : '\x1b[47m' } def __forced_init(self): self.MODE = self.__ansi_forced_MODE self.FG_COLOR = self.__ansi_forced_FG_COLOR self.BG_COLOR = self.__ansi_forced_BG_COLOR def reinit(self, term_stream=None, color='auto'): """Reinitializes the :class:`Term`. :param term_stream: the terminal stream that the :class:`Term` should be initialized to use. If term_stream is not given, :attr:`sys.stdout` is used. :param color: when to colorize output. Valid values are 'always', 'auto', and 'never'. 'always' will use ANSI codes to always colorize output, 'auto' will decide whether do colorize depending on the terminal, and 'never' will never colorize. """ self.__enabled = True self.lines = 24 if color == 'always': self.__forced_init() return # Output modes: self.MODE = { 'bold' : '', 'blink' : '', 'dim' : '', 'reverse' : '', 'underline' : '', 'normal' : '' } # Colours self.FG_COLOR = { 'black' : '', 'blue' : '', 'green' : '', 'cyan' : '', 'red' : '', 'magenta' : '', 'yellow' : '', 'white' : '' } self.BG_COLOR = { 'black' : '', 'blue' : '', 'green' : '', 'cyan' : '', 'red' : '', 'magenta' : '', 'yellow' : '', 'white' : '' } if color == 'never': self.__enabled = False return assert color == 'auto' # If the stream isn't a tty, then assume it has no capabilities. if not term_stream: term_stream = sys.stdout if not term_stream.isatty(): self.__enabled = False return # Check the terminal type. If we fail, then assume that the # terminal has no capabilities. try: curses.setupterm(fd=term_stream.fileno()) except Exception: self.__enabled = False return self._ctigetstr = curses.tigetstr self.lines = curses.tigetnum('lines') # Look up string capabilities. for cap_name in self.MODE: mode = cap_name if cap_name in self.__cap_names: cap_name = self.__cap_names[cap_name] self.MODE[mode] = self._tigetstr(cap_name) # Colors set_fg = self._tigetstr('setf').encode('utf-8') if set_fg: for (color, val) in self.__colors.items(): self.FG_COLOR[color] = curses.tparm(set_fg, val).decode() or '' set_fg_ansi = self._tigetstr('setaf').encode('utf-8') if set_fg_ansi: for (color, val) in self.__ansi_colors.items(): fg_color = curses.tparm(set_fg_ansi, val).decode() or '' self.FG_COLOR[color] = fg_color set_bg = self._tigetstr('setb').encode('utf-8') if set_bg: for (color, val) in self.__colors.items(): self.BG_COLOR[color] = curses.tparm(set_bg, val).decode() or '' set_bg_ansi = self._tigetstr('setab').encode('utf-8') if set_bg_ansi: for (color, val) in self.__ansi_colors.items(): bg_color = curses.tparm(set_bg_ansi, val).decode() or '' self.BG_COLOR[color] = bg_color def __init__(self, term_stream=None, color='auto'): self.reinit(term_stream, color) def _tigetstr(self, cap_name): # String capabilities can include "delays" of the form "$<2>". # For any modern terminal, we should be able to just ignore # these, so strip them out. cap = self._ctigetstr(cap_name) or '' if dnf.pycomp.is_py3bytes(cap): cap = cap.decode() return re.sub(r'\$<\d+>[/]?', '', cap) def color(self, color, s): """Colorize string with color""" return (self.MODE[color] + str(s) + self.MODE['normal']) def bold(self, s): """Make string bold.""" return self.color('bold', s) def sub(self, haystack, beg, end, needles, escape=None, ignore_case=False): """Search the string haystack* for all occurrences of any string in the list needles. Prefix each occurrence with beg, and postfix each occurrence with end, then return the modified string. For example:: >>> yt = Term() >>> yt.sub('spam and eggs', 'x', 'z', ['and']) 'spam xandz eggs' This is particularly useful for emphasizing certain words in output: for example, calling :func:`sub` with beg = MODE['bold'] and end = MODE['normal'] will return a string that when printed to the terminal will appear to be haystack with each occurrence of the strings in needles in bold face. Note, however, that the :func:`sub_mode`, :func:`sub_bold`, :func:`sub_fg`, and :func:`sub_bg` methods provide convenient ways to access this same emphasizing functionality. :param haystack: the string to be modified :param beg: the string to be prefixed onto matches :param end: the string to be postfixed onto matches :param needles: a list of strings to add the prefixes and postfixes to :param escape: a function that accepts a string and returns the same string with problematic characters escaped. By default, :func:`re.escape` is used. :param ignore_case: whether case should be ignored when searching for matches :return: haystack with beg prefixing, and end postfixing, occurrences of the strings in needles """ if not self.__enabled: return haystack if not escape: escape = re.escape render = lambda match: beg + match.group() + end for needle in needles: pat = escape(needle) flags = re.I if ignore_case else 0 haystack = re.sub(pat, render, haystack, flags=flags) return haystack def sub_norm(self, haystack, beg, needles, *kwds): """Search the string haystack* for all occurrences of any string in the list needles. Prefix each occurrence with beg, and postfix each occurrence with self.MODE['normal'], then return the modified string. If beg is an ANSI escape code, such as given by self.MODE['bold'], this method will return haystack with the formatting given by the code only applied to the strings in needles. :param haystack: the string to be modified :param beg: the string to be prefixed onto matches :param end: the string to be postfixed onto matches :param needles: a list of strings to add the prefixes and postfixes to :return: haystack with beg prefixing, and self.MODE['normal'] postfixing, occurrences of the strings in needles """ return self.sub(haystack, beg, self.MODE['normal'], needles, kwds) def sub_mode(self, haystack, mode, needles, kwds): """Search the string haystack for all occurrences of any string in the list needles. Prefix each occurrence with self.MODE[mode], and postfix each occurrence with self.MODE['normal'], then return the modified string. This will return a string that when printed to the terminal will appear to be haystack with each occurrence of the strings in needles in the given mode. :param haystack: the string to be modified :param mode: the mode to set the matches to be in. Valid values are given by self.MODE.keys(). :param needles: a list of strings to add the prefixes and postfixes to :return: haystack with self.MODE[mode] prefixing, and self.MODE['normal'] postfixing, occurrences of the strings in needles """ return self.sub_norm(haystack, self.MODE[mode], needles, kwds) def sub_bold(self, haystack, needles, kwds): """Search the string haystack for all occurrences of any string in the list needles. Prefix each occurrence with self.MODE['bold'], and postfix each occurrence with self.MODE['normal'], then return the modified string. This will return a string that when printed to the terminal will appear to be haystack with each occurrence of the strings in needles in bold face. :param haystack: the string to be modified :param needles: a list of strings to add the prefixes and postfixes to :return: haystack with self.MODE['bold'] prefixing, and self.MODE['normal'] postfixing, occurrences of the strings in needles """ return self.sub_mode(haystack, 'bold', needles, kwds) def sub_fg(self, haystack, color, needles, kwds): """Search the string haystack for all occurrences of any string in the list needles. Prefix each occurrence with self.FG_COLOR[color], and postfix each occurrence with self.MODE['normal'], then return the modified string. This will return a string that when printed to the terminal will appear to be haystack with each occurrence of the strings in needles in the given color. :param haystack: the string to be modified :param color: the color to set the matches to be in. Valid values are given by self.FG_COLOR.keys(). :param needles: a list of strings to add the prefixes and postfixes to :return: haystack with self.FG_COLOR[color] prefixing, and self.MODE['normal'] postfixing, occurrences of the strings in needles """ return self.sub_norm(haystack, self.FG_COLOR[color], needles, kwds) def sub_bg(self, haystack, color, needles, kwds): """Search the string haystack for all occurrences of any string in the list needles. Prefix each occurrence with self.BG_COLOR[color], and postfix each occurrence with self.MODE['normal'], then return the modified string. This will return a string that when printed to the terminal will appear to be haystack with each occurrence of the strings in needles highlighted in the given background color. :param haystack: the string to be modified :param color: the background color to set the matches to be in. Valid values are given by self.BG_COLOR.keys(). :param needles: a list of strings to add the prefixes and postfixes to :return: haystack with self.BG_COLOR[color] prefixing, and self.MODE['normal'] postfixing, occurrences of the strings in needles """ return self.sub_norm(haystack, self.BG_COLOR[color], needles, **kwds)	14.34 KB	2022-09-09 11:21
📄 utils.py # Copyright (C) 2016 Red Hat, Inc. # # 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 Library 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. """Various utility functions, and a utility class.""" from __future__ import absolute_import from __future__ import unicode_literals from dnf.cli.format import format_number from dnf.i18n import _ import dnf.util import logging import os import time _USER_HZ = os.sysconf(os.sysconf_names['SC_CLK_TCK']) logger = logging.getLogger('dnf') def jiffies_to_seconds(jiffies): """Convert a number of jiffies to seconds. How many jiffies are in a second is system-dependent, e.g. 100 jiffies = 1 second is common. :param jiffies: a number of jiffies :return: the equivalent number of seconds """ return int(jiffies) / _USER_HZ def seconds_to_ui_time(seconds): """Return a human-readable string representation of the length of a time interval given in seconds. :param seconds: the length of the time interval in seconds :return: a human-readable string representation of the length of the time interval """ if seconds >= 60 * 60 * 24: return "%d day(s) %d:%02d:%02d" % (seconds // (60 * 60 * 24), (seconds // (60 * 60)) % 24, (seconds // 60) % 60, seconds % 60) if seconds >= 60 * 60: return "%d:%02d:%02d" % (seconds // (60 * 60), (seconds // 60) % 60, (seconds % 60)) return "%02d:%02d" % ((seconds // 60), seconds % 60) def get_process_info(pid): """Return info dict about a process.""" pid = int(pid) # Maybe true if /proc isn't mounted, or not Linux ... or something. if (not os.path.exists("/proc/%d/status" % pid) or not os.path.exists("/proc/stat") or not os.path.exists("/proc/%d/stat" % pid)): return ps = {} with open("/proc/%d/status" % pid) as status_file: for line in status_file: if line[-1] != '\n': continue data = line[:-1].split(':\t', 1) if len(data) < 2: continue data[1] = dnf.util.rtrim(data[1], ' kB') ps[data[0].strip().lower()] = data[1].strip() if 'vmrss' not in ps: return if 'vmsize' not in ps: return boot_time = None with open("/proc/stat") as stat_file: for line in stat_file: if line.startswith("btime "): boot_time = int(line[len("btime "):-1]) break if boot_time is None: return with open('/proc/%d/stat' % pid) as stat_file: ps_stat = stat_file.read().split() ps['start_time'] = boot_time + jiffies_to_seconds(ps_stat[21]) ps['state'] = {'R' : _('Running'), 'S' : _('Sleeping'), 'D' : _('Uninterruptible'), 'Z' : _('Zombie'), 'T' : _('Traced/Stopped') }.get(ps_stat[2], _('Unknown')) return ps def show_lock_owner(pid): """Output information about process holding a lock.""" ps = get_process_info(pid) if not ps: msg = _('Unable to find information about the locking process (PID %d)') logger.critical(msg, pid) return msg = _(' The application with PID %d is: %s') % (pid, ps['name']) logger.critical("%s", msg) logger.critical(_(" Memory : %5s RSS (%5sB VSZ)"), format_number(int(ps['vmrss']) * 1024), format_number(int(ps['vmsize']) * 1024)) ago = seconds_to_ui_time(int(time.time()) - ps['start_time']) logger.critical(_(' Started: %s - %s ago'), dnf.util.normalize_time(ps['start_time']), ago) logger.critical(_(' State : %s'), ps['state']) return	4.41 KB	2022-09-09 11:21

Terminal