~stefanor/+junk/ubuntu-sponsorship-miner

#!/usr/bin/env python

import cgi

import cgitb

import re

import urllib

import genshi.template

import genshi.template.text

import psycopg2

import psycopg2.extensions

URL = 'ubuntu-sponsorships.cgi'

STATIC_URL = '../ubuntu-sponsorships'

STATIC_PATH = '../htdocs/ubuntu-sponsorships'

DB = 'service=udd'

LIMIT = 1000

class AttrDict(dict):

    """Dictionary with attribute access"""

    def __init__(self, **kwargs):

        for key, value in kwargs.iteritems():

            self[key] = value

    def __getattr__(self, name):

        try:

            return self[name]

        except KeyError, e:

            raise AttributeError(e)

def get_template(name, type_='html'):

    """Load a template"""

    loader = genshi.template.TemplateLoader(STATIC_PATH)

    kwargs = {}

    if type_ == 'text':

        kwargs['cls'] = genshi.template.text.NewTextTemplate

    return loader.load(name, **kwargs)

def locate_sponsorships(sponsor, sponsor_search, sponsoree, sponsoree_search):

    """Query UDD for sponsorhips. Yields rows"""

    # String-substituted into query:

    assert sponsor_search in ('name', 'email')

    assert sponsoree_search in ('name', 'email')

    # Make sure we are dealing with Unicode:

    psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)

    psycopg2.extensions.register_type(psycopg2.extensions.UNICODEARRAY)

    conn = psycopg2.connect(DB)

    conn.set_client_encoding('UTF-8')

    cur = conn.cursor()

    sponsor = sponsor.replace('*', '%')

    if sponsor == '':

        sponsor = '%'

    sponsoree = sponsoree.replace('*', '%')

    if sponsoree == '':

        sponsoree = '%'

    cur.execute("""

        SELECT source, version, date, changed_by, changed_by_name,

               changed_by_email, signed_by, signed_by_name, signed_by_email,

               distribution,

               array_to_string(ARRAY(

                 SELECT bug

                 FROM ubuntu_upload_history_launchpad_closes

                 WHERE source = ubuntu_upload_history.source

                   AND version = ubuntu_upload_history.version

               ), ' ') AS fixed

        FROM ubuntu_upload_history

        WHERE lower(changed_by_%s) LIKE lower(%%s)

          AND lower(signed_by_%s) LIKE lower(%%s)

          AND signed_by_name <> changed_by_name

        ORDER BY date

        LIMIT %i;

    """ % (sponsoree_search, sponsor_search, LIMIT)

    , (sponsoree, sponsor))

    keys = ('source version date changed_by changed_by_name changed_by_email '

            'signed_by signed_by_name signed_by_email distribution fixed'

           ).split()

    for row in cur.fetchall():

        upload = AttrDict(**dict(zip(keys, row)))

        yield upload

def mine_sponsorships(sponsorships):

    """Iterate over sponsorships and guess what we can

    yields sponsorhips

    """

    resyncable_re = re.compile(r'[-\d](fakesync|build)\d+$')

    ubuver_re = re.compile(r'[-\d](\d+)?ubuntu(\d+)$')

    for upload in sponsorships:

        upload['fixed'] = upload['fixed'].split()

        upload['release'] = upload.distribution.split('-')[0]

        info = set()

        if upload.distribution.endswith('-proposed'):

            info.add('sru')

        if upload.distribution.endswith('-security'):

            info.add('security')

        m = resyncable_re.search(upload.version)

        if m is not None:

            if m.group(1) == 'fakesync':

                info.add('sync')

            elif m.group(1) == 'build':

                info.add('rebuild')

            else:

                info.add(m.group(1))

        m = ubuver_re.search(upload.version)

        if m is not None:

            if m.group(2) == '1':

                if m.group(1) == '0':

                    info.add('upgrade')

        upload['info'] = sorted(info)

        # Things we can't determine without changelog / lpapi:

        # sync, merge

        yield upload

def display_sponsorships(sponsor, sponsor_search, sponsoree, sponsoree_search,

                         type_):

    """Return rendering of sponsorships"""

    if type_ == 'html':

        template = get_template('sponsorships.xml')

    else:

        template = get_template('sponsorships.txt', 'text')

    params = {

              'static': STATIC_URL,

              'url': URL,

              'text_url': URL + '?' + urllib.urlencode({

                  'render': 'text',

                  'sponsor': sponsor,

                  'sponsor_search': sponsor_search,

                  'sponsoree': sponsoree,

                  'sponsoree_search': sponsoree_search,

              }),

              'sponsorships': mine_sponsorships(locate_sponsorships(

                  sponsor,

                  sponsor_search,

                  sponsoree,

                  sponsoree_search,

              )),

             }

    return template.generate(**params).render(type_)

def display_form():

    """Return rendered search form"""

    template = get_template('form.xml')

    params = {

              'static': STATIC_URL,

              'url': URL,

             }

    return template.generate(**params).render('html')

def main():

    cgitb.enable()

    form = cgi.FieldStorage()

    body = 'ERROR: No body'

    render = form.getfirst('render', '')

    sponsor = form.getfirst('sponsor', '')

    sponsoree = form.getfirst('sponsoree', '')

    if render and (sponsor or sponsoree):

        sponsor_search = form.getfirst('sponsor_search')

        if sponsor_search not in ('name', 'email'):

            sponsor_search = 'name'

        sponsoree_search = form.getfirst('sponsoree_search')

        if sponsoree_search not in ('name', 'email'):

            sponsoree_search = 'name'

        if render not in ('html', 'text'):

            render = 'html'

        body = display_sponsorships(sponsor, sponsor_search, sponsoree,

                                    sponsoree_search, render)

    else:

        render = 'html'

        body = display_form()

    if render == 'text':

        print "Content-Type: text/plain; charset=utf-8"

    else:

        print "Content-Type: text/html; charset=utf-8"

    print ""

    print body

if __name__ == '__main__':

    main()