Gist to analyze a workflow's runs on a repo

AnalyzeGitHubWorkflows.py

import datetime
from github import Github
from helper import fetch_jobs

PAT_TOKEN_PATH = "/Users/shubham1172/.pat_token"
DAPR_CLI_REPO = "dapr/dapr"
LOOKBACK_DAYS = 3

# read token from file
with open(PAT_TOKEN_PATH, "r") as f:
    token = f.read().strip()

if not token:
    raise ValueError("Token not found in file")

g = Github(token)

# Get the repository
repo = g.get_repo(DAPR_CLI_REPO)

# Uncomment below to get the workflow id
# workflows = repo.get_workflows()
# for workflow in workflows:
#     if workflow.name == "dapr-test":
#         print(f"Workflow: {workflow.name}: {workflow.id}")
workflow_id = 4432

# Calculate the date LOOKBACK_DAYS days ago
lookback_days_ago = datetime.datetime.now() - datetime.timedelta(days=LOOKBACK_DAYS)

successes = 0
failures = 0

# Maps job name to a dictionary of status to count
jobs_dict = {}
failed_steps = {}

# Get the workflow runs from the past 3 days
workflow_runs = repo.get_workflow_runs()
for run in workflow_runs:
    if run.created_at < lookback_days_ago:
        break

    if run.workflow_id == workflow_id:
        print("Looking at run created at: ", run.created_at)
        if run.conclusion == "success":
            successes += 1
        elif run.conclusion == "failure":
            failures += 1
            jobs = fetch_jobs(run.jobs_url, token=token)
            for job in jobs:
                if job.name not in jobs_dict:
                    jobs_dict[job.name] = {}
                if job.conclusion not in jobs_dict[job.name]:
                    jobs_dict[job.name][job.conclusion] = 0
                jobs_dict[job.name][job.conclusion] += 1

                if job.conclusion == "failure":
                    if job.name not in failed_steps:
                        failed_steps[job.name] = []
                    failed_steps[job.name].append(job.first_failed_step)

print("\n")
print("=========================================")
print(f"Summary (past {LOOKBACK_DAYS} days):")
print(f"Successes: {successes}")
print(f"Failures: {failures}")
print(f"    Success rate: {successes / (successes + failures) * 100:.2f}%")
print("\n")
print("Job-wise summary:")
for job_name, job_dict in jobs_dict.items():
    print(f"Job: {job_name}")
    for status, count in job_dict.items():
        print(f"  -  {status}: {count}")
print("\n")
print("Failed steps summary:")
for job_name, step_names in failed_steps.items():
    print(f"Job: {job_name}")
    step_count = {}
    for step_name in step_names:
        if step_name not in step_count:
            step_count[step_name] = 0
        step_count[step_name] += 1
    for step_name, count in step_count.items():
        print(f"  -  {step_name}: {count}")
        
""" 
Sample output:

=========================================
Summary (past 3 days):
Successes: 0
Failures: 15
    Success rate: 0.00%


Job-wise summary:
Job: Deploy test infrastructure
  -  success: 10
  -  failure: 5
Job: Build for  linux on amd64
  -  success: 10
  -  failure: 5
Job: Build for  windows on amd64
  -  success: 11
  -  failure: 4
Job: End-to-end linux on amd64 tests
  -  success: 6
  -  failure: 3
Job: End-to-end windows on amd64 tests
  -  failure: 9
Job: Clean up Azure resources
  -  success: 15
Job: End-to-end ${{ matrix.target_os }} on ${{ matrix.target_arch }} tests
  -  skipped: 6


Failed steps summary:
Job: End-to-end windows on amd64 tests
  -  Preparing AKS cluster for test: 9
Job: End-to-end linux on amd64 tests
  -  Run E2E tests: 3
Job: Build for  linux on amd64
  -  Login to Azure Container Registry for Linux or Mac: 5
Job: Deploy test infrastructure
  -  Deploy the test cluster: 5
Job: Build for  windows on amd64
  -  Login to Azure Container Registry for Windows: 4
"""