jira-pr.py

#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "PyGithub",
#     "jira",
#     "requests",
# ]
# ///
#
# With the help of Gemini and my own brain made from flesh and blood

import argparse
import json
import os
import re
from getpass import getpass

import requests
from github import Github
from jira import JIRA


def setup_arg_parser():
    """
    Sets up the command-line argument parser.
    """
    parser = argparse.ArgumentParser(
        description="Check merged GitHub PRs for Jira ticket links and their status, with optional Slack notifications.",
        formatter_class=argparse.RawTextHelpFormatter,
    )

    # --- GitHub Arguments ---
    parser.add_argument(
        "--github-token",
        default=os.environ.get("GITHUB_TOKEN"),
        help="GitHub Personal Access Token. Can also be set via GITHUB_TOKEN environment variable.",
    )
    parser.add_argument(
        "--github-repo",
        help="Target GitHub repository in 'owner/name' format (e.g., 'my-org/my-project').",
        default=os.environ.get("GITHUB_REPO"),
    )

    parser.add_argument(
        "--closed-status",
        nargs="+",
        default=["Done", "Closed", "Verified", "On QA"],
        help="Space-separated list of closed Jira issue statuses to ignore.",
    )

    # --- Jira Arguments ---
    parser.add_argument(
        "--jira-server",
        default=os.environ.get("JIRA_SERVER", "https://issues.redhat.com"),
        help="URL of the Jira server. Defaults to 'https://issues.redhat.com'.",
    )
    parser.add_argument(
        "--jira-user",
        default=os.environ.get("JIRA_USER"),
        help="Jira username (email). Can also be set via JIRA_USER environment variable.",
    )
    parser.add_argument(
        "--jira-token",
        default=os.environ.get("JIRA_TOKEN"),
        help="Jira API Token. Can also be set via JIRA_TOKEN environment variable.",
    )
    parser.add_argument(
        "--jira-prefix",
        default="https://issues.redhat.com/browse/",
        help="The URL prefix for Jira tickets to find in PR descriptions.",
    )

    # --- Slack Argument ---
    parser.add_argument(
        "--slack-webhook-url",
        default=os.environ.get("SLACK_WEBHOOK_URL"),
        help="Slack Incoming Webhook URL to post results. Can also be set via SLACK_WEBHOOK_URL environment variable.",
    )

    # --- Mode Subparsers ---
    subparsers = parser.add_subparsers(
        dest="mode", required=True, help="Mode of operation"
    )

    parser_last = subparsers.add_parser(
        "last", help="Check the last N merged pull requests."
    )
    parser_last.add_argument(
        "-n",
        "--count",
        type=int,
        default=100,
        help="Number of recent merged PRs to check. Default is 100.",
    )

    parser_release = subparsers.add_parser(
        "release", help="Check PRs between two releases/tags."
    )
    parser_release.add_argument(
        "--from-tag", required=True, help="The starting tag/release (older)."
    )
    parser_release.add_argument(
        "--to-tag", required=True, help="The ending tag/release (newer)."
    )

    return parser


def find_jira_tickets(text, jira_prefix):
    """
    Finds all Jira ticket keys in a given text based on the URL prefix or standalone key.
    """
    url_pattern = re.compile(re.escape(jira_prefix) + r"([A-Z][A-Z0-9]+-\d+)")
    key_pattern = re.compile(r"\b([A-Z][A-Z0-9]+-\d+)\b")
    tickets_from_url = url_pattern.findall(text)
    tickets_from_key = key_pattern.findall(text)
    return list(set(tickets_from_url + tickets_from_key))


def send_slack_notification(webhook_url, results, repo_name):
    """
    Formats and sends the results to a Slack webhook.
    """
    if not results:
        print("No issues to report to Slack.")
        return

    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": f":warning: Jira Status Check for {repo_name}",
                "emoji": True,
            },
        },
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": "The following merged Pull Requests are linked to Jira tickets that are not yet in a '**Done**' or '**Closed**' state. Please review them.",
            },
        },
        {"type": "divider"},
    ]

    # Create a formatted bullet point for each result
    for res in results:
        block_text = (
            f"• *<"
            f"{res['pr_url']}|PR #{res['pr_number']}>*: `{res['pr_title']}`\n"
            f"  :jira: *<"
            f"{res['jira_url']}|{res['jira_key']}>* is in status: `{res['jira_status']}`"
        )
        blocks.append(
            {"type": "section", "text": {"type": "mrkdwn", "text": block_text}}
        )

    payload = {"blocks": blocks}

    try:
        response = requests.post(
            webhook_url,
            data=json.dumps(payload),
            headers={"Content-Type": "application/json"},
        )
        response.raise_for_status()  # Raise an exception for bad status codes
        print("Successfully sent notification to Slack.")
    except requests.exceptions.RequestException as e:
        print(f"\033[91mERROR: Failed to send Slack notification. Reason: {e}\033[0m")


