~ubuntu-dev/ubuntu-sponsoring/trunk

#!/usr/bin/python

import cgi
import datetime
import email.utils
import json
import operator
import optparse
import os
import re
import sys

from lazr.restfulclient.errors import NotFound, ServerError
import launchpad

# { a : b }
# Look for merge proposals / bug subcriptions for a and output them to b's
# page. The default (None) is to use a.
EXTRA_TEAMS = { # ubuntu-branches is useless until the git clone ubuntu initiative gets going
#    "ubuntu-branches": None,
    "ubuntu-dev": None,
    "ubuntu-packaging-guide-team": None,
    "unity-settings-daemon-team": "ubuntu-desktop",
    "unity-control-center-team": "ubuntu-desktop",
    "ubuntu-security": "ubuntu-security-sponsors",
}

# { a : b }
# Items against packages in b should go to a's page
SETS = {
    "desktop-core": "ubuntu-desktop",
}

SPONSOR_TEAMS = {
    "general":  "ubuntu-sponsors",
    "security": "ubuntu-security-sponsors"
}

BLACKLISTED_REQUESTER = (
    "package-import",
)

# A package that is outside of main and package sets
UNIVERSE_PACKAGE = "mathjax-docs"


def htmlify_name(person, package, devs, distribution):
    if not person:
        return ""
    name = person.name
    if name not in devs:
        return name
    if package == "ubuntu":
        package = UNIVERSE_PACKAGE
    if launchpad.person_can_upload(distribution, person, package):
        return "<b>%s</b>" % name
    return "<i>%s</i>" % name


def classify_name(prefix, name):
    name = name.upper()
    if prefix == "status":
        name = re.sub(" ", "", name)
    else:
        name = re.sub(" ", "_", name)
    return prefix + name


class UTC(datetime.tzinfo):
    """UTC"""

    def utcoffset(self, _):
        return datetime.timedelta(0)

    def tzname(self, _):
        return "UTC"

    def dst(self, _):
        return datetime.timedelta(0)


