summary-pr-report

#!/usr/bin/env python
# A script which prepares a report of the pending PRs
from argparse import ArgumentParser
from datetime import datetime, timedelta
from alibot_helpers.github_utilities import GithubCachedClient

def from_iso(d):
  return datetime.strptime(d, "%Y-%m-%dT%H:%M:%SZ")

def parse_args():
  parser = ArgumentParser()
  parser.add_argument("repo_name", help="Name of the repository to check")
  return parser.parse_args()

def in_the_last_hour(t):
  return datetime.now()-t > timedelta(hours=1)

def getPrStatus(cgh, repo_name, ref):
  statuses = cgh.get("/repos/{repo_name}/commits/{ref}/statuses",
                     repo_name=repo_name,
                     ref=ref)
  statuses = [(x["context"], {"state": x["state"],
                              "created_at": from_iso(x["created_at"]),
                              "updated_at": from_iso(x["updated_at"])}) for x in statuses]
  return dict(reversed(statuses))

if __name__ == "__main__":
  args = parse_args()
  with GithubCachedClient() as cgh:
    openIssues = cgh.get('/repos/{repo_name}/pulls?state=open',
                         repo_name=args.repo_name)
    # Print PRs which have an error state for more than 1h
    error_prs = []
    for issue in openIssues:
      statuses = getPrStatus(cgh, args.repo_name, issue["head"]["sha"])
      error_states = [x for x in statuses.values() if x["state"] == "error" and in_the_last_hour(x["updated_at"])]
      if error_states:
        error_prs.append("* [{number}]: {title}".format(number=issue["number"], title=issue["title"]))

    pending_prs = []
    for issue in openIssues:
      statuses = getPrStatus(cgh, args.repo_name, issue["head"]["sha"])
      error_states = [x for x in statuses.values() if x["state"] == "pending" and in_the_last_hour(x["updated_at"])]
      if error_states:
        pending_prs.append("* [{number}]: {title}".format(number=issue["number"], title=issue["title"]))

    if error_prs:
      print("## The following PRs have an error since more than 1h")
      print("\n".join(error_prs))

    if pending_prs:
      print("## The following PRs have are pending since more than 1h")
      print("\n".join(pending_prs))