~mabac/launchpad-work-items-tracker/better-error-contacts

#!/usr/bin/python
# -*- coding: utf-8
#
# Create HTML reports from a work item database.
#
# Copyright (C) 2010, 2011 Canonical Ltd.
# License: GPL-3

import optparse

from report_tools import escape_url
import report_tools
from roadmap_health import (
    card_health_checks,
)
from lpworkitems.models import (
    ROADMAP_STATUSES_MAP,
    ROADMAP_ORDERED_STATUSES,
)


class WorkitemTarget(object):

    def __init__(self, todo=0, blocked=0, inprogress=0, postponed=0,
                 done=0):
        self.todo = todo
        self.blocked = blocked
        self.inprogress = inprogress
        self.postponed = postponed
        self.done = done

    @property
    def percent_complete(self):
        completed = self.postponed + self.done
        if self.total < 1:
            return 0
        return int((float(completed)/self.total)*100 + 0.5)

    @property
    def total(self):
        total = self.postponed + self.done + self.todo + self.blocked + self.inprogress
        return total


class Blueprint(WorkitemTarget):

    def __init__(self, name, url, complexity=0,
                 todo=0, blocked=0, inprogress=0, postponed=0,
                 done=0, implementation=None, priority=None,
                 status=None):
        super(Blueprint, self).__init__(
            todo=todo, blocked=blocked, inprogress=inprogress,
            postponed=postponed, done=done)
        self.name = name
        self.url = url
        self.complexity = complexity
        self.implementation = implementation
        self.priority = priority
        self.status = status


class Assignee(WorkitemTarget):

    def __init__(self, name, url, complexity=0, todo_wis=[],
                 blocked_wis=[], inprogress_wis=[], postponed_wis=[],
                 done_wis=[]):
        super(Assignee, self).__init__(
            todo=len(todo_wis), blocked=len(blocked_wis),
            inprogress=len(inprogress_wis), postponed=len(postponed_wis),
            done=len(done_wis))
        self.name = name
        self.url = url
        self.complexity = complexity
        self.todo_wis = todo_wis
        self.blocked_wis = blocked_wis
        self.inprogress_wis = inprogress_wis
        self.postponed_wis = postponed_wis
        self.done_wis = done_wis


class BlueprintGroup(Blueprint):

    def __init__(self, name, url, area, complexity=0, todo=0, blocked=0,
                 inprogress=0, postponed=0, done=0, implementation=None,
                 priority=None, status=None):
        super(BlueprintGroup, self).__init__(name, url,
            complexity=complexity, todo=todo, blocked=blocked,
            inprogress=inprogress, postponed=postponed,
            done=done, implementation=implementation, priority=priority,
            status=status)
        self.area = area
        self.specs = []


class Area(object):

    def __init__(self, name):
        self.name = name
        self.groups = []

    def has_priority(self):
        return any([g.priority for g in self.groups])


class Workitem(object):
    """A single workitem"""

    def __init__(self, description, status, blueprint, assignee=None):
        self.description = description
        self.status = status
        self.blueprint = blueprint
        self.assignee = assignee


def spec_group_completion(db, team, milestone_collection=None):
    data = report_tools.spec_group_completion(db, team, milestone_collection=milestone_collection)
    if not data:
        return dict(areas=[], group_completion_series=[], groups=[])
    groups = []
    by_area = {}
    for bp_name, i in data.items():
        area_name = i['area']
        if area_name == "ad":
            area_name = "advanced development"
        area_name = " ".join([a.capitalize() for a in area_name.split()])
        if area_name not in by_area:
            area = Area(area_name)
            by_area[area_name] = area
        else:
            area = by_area[area_name]
        group = BlueprintGroup(bp_name, i['url'], area,
                todo=i['todo'], blocked=i['blocked'],
                inprogress=i['inprogress'],
                postponed=i['postponed'], done=i['done'],
                implementation=i['implementation'],
                priority=i['priority'], status=i['status'])
        groups.append(group)
        area.groups.append(group)
    return dict(areas=sorted(by_area.values(), key=lambda x: x.name),
                groups=sorted(groups, key=lambda x: report_tools.priority_value(x.priority), reverse=True))