class SponsoringItem(object):
    _uploader_memo = {}

    def __init__(self, distribution, team, link, link_desc, release="",
                 package="", s_types=set(), last_comment="", status="",
                 importance="", title="", tags=set(), date_created=None,
                 time_in_queue=None, verbose=False):
        self.team = team
        self.link = link
        self.link_desc = link_desc
        self.release = release
        self.s_types = s_types
        self.package = package
        self.last_comment = last_comment
        self.status = status
        self.importance = importance
        self.title = title
        self.tags = tags
        self.distribution = distribution
        self.current_series_name = self.distribution.current_series.name
        self.origin = None
        self.date_created = date_created
        if self.date_created is None:
            self.date_created = datetime.datetime.utcnow()
        self.time_in_queue = time_in_queue
        self.freshness = self.get_freshness()
        self.components = None
        self.sets = None
        if verbose:
            print "Added %s: %s for %s" % (self.package, self.link, self.team)

    def is_sync(self):
        """Returns true if the sponsoring item is a sync request."""
        title = self.title.lower()
        return 'sync' in self.tags or \
               title.startswith("please sync ") or title.startswith("sync ")

    def is_merge(self):
        """Returns true if the sponsoring item is a merge request."""
        title = self.title.lower()
        return 'merge' in self.tags or \
               self.link_desc == "debian-merge" or \
               title.startswith("please merge ") or title.startswith("merge ")

    def is_upgrade(self):
        """Returns true if the item provides a new upstream release."""
        return 'new upstream version' in self.title.lower() or \
               self.link_desc == "new-upstream" or \
               'upgrade' in self.tags \
               or 'upgrade-software-version' in self.tags

    def is_sru(self):
        """Determine whether the item is a stable release update (SRU)."""
        stable_release = [r for r in self.release
                          if r != self.current_series_name]
        return 'sru' in self.tags or stable_release

    def get_uploaders(self):
        """Get the set of all people (including teams) who can upload this
        package. """
        def permission_to_teams(perm):
            out = set()

            # The SETS dictionary contains overrides
            if perm.package_set_name is not None:
                if perm.distro_series_name != self.current_series_name:
                    return out

                toadd = SETS.get(perm.package_set_name, perm.person)
            else:
                toadd = perm.person

            teams = SponsoringItem._uploader_memo.get(
                toadd.name,
                [p.name for p in toadd.participants if p.is_team] +
                [toadd.name])
            SponsoringItem._uploader_memo[toadd.name] = teams

            out.update(teams)

            return out

        output = set()
        archive = self.distribution.main_archive

        seen_packagesets = set()

        for component in self.get_components():
            for perm in archive.getUploadersForComponent(
                    component_name=component):
                output.update(permission_to_teams(perm))

        try:
            for perm in archive.getPackagesetsForSource(
                    sourcepackagename=self.package):
                if perm.package_set_name not in seen_packagesets:
                    s = permission_to_teams(perm)
                    if s:
                        output.update(s)
                        seen_packagesets.add(perm.package_set_name)

            for perm in archive.getUploadersForPackage(
                    source_package_name=self.package):
                output.update(permission_to_teams(perm))
        except NotFound:
            pass

        return output

    def get_components(self):
        if self.components is None:
            self.components = set()
            if len(self.release) == 0:
                # Assume latest series
                current_series = self.distribution.current_series
                package = current_series.getSourcePackage(name=self.package)
                if package is not None:
                    component = package.latest_published_component_name
                    if component is None:
                        # let's assume universe
                        self.components.add('universe')
                    else:
                        self.components.add(component)
            for release in self.release:
                series = self.distribution.getSeries(name_or_version=release)
                package = series.getSourcePackage(name=self.package)
                if package is not None:
                    component = package.latest_published_component_name
                    if component is not None:
                        self.components.add(component)
            if len(self.components) == 0:
                self.components.add('universe')  # let's assume universe
        return self.components

    def get_sets(self, packages):
        if self.sets is None:
            self.sets = set()
            if self.team == "security":
                self.sets.add("security")
            elif self.team == "universe" or self.package.lower() == "ubuntu":
                self.sets.add("unseeded")
            else:
                for packageset in packages.keys():
                    if self.package in packages[packageset]:
                        self.sets.add(packageset)
                if len(self.sets) == 0:
                    self.sets.add("unseeded")
        return self.sets

    def get_time_in_queue(self):
        if self.time_in_queue:
            time_in_queue = self.time_in_queue
        else:
            time_in_queue = datetime.datetime(1970, 1, 1, tzinfo=UTC())
        return time_in_queue

    def get_freshness(self):
        today = datetime.datetime.today().replace(tzinfo=UTC())
        one_week_old = today - datetime.timedelta(weeks=1)
        two_weeks_old = today - datetime.timedelta(weeks=2)
        three_weeks_old = today - datetime.timedelta(weeks=3)
        if self.time_in_queue < three_weeks_old:
            freshness = "oldest"
        elif self.time_in_queue < two_weeks_old:
            freshness = "older"
        elif self.time_in_queue < one_week_old:
            freshness = "old"
        else:
            freshness = ""
        return freshness

    def html(self, devs, packages):
        if self.importance:
            importance = '<span class="%s">%s</span>' % \
                         (classify_name("importance", self.importance),
                          self.importance)
        else:
            importance = ""
        if type(self.status) == list:
            status = self.status[0]
            if len(self.status) > 1:
                votes = [
                    '<span class="%s">%s</span>' %
                    (classify_name("vote", v), v) for v in self.status[1:]]
                status += ", (%s)" % ", ".join(votes)
        elif self.status:
            status = '<span class="%s">%s</span>' % \
                     (classify_name("status", self.status), self.status)
        else:
            status = ""
        if 'upstream' in self.s_types:
            if self.last_comment:
                last_comment = self.last_comment.name
            else:
                last_comment = ""
        else:
            last_comment = htmlify_name(self.last_comment, self.package, devs,
                                        self.distribution)
        package_notes = []
        if self.release:
            package_notes += list(self.release)

        s_types = self.s_types
        if 'upstream' in s_types:
            package_notes.append('upstream')
            s_types.remove('upstream')
        if self.is_sync():
            s_types.add("sync")
        elif self.is_merge():
            s_types.add("merge")
        elif self.is_upgrade():
            s_types.add("upgrade")
        if self.is_sru():
            s_types.add("sru")
        if self.link.startswith("https://launchpad.net/bugs/"):
            t = "bugs"
        else:
            t = "branches"
        self.origin = "%s %s" % (",".join(self.get_sets(packages)), t)

        # workaround for old pylplib, ditch once qa.ubuntu.com is
        # upgraded
        if type(self.date_created) in [str, unicode]:
            date_string = self.date_created.split("T")[0]
        else:
            date_string = self.date_created.strftime("%F")

        time_in_queue = "unknown"
        if self.time_in_queue:
            time_in_queue = self.time_in_queue.strftime("%F")

        return """
<tr class="%(freshness)s">
        <td>%(time_in_queue)s</td>
        <td>%(title)s</td>
        <td><a href='%(link_url)s'>%(link_desc)s</a></td>
        <td>%(types)s</td>
        <td>%(package)s (%(package_note)s)</td>
        <td>%(origin)s</td>
        <td>%(last_commenter)s</td>
        <td>%(status)s</td>
        <td>%(importance)s</td>
        <td>%(components)s</td>
        <td>%(date_created)s</td>
</tr>""" % {
            'freshness': self.freshness,
            'time_in_queue': time_in_queue,
            'title': cgi.escape(self.title),
            'link_url': self.link,
            'link_desc': self.link_desc,
            'types': ', '.join(s_types),
            'package': self.package,
            'package_note': ', '.join(package_notes),
            'origin': self.origin,
            'last_commenter': last_comment,
            'status': status,
            'importance': importance,
            'components': ", ".join(sorted(self.get_components())),
            'date_created': date_string,
        }

    def prepare_json(self):
        mapping = {
                   "title": "description",
                   "link": "link",
                   "package": "source_package",
                   "link_desc": "short_description",
                   "importance": "severity",
                  }
        item_dict = {}
        for key in mapping.keys():
            item_dict[mapping[key]] = self.__dict__[key]
        return item_dict


