GithubWebhook解析サービス

2023-10-26 2024-12-25 约 3200 字预计阅读 7 分钟 - 次阅读

1 Webhook

WebhookはHTTPに基づくコールバック関数で、2つのアプリケーションプログラミングインターフェース（API）間で軽量なイベント駆動通信を実現できます。クライアントはサーバーAPIにユニークなURLを提供し、知りたいイベントを指定します。Webhookを設定すると、クライアントはサーバーをポーリングする必要がなくなり、指定されたイベントが発生したときに、サーバーが自動的に関連するペイロードをクライアントのWebhook URLに送信します。

Webhookは自動化を実現するための重要な要素であり、実装が簡単です（HTTPリクエスト1回で、どこにでも埋め込むことができます）。国内の飛書、钉钉、企業微信などはすべてWebhookによるメッセージプッシュをサポートしています。

これが利点ですが、欠点はHTTPリクエストのパラメータが統一されていないことです。これは非常にシンプルであるため、プロトコル規格がなく、プラットフォームごとに異なるためです。例えば、あるプラットフォームではメッセージが「msg」と呼ばれるのに対し、他のプラットフォームでは「message」または「text」となることがあります。したがって、Webhookの機能を十分に活用するには、自分のWebhook解析処理サービスを構築する必要がありますが、これはそれほど複雑ではありません。この記事では、Github Webhookを例に、Flaskを基に解析し企業微信にプッシュする方法を紹介します。

しかし、もし高度にカスタマイズされたニーズがなく、Github Webhookだけを使用し他のプラットフォームを使用しない場合は、Dinglingを使用することができます。これはGithub Webhookを企業微信に転送できます。DinglingはMarkdown形式のメッセージのみをプッシュできる（微信で直接閲覧できない）ため、私はそれを使用しませんでした。

2 CloudFlare Webhook

CloudFlare Webhookのメッセージボディは非常にシンプルで、以下のようになります：

1

{"text":"CloudFlareは業界の良心です！"}

Flaskを使用して解析し、メッセージを企業微信に転送します：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


from flask import Flask, request, jsonify,abort
import requests
from config import config
app = Flask(__name__)

@app.route('/cloudflare', methods=['POST'])
def cloudflare():
  if request.method == 'POST':
    print(request.json)
    text = request.json['text']
    data = {
        "msgtype": "text",
        "text": {
            "content": text
        }
        }
    r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_3'], json=data)
    print(r.json())
    return 'success', 200
  else:
    abort(400)
 
if __name__ == '__main__':
    app.run(port=5700,host='0.0.0.0',debug=True)

Flaskプロセスをサーバーにデプロイし、Cloudflareにアドレスを入力します。以下の図のように：

保存してテストした後、企業微信で簡単なテストメッセージを受け取ることができるはずです。

3 Github Webhook

CloudFlareのWebhookが短いメッセージであるとすれば、Githubはサイト全体の情報をすべて送信したいかのように、数十個のパラメータがありますが、自然言語は一つもありません。そのため、自分のニーズに応じて解析する必要があります。次のコードのロジックは、まずイベントタイプを判断し、次にイベントタイプに基づいて必要な情報を抽出して転送することです。以下の解析内容には、push、release、issue、pullなどの動作が含まれます。

  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
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254


