python git post-commit hook for Git/Asana integration

asana-post-commit.py

#!/usr/bin/env python

# git post-commit hook for linking git commits to asana tasks
# (inspired / adapted in part from https://github.com/Darunada/git-asana-post-commit-hook)
#
# Tested with Python 2.7 and Python 3.5
#
# INSTALLATION
# - Copy this script to .git/hooks/post-commit in your local repository
#   (be sure the script is executable)
# - Install python dependencies:
#       pip install asana colorama
# - Create an Asana personal access token under
#   Profile settings -> Apps -> Manage Developer Apps
# - Store that token in your git config:
#      git config --global asana.token `###`
#
# USAGE
# - Reference Asana tasks by id or URL in your commit message with an
#   optional verb, e.g.
#   [Fixes #1234] or [Implements https://app.asana.com/0/1234/4567]
#
# If the referenced task is unassigned, it will be assigned to the
# current user.  If the task has no development status tags it will
# be tagged as "dev in progress" or "dev complete" if there is a verb
# indicating the task is done.
#
# NOTES
# - currently only supports one task reference per commit


import asana
import subprocess
import re
# NOTE: colorama could probably be made optional fairly easily
import colorama


colorama.init(autoreset=True)

# recognized terms that can be used to designate a commit
# as fixing or re-opening the specified task, e.g.
# [Fixes #123]
VERBS = {
    # each term will look for optional endings: s, es, ed, ing
    'complete': ['fix', 'close', 'clos', 'finish', 'implement'],
    'break': ['break', 'reopen']
}

# development status (currently tracking in Asana as tags)
DEV_COMPLETE = 'dev complete'
DEV_IN_PROGRESS = 'dev in progress'

STATUSES = ['design in progress', DEV_IN_PROGRESS, DEV_COMPLETE,
    'awaiting testing', 'accepted', 'rejected']
    #  TBD: do we need accepted/complete?  Mark as done in asana
    # to mark complete?


def get_cmd_output(cmd):
    # run a system command and return the resulting output as a string
    # with any newline stripped off the end
    # takes command as a single string, then splits into an array
    # as needed by subprocess
    return subprocess.check_output(cmd.split()).decode().strip()


def git_remote_url():
    # determine the base remote url on github
    # git config --get  remote.origin.url
    # returns either:
    #   [email protected]:Princeton-CDH/sync-test.git
    # or
    #   https://github.com/torvalds/linux.git
    remote_url = get_cmd_output('git config --get remote.origin.url')
    # clean up so it can be used as a repo url base for the commit hash
    if remote_url.startswith('[email protected]:'):
        remote_url = remote_url.replace('[email protected]:', 'https://github.com/')
    if remote_url.endswith('.git'):
        remote_url = remote_url[:-(len('.git'))]
    return remote_url


def workspace_tags(client, workspace_id):
    # asana returns a list of tags with name and id; convert
    # into a name-based lookup dict for convenience
    tags = client.tags.find_all({'workspace': workspace_id})
    tag_info = dict((tag['name'], tag['id']) for tag in tags)
    return tag_info