def sponsors_subscribed_date(bug):
    for activity in reversed(bug.activity_collection):
        subscribed_teams = ["added subscriber Ubuntu Security Sponsors Team",
                            "added subscriber Ubuntu Sponsors Team"]
        if activity.message in subscribed_teams:
            return activity.datechanged
    # If there was no activity subscribing sponsors, they must have
    # been subscribed when the bug was opened
    return bug.date_created


def get_bugs(lp, distribution, team, verbose=False):
    active_releases = [a for a in distribution.series if a.active] + \
                      [distribution]
    tasks = set()
    lp_team = lp.people[SPONSOR_TEAMS[team]]
    for release in active_releases:
        release_name = release.name
        release_tasks = release.searchTasks(bug_subscriber=lp_team,
                                            omit_targeted=False)
        if release != distribution:
            nominated_tasks = distribution.searchTasks(nominated_for=release,
                                                       bug_subscriber=lp_team)
        team_tasks = [task for task in release_tasks] + \
                     [task for task in nominated_tasks]
        for task in team_tasks:
            bug = task.bug
            bug_number = bug.id
            same_bug = [a for a in tasks if a.link_desc == bug_number]
            package = task.bug_target_name.split()[0]
            if same_bug:
                if release_name != "ubuntu":
                    same_bug[0].release.add(release_name)
            else:
                comments = bug.messages
                if comments:
                    last_comment = comments[len(comments)-1].owner
                else:
                    last_comment = bug.owner
                release = set()
                if release_name != "ubuntu":
                    release.add(release_name)
                link = "https://launchpad.net/bugs/%s" % bug_number
                time_in_queue = sponsors_subscribed_date(bug)
                tasks.add(SponsoringItem(distribution=distribution,
                                         team=team,
                                         link=link,
                                         link_desc=bug_number,
                                         release=release,
                                         package=package,
                                         s_types=set(),
                                         last_comment=last_comment,
                                         status=task.status,
                                         importance=task.importance,
                                         title=bug.title,
                                         tags=set(bug.tags),
                                         date_created=bug.date_created,
                                         time_in_queue=time_in_queue,
                                         verbose=verbose))
    return tasks