@app.route('/github', methods=['POST'])
def github():
    githubEvent = request.headers['x-github-event']
    if githubEvent == 'push':
        respority = request.json['repository']['name']
        branch = request.json['ref'].split('/')[-1]
        commits = request.json['commits']
        for commit in commits:
            author = commit['author']['name']
            message = commit['message']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人提交了代码\n分支：{branch}\n作者：{author}\n提交信息：{message}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'], json=data)
        return 'success', 200
    if githubEvent == 'issues':
        action = request.json['action']
        if action == 'opened':
            respority = request.json['repository']['name']
            issue = request.json['issue']
            title = issue['title']
            body = issue['body']
            user = issue['user']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人提交了 issue\n标题：{title}\n内容：{body}\n提交者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'], json=data)
        if action == 'closed':
            respority = request.json['repository']['name']
            issue = request.json['issue']
            title = issue['title']
            body = issue['body']
            user = issue['user']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人关闭了 issue\n标题：{title}\n内容：{body}\n提交者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'], json=data)
        if action == 'reopened':
            respority = request.json['repository']['name']
            issue = request.json['issue']
            title = issue['title']
            body = issue['body']
            user = issue['user']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人重新打开了 issue\n标题：{title}\n内容：{body}\n提交者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'], json=data)
        return 'success', 200
    if githubEvent == 'pull_request':
        action = request.json['action']
        if action == 'opened':
            respority = request.json['repository']['name']
            pull_request = request.json['pull_request']
            title = pull_request['title']
            body = pull_request['body']
            user = pull_request['user']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人提交了 pull request\n标题：{title}\n内容：{body}\n提交者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'], json=data)
        if action == 'closed':
            respority = request.json['repository']['name']
            pull_request = request.json['pull_request']
            title = pull_request['title']
            body = pull_request['body']
            user = pull_request['user']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人关闭了 pull request\n标题：{title}\n内容：{body}\n提交者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'], json=data)
        if action == 'reopened':
            respority = request.json['repository']['name']
            pull_request = request.json['pull_request']
            title = pull_request['title']
            body = pull_request['body']
            user = pull_request['user']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人重新打开了 pull request\n标题：{title}\n内容：{body}\n提交者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'], json=data)
        return 'success', 200
    if githubEvent == 'release':
        action = request.json['action']
        if action == 'published':
            respority = request.json['repository']['name']
            release = request.json['release']
            tag_name = release['tag_name']
            body = release['body']
            user = release['author']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人发布了 release\n版本：{tag_name}\n内容：{body}\n发布者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'], json=data)
        return 'success', 200
    if githubEvent == 'ping':
        data = {
            "msgtype": "text",
            "text": {
                "content": "github webhook test"
            }
        }
        r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'], json=data)
        return 'success', 200
    if githubEvent == 'create':
        ref_type = request.json['ref_type']
        if ref_type == 'branch':
            respority = request.json['repository']['name']
            branch = request.json['ref']
            user = request.json['sender']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人创建了分支\n分支：{branch}\n创建者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'],
                              json=data)
        if ref_type == 'tag':
            respority = request.json['repository']['name']
            tag = request.json['ref']
            user = request.json['sender']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人创建了 tag\ntag：{tag}\n创建者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'],
                              json=data)
        return 'success', 200
    if githubEvent == 'delete':
        ref_type = request.json['ref_type']
        if ref_type == 'branch':
            respority = request.json['repository']['name']
            branch = request.json['ref']
            user = request.json['sender']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人删除了分支\n分支：{branch}\n删除者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'],
                              json=data)
        if ref_type == 'tag':
            respority = request.json['repository']['name']
            tag = request.json['ref']
            user = request.json['sender']['login']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人删除了 tag\ntag：{tag}\n删除者：{user}"
                }
            }
            r = requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'],
                              json=data)
        return 'success', 200
    if githubEvent == 'pull_request_review':
        action = request.json['action']
        if action == 'submitted':
            respority = request.json['repository']['name']
            pull_request = request.json['pull_request']
            title = pull_request['title']
            body = pull_request['body']
            user = pull_request['user']['login']
            review = request.json['review']
            state = review['state']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人提交了 pull request review\n标题：{title}\n内容：{body}\n提交者：{user}\n状态：{state}"
                }
            }
            r = requests.post(
                'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'],
                json=data)
        return 'success', 200
    if githubEvent == 'pull_request_review_comment':
        action = request.json['action']
        if action == 'created':
            respority = request.json['repository']['name']
            pull_request = request.json['pull_request']
            title = pull_request['title']
            body = pull_request['body']
            user = pull_request['user']['login']
            comment = request.json['comment']
            comment_body = comment['body']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人提交了 pull request review comment\n标题：{title}\n内容：{body}\n提交者：{user}\n内容：{comment_body}"
                }
            }
            r = requests.post(
                'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'],
                json=data)
        return 'success', 200
    # if githubEvent == 'check_run':
    #     action = request.json['action']
    #     if action == 'created':
    #         respority = request.json['repository']['name']
    #         check_run = request.json['check_run']
    #         name = check_run['name']
    #         conclusion = check_run['conclusion']
    #         data = {
    #             "msgtype": "text",
    #             "text": {
    #                 "content": f"{respority} 有人提交了 check run\n名称：{name}\n状态：{conclusion}"
    #             }
    #         }
    #         r = requests.post(
    #             'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'],
    #             json=data)
    #     return 'success', 200
    if githubEvent == 'check_suite':
        action = request.json['action']
        if action == 'completed':
            respority = request.json['repository']['name']
            check_suite = request.json['check_suite']
            conclusion = check_suite['conclusion']
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{respority} 有人提交了 check suite\n状态：{conclusion}"
                }
            }
            r = requests.post(
                'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='+config['API_KEY_2'],
                json=data)
        return 'success', 200