def spec_completion(db, team, milestone_collection=None):
    data = report_tools.blueprint_completion(db, team, milestone_collection=milestone_collection)

    blueprints = []
    all_workitems = WorkitemTarget()
    essential_workitems = WorkitemTarget()
    high_workitems = WorkitemTarget()
    medium_workitems = WorkitemTarget()
    low_workitems = WorkitemTarget()
    unknown_workitems = WorkitemTarget()
    for bp_name, i in data.items():
        bp = Blueprint(bp_name, i['url'], complexity=i['complexity'],
                      todo=i['todo'], blocked=i['blocked'],
                      inprogress=i['inprogress'],
                      postponed=i['postponed'], done=i['done'],
                      implementation=i['implementation'],
                      priority=i['priority'], status=i['status'])
        def add_totals(container):
            container.todo += bp.todo
            container.blocked += bp.blocked
            container.inprogress += bp.inprogress
            container.postponed += bp.postponed
            container.done += bp.done
        add_totals(all_workitems)
        by_importance_workitems = unknown_workitems
        if i['priority'] == "Essential":
            by_importance_workitems = essential_workitems
        elif i['priority'] == "High":
            by_importance_workitems = high_workitems
        elif i['priority'] == "Medium":
            by_importance_workitems = medium_workitems
        elif i['priority'] == "Low":
            by_importance_workitems = low_workitems
        add_totals(by_importance_workitems)
        blueprints.append(bp)

    # avoid showing empty columns
    specs_have_status = any([bp.status for bp in blueprints])
    specs_have_complexity = any([bp.complexity > 0 for bp in blueprints])

    blueprints.sort(
        key=lambda bp: bp.percent_complete*100+report_tools.priority_value(bp.priority),
        reverse=True)

    return dict(
        specs_have_status=specs_have_status,
        specs_have_complexity=specs_have_complexity,
        blueprints=blueprints,
        all_workitems=all_workitems,
        essential_workitems=essential_workitems,
        high_workitems=high_workitems,
        medium_workitems=medium_workitems,
        low_workitems=low_workitems,
        unknown_workitems=unknown_workitems,
        )

def get_assignee_completion(db, team=None, group=None, milestone_collection=None):
    data = report_tools.assignee_completion(db, team=team, group=group, milestone_collection=milestone_collection)

    def _sort(a, b):
        # Takes [blueprint, workitem, priority, url] in a and b.
        # first by priority
        relative = cmp(report_tools.priority_value(b[2]), report_tools.priority_value(a[2]))
        if relative != 0:
            return relative
        # then by blueprint
        return cmp(a[0], b[0])

    assignees = []
    for name, info in data.items():
        name = name or team or 'nobody'
        name_html = escape_url(name or team or 'nobody')
        if not milestone_collection or not milestone_collection.single_milestone or isinstance(team, report_tools.user_string):
            url = '%s/~%s/+specs?role=assignee' % (report_tools.blueprints_base_url, name_html)
        else:
            target = name_html + '-' + milestone_collection.name
            url = 'u/%s.html' % target
        assignees.append(
            Assignee(name, url, complexity=info['complexity'],
                     todo_wis=sorted(info['todo'], cmp=_sort),
                     blocked_wis=info['blocked'],
                     inprogress_wis=info['inprogress'],
                     postponed_wis=info['postponed'],
                     done_wis=info['done']))
    return assignees


def by_assignee(db, team=None, group=None, milestone_collection=None):
    assignees = get_assignee_completion(
        db, team=team, group=group, milestone_collection=milestone_collection)
    assignees.sort(key=lambda a: a.name, reverse=False)

    # avoid showing empty columns
    assignees_have_complexity = any([a.complexity > 0 for a in assignees])

    # avoid showing empty columns
    workitems_have_priority = False
    for assignee in assignees:
        for status in report_tools.valid_states:
            for wi in getattr(assignee, status+"_wis", []):
                if wi[2]:
                    workitems_have_priority = True
                    break
        if workitems_have_priority:
            break

    return dict(
        assignees_have_complexity=assignees_have_complexity,
        assignees=assignees,
        workitems_have_priority=workitems_have_priority,
        valid_states=report_tools.valid_states)


def member_completion(db, team, milestone):
    if team is not None:
        members = report_tools.subteams(db, team)
    else:
        members = report_tools.all_teams(db)
    return dict(members=sorted(members))


def blueprint_count(db, team, milestone):
    return dict(blueprint_count=report_tools.blueprint_count(db, team=team, milestone=milestone))