def get_branches(lp, distribution, team, verbose=False):
    ubuntu = lp.distributions['ubuntu']
    tasks = set()
    lp_team = lp.people[team]
    try:
        proposals = [p for p
                     in lp_team.getRequestedReviews(status="Needs review")
                     if p.registrant.name not in BLACKLISTED_REQUESTER]
    except ServerError:
        try:
            # Dirty workaround for 1033446
            anon_lp = launchpad.lp_login(anon=True)
            lp_team = anon_lp.people[team]
            proposals = [
                p for p
                in lp_team.getRequestedReviews(status="Needs review")
                if p.registrant.name not in BLACKLISTED_REQUESTER]
        except ServerError:
            proposals = []
    for proposal in proposals:
        releases = set()
        s_types = set()
        status = ""
        last_comment = ""
        if proposal.target_branch.sourcepackage:
            package = proposal.target_branch.sourcepackage.name
            distribution = proposal.target_branch.sourcepackage.distribution
            if distribution.name != 'ubuntu':
                continue
            series = proposal.target_branch.sourcepackage.distroseries.name
            if (series != distribution.current_series.name and
                    series != "ubuntu"):
                releases.add(series)
        else:
            try:
                package = proposal.target_branch.project.name
            except NotFound:
                # No project, let's skip this
                continue
            distribution = ubuntu
            s_types.add('upstream')
        tags = set()
        if not [a for a in tasks if a.link == proposal.web_link]:
            try:
                comments = [a for a in proposal.all_comments]
            except NotFound:
                comments = []
            if comments:
                status = ["%s comments" % (len(comments))]
                last_comment = comments[-1].author
            else:
                last_comment = proposal.registrant
            try:
                votes = [a for a in proposal.votes if a.comment is not None]
            except NotFound:
                votes = []
            if votes:
                status += [a.comment.vote for a in votes]
            try:
                title = proposal.source_branch.display_name
            except NotFound:
                title = 'MP source branch display name unavailable'
            try:
                link_desc = proposal.source_branch.name
            except NotFound:
                link_desc = 'MP source branch name unavailable'
            tasks.add(SponsoringItem(distribution=distribution,
                                     team=team,
                                     link=proposal.web_link,
                                     link_desc=link_desc,
                                     release=releases,
                                     s_types=s_types,
                                     package=package,
                                     last_comment=last_comment,
                                     status=status,
                                     importance="",
                                     title=title,
                                     tags=tags,
                                     date_created=proposal.date_created,
                                     time_in_queue=proposal.date_created,
                                     verbose=verbose))
    return tasks


def get_packageset_dict(sponsoring_items, packages):
    packageset_dict = dict()
    for item in sponsoring_items:
        for packageset in item.get_sets(packages):
            if packageset in packageset_dict:
                packageset_dict[packageset] += 1
            else:
                packageset_dict[packageset] = 1
    return packageset_dict


