path: root/PackageOwners.py

#!/usr/bin/python
# -*- mode: Python; indent-tabs-mode: nil; -*-
#
# 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., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.

import commands
import errno
import os, sys, time
import shutil
import tempfile
from urllib import FancyURLopener


class AccountsURLopener(FancyURLopener):
    """Subclass of urllib.FancyURLopener to allow passing http basic auth info"""
    def __init__(self, username, password):
        FancyURLopener.__init__(self)
        self.username = username
        self.password = password

    def prompt_user_passwd(self, host, realm):
        return (self.username, self.password)


class PackageOwners:
    """interface to Fedora package owners list (and Fedora Extras owners/owners.list file)"""

    def __init__(self):
        self.dict = {}
        self.how = 'unknown'


    def FromURL(self, retries=3, retrysecs=300, url='https://admin.fedoraproject.org/pkgdb/acls/bugzilla?tg_format=plain',
                pkgdb=True, username=None, password=None):
        # old url='http://cvs.fedora.redhat.com/viewcvs/*checkout*/owners/owners.list?root=extras'
        if pkgdb:
            self.how = 'pkgdb'
        else:
            self.how = 'url'
        self.url = url
        self.retries = retries
        self.retrysecs = retrysecs
        self.username = username
        self.password = password
        return self._refresh()


    def FromCVS(self, retries=3, retrysecs=300, command='LC_ALL=C CVS_RSH=ssh cvs -f -d :pserver:anonymous@cvs.fedora.redhat.com:/cvs/extras co owners', workdir=''):
        self.how = 'cvs'
        self.command = command
        self.retries = retries
        self.retrysecs = retrysecs
        self.workdir = workdir
        self.ownersfile = os.path.join('owners', 'owners.list')
        self.cwdstack = []
        return self._refresh()


    def __getitem__(self,rpmname):
        """return e-mail address from initialowner field"""
        return self.GetOwner(rpmname)


    def GetOwner(self,rpmname):
        """return e-mail address from initialowner field"""
        try:
            r = self.dict[rpmname]['mailto']
        except KeyError:
            r = ''
        return r


    def GetOwners(self,rpmname):
        """return list of e-mail addresses from initialowner+initialcclist fields"""
        r = self.GetCoOwnerList(rpmname)
        r2 = self.GetOwner(rpmname)
        if len(r2):
            r.append(r2)
        return r


    def GetCoOwnerList(self,rpmname):
        """return list of e-mail addresses from initialcclist field"""
        try:
            r = self.dict[rpmname]['cc']
        except KeyError:
            r = []
        return r


    def _enterworkdir(self):
        self.cwdstack.append( os.getcwd() )
        if self.workdir != '':
            os.chdir(self.workdir)


    def _leaveworkdir(self):
        if len(self.cwdstack):
            os.chdir( self.cwdstack.pop() )


    def _refresh(self):
        self.dict = {}  # map package name to email address, dict[name]
        return self._download()


    def _parse(self,ownerslist):
        for line in ownerslist:
            if line.startswith('#') or line.isspace():
                continue
            try:
                (repo,pkgname,summary,emails,qacontact,cc) = line.rstrip().split('|')
                # The PkgDb includes repo's other than Fedora (Fedora EPEL,
                # Fedora OLPC, and Red Hat Linux, for example).  Skip them.
                if repo != 'Fedora':
                    continue
                def fixaddr(a):
                    # Old Fedora CVS owners.list contains e-mail addresses.
                    # PkgDb plain output contains usernames only.
                    if not self.how == 'pkgdb':
                        return a
                    if not self.usermap.has_key(a):
                        return a
                    return self.usermap[a]

                addrs = []
                mailto = '' # primary pkg owner
                if len(emails):
                    if emails.find(',')>=0:
                        (addrs) = emails.split(',')
                        mailto = addrs[0]
                        addrs = addrs[1:]
                    else:
                        mailto = emails
                    mailto = fixaddr(mailto)

                ccaddrs = []
                if len(cc):
                    (ccaddrs) = cc.split(',')
                addrs += ccaddrs
                addrs = map(lambda a: fixaddr(a), addrs)

                self.dict[pkgname] = {
                    'mailto' : mailto,
                    'cc' : addrs
                    }
            except:
                print 'ERROR: owners.list is broken'
                print line


    def _downloadfromcvs(self):
        self._enterworkdir()
        # Dumb caching. Check that file exists and is "quite recent".
        cached = False
        try:
            fstats = os.stat(self.ownersfile)
            if ( fstats.st_size and
                 ((time.time() - fstats.st_ctime) < 3600*2) ):
                cached = True
        except OSError:
            pass

        if not cached:
            # Remove 'owners' directory contents, if it exists.
            for root, dirs, files in os.walk( 'owners', topdown=False ):
                for fname in files:
                    os.remove(os.path.join( root, fname ))
                for dname in dirs:
                    os.rmdir(os.path.join( root, dname ))
            # Retry CVS checkout a few times.
            for count in range(self.retries):
                (rc, rv) = commands.getstatusoutput(self.command)
                if not rc:
                    break
                print rv
                time.sleep(self.retrysecs)
            if rc:
                # TODO: customise behaviour on error conditions
                self._leaveworkdir()
                return False

        try:
            f = file( self.ownersfile )
        except IOError, (err, strerr):
            print 'ERROR: %s' % strerr
            # TODO: customise behaviour on error conditions
            self._leaveworkdir()
            return err
        ownerslist = f.readlines()
        f.close()
        self._parse(ownerslist)
        self._leaveworkdir()
        return True


    def _getlinesfromurl(self,url):
        err = 0
        strerr = ''
        # Retry URL download a few times.
        for count in range(self.retries):
            if count != 0:
                time.sleep(self.retrysecs)
            try:
                opener = AccountsURLopener(self.username, self.password)
                f = opener.open(url)
                rc = 0
                if 'www-authenticate' in f.headers:
                    rc = 1
                    strerr = 'Authentication is required to access %s' % url
                break
            except IOError, (_err, _strerr):
                rc = 1
                print url
                print _strerr
                (err,strerr) = (_err,_strerr)
        if rc:
            raise IOError, (err, strerr)
        else:
            l = f.readlines()
            f.close()
            return l


    def _downloadfromurl(self):
        self._parse(self._getlinesfromurl(self.url))
        return True


    def _downloadfrompkgdb(self):
        fasdump = self._getlinesfromurl('https://admin.fedoraproject.org/accounts/dump-group.cgi')
        self.usermap = {}
        for line in fasdump:
            fields = line.split(',')
            user = fields[0]
            addr = fields[1]
            if (addr.find('@') < 0):  # unexpected, no addr
                print 'No email in:', line
                raise Exception
            self.usermap[user] = addr
        self._parse(self._getlinesfromurl(self.url))
        return True


    def _download(self):
        if self.how == 'url':
            return self._downloadfromurl()
        elif self.how == 'pkgdb':
            return self._downloadfrompkgdb()
        elif self.how == 'cvs':
            return self._downloadfromcvs()
        else:
            self.__init__()
            return False