Query reviews on phabricator

phabricator_query.py

#!/usr/bin/env python3
import argparse
import datetime
import json
import urllib.parse as url_parse
import urllib.request as url_request
from collections import defaultdict

api_token = "TOKEN"


def conduit(method, data, after=None, **kwargs):
    req = url_request.Request(
        f"https://phabricator.services.mozilla.com//api/{method}",
        method="POST",
        data=url_parse.urlencode(
            {
                "params": json.dumps(
                    {
                        **kwargs,
                        "__conduit__": {"token": api_token},
                        "after": after,
                    }
                ),
                "output": "json",
                "__conduit__": True,
            }
        ).encode(),
    )

    with url_request.urlopen(req) as r:
        res = json.load(r)
    if res["error_code"] and res["error_info"]:
        raise Exception(res["error_info"])

    if res["result"]["cursor"]["after"] is not None:
        # print(f'Fetching new page (from {res["result"]["cursor"]["after"]})')
        conduit(method, data, res["result"]["cursor"]["after"], **kwargs)

    if "results" in data:
        data["results"].extend(res["result"]["data"])
    else:
        data["results"] = res["result"]["data"]


def ymd(value):
    try:
        return datetime.datetime.strptime(value, "%Y-%m-%d")
    except ValueError:
        raise argparse.ArgumentTypeError(f"Invalid YYYY-MM-DD date: {value}")


def get_revisions(type, user, data, constraints):
    revisions = {}
    conduit(
        "differential.revision.search",
        revisions,
        constraints=constraints,
        order="newest",
    )

    revisions = revisions["results"]
    if not revisions:
        return

    revisions = sorted(revisions, key=lambda d: d["fields"]["dateCreated"])

    for revision in revisions:
        fields = revision["fields"]
        date_created = datetime.datetime.utcfromtimestamp(fields["dateCreated"])
        key = date_created.strftime("%Y-%m")
        if type not in data[user][key]:
            data[user][key][type] = []
        rev = (
            f'D{revision["id"]:5} {date_created.strftime("%Y-%m-%d")} {fields["title"]}'
        )
        data[user][key][type].append(rev)


def print_revisions(data, start_date, verbose):
    rev_details = {}
    for user, user_data in data.items():
        rev_details[user] = {
            "authored": [],
            "reviewed": [],
        }
        print(f"\n\nActivity for {user} since {start_date}")
        authored = 0
        reviewed = 0
        print("\nDetails (authored, reviewed):")
        for period, period_data in user_data.items():
            print(
                f"{period}: {len(period_data['authored'])}, {len(period_data['reviewed'])}"
            )
            authored += len(period_data["authored"])
            reviewed += len(period_data["reviewed"])
            rev_details[user]["authored"].extend(period_data["authored"])
            rev_details[user]["reviewed"].extend(period_data["reviewed"])

        print(f"\nTotal authored: {authored}")
        print(f"Total reviewed: {reviewed}")

    if verbose:
        for user, user_data in rev_details.items():
            print(f"\n----\nDetailed data for {user}")
            for type, type_data in user_data.items():
                print(f"\nList of {type} revisions ({len(type_data)}):")
                for rev in type_data:
                    print(f"  {rev}")


def get_user_phids():
    constraints = {
        "usernames": [
            "bolsson",
            "flod",
        ],
    }
    user_data = {}
    conduit("user.search", user_data, constraints=constraints)
    users = []
    for u in user_data["results"]:
        users.append(
            {
                "user": u["fields"]["username"],
                "name": u["fields"]["realName"],
                "phid": u["phid"],
            }
        )

    return users


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--since", "-s", type=ymd, help="Start date (defaults to 4 weeks ago)"
    )
    parser.add_argument(
        "--verbose", "-v", help="Print list of revisions", action="store_true"
    )
    args = parser.parse_args()
    if not args.since:
        args.since = datetime.datetime.today() - datetime.timedelta(weeks=4)
    since = args.since.replace(hour=0, minute=0, second=0, microsecond=0)

    users = get_user_phids()
    since = int(since.timestamp())

    recursivedict = lambda: defaultdict(recursivedict)
    data = recursivedict()
    for u in users:
        get_revisions(
            "authored",
            u["user"],
            data,
            dict(authorPHIDs=[u["phid"]], createdStart=since),
        )
        get_revisions(
            "reviewed",
            u["user"],
            data,
            dict(reviewerPHIDs=[u["phid"]], createdStart=since),
        )

    print_revisions(data, args.since.strftime("%Y-%m-%d"), args.verbose)


if __name__ == "__main__":
    main()