Add deployment notes to github utils script

utils/github_utils.py

@@ -8,6 +8,7 @@
 import click
 import json
 import subprocess
+from datetime import datetime
 from time import sleep
 from typing import NoReturn
 
@@ -89,7 +90,6 @@ def release_notes(pull_request_number: int) -> NoReturn:
     dispatch_pr_url = "https://github.com/Netflix/dispatch/pull/"
     exclude_bot_authors = True
     exclude_labels = ["skip-changelog", "UI/UX", "javascript"]
-    gh_pr_list_merged_command = 'gh pr list -s merged --json "title,author,number,labels" -L 2000'
     sections = {
         "bug": "",
         "dependencies": "",
@@ -105,7 +105,9 @@ def release_notes(pull_request_number: int) -> NoReturn:
     }
 
     click.echo(f"Fetching list of merged PRs since #{pull_request_number}...")
-    pull_requests = json.loads(run_command(gh_pr_list_merged_command))
+    # Fetch only merged PRs for release notes
+    gh_command = 'gh pr list -s merged --json "title,author,number,labels" -L 2000'
+    pull_requests = json.loads(run_command(gh_command))
 
     if not pull_requests:
         click.echo(f"No PRs merged since #{pull_request_number}.")
@@ -157,5 +159,101 @@ def release_notes(pull_request_number: int) -> NoReturn:
     )
 
 
+def fetch_pull_requests(repo: str = "Netflix/dispatch", limit: int = 2000) -> list[dict]:
+    """Fetch pull requests from repository using gh CLI."""
+    gh_command = f'gh pr list --repo {repo} --state all --limit {limit} --json number,title,state,createdAt,author,url,headRefName,labels'
+    output = run_command(gh_command)
+    return json.loads(output)
+
+
+def categorize_pull_requests(prs: list[dict]) -> dict[str, list[dict]]:
+    """Categorize pull requests by their type."""
+    categories = {
+        'Fixes': [],
+        'Features': [],
+        'Refactors': []
+    }
+
+    for pr in prs:
+        title = pr['title']
+        first_word = title.split('(')[0].split(':')[0].lower()
+
+        if first_word in ['fix', 'fixes']:
+            categories['Fixes'].append(pr)
+        elif first_word in ['refactor', 'refactoring']:
+            categories['Refactors'].append(pr)
+        else:
+            categories['Features'].append(pr)
+
+    return categories
+
+
+def clean_title(title: str) -> str:
+    """Clean PR title by removing prefix and capitalizing."""
+    if ':' in title:
+        title = title.split(':', 1)[1].strip()
+    return title[0].upper() + title[1:] if title else title
+
+
+def format_deployment_prs(prs: list[dict]) -> None:
+    """Format and print pull requests for deployment announcements."""
+    if not prs:
+        print("No pull requests found")
+        return
+
+    today = datetime.now().strftime("%b %d")
+    print(f":announcement-2549: *Dispatch deployment to production today* ({today}) at [TIME] PT. Expect brief downtime.")
+    print()
+
+    # Filter out chore PRs before categorizing
+    non_chore_prs = []
+    for pr in prs:
+        title = pr['title']
+        first_word = title.split('(')[0].split(':')[0].lower()
+        if first_word not in ['chore', 'deps', 'deps-dev']:
+            non_chore_prs.append(pr)
+
+    categories = categorize_pull_requests(non_chore_prs)
+
+    for category, pr_list in categories.items():
+        if pr_list:
+            print(f"*{category}*")
+            for pr in pr_list:
+                title = clean_title(pr['title'])
+                print(f"• {title} ([#{pr['number']}]({pr['url']}))")
+            print()
+
+
+@cli.command()
+@click.option(
+    "--pr-number",
+    "-n",
+    required=True,
+    type=int,
+    help="PR number to start from (inclusive)"
+)
+@click.option(
+    "--repo",
+    default="Netflix/dispatch",
+    help="Repository in format owner/repo (default: Netflix/dispatch)"
+)
+def deployment_notes(pr_number: int, repo: str) -> NoReturn:
+    """Generate deployment notes for PRs starting from a specified PR number."""
+    try:
+        # Fetch all PRs
+        prs = fetch_pull_requests(repo, limit=100)
+
+        # Filter PRs with number >= pr_number
+        filtered_prs = [pr for pr in prs if pr["number"] >= pr_number]
+
+        # Sort by PR number (descending)
+        filtered_prs.sort(key=lambda x: x["number"], reverse=True)
+
+        format_deployment_prs(filtered_prs)
+
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)
+
+
 if __name__ == "__main__":
     cli()