def process_commit():
    # process the last git commit and add a comment to an Asana ticket
    # if one ir referenced in the commit message; update the ticket
    # assignment and status as appropriate

    # asana access token should be configured as a git config option
    try:
        token = get_cmd_output('git config asana.token')
    except subprocess.CalledProcessError:
        print('%sError! Asana token is not set in git config.' % colorama.Fore.RED)
        print('Create a personal access token and set in git config: git config --global asana.token ###')
        exit(-1)

    # initialize asana api client with access token
    client = asana.Client.access_token(token)

    # get the details from the last commit
    commit_info = get_cmd_output('git log -1 --pretty=format:"%h|%an|%s|%b|%ci" HEAD')
    # split git log info into pieces for processing
    commit_hash, author, commit_msg, commit_msg_body, date = commit_info.split('|')

    # handle single id for now; match #idnum OR full asana url
    # capture full task url if present
    regex = re.compile('\[((?P<verb>\w+)* *(?P<task_url>(#|https://app.asana.com/0/\d+/)(?P<id>\d+)))+\]')
    # TODO: handle multiple ids
    # e.g. [#123 #456]  [Fixes #123]
    # [Fixes #123, #456] - does it fix both?
    # [Fixes #245; #456]

    # first look in main commit message
    matches = regex.search(commit_msg)
    # if no matches were found there, look in commit message body
    if not matches:
        matches = regex.search(commit_msg_body)

    # if no match in commit message or body, then there is nothing to be done
    if not matches:
        return

    matchdict = matches.groupdict()
    task_id = matchdict['id']
    verb = matchdict['verb']
    if verb is not None:
        verb = verb.lower()
    msg = '\n'.join([commit_msg, commit_msg_body])

    # remove task id reference from commit message body
    # before adding as asana comment
    msg = regex.sub('', msg)

    # determine base remote url on github to generate link to this commit
    remote_url = git_remote_url()

    # generate a comment to add to the ticket
    # - comment format adapted from Pivotal Tracker GitHub integration:
    #     <b>Commit by username</b> <sm>on date</sm>
    #     commit comment
    #     github commit link
    txt_comment = '''Commit by %(user)s on %(date)s

    %(msg)s
    %(url)s
    '''
    # generate link to commit on github
    commit_url = '%s/commit/%s' % (remote_url, task_id)
    comment_data = {
        'user': author,
        'date': date,
        'msg': msg,
        'url': commit_url
    }
    comment_text = txt_comment % comment_data
    # add a comment on the referenced task
    resp = client.stories.create_on_task(task_id, {'text': comment_text})

    # retrieve the task object for additional updates
    task = client.tasks.find_by_id(task_id)

    # get workspace id for the task to find all tags
    # TOOD: check that this also works for subtasks
    workspace_id = task['workspace']['id']
    tags = workspace_tags(client, workspace_id)

    action_msgs = []
    # if task is not yet assigned, assign to current user
    if not task['assignee']:
        client.tasks.update(task_id, {'assignee': 'me'})
        action_msgs.append('assigned to you')

    # update tags for story based on verb, if specified
    verb_stem = None
    if verb:

        # strip off endings for simpler matching of supported terms
        verb_stem = re.sub('(es|ed|ing|s)$', '', verb.lower())

        if verb_stem in VERBS['complete']:
            tag_id = tags.get(DEV_COMPLETE, None)
            if tag_id is None:
                newtag = client.tags.create({'workspace': workspace_id,
                    'name': DEV_COMPLETE})
                tag_id = newtag['id']
            client.tasks.add_tag(task_id, {'tag': tag_id})
            action_msgs.append('tagged as "%s"' % DEV_COMPLETE)
            # if previously tagged as dev in progress, remove that tag
            task_tags = [tag['name'] for tag in task['tags']]
            if DEV_IN_PROGRESS in task_tags:
                # if the tag is in use, then must already be defined in
                # the current workspace, so we should have the id
                client.tasks.remove_tag(task_id, {'tag', tags[DEV_IN_PROGRESS]})

        elif verb_stem in VERBS['break']:
            # is there an appropriate tag here?
            client.tasks.update(task_id, {'completed': False})
            action_msgs.append('reopened')

    # otherwise, mark as development in progress unless
    # task already has a development status tag
    else:
        task_tags = [tag['name'] for tag in task['tags']]
        if not any([status in task_tags for status in STATUSES]):
            # no current dev status - mark as in progress
            tag_id = tags.get(DEV_IN_PROGRESS, None)
            # create the tag if it doesn't yet exist
            if tag_id is None:
                newtag = client.tags.create({'workspace': workspace_id,
                    'name': DEV_IN_PROGRESS})
                tag_id = newtag['id']
            client.tasks.add_tag(task_id, {'tag': tag_id})
            action_msgs.append('tagged as "%s"' % DEV_IN_PROGRESS)

    # if comment was added successfully, display what was done
    # includes task title and any actions
    if resp['created_at']:
        action = ''
        if action_msgs:
            action = '\n(%s)' % ', '.join(action_msgs)
        print('%sAdded comment to "%s"%s' % (colorama.Fore.GREEN,
            task['name'], action))


if __name__ == '__main__':
    process_commit()