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()
|