| Server IP : 66.29.132.122 / Your IP : 13.59.72.254 Web Server : LiteSpeed System : Linux business142.web-hosting.com 4.18.0-553.lve.el8.x86_64 #1 SMP Mon May 27 15:27:34 UTC 2024 x86_64 User : admazpex ( 531) PHP Version : 7.2.34 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/proc/self/root/proc/thread-self/root/proc/thread-self/root/proc/self/root/proc/thread-self/root/proc/self/root/proc/self/root/usr/lib/python3.6/site-packages/dnf/cli/ |
Upload File : |
# 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(pkg.name for pkg in packages)
print('\n' + _('Group: %s') % 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(group.name for group in groups)
print(_('Environment Group: %s') % 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):
name = group.getName()
return (name if name else _("<name-unset>"), "", "", "", "", "", "")
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 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.sack._rpmdb_version()
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)
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_done*long(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)