def generate_page(items, devs, packages, other_teams=[], team=None):
    if team is None:
        filename = "index.html"
        json_filename = "sponsoring.json"
    else:
        filename = "%s.html" % team
        json_filename = "%s.json" % team

    date = email.utils.formatdate()

    subst = dict()
    subst["LAST_UPDATED"] = date
    subst["TABLE"] = ""
    for item in sorted(items, key=operator.methodcaller("get_time_in_queue")):
        subst["TABLE"] += item.html(devs, packages)
    subst["TOTAL"] = len(items)
    subst["SYNCS"] = len([x for x in items if x.is_sync()])
    subst["MERGES"] = len([x for x in items if x.is_merge()])
    subst["UPGRADES"] = len([x for x in items if x.is_upgrade()])
    subst["SRUS"] = len([x for x in items if x.is_sru()])
    subst["OTHERS"] = len([x for x in items
                           if not (x.is_sync() or x.is_merge() or
                                   x.is_upgrade() or x.is_sru())])
    subst["MAIN"] = len([x for x in items
                         if "main" in x.get_components()])
    subst["RESTRICTED"] = len([x for x in items
                               if "restricted" in x.get_components()])
    subst["UNIVERSE"] = len([x for x in items
                             if "universe" in x.get_components()])
    subst["MULTIVERSE"] = len([x for x in items
                               if "multiverse" in x.get_components()])
    subst["PACKAGE_SETS"] = ""
    packageset_dict = get_packageset_dict(items, packages)
    for packageset in sorted(packageset_dict):
        subst["PACKAGE_SETS"] += "	<li>%s: %i</li>\n" % \
                                 (packageset, packageset_dict[packageset])
    subst["OTHER_TEAMS"] = ""
    for other_team in other_teams:
        subst["OTHER_TEAMS"] += '	<li><a href="%s">%s</a></li>\n' % \
                                 ("%s.html" % other_team, other_team)

    html = open("template.html").read()
    for pattern, substitution in subst.iteritems():
        html = re.sub("@" + pattern + "@", unicode(substitution), html)

    if os.path.exists(filename):
        os.remove(filename)
    f = open(filename, "w")
    f.write(html.encode("utf-8"))
    f.close()

    json_items = []
    for item in items:
        json_items += [item.prepare_json()]
    if os.path.exists(json_filename):
        os.remove(json_filename)
    f = open(json_filename, "w")
    f.write(json.dumps(json_items))
    f.close()


def main():
    script_name = os.path.basename(sys.argv[0])
    usage = "%s [options]" % (script_name)
    parser = optparse.OptionParser(usage=usage)
    parser.add_option("-v", "--verbose", help="print more information",
                      dest="verbose", action="store_true", default=False)
    options = parser.parse_args()[0]

    # We have to login authenticated for archive.getPermissionsForPerson()
    # to return any permissions:
    lp = launchpad.lp_login()
    distribution = lp.distributions['ubuntu']
    sponsoring_items = dict()
    # 1. Get all sponsor items
    # 1a. Get bugs and branches for SPONSOR_TEAMS
    # 1b. Get branches for all uploading teams
    # 2. Go over the items from 1a, find out who can upload (use devel if
    #    necessary), output to that page
    all_uploaders = set([
        perm.person.name for perm
        in distribution.main_archive.getAllPermissions()
        if (perm.person.is_team and
            perm.permission == "Archive Upload Rights")]) \
        | set(EXTRA_TEAMS.keys())
    for team in SPONSOR_TEAMS:
        sponsoring_items[team] = get_bugs(
            lp, distribution, team, options.verbose)
        branches = get_branches(lp, distribution, team, options.verbose)
        sponsoring_items[team] = \
            sponsoring_items.get(team, set()) | branches
    for team in all_uploaders:
        output_team = EXTRA_TEAMS.get(team, SPONSOR_TEAMS["general"]) or team
        branches = get_branches(lp, distribution, team, options.verbose)
        sponsoring_items[output_team] = \
            sponsoring_items.get(output_team, set()) | branches

    all_sponsoring_items = list(set.union(*sponsoring_items.values()))

    packages = launchpad.get_packagesets(lp, distribution)

    # reverse the SPONSOR_TEAMS map so we can go back to page from team name
    reversed_sponsors = dict(zip(SPONSOR_TEAMS.values(), SPONSOR_TEAMS.keys()))

    for item in all_sponsoring_items:
        uploaders = item.get_uploaders() & all_uploaders
        if options.verbose:
            print "Uploaders for %s: %s" % (item.package, ", ".join(uploaders))
        for uploader in uploaders:
            output_team = reversed_sponsors.get(uploader, uploader)
            output_set = sponsoring_items.get(output_team, set())
            output_set.add(item)
            sponsoring_items[output_team] = output_set

    devs = [a.name for a in lp.people["ubuntu-dev"].participants]

    generate_page(
        all_sponsoring_items, devs, packages,
        [t for t in sponsoring_items.keys()
         if len(sponsoring_items[t]) > 0])

    for team, items in sponsoring_items.iteritems():
        if len(items) > 0:
            generate_page(
                items, devs, packages,
                [t for t in sponsoring_items.keys()
                 if t != team and len(sponsoring_items[t]) > 0], team)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nUser abort.")
        sys.exit(2)