~statik/hydrazine/packaging

#! /usr/bin/python

# Copyright (C) 2010 Martin Pool

"""A text-mode interactive Launchpad client"""

import cmd

import httplib2

import optparse

import os

import subprocess

import sys

import webbrowser

import hydrazine

import launchpadlib.launchpad

import lazr.restfulclient.errors

class HydrazineCmd(cmd.Cmd):

    def __init__(self):

        cmd.Cmd.__init__(self)

        self.bug = None

        self.pillar = None

        self.task_list = None

    def _connect(self):

        self.session = hydrazine.create_session()

    def do_bug(self, bug_number):

        """Open bug by number"""

        try:

            bug_number = int(bug_number)

        except ValueError:

            print 'usage: bzr NUMBER'

            return

        try:

            the_bug = self.session.bugs[bug_number]

        except KeyError:

            print 'no such bug?'

            return

        self._select_bug(the_bug)

    def do_comment(self, line):

        """Post a comment to the current bug."""

        if self._needs_bug(): return

        if not line:

            print "Please specify a comment"

            return

        result = self.bug.newMessage(content=line)

        print "Posted message: %s" % result

    def do_description(self, nothing):

        """Show bug description"""

        if self._needs_bug():

            return

        print self.bug.description

    def do_duplicate(self, duplicate_id):

        """Mark as a duplicate"""

        if self._needs_bug():

            return

        try:

            duplicate_id = int(duplicate_id)

        except ValueError:

            print 'usage: duplicate BUG_NUMBER'

            return

        # XXX: could just synthesize a URL, which might be faster; probably

        # need to make sure the root lines up correctly

        try:

            duplicate_bug = self.session.bugs[duplicate_id]

        except KeyError:

            print 'no such bug?'

            return

        print 'marking %d as a duplicate of %d' % (self.current_bug_number,

            duplicate_bug.id)

        print '    "%s"' % duplicate_bug.title

        self.bug.markAsDuplicate(duplicate_of=duplicate_bug)

    def do_EOF(self, what):

        return True

    def do_importance(self, line):

        """Set importance"""

        task = self._needs_single_task()

        if task is None:

            return

        new_importance = canonical_importance(line)

        if new_importance is None:

            return

        print 'changing importance %s => %s' % (task.importance, new_importance)

        task.importance = new_importance

        print '**** before save'

        if opts.debug:

            print task._wadl_resource._definition.representation

        try:

            task.lp_save()

        except:

            print '**** got error'

            if opts.debug:

                print task._wadl_resource._definition.representation

    def do_next(self, ignored):

        """Go to the next bug in the list"""

        if self.task_list is None:

            print 'no list loaded; use select_new etc'

            return

        self.search_index += 1

        bug_task = self.task_list[self.search_index]

        self._select_bug(bug_task.bug)

    def do_official_tags(self, ignored):

        """Show the official tags for the current pillar."""

        if self._needs_pillar(): return

        print 'Official bug tags for %s' % self.pillar.name

        tags = self.pillar.official_bug_tags

        _show_columnated(tags)

    def do_open(self, ignored):

        """Open the current bug in a web browser"""

        if self._needs_bug():

            return

        webbrowser.open(web_url(self.bug))

    def do_pillar(self, pillar_name):

        """Select a pillar (project, etc)"""

        self._select_pillar(self._find_pillar(pillar_name))

    def do_refresh(self, ignored):

        """Reload current bug."""

        if self._needs_bug(): return

        self.bug.lp_refresh()

        self._show_bug(self.bug)

    def do_retarget(self, to_pillar):

        """Change a bug from this pillar to another."""

        task = self._needs_single_task()

        if task is None: return

        if not to_pillar:

            print 'usage: retarget TO_PILLAR'

            return

        new_target = self._find_pillar(to_pillar)

        if new_target is None:

            print 'no such product?'

            return

        print 'change target of bug %s' % (task.bug.id,)

        print '  from: %s' % (task.target,)

        print '    to: %s' % (new_target,)

        task.target = new_target

        task.lp_save()

    def do_select_new(self, ignored):

        """Select the list of new bugs in the current pillar"""

        if self._needs_pillar(): return

        self.task_list = self.pillar.searchTasks(status="New",

            order_by=['-datecreated'])

        self.task_list_index = 0

        try:

            first_bug_task = self.task_list[0]

            self.search_index = 0

        except IndexError:

            print "No bugtasks found"

        self._select_bug(first_bug_task.bug)

    def do_show(self, ignored):

        """Show the header of the current bug"""

        if self._needs_bug():

            return

        self._show_bug(self.bug)

    def do_title(self, new_title):

        """Change the title of the current bug.

example: 

    title bzr diff should warn if tree is out of date with branch

        """

        if self._needs_bug():

            return

        print 'changing title of bug %d to "%s"' % (self.bug.id, new_title)

        print '  old title "%s"' % (self.bug.title)

        self.bug.title = new_title

        self.bug.lp_save()

    def do_quit(self, ignored):

        return True

    def do_status(self, line):

        """Change status of the current bug"""

        task = self._needs_single_task()

        if task is None:

            return

        new_status = canonical_status(line)

        if new_status is None:

            return

        print 'changing status %s => %s' % (task.status, new_status)

        task.status = new_status

        task.lp_save()

    def do_tags(self, line):

        """Show, add or remove bug tags.

example: 

    tags +easy -crash

If no arguments are given, show the current tags.

Otherwise, add or remove the given tags.

"""

        if self._needs_bug(): return

        if not line.strip():

            print 'bug %d tags: %s' % (self.bug.id, ' '.join(self.bug.tags))

            return

        to_add = []

        to_remove = []

        for word in line.split():

            if word[0] == '+':

                to_add.append(word[1:])

            elif word[0] == '-':

                to_remove.append(word[1:])

            else:

                # XXX: not sure, should we just set it?

                to_add.append(word)

        old_tags = list(self.bug.tags)

        new_tags = old_tags[:]

        for a in to_add:

            if a not in new_tags:

                new_tags.append(a)

        for a in to_remove:

            if a in new_tags:

                new_tags.remove(a)

        print 'changing bug %d tags' % self.bug.id

        print '  from: %s' % ' '.join(old_tags)

        print '    to: %s' % ' '.join(new_tags)

        self.bug.tags = new_tags

        self.bug.lp_save()

    def do_triage(self, line):

        """Set tags, status, and importance.

example: 

    triage confirmed wishlist +foo +bar

        """

        if self._needs_bug(): return

        task = self._needs_single_task()

        if not task:

            print 'no task selected'

            return

        for w in line.split():

            if w[0] == '+':

                self.bug.tags.append(w[1:])

                continue

            importance = canonical_importance(w)

            if importance:

                task.importance = importance

                continue

            status = canonical_status(w)

            if status:

                task.status = status

                continue

        if self.bug._dirty_attributes:

            self.bug.lp_save()

        if task._dirty_attributes:

            task.lp_save()

    def onecmd(self, cmdline):

        # run the command protected against stupid

        # errors caused by eg <https://bugs.edge.launchpad.net/bugs/341950>

        try:

            # can't use super because Cmd is an old-style

            # class

            cmd.Cmd.onecmd(self, cmdline)

        except lazr.restfulclient.errors.RestfulError, e:

            print e

            pass

    def _needs_bug(self):

        if self.bug is None:

            print 'no bug selected'

            return True

    def _needs_pillar(self):

        if self.pillar is None:

            print 'no pillar selected'

            return True

    def _needs_single_task(self):

        """Return the single task for the current bug in the current pillar, or None"""

        if self.bug is None:

            print 'no bug selected'

            return None

        tasks = list(self.bug.bug_tasks)

        if self.pillar is None:

            if len(tasks) == 1:

                # no pillar; assume this is ok

                return tasks[0]

            else:

                print 'This bug has multiple tasks; please choose a pillar'

                return None

        else:

            for t in tasks:

                if t.target == self.pillar:

                    return t

            else:

                print 'No task for %s in %s' % (self.pillar, self.bug)

                return None

    @property 

    def prompt(self):

        p = 'hydrazine(%s) ' % (self.short_service_root,)

        if self.bug is not None:

            p += '#%d ' % (self.current_bug_number,)

        if self.pillar is not None:

            p += 'in %s ' % self.pillar.name

        # would like to highlight the prompt, but Cmd doesn't seem to have a

        # way to know some characters are not visible, therefore repainting is

        # messed

        if p[-1] == ' ':

            p = p[:-1]

        return p + '> '

    def _select_bug(self, the_bug):

        self.bug = the_bug

        self.current_bug_number = the_bug.id

        self._show_bug(self.bug)

    def _find_pillar(self, pillar_name):

        pillar_collection = self.session.pillars.search(text=pillar_name)

        try:

            return pillar_collection[0]

        except IndexError:

            print "No such pillar?"

            return

    def _select_pillar(self, pillar):

        self.pillar = pillar

        if pillar is None:

            print "no pillar selected"

        else:

            print "  %s" % self.pillar

    def _show_bug(self, bug):

        print 'bug: %d: %s' % (bug.id, bug.title)

        if bug.duplicate_of:

            print '  duplicate of bug %d' % (bug.duplicate_of.id,)

        else:

            for task in bug.bug_tasks:

                print '  affects %-40s %14s %s' % (

                    task.bug_target_name, task.status, task.importance,)

            print '  tags: %s' % ' '.join(bug.tags)