def get_group(db, group_name):
    info = report_tools.spec_group_info(db, group_name)
    if info is None:
        # Means either no group, or no workitems.
        # We know that it likely means no workitems, but we should
        # do two queries to allow us to represent that better
        return BlueprintGroup(group_name, None, None)
    area = Area(info['area'])
    group = BlueprintGroup(group_name, info['url'], area,
            todo=info['todo'], blocked=info['blocked'],
            inprogress=info['inprogress'], postponed=info['postponed'],
            done=info['done'], implementation=info['implementation'],
            priority=info['priority'], status=info['status'])
    for spec_name, spec_info in info['specs'].items():
        spec = Blueprint(spec_name, spec_info['url'],
                      complexity=0,
                      todo=spec_info['todo'],
                      blocked=spec_info['blocked'],
                      inprogress=spec_info['inprogress'],
                      postponed=spec_info['postponed'],
                      done=spec_info['done'],
                      implementation=spec_info['implementation'],
                      priority=spec_info['priority'],
                      status=spec_info['status'])
        group.specs.append(spec)
    return group


def workitems_in_status(db, status, team=None, milestone_collection=None):
    """Get a list of the workitems in a particular status.

    The list can be filtered by assigned user/team, and/or by milestone if desired.
    """
    data = report_tools.assignee_completion(
        db, team=team, milestone_collection=milestone_collection)
    workitems = []
    for assignee in data:
        if status in data[assignee]:
            for wi_info in data[assignee][status]:
                bp_name, description, priority, url = wi_info
                blueprint = Blueprint(bp_name, url, priority=priority)
                workitems.append(
                    Workitem(description, status, blueprint,
                        assignee=assignee))
    return workitems


def list_people(db, team=None):
    team_info = report_tools.team_information(db, team=team)
    if team is None:
        people = set()
        for new_people in team_info.values():
            for new_person in new_people:
                people.add(new_person)
    else:
        people = set(team_info.get(team, []))
    return sorted(list(people))


