phlint.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from phabricator import Phabricator
from datetime import datetime, timedelta


class Task(object):

    def __init__(self, maniphest, transactions, project, default_column, base_url):
        self.maniphest = maniphest
        self.transactions = transactions
        self.project = project
        self.column = self.get_task_column(default_column)
        self.base_url = base_url

    def get_task_column(self, default_column):
        for transaction in self.transactions:
            if transaction['transactionType'] == 'core:columns':
                column_info = transaction['newValue'][0]
                if column_info['boardPHID'] == self.project['phid']:
                    for column in self.project['columns']:
                        if column['phid'] == column_info['columnPHID']:
                            return column
        return default_column

    def get_days_since_prioritized(self):
        since = self.maniphest['fields']['dateCreated']
        for transaction in self.transactions:
            if transaction['transactionType'] == 'priority':
                since = int(transaction['dateCreated'])
                break
        return (datetime.now() - datetime.fromtimestamp(since)).days

    def get_days_since_moved(self):
        since = self.maniphest['fields']['dateCreated']
        for transaction in self.transactions:
            if transaction['transactionType'] == 'core:columns':
                column_info = transaction['newValue'][0]
                if column_info['boardPHID'] == self.project['phid']:
                    since = int(transaction['dateCreated'])
                    break
        return (datetime.now() - datetime.fromtimestamp(since)).days

    def get_days_without_activity(self):
        since = self.maniphest['fields']['dateCreated']
        for transaction in self.transactions:
            since = int(transaction['dateCreated'])
            break
        return (datetime.now() - datetime.fromtimestamp(since)).days

    def name(self):
        return self.maniphest['fields']['name']

    def project_name(self):
        return self.project['fields']['name']

    def column_name(self):
        return self.column['fields']['name']

    def priority(self):
        return self.maniphest['fields']['priority']['name']

    def owner_phid(self):
        return self.maniphest['fields']['ownerPHID']

    def url(self):
        return '%sT%d' % (self.base_url, self.maniphest['id'])

    def info(self):
        return '[%s::%s] %s {%s}' % (self.project_name(), self.column_name(), self.name(), self.priority())


class TaskSet(object):

    def __init__(self, tasks, conditions):
        self.tasks = tasks
        self.conditions = conditions

    def in_column(self, columns):
        if isinstance(columns, str):
            columns = [columns]
        return TaskSet(
            list(filter(
                lambda t: t.column_name() in columns,
                self.tasks
            )),
            {**self.conditions, **{
                'in_column': columns
            }}
        )

    def not_in_column(self, columns):
        if isinstance(columns, str):
            columns = [columns]
        return TaskSet(
            list(filter(
                lambda t: t.column_name() not in columns,
                self.tasks
            )),
            {**self.conditions, **{
                'not_in_column': columns
            }}
        )

    def with_priority(self, priorities):
        if isinstance(priorities, str):
            priorities = [priorities]
        return TaskSet(
            list(filter(
                lambda t: t.priority() in priorities,
                self.tasks
            )),
            {**self.conditions, **{
                'with_priority': priorities
            }}
        )

    def count(self):
        return ValueSet('Task count', [
            (len(self.tasks), self)
        ])

    def priority(self):
        return ValueSet('Priority', [
            (t.priority(), t)
            for t in self.tasks
        ])

    def owner(self):
        return ValueSet('Owner', [
            (t.owner_phid(), t)
            for t in self.tasks
        ])

    def days_since_prioritized(self):
        return ValueSet('Days since prioritized', [
            (t.get_days_since_prioritized(), t)
            for t in self.tasks
        ])

    def days_since_moved(self):
        return ValueSet('Days since moved', [
            (t.get_days_since_moved(), t)
            for t in self.tasks
        ])

    def days_without_activity(self):
        return ValueSet('Days without activity', [
            (t.get_days_without_activity(), t)
            for t in self.tasks
        ])

    def info(self):
        return 'Task set %s' % str(self.conditions)


class ValueSet(object):

    def __init__(self, description, values):
        self.description = description
        self.values = values

    def equal(self, reference_value):
        for value, subject in self.values:
            if value != reference_value:
                print(subject.info())
                if isinstance(subject, Task):
                    print(subject.url())
                print('>>> %s (%s) should equal %s.' % (self.description, str(value), str(reference_value)))
                print()

    def not_equal(self, reference_value):
        for value, subject in self.values:
            if value == reference_value:
                print(subject.info())
                if isinstance(subject, Task):
                    print(subject.url())
                print('>>> %s should not equal %s.' % (self.description, str(reference_value)))
                print()

    def less_or_equal(self, reference_value):
        for value, subject in self.values:
            if value > reference_value:
                print(subject.info())
                if isinstance(subject, Task):
                    print(subject.url())
                print('>>> %s (%s) should be less or equal than %s.' % (self.description, str(value), str(reference_value)))
                print()


def tasks_in_project(project_name, default_column_name):
    print('Loading Phabricator data...')
    print()
    phab = Phabricator(timeout=300)
    phab.update_interfaces()
    project = _get_project(phab, project_name)
    child_projects = _get_child_projects(phab, project)
    default_column = _get_column(project, default_column_name)
    maniphests = _get_maniphests(phab, project)
    for child_project in child_projects:
        maniphests_to_exclude = _get_maniphests(phab, child_project)
        phids_to_exclude = [m['phid'] for m in maniphests_to_exclude]
        maniphests = [m for m in maniphests if m['phid'] not in phids_to_exclude]
    transactions = _get_transactions(phab, maniphests)
    tasks = [
        Task(m, t, project, default_column, phab.host[0:-4])
        for m, t in zip(maniphests, transactions)
    ]
    return TaskSet(tasks, {'project': project_name})


def _get_project(phab, project_name):
    project_query = {'query': project_name}
    results = phab.project.search(constraints=project_query)['data']
    for project in results:
        if project['fields']['name'] == project_name:
            columns_query = {'projects': [project['phid']]}
            results = phab.project.column.search(constraints=columns_query)
            project['columns'] = results['data']
            return project
    raise Exception('Could not find project %s.' % project_name)


def _get_child_projects(phab, project):
    query = {'parents': [project['phid']]}
    return phab.project.search(constraints=query)['data']


def _get_column(project, column_name):
    for column in project['columns']:
        if column['fields']['name'] == column_name:
            return column
    raise Exception('Could not find column %s.' % project_name)


def _get_maniphests(phab, project):
    query = {'projects': [project['phid']], 'statuses': ['open']}
    results = phab.maniphest.search(constraints=query)
    maniphests = results['data']
    while results['cursor']['after'] is not None:
        results = phab.maniphest.search(
            constraints=query,
            after=results['cursor']['after']
        )
        maniphests.extend(results['data'])
    maniphests.sort(key=lambda t: t['id'])
    return maniphests


def _get_transactions(phab, maniphests):
    ids = [int(m['id']) for m in maniphests]
    transactions = list(phab.maniphest.gettasktransactions(ids=ids).values())
    transactions.sort(key=lambda t: int(t[0]['taskID']))
    for i in range(len(transactions)):
        transactions[i].sort(key=lambda t: -int(t['dateCreated']))
    return transactions