def process_pull_requests(pr_list, jira_client, jira_prefix, closed_status=None):
    """
    Iterates through a list of Pull Requests, collecting Jira issues that need attention.
    """
    if closed_status is None:
        closed_status = []
    if not pr_list:
        print("No pull requests found to process.")
        return []

    results_for_slack = []

    for pr in pr_list:
        pr_body = pr.body if pr.body else ""
        ticket_keys = find_jira_tickets(f"{pr.title} {pr_body}", jira_prefix)

        if not ticket_keys and (
            pr.title.lower().startswith("feat") or pr.title.lower().startswith("fix")
        ):
            # This could also be a result to add to Slack if desired
            continue

        for key in ticket_keys:
            try:
                issue = jira_client.issue(key)
                status = issue.fields.status.name
                if status in closed_status:
                    continue

                # Print to console for immediate feedback
                result_string = f"Closed PR: {pr.html_url} is attached to jira {jira_prefix}{key} that has status {status}"
                print(f"\033[93m[ATTENTION]\033[0m {result_string}")

                # Collect details for Slack notification
                results_for_slack.append(
                    {
                        "pr_number": pr.number,
                        "pr_title": pr.title,
                        "pr_url": pr.html_url,
                        "jira_key": key,
                        "jira_url": f"{jira_prefix}{key}",
                        "jira_status": status,
                    }
                )

            except Exception as e:
                print(
                    f"\033[91m-> ERROR: Could not fetch Jira ticket {key}. Reason: {e}\033[0m"
                )

    return results_for_slack


def main():
    """
    Main function to orchestrate the script's execution.
    """
    parser = setup_arg_parser()
    args = parser.parse_args()

    github_token = args.github_token or getpass(
        "Enter your GitHub Personal Access Token: "
    )
    jira_token = args.jira_token or getpass("Enter your Jira API Token: ")

    try:
        gh = Github(github_token)
        if not args.github_repo:
            raise ValueError("GitHub repository must be specified.")
        repo = gh.get_repo(args.github_repo)
    except Exception as e:
        print(
            f"\033[91mERROR: Could not connect to GitHub or find repository '{args.github_repo}'.\033[0m"
        )
        print(f"Reason: {e}")
        return

    try:
        jira_options = {"server": args.jira_server}
        # Corrected to use basic_auth as expected by the jira library
        jira_client = JIRA(options=jira_options, token_auth=jira_token)
        _ = jira_client.myself()  # Verify connection
    except Exception as e:
        print("\033[91mERROR: Could not connect to Jira.\033[0m")
        print(f"Reason: {e}")
        return

    pull_requests_to_check = []
    print("\nFetching pull requests...")

    if args.mode == "last":
        try:
            pulls = repo.get_pulls(state="closed", sort="updated", direction="desc")
            merged_pulls = []
            for pr in pulls:
                if len(merged_pulls) >= args.count:
                    break
                if pr.merged:
                    merged_pulls.append(pr)
            pull_requests_to_check = merged_pulls
            print(
                f"Found {len(pull_requests_to_check)} most recently merged pull requests."
            )
        except Exception as e:
            print(f"\033[91mERROR: Could not fetch pull requests. {e}\033[0m")
            return

    elif args.mode == "release":
        try:
            comparison = repo.compare(args.from_tag, args.to_tag)
            print(
                f"Comparing releases '{args.from_tag}' -> '{args.to_tag}'. Found {len(comparison.commits)} commits."
            )
            pr_numbers = set()
            all_prs = []
            for commit in comparison.commits:
                for pr in commit.get_pulls():
                    if pr.number not in pr_numbers and pr.merged:
                        pr_numbers.add(pr.number)
                        all_prs.append(pr)
            pull_requests_to_check = all_prs
            print(
                f"Found {len(pull_requests_to_check)} unique merged pull requests in this release range."
            )
        except Exception as e:
            print(
                f"\033[91mERROR: Could not compare releases. Check tags '{args.from_tag}' and '{args.to_tag}'. {e}\033[0m"
            )
            return

    # Process PRs and get results
    slack_results = process_pull_requests(
        pull_requests_to_check, jira_client, args.jira_prefix, args.closed_status
    )

    # If a webhook is provided and there are results, send the notification
    if args.slack_webhook_url and slack_results:
        print("\nSending results to Slack...")
        send_slack_notification(args.slack_webhook_url, slack_results, args.github_repo)
    elif args.slack_webhook_url:
        print("\nNo results to send to Slack.")


if __name__ == "__main__":
    main()