class ReportWriter(object):

    def template_data(self, db, opts):
        data = {}
        if opts.root:
            root = opts.root
            if not root.endswith("/"):
                root += "/"
        else:
            root = "/"
        data["root"] = root
        data["real_names"] = report_tools.real_names(db)
        data.update(report_tools.days_left(db, milestone=opts.milestone))
        return data

    def get_milestone_collection(self, db, opts):
        milestone_collection = None
        if opts.milestone:
            milestone_collection = report_tools.get_milestone(db, opts.milestone)
        elif opts.date:
            milestone_collection = report_tools.MilestoneGroup(
                report_tools.date_to_python(opts.date))
        return milestone_collection

    def burndown(self, db, opts):
        '''Print work item status as HTML.'''

        if not opts.title:
            if opts.team:
                title = opts.team
            else:
                title = 'all teams'
        else:
            title = opts.title

        milestone_collection = self.get_milestone_collection(db, opts)
        if milestone_collection is not None:
            title += ' (%s)' % milestone_collection.name

        data = self.template_data(db, opts)
        data.update(dict(team_name=title, chart_url=opts.chart_url))
        data.update(
            spec_group_completion(db, opts.team, milestone_collection=milestone_collection))
        data.update(spec_completion(db, opts.team, milestone_collection=milestone_collection))
        data.update(by_assignee(db, team=opts.team, milestone_collection=milestone_collection))
        data.update(time_stats(db, team=opts.team, milestone_collection=milestone_collection))
        if milestone_collection is not None:
            data.update(dict(page_type="milestones"))
        elif isinstance(opts.team, report_tools.user_string):
            data.update(dict(page_type="people"))
        else:
            data.update(dict(page_type="teams"))
        print report_tools.fill_template(
            "burndown.html", data, theme=opts.theme)

    def overview(self, db, opts):
        if not opts.title:
            if opts.team:
                title = opts.team
            else:
                title = 'all teams'
        else:
            title = opts.title

        milestone_collection = self.get_milestone_collection(db, opts)

        if milestone_collection is not None:
            title += ' (%s)' % milestone_collection.name

        data = self.template_data(db, opts)
        data.update(dict(team_name=title, chart_url=opts.chart_url))
        data.update(spec_group_completion(db, None, milestone_collection=milestone_collection))
        data.update(spec_completion(db, opts.team, milestone_collection=milestone_collection))
        data.update(blueprint_count(db, opts.team, opts.milestone))
        data.update(dict(page_type="overview"))
        print report_tools.fill_template(
            "overview.html", data, theme=opts.theme)

    def group_overview(self, db, opts):
        if not opts.group:
            raise AssertionError(
                "Must specify group for group_overview report.")

        data = self.template_data(db, opts)
        data.update(dict(chart_url=opts.chart_url))
        data["group"] = get_group(db, opts.group)
        data.update(by_assignee(db, group=opts.group))
        data.update(dict(page_type="overview"))
        print report_tools.fill_template(
            "group_overview.html", data, theme=opts.theme)

    def team_list(self, db, opts):
        data = self.template_data(db, opts)
        data.update(member_completion(db, opts.team, opts.milestone))
        data.update(dict(page_type="teams"))
        print report_tools.fill_template(
            "team_list.html", data, theme=opts.theme)

    def person_list(self, db, opts):
        data = self.template_data(db, opts)
        data.update(people=list_people(db, team=opts.team))
        data.update(dict(page_type="people"))
        print report_tools.fill_template(
            "person_list.html", data, theme=opts.theme)

    def milestone_list(self, db, opts):
        data = self.template_data(db, opts)
        data.update(
            milestone_groups=report_tools.milestone_groups(
                db, team=opts.team))
        data.update(dict(page_type="milestones"))
        print report_tools.fill_template(
            "milestone_list.html", data, theme=opts.theme)

    def about(self, db, opts):
        data = self.template_data(db, opts)
        data.update(dict(page_type="about"))
        print report_tools.fill_template(
            "about.html", data, theme=opts.theme)

    def workitem_list(self, db, opts):
        """Generate a page listing workitems in a particular status."""
        assert opts.status is not None, (
            "Must pass --status for workitem_list report-type")
        milestone_collection = self.get_milestone_collection(db, opts)
        workitems = workitems_in_status(
            db, opts.status, team=opts.team,
            milestone_collection=milestone_collection)
        data = self.template_data(db, opts)
        data.update(dict(status=opts.status))
        data.update(dict(workitems=workitems))
        data.update(dict(page_type="overview"))
        print report_tools.fill_template(
            "workitem_list.html", data, theme=opts.theme)

    def roadmap_page(self, store, opts):
        if opts.lane is None:
            print "<h1>Error, no lane specified.</h1>"
        if not opts.title:
            title = opts.lane
        else:
            title = opts.title

        data = self.template_data(store, opts)
        lane = report_tools.lane(store, opts.lane)
        lanes = report_tools.lanes(store)
        statuses = []
        bp_status_totals = {'Completed': 0, 'Total': 0, 'Percentage': 0}
        for status, cards in report_tools.statuses(store, lane):
            cards_with_bps = []
            for card in cards:
                report_tools.check_card_health(store, card_health_checks, card)
                blueprint_status_counts = report_tools.card_bp_status_counts(
                    store, card.roadmap_id)
                total = sum(blueprint_status_counts.values())
                bp_percentages = dict.fromkeys(ROADMAP_ORDERED_STATUSES, 0)
                bp_status_totals['Completed'] += \
                    blueprint_status_counts['Completed']
                for key in ROADMAP_STATUSES_MAP:
                    bp_status_totals['Total'] += blueprint_status_counts[key]
                    if total > 0:
                        bp_percentages[key] = (
                            100.0 * blueprint_status_counts[key] / total)

                cards_with_bps.append({'card': card,
                                       'bp_statuses': blueprint_status_counts,
                                       'bp_percentages': bp_percentages})
            statuses.append(dict(name=status, cards=cards_with_bps))
        if bp_status_totals['Total'] > 0:
            bp_status_totals['Percentage'] = (100 * bp_status_totals['Completed'] /
                bp_status_totals['Total'])

        data.update(dict(statuses=statuses))
        data.update(dict(bp_status_totals=bp_status_totals))
        data.update(dict(status_order=ROADMAP_ORDERED_STATUSES))
        data.update(dict(page_type="roadmap_lane"))
        data.update(dict(lane_title=title))
        data.update(dict(lanes=lanes))
        data.update(dict(chart_url=opts.chart_url))
        print report_tools.fill_template(
            "roadmap_lane.html", data, theme=opts.theme)


    def roadmap_card(self, store, opts):
        if opts.card is None:
            print "<h1>Error, no card specified.</h1>"

        data = self.template_data(store, opts)
        card = report_tools.card(store, int(opts.card)).one()
        health_checks = report_tools.check_card_health(store, card_health_checks, card)
        lane = report_tools.lane(store, None, id=card.lane_id)

        if not opts.title:
            title = card.name
        else:
            title = opts.title

        blueprints = report_tools.card_blueprints_by_status(store, card.roadmap_id)
        bp_status_totals = {'Completed': 0, 'Total': 0, 'Percentage': 0}
        bp_status_totals['Total'] = (len(blueprints['Planned']) +
                                     len(blueprints['Blocked']) +
                                     len(blueprints['In Progress']) +
                                     len(blueprints['Completed']))
        bp_status_totals['Completed'] = len(blueprints['Completed'])
        if bp_status_totals['Total'] > 0:
            bp_status_totals['Percentage'] = (100 * bp_status_totals['Completed'] /
                bp_status_totals['Total'])

        card_has_blueprints = bp_status_totals['Total'] > 0

        status_order = ROADMAP_ORDERED_STATUSES[:]
        status_order.reverse()

        data.update(dict(page_type="roadmap_card"))
        data.update(dict(card_title=title))
        data.update(dict(card=card))
        data.update(dict(health_checks=health_checks))
        data.update(dict(lane=lane.name))
        data.update(dict(status_order=status_order))
        data.update(dict(blueprints=blueprints))
        data.update(dict(bp_status_totals=bp_status_totals))
        data.update(dict(card_has_blueprints=card_has_blueprints))

        print report_tools.fill_template(
            "roadmap_card.html", data, theme=opts.theme)