def canonical_importance(from_importance):

    real_importances = ['Critical', 'High', 'Medium', 'Low', 'Wishlist', 'Undecided']

    return canonical_enum(from_importance, real_importances)

def canonical_status(entered):

    return canonical_enum(entered,

        ['Confirmed', 'Triaged', 'Fix Committed', 'Fix Released', 'In Progress',

         "Won't Fix", "Incomplete", "Invalid", "New"])

def canonical_enum(entered, options):

    def squish(a):

        return a.lower().replace(' ', '')

    for i in options:

        if squish(i) == squish(entered):

            return i

    return None

def _show_columnated(tags):

    tags = tags[:]

    longest = max(map(len, tags))

    cols = int(os.environ.get('COLUMNS', '80'))

    per_row = max(int((cols-1)/(longest + 1)), 1)

    i = 0

    while tags:

        t = tags.pop(0)

        print '%-*s' % (longest, t),

        i += 1

        if i == per_row:

            print

            i = 0

    if i != 0:

        print

def web_url(launchpad_object):

    """Translate from an object's api url to the web page url"""

    # very dodgy; see https://bugs.launchpad.net/launchpadlib/+bug/316694

    return launchpad_object.self_link.replace('api.', '', 1).replace('/beta/', '/', 1)

def main(argv):

    parser = optparse.OptionParser()

    parser.add_option('--staging', action='store_const',

        const='staging',

        dest='short_service_root')

    parser.add_option('--debug', action='store_true',

        dest='debug',

        help='Show trace of API calls')

    parser.add_option('-c', '--command',

        action='append',

        dest='commands',

        help='Run this command before starting interactive mode (may be repeated)',

        metavar='COMMAND',

        )

    parser.set_defaults(short_service_root='edge')

    global opts, args

    opts, args = parser.parse_args(argv)

    hydrazine.service_root = dict(

        edge=launchpadlib.launchpad.EDGE_SERVICE_ROOT,

        staging=launchpadlib.launchpad.STAGING_SERVICE_ROOT,

        )[opts.short_service_root]

    if opts.debug:

        # debuglevel only takes effect when the connection is opened, so we can't

        # trivially change it while the program is running

        # see <https://bugs.edge.launchpad.net/launchpadlib/+bug/520219>

        httplib2.debuglevel = int(not httplib2.debuglevel)

    cmd = HydrazineCmd()

    cmd.short_service_root = opts.short_service_root

    cmd._connect()

    for c in opts.commands or []:

        print '> ' + c

        if cmd.onecmd(c):

            break

    else:

        # run cmdloop unless eg '-c quit' caused us to exit already

        cmd.cmdloop()

if __name__ == '__main__':

    main(sys.argv)