issues.py

import json
import os
import requests

from datetime import datetime
from dotenv import load_dotenv
from alive_progress import alive_bar

load_dotenv()

label_v9 = "Fluent UI react-components (v9)"
label_epic = "Type: Epic"
label_feature = "Type: Feature"
label_bug = "Type: Bug :bug:"
label_needs_backlog_grooming = "Needs: Backlog review"
label_a11y = "Area: Accessibility"
label_soft_close = "Resolution: Soft Close"
label_parter_ask = "Partner Ask"
label_needs_triage = "Needs: Triage :mag:"

token = os.environ["GITHUB_TOKEN"]
repo = "microsoft/fluentui"


def run_query(query):
    url = "https://api.github.com/graphql"
    headers = {"Authorization": f"bearer {token}"}
    response = requests.post(url, json={"query": query}, headers=headers)
    response.raise_for_status()

    return response.json()


def generate_graphql_query(repo, after_query):
    return f"""
    {{
      search(query: "repo:{repo} is:issue label:\\"{label_v9}\\"", type: ISSUE, first: 100, {after_query}) {{
        pageInfo {{
          endCursor
          hasNextPage
        }}
        edges {{
          node {{
            ... on Issue {{
              title
              number
              createdAt
              closedAt
              state
              labels(first: 100) {{
                nodes {{
                  name
                }}
              }}
              timelineItems(first: 100, itemTypes: [LABELED_EVENT, UNLABELED_EVENT]) {{
                nodes {{
                  __typename
                  ... on LabeledEvent {{
                    createdAt
                    label {{
                      name
                    }}
                  }}
                  ... on UnlabeledEvent {{
                    createdAt
                    label {{
                      name
                    }}
                  }}
                }}
              }}
            }}
          }}
        }}
      }}
    }}
    """


def fetch_all_issues():
    issues = []
    has_next_page = True
    end_cursor = None

    after_query = ""

    with alive_bar(0, title="Fetching issues", unit=" pages") as bar:
        while has_next_page:
            query = generate_graphql_query(repo, after_query)
            result = run_query(query)

            issues.extend([edge["node"]
                          for edge in result["data"]["search"]["edges"]])

            page_info = result["data"]["search"]["pageInfo"]

            has_next_page = page_info["hasNextPage"]

            if has_next_page:
                end_cursor = f'"{page_info["endCursor"]}"'
                after_query = "after: " + end_cursor

            bar()

    return issues


def get_issues(from_file=False):
    issues = []

    data_folder = "data"
    data_file = os.path.join(data_folder, "issues.json")

    if not os.path.exists(data_folder):
        os.makedirs(data_folder)

    if from_file:
        with open(data_file, "r") as json_file:
            issues = json.load(json_file)
    else:
        issues = fetch_all_issues()
        with open(data_file, "w") as json_file:
            json.dump(issues, json_file, indent=2)

    return issues


def normalize_issues(issues):
    normalized_issues = []

    for issue in issues:
        labels = set([label["name"] for label in issue["labels"]["nodes"]])
        timeline_items = []

        if "timelineItems" in issue:
            for timeline_item in issue["timelineItems"]["nodes"]:
                formatted_item = {
                    "type": timeline_item["__typename"],
                    "createdAt": datetime.strptime(
                        timeline_item["createdAt"], "%Y-%m-%dT%H:%M:%SZ"
                    ),
                }

                if "label" in timeline_item:
                    formatted_item["label"] = timeline_item["label"]["name"]

                timeline_items.append(formatted_item)

        normalized_issue = {
            "title": issue["title"],
            "number": issue["number"],
            "createdAt": datetime.strptime(issue["createdAt"], "%Y-%m-%dT%H:%M:%SZ"),
            "closedAt": (
                datetime.strptime(issue["closedAt"], "%Y-%m-%dT%H:%M:%SZ")
                if issue["closedAt"]
                else None
            ),
            "state": issue["state"],
            "labels": labels,
            "timelineItems": timeline_items,
        }

        normalized_issues.append(normalized_issue)

    return normalized_issues


if __name__ == "__main__":
    issues = get_issues(True)
    normalized_issues = normalize_issues(issues)

    print(normalized_issues[-1])