~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)