GitHub Pull Request 合并时在 Slack 上通知我

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

import requests
import argparse
import time
import re

# Could be none or personal access token https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api
GITHUB_TOKEN="<your_github_token>"
# On how to get webhooks token https://api.slack.com/messaging/webhooks
SLACK_WEBHOOK="https://hooks.slack.com/services/XXXXXXXXXXX/YYYYYYYYYY/ZZZZZZZZZZZZZZ"
WAIT_DURATION_SEC=600

def parse_args():
    parser = argparse.ArgumentParser(description='Check if a GitHub pull request was merged and send a Slack message.')
    parser.add_argument('pr_url', help='GitHub pull request URL.')
    return parser.parse_args()

def parse_pr_url(pr_url):
    """
    Parses the PR URL and returns the owner, repository, and PR number.

    Example URL: https://github.com/owner/repo/pull/123
    """
    pattern = r'https?://github\.com/(?P<owner>[^/]+)/(?P<repo>[^/]+)/pull/(?P<pr_number>\d+)'
    match = re.match(pattern, pr_url)
    if match:
        return match.group('owner'), match.group('repo'), int(match.group('pr_number'))
    else:
        raise ValueError('Invalid GitHub pull request URL.')

def check_if_pr_merged(owner, repo, pr_number, github_token=None):
    url = f'https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}'
    headers = {'Accept': 'application/vnd.github.v3+json'}
    if github_token:
        headers['Authorization'] = f'token {github_token}'
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        pr_data = response.json()
        return pr_data.get('merged', False)
    else:
        print(f'Error fetching PR data: {url} {response.status_code} {response.reason}')
        return False

def send_slack_message(slack_webhook_url, message):
    payload = {'text': message}
    response = requests.post(slack_webhook_url, json=payload)
    if response.status_code != 200:
        print(f'Error sending Slack message: {slack_webhook_url} {response.status_code} {response.reason}')

def main():
    args = parse_args()
    try:
        owner, repo, pr_number = parse_pr_url(args.pr_url)
    except ValueError as e:
        print(e)
        return

    print(f"Monitoring PR #{pr_number} in repository '{owner}/{repo}' for merge status.")

    while True:
        is_merged = check_if_pr_merged(owner, repo, pr_number, GITHUB_TOKEN)

        if is_merged:
            message = f'Pull request {args.pr_url} has been merged.'
            send_slack_message(SLACK_WEBHOOK, message)
            break
        else:
            time.sleep(WAIT_DURATION_SEC)

if __name__ == '__main__':
    main()

1

ntfslk https://github.com/owner/repo/pull/123 &