Server : Apache
System : Linux server1.cgrithy.com 3.10.0-1160.95.1.el7.x86_64 #1 SMP Mon Jul 24 13:59:37 UTC 2023 x86_64
User : nobody ( 99)
PHP Version : 8.1.23
Disable Function : NONE
Directory : /usr/bin/
|
Server : Apache System : Linux server1.cgrithy.com 3.10.0-1160.95.1.el7.x86_64 #1 SMP Mon Jul 24 13:59:37 UTC 2023 x86_64 User : nobody ( 99) PHP Version : 8.1.23 Disable Function : NONE Directory : /usr/bin/ |
#!/usr/bin/python
# 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.
# (c) Copyright Seth Vidal 2004
# need hdropen, dir traversing, version comparison, and getopt (eventually)
# this should take a dir, traverse it - build a dict of foo[(name, arch)] = [/path/to/file/that/is/highest, /path/to/equalfile]
import os
import sys
import rpm
import fnmatch
import string
import rpmUtils
from yum import misc
from optparse import OptionParser
def errorprint(stuff):
print >> sys.stderr, stuff
def getFileList(path, ext, filelist):
"""Return all files in path matching ext, store them in filelist, recurse dirs
return list object"""
extlen = len(ext)
try:
dir_list = os.listdir(path)
except OSError, e:
errorprint('Error accessing directory %s, %s' % (path, str(e)))
return []
for d in dir_list:
if os.path.isdir(path + '/' + d):
filelist = getFileList(path + '/' + d, ext, filelist)
else:
if string.lower(d[-extlen:]) == '%s' % (ext):
newpath = os.path.normpath(path + '/' + d)
filelist.append(newpath)
return filelist
def trimRpms(rpms, excludeGlobs):
# print 'Pre-Trim Len: %d' % len(rpms)
badrpms = []
for fn in rpms:
for glob in excludeGlobs:
if fnmatch.fnmatch(fn, glob):
# print 'excluded: %s' % fn
if fn not in badrpms:
badrpms.append(fn)
for fn in badrpms:
if fn in rpms:
rpms.remove(fn)
# print 'Post-Trim Len: %d' % len(rpms)
return rpms
def parseargs(args):
usage = """
repomanage: manage a directory of rpm packages. returns lists of newest
or oldest packages in a directory for easy piping to xargs
or similar programs.
repomanage [--old] [--new] path.
"""
parser = OptionParser(usage=usage)
# new is only used to make sure that the user is not trying to get both
# new and old, after this old and not old will be used.
# (default = not old = new)
parser.add_option("-o", "--old", default=False, action="store_true",
help='print the older packages')
parser.add_option("-n", "--new", default=False, action="store_true",
help='print the newest packages')
parser.add_option("-s", "--space", default=False, action="store_true",
help='space separated output, not newline')
parser.add_option("-k", "--keep", default=1, dest='keep', action="store",
help='newest N packages to keep - defaults to 1')
parser.add_option("-c", "--nocheck", default=0, action="store_true",
help='do not check package payload signatures/digests')
(opts, args)= parser.parse_args()
if opts.new and opts.old:
errorprint('\nPass either --old or --new, not both!\n')
print parser.format_help()
sys.exit(1)
if len(args) > 1:
errorprint('Error: Only one directory allowed per run.')
print parser.format_help()
sys.exit(1)
if len(args) < 1:
errorprint('Error: Must specify a directory to index.')
print parser.format_help()
sys.exit(1)
return (opts, args)
def main(args):
(options, args) = parseargs(args)
mydir = args[0]
rpmList = []
rpmList = getFileList(mydir, '.rpm', rpmList)
verfile = {}
pkgdict = {} # hold all of them - put them in (n,a) = [(e,v,r),(e1,v1,r1)]
keepnum = int(options.keep)*(-1) # the number of items to keep
if len(rpmList) == 0:
errorprint('No files to process')
sys.exit(1)
ts = rpm.TransactionSet()
if options.nocheck:
ts.setVSFlags(~(rpm._RPMVSF_NOPAYLOAD))
else:
ts.setVSFlags(~(rpm.RPMVSF_NOMD5|rpm.RPMVSF_NEEDPAYLOAD))
for pkg in rpmList:
try:
hdr = rpmUtils.miscutils.hdrFromPackage(ts, pkg)
except rpmUtils.RpmUtilsError, e:
msg = "Error opening pkg %s: %s" % (pkg, str(e))
errorprint(msg)
continue
pkgtuple = rpmUtils.miscutils.pkgTupleFromHeader(hdr)
(n,a,e,v,r) = pkgtuple
del hdr
if (n,a) not in pkgdict:
pkgdict[(n,a)] = []
pkgdict[(n,a)].append((e,v,r))
if pkgtuple not in verfile:
verfile[pkgtuple] = []
verfile[pkgtuple].append(pkg)
for natup in pkgdict.keys():
evrlist = pkgdict[natup]
if len(evrlist) > 1:
evrlist = misc.unique(evrlist)
evrlist.sort(rpmUtils.miscutils.compareEVR)
pkgdict[natup] = evrlist
del ts
# now we have our dicts - we can return whatever by iterating over them
outputpackages = []
#if new
if not options.old:
for (n,a) in pkgdict.keys():
evrlist = pkgdict[(n,a)]
if len(evrlist) < abs(keepnum):
newevrs = evrlist
else:
newevrs = evrlist[keepnum:]
for (e,v,r) in newevrs:
for pkg in verfile[(n,a,e,v,r)]:
outputpackages.append(pkg)
if options.old:
for (n,a) in pkgdict.keys():
evrlist = pkgdict[(n,a)]
if len(evrlist) < abs(keepnum):
continue
oldevrs = evrlist[:keepnum]
for (e,v,r) in oldevrs:
for pkg in verfile[(n,a,e,v,r)]:
outputpackages.append(pkg)
outputpackages.sort()
for pkg in outputpackages:
if options.space:
print '%s' % pkg,
else:
print pkg
def usage():
print """
repomanage [--old] [--new] path
-o --old - print the older packages
-n --new - print the newest packages
-s --space - space separated output, not newline
-k --keep - newest N packages to keep - defaults to 1
-c --nocheck - do not check package payload signatures/digests
-h --help - duh
By default it will output the full path to the newest packages in the path.
"""
if __name__ == "__main__":
if len(sys.argv) < 1:
usage()
sys.exit(1)
else:
main(sys.argv[1:])