rpmclique.py

#!/usr/bin/python3

import rpm
#from rpmUtils import miscutils

ts = rpm.TransactionSet('/')

def U(b):
    return str(b)

def identifier(h):
    return U(h.name) + '-' + U(h.version) + '-' + U(h.release)

class Clique:
    def __init__(self, first):
        self.packages = set()
        self.packages.add(first)

    def takeover(self, victim):
        self.packages |= victim.packages
        for p in victim.packages:
            p.clique = self

    def merge(self, other):
        if len(self.packages) >= len(other.packages):
            self.takeover(other)
        else:
            other.takeover(self)

    def children(self):
        for p in self.packages:
            for pp in p.children:
                yield pp.clique

    def unique_children(self):
        return set(self.children())

    def is_parent(self):
        for c in self.children():
            if c != self:
                return True
        return False

    def name(self):
        l = len(self.packages)
        n = self.packages.__iter__().next().short_name
        if l == 1:
            return n
        else:
            return n + " + " + str(l-1)

class Package:
    def __init__(self, n, s):
        self.name = n
        self.short_name = s
        self.children = set()
        self.clique = Clique(self)
        # 0 = not yet processed.
        # 1 = being processed.
        # 2 = done.
        self.state = 0

    packages = { }
    requires = { }

    @classmethod
    def get(self, h):
        name = identifier(h)
        if not name in self.packages:
            self.packages[name] = Package(name, U(h.name) + '.' + U(h.arch))
        return self.packages[name]

    def update(self, h):
        for r in h[rpm.RPMTAG_PROVIDES]:
            if r in self.requires:
                self.children |= self.requires[r]
        for r in h[rpm.RPMTAG_FILENAMES]:
            if r in self.requires:
                self.children |= self.requires[r]

def ignore(h):
    return h.name == 'gpg-pubkey' or h.arch.endswith('86') or \
        h.name.endswith('-devel') or h.name.endswith('-debuginfo')

# First build the package list indexed by requires...
for h in ts.dbMatch():
    if ignore(h):
        continue
    p = Package.get(h)
    for r in h[rpm.RPMTAG_REQUIRENAME]:
        Package.requires.setdefault(r, set()).add(p)

for h in ts.dbMatch():
    if ignore(h):
        continue
    Package.get(h).update(h)

# Go through each package, recursively following children, and collapsing
# cycles.

def process(p, stack):
    if p.state == 1:
        # We have a cycle.
        for pp in reversed(stack):
            if pp == p:
                break
            p.clique.merge(pp.clique)
        else:
            abort()
        return
    assert p.state == 0
    stack.append(p)
    p.state = 1
    for pp in p.children:
        if pp.state != 2:
            process(pp, stack)
    assert p.state == 1
    p.state = 2
    o = stack.pop()
    assert o == p

for p in Package.packages.values():
    if p.state != 2:
        process(p, [])

allcliques = set(p.clique for p in Package.packages.values())

print("#packages =", len(Package.packages))
print("#cliques =", len(allcliques))

terminal = [ c for c in allcliques if not c.is_parent() ]

print("#terminal = ", len(terminal))

#for c in terminal:
#    print " ".join(p.short_name for p in c.packages)

for c in allcliques:
    if not c.is_parent():
        print(" ".join(p.short_name for p in c.packages))