class WorkitemsOnDate(object):

    def __init__(self, date, done, delta_done, todo, delta_todo):
        self.date = date
        self.done = done
        self.delta_done = delta_done
        self.todo = todo
        self.delta_todo = delta_todo


def time_stats(db, team=None, milestone_collection=None):
    data = report_tools.workitems_over_time(
        db, team=team, milestone_collection=milestone_collection)

    prev_done = None
    prev_todo = None

    workitems_by_day = []
    for (date, states) in sorted(data.iteritems(), key=lambda k: k[0]):
        todo = states.get('todo', 0) + states.get('inprogress', 0)
        done = states.get('done', 0) + states.get('postponed', 0)
        if prev_done:
            delta_done = done - prev_done
            if delta_done < 0:
                delta_done = ' (' + str(delta_done) + ')'
            else:
                delta_done = ' (+' + str(delta_done) + ')'
        else:
            delta_done = ''
        if prev_todo:
            delta_todo = todo - prev_todo
            if delta_todo < 0:
                delta_todo = ' (' + str(delta_todo) + ')'
            else:
                delta_todo = ' (+' + str(delta_todo) + ')'
        else:
            delta_todo = ''
        prev_todo = todo
        prev_done = done
        workitems_by_day.append(WorkitemsOnDate(date, done, delta_done,
            todo, delta_todo))
    return dict(workitems_by_day=workitems_by_day)


#
# main
#

if __name__ == '__main__':
    report_tools.fix_stdouterr()

    # argv parsing
    optparser = optparse.OptionParser()
    optparser.add_option('-d', '--database',
        help='Path to database', dest='database', metavar='PATH')
    optparser.add_option('-t', '--team',
            help='Restrict report to a particular team', dest='team')
    optparser.add_option('-m', '--milestone',
            help='Restrict report to a particular milestone', dest='milestone')
    optparser.add_option('--chart', default='burndown.svg',
            help='(Relative) URL to burndown chart', dest='chart_url')
    optparser.add_option('-u', '--user', 
            help='Run for this user', dest='user')
    optparser.add_option('--title', 
            help='Set custom title')
    optparser.add_option('--report-type', default="burndown",
            dest="report_type")
    optparser.add_option('--group', dest="group",
            help="Select the group for a group_overview report.")
    optparser.add_option('--root', dest="root",
            help="Root URL for the charts")
    optparser.add_option('--status', dest="status",
            help="Workitem status to consider for workitem_list report-type")
    optparser.add_option('--date', dest="date",
            help="Include all milestones targetted to this date.")
    optparser.add_option('--theme', dest="theme",
            help="The theme to use.", default="linaro")
    optparser.add_option('--lane',
        help='Roadmap lane', dest='lane')
    optparser.add_option('--card',
        help='Roadmap card', dest='card')

    (opts, args) = optparser.parse_args()
    if not opts.database:
        optparser.error('No database given')
    if opts.user and  opts.team:
        optparser.error('team and user options are mutually exclusive')
    if opts.milestone and opts.date:
        optparser.error('milestone and date options are mutually exclusive')

    # The typing allows polymorphic behavior
    if opts.user:
        opts.team = report_tools.user_string(opts.user)
    elif opts.team:
        opts.team = report_tools.team_string(opts.team)

    store = report_tools.get_store(opts.database)

    writer = ReportWriter()
    method = getattr(writer, opts.report_type, None)
    if method is None:
        optparser.error('Unknown report type: %s' % opts.report_type)
    method(store, opts)