共计 3599 个字符,预计需要花费 9 分钟才能阅读完成。
本文同步公布于字节话云公众号。
背景
很多人都有本人的 GitHub 我的项目,可能须要将仓库中的事件主动告诉到钉钉中。只管钉钉群里能够增加专门的 GitHub 机器人,但它所反对的事件类型十分无限,根本只有 push 相干的事件。想要反对更多的事件,不如本人入手打造一个机器人。
于是,实现了一个 GitHub to DingTask 的我的项目。该我的项目通过阿里云的函数计算服务部署,无需任何金钱老本,就能部署这个告诉服务。
实现思路
外围思路就是在钉钉群中增加自定义机器人,实现一段 WebHook 的逻辑,依据接管到的 GitHub 事件类型的不同向自定义机器人发送不同的音讯。而这段 WebHook 逻辑须要有公网地址可被调用,各大云服务的函数计算服务每月会提供肯定的收费调用次数,是寄存 WebHook 逻辑的首选。
实现原理
钉钉增加自定义机器人
在钉钉群里增加自定义机器人,在机器人的平安设置中抉择“加签”,记下密钥并点击“实现”,在实现界面中记下机器人的 WebHook 地址。
告诉逻辑
GitHub 的 WebHook 事件的事件体是个 Json 字典,那么能够结构一个 DingTalkNotifier
类接管事件体 payload
,并提供 notify()
办法用于依据事件内容将具体音讯告诉到钉钉群。而告诉到钉钉群有现成的开源 Python SDK——DingtalkChatbot 能够应用,就省得咱们翻阅钉钉自定义机器人的技术文档去封装实现了。
WebHook 告诉的外围代码如下:
import logging
import conf
from dingtalkchatbot.chatbot import DingtalkChatbot
class DingTalkNotifier(object):
def __init__(self, payload: dict):
self.payload = payload
self.action = self.payload.get('action')
self.action_prep = 'to' if self.action in ('created', 'opened', 'submitted', None) else 'of'
self.sender = sender = payload.get('sender') or {}
self.sender_full_name = sender.get('login')
self.sender_page = sender.get('html_url')
self._md_sender = f'[{self.sender_full_name}]({self.sender_page})'
self.repo = repo = payload.get('repository') or {}
self.repo_full_name = repo.get('full_name')
self.repo_page = repo.get('html_url')
self.repo_language = repo.get('language')
self.repo_star_count = repo.get('stargazers_count')
self._md_repo = f'[{self.repo_full_name}]({self.repo_page})'
self.bot = DingtalkChatbot(conf.webhook, conf.secret)
def notify(self):
logging.info(f'Preparing notification: {self.payload}')
if 'pull_request' in self.payload:
self._notify_pull_request()
elif 'head_commit' in self.payload:
self._notify_push()
elif 'issue' in self.payload:
self._notify_issue()
elif 'starred_at' in self.payload:
self._notify_star()
elif 'forkee' in self.payload:
self._notify_fork()
elif 'discussion' in self.payload:
self._notify_discussion()
def _notify_pull_request(self):
pr = self.payload['pull_request']
pr_page = pr['html_url']
pr_number = pr['number']
pr_title = pr['title']
pr_body = pr['body'] or ''review = self.payload.get('review')
comment = self.payload.get('comment')
if review:
pr_review_page = review['html_url']
review_body = review['body'] or ''
self.bot.send_markdown(
title='Pull Request Review',
text=f'{self._md_sender} has {self.action} a pull request review {self.action_prep} {self._md_repo}\n\n'
f'[#{pr_number} {pr_title}]({pr_review_page})\n\n'
f'> {review_body}'
)
elif comment:
comment_page = comment['html_url']
comment_body = comment['body'] or ''
self.bot.send_markdown(
title='Issue Comment',
text=f'{self._md_sender} has {self.action} a pull request review comment'
f'{self.action_prep} {self._md_repo}\n\n'
f'[#{pr_number} {pr_title}]({comment_page})\n\n'
f'> {comment_body}'
)
else:
self.bot.send_markdown(
title='Pull Request',
text=f'{self._md_sender} has {self.action} a pull request {self.action_prep} {self._md_repo}\n\n'
f'[#{pr_number} {pr_title}]({pr_page})\n\n'
f'> {pr_body}'
)
在上述代码中:
DingTalkNotifier
的构造函数接管payload
变量,即 GitHub WebHook 的事件体。构造函数中将事件体中常见的sender
和repository
两块数据做了初步解析,以供后续发送音讯时组成须要的内容。此外,应用DingtalkChatbot(conf.webhook, conf.secret)
传入钉钉机器人的 WebHook 地址和密钥,初始化了钉钉机器人类,用来发送钉钉音讯。notify()
办法用来发送钉钉音讯。该办法中,依据self.payload
的不同特色,决定调用不同的音讯发送办法。例如,当事件体中蕴含pull_request
键时,就调用self._notify_pull_request()
发送 Pull Request 音讯。_notify_pull_request()
办法是具体的音讯告诉实现。该办法从事件体中获取须要的信息,进一步判断事件体的类型,再应用self.bot.send_markdown()
向钉钉群发送 MarkDown 格局的音讯。
部署告诉服务
写好的告诉逻辑,就须要将此逻辑部署到服务器上,并提供可供调用的 URL,以供 GitHub WebHook 调用。为了不便,也为了收费,这里抉择阿里云的函数计算服务(FC)进行部署。FC 的长处在于每月有 100 万次的收费调用额度,且提供了 HTTP 触发器,也就意味着提供了可供调用的 URL,就省得再去申请域名了。
FC 有专门的部署工具,叫做 Serverless Devs。在装置和配置好此工具后,将我的项目中 notification/conf.py
中的 webhook
和 secret
变量值替换为钉钉机器人的 WebHook 和密钥,就能够执行如下命令进行部署:
s github-notification deploy
部署实现后,拜访 FC 控制台,获取公网拜访地址,将之作为 GitHub 仓库的 WebHook,就可能欢快地承受来自 GitHub 仓库的音讯告诉了。
当 GitHub 仓库产生事件时,钉钉群的机器人就会推送相干的音讯,成果如下: