Display basic info about recent unbreak-now tasks from Phabricator projects

show_ubns.py

#!/usr/bin/python3
'''
I needed a script to show me all the ubns in the last few days on a couple
projects, on demand. So, here it is.
'''
import getopt
import os
import sys
import json
import time
import requests

# pylint: disable=broad-except

PHAB_URL = "https://phabricator.wikimedia.org/api"

# PHID-PROJ-va37kiwzbgywahiczwjk       Platform Team Workboards (Clinic Duty Team)
# PHID-PROJ-4uc7r7pdosfsk55qg7f6       Wikimedia-production-error
# PHID-PROJ-2b7oz62ylk3jk4aus262       Platform Engineering
PROJECTS = ["PHID-PROJ-va37kiwzbgywahiczwjk", "PHID-PROJ-2b7oz62ylk3jk4aus262",
            "PHID-PROJ-4uc7r7pdosfsk55qg7f6"]


class TaskRetriever():
    '''
    assemble conduit queries, make requests, check the output, display it
    '''
    def __init__(self, phaburl, projects, days, verbose):
        '''
        looking for every ubn in the given projects created no more than <days> ago
        '''
        self.verbose = verbose
        self.token = TaskRetriever.read_phab_token()
        self.phaburl = phaburl
        self.sess = TaskRetriever.setup_session()
        self.days = days
        self.projects = projects

    @staticmethod
    def read_phab_token():
        '''
        token file consists of lines
        MyUserName:phab_token_here
        blank lines and comments will be skipped
        This token is required for access to the Conduit API for Phab
        '''
        try:
            tokenfile = os.path.join(os.getenv("HOME"), '.phabenv')
            contents = open(tokenfile).read()
            lines = contents.splitlines()
            # skip comments, empty lines
            lines = [line for line in lines if line and not line.startswith('#')]
            if not lines:
                sys.stderr.write('no phab token found in file {tf}\n'.format(tf=tokenfile))
                return None
            return lines[0].split(':')[1]
        except Exception:
            sys.stderr.write('failed to read phab token from {tf}\n'.format(tf=tokenfile))
            return None

    @staticmethod
    def setup_session():
        '''
        we probably don't need a session since we only do one request. but
        for the future, there's always feeping creaturism.
        '''
        sess = requests.Session()
        sess.headers.update(
            {"User-Agent": "show-ubns.py/0.1 (WMF atg testing)",
             "Accept": "application/json"})
        return sess

    @staticmethod
    def string_convert(items):
        '''
        turning params into printable things: mixed lists become lists of strings
        '''
        return [str(item) if isinstance(item, int) else item for item in items]

    @staticmethod
    def flatten(items):
        '''
        turning params into printable strings: single ints become strings,
        and lists are converted by another method
        '''
        if isinstance(items, str):
            return items
        if isinstance(items, int):
            return str(items)
        items = TaskRetriever.string_convert(items)
        return ','.join(items)

    @staticmethod
    def printable_params(params):
        '''
        turn params into printable strings for display
        '''
        try:
            return '|'.join([param + '=' + TaskRetriever.flatten(params[param])
                             for param in params])
        except Exception:
            print("params are", params)
            return 'woops (failed to format params)'

    @staticmethod
    def display_task_info(task):
        '''
        show the bare minimum about a task: id, title, description
        '''
        print("task id:", task['id'])
        print("title:", task['fields']['name'])
        print("description:", task['fields']['description']['raw'])

    def project_search(self):
        '''
        this was going to be how I found the proj id but in the end this is broken, see
        https://discourse.phabricator-community.org/t/phabricator-project-search-api-broken-after-2020-16-release/3904/12
        so instead, I found some show all or some feed button on the project page and copied
        the proj id from the url :-P
        '''
        constraints = {'constraints[watchers][0]': 'PHID-USER-gpb2ynmxbgmy4ld2555w'}
        results = self.query('project.search', constraints)
        if results is None or not results['result']['data']:
            print('no project info. sad!')
            sys.exit(1)
        print(results['result']['data'])
        sys.exit(1)

    @staticmethod
    def remove_dup_tasks(tasks):
        '''
        grab only one copy of each task in the list, and
        return the new list
        '''
        no_dups = []
        task_ids = []
        for task in tasks:
            if task['id'] not in task_ids:
                task_ids.append(task['id'])
                no_dups.append(task)
        return no_dups

    def task_search(self):
        '''
        find all the tasks more recent than x days ubns for the given projects
        and display info about them in no particular order
        '''
        to_return = []
        epoch = int(time.time()) - (60 * 60 * 24 * self.days)
        constraints = {'constraints[statuses][0]': 'open',
                       'constraints[priorities][0]': '100',
                       'constraints[createdStart]': str(epoch)}
        for project in enumerate(self.projects):
            constraints['constraints[projects][0]'] = project
            # when phab is given a list of projects as a constraint,
            # it expects the task to be in ALL of them instead of ONE
            # of them, grrrr. so, one query per project >_<
            results = self.query('maniphest.search', constraints)
            if results and results['result']['data']:
                to_return.extend(results['result']['data'])
        if results is None or not results['result']['data']:
            print('no UBN tasks today!')
            sys.exit(1)
        return self.remove_dup_tasks(to_return)

    def query(self, module, params):
        '''
        run a query against a given phab conduit module, get the results as json,
        and return the results
        '''
        url = self.phaburl + '/' + module.lstrip('/')
        params['api.token'] = self.token
        if self.verbose:
            print("query:", module, TaskRetriever.printable_params(params))
        response = self.sess.post(url, data=params, timeout=15)
        if response.status_code != 200:
            sys.stderr.write("failed to get content with response code %s (%s) for %s, %s\n" %
                             (response.status_code, response.reason, module,
                              TaskRetriever.printable_params(params)))
        contents = response.content
        try:
            json_contents = json.loads(contents)
        except Exception:
            print("failed to get json contents back for %s, %s\n" % (module, '|'.join(params)))
            print("got: ", contents)
            return None
        if 'error_code' in json_contents and json_contents['error_code'] is not None:
            # example:
            # {"result":null,"error_code":"ERR-INVALID-SESSION",
            #  "error_info":"Session key is not present."}
            print("error reported for request for %s, %s\n" % (
                module, TaskRetriever.printable_params(params)))
            print("got: ", contents)
            return None
        if self.verbose > 1:
            print("json contents from query:", json_contents)
        return json_contents


def usage(message=None):
    '''
    display a helpful usage message with
    an optional introductory message first
    '''

    if message is not None:
        sys.stderr.write(message)
        sys.stderr.write("\n")
    usage_message = """
Usage: show_ubns.py [--phaburl phaburl] [--projects <name>[,<name>...]]
                    [--days <num>] [--verbose] | --help

This script shows unbreak now tasks from specified projects.

Arguments:
  --phaburl   (-p):  url to phabricator instance
                     default: https://phabricator.wikimedia.org/api
  --projects  (-P):  comma-separated list of project ids
                     default: PHID-PROJ-va37kiwzbgywahiczwjk,PHID-PROJ-2b7oz62ylk3jk4aus262
  --days      (-d):  number of days back to search
                     default: 5
  --verbose   (-v):  print a few progress messages while running
  --help      (-h):  display this help message

Setup:
  This script needs a phabricator api token in order to run.
  Create the file .phabenv in the directory where you run this script;
  it should contain the single line
  MyUserNameOrOtherNiceWordHere:api-token-here

"""
    sys.stderr.write(usage_message)
    sys.exit(1)


def get_args():
    """
    read, parse and return command line options
    """
    phaburl = PHAB_URL
    days = '5'
    projects = None
    project_list = PROJECTS
    verbose = 0

    try:
        (options, remainder) = getopt.gnu_getopt(
            sys.argv[1:], "d:p:vh", ["days=", "phaburl=", "projects=", "verbose", "help"])
    except getopt.GetoptError as err:
        usage("Unknown option specified: " + str(err))

    for (opt, val) in options:
        if opt in ["-d", "--days"]:
            days = val
        elif opt in ["-p", "--phaburl"]:
            phaburl = val
        elif opt in ["-P", "--projects"]:
            projects = val
        elif opt in ["-v", "--verbose"]:
            verbose += 1
        elif opt in ["-h", "--help"]:
            usage("Help for this script")

    if len(remainder) > 0:
        usage("Unknown option(s) specified: <%s>" % remainder[0])
    if not days.isdigit():
        usage("days must be a number")
    if projects:
        project_list = projects.split(",")
    return phaburl, int(days), project_list, verbose


def do_main():
    '''
    entry point
    display all UBN tasks of a given project
    '''
    phaburl, days, projects, verbose = get_args()
    phaburl = phaburl.rstrip('/')
    getter = TaskRetriever(phaburl, projects, days, verbose)
    if not getter.token:
        usage("Where is your token, silly person?")
        sys.exit(1)

    # results = getter.project_search()
    # print(results)

    results = getter.task_search()
    for task in results:
        TaskRetriever.display_task_info(task)


if __name__ == '__main__':
    do_main()