103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
import time
|
|
import hmac
|
|
import hashlib
|
|
import base64
|
|
import urllib.parse
|
|
import requests
|
|
import json
|
|
import git
|
|
import pathlib
|
|
import sys
|
|
import getopt
|
|
from configparser import ConfigParser
|
|
|
|
|
|
def generate_url(secret, webhook):
|
|
timestamp = str(round(time.time() * 1000))
|
|
secret_enc = secret.encode("utf-8")
|
|
string_to_sign = "{}\n{}".format(timestamp, secret)
|
|
string_to_sign_enc = string_to_sign.encode("utf-8")
|
|
hmac_code = hmac.new(
|
|
secret_enc, string_to_sign_enc, digestmod=hashlib.sha256
|
|
).digest()
|
|
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
|
|
return webhook + "×tamp=" + timestamp + "&sign=" + sign
|
|
|
|
def get_last_commit_info(repoPath):
|
|
commitInfo = dict()
|
|
repo = git.Repo(repoPath)
|
|
commitInfo['id'] = repo.commit().hexsha
|
|
commitInfo['author'] = repo.commit().author.name
|
|
commitInfo['email'] = repo.commit().author.email
|
|
commitInfo['time'] = repo.commit().authored_datetime.strftime("%Y-%m-%d %H:%M:%S")
|
|
commitInfo['message'] = repo.commit().message.strip()
|
|
return commitInfo
|
|
|
|
def push_message(url, projectName, commitInfo):
|
|
|
|
header = {"Content-Type": "application/json;charset=UTF-8"}
|
|
data = {
|
|
"msgtype": "markdown",
|
|
"markdown": {
|
|
"title": "仓库变更通知",
|
|
"text": "**项目名称**: %s \n\n" % projectName
|
|
+ "**提交Id**: %s \n\n" % commitInfo['id']
|
|
+ "**提交人**: %s \\<%s\\> \n\n" % (commitInfo['author'], commitInfo['email'])
|
|
+ "**提交时间**: %s \n\n" % commitInfo['time']
|
|
+ "**提交信息**: %s \n\n" % commitInfo['message'],
|
|
},
|
|
"at": {"atMobiles": ["18368802918"], "isAtAll": False},
|
|
}
|
|
|
|
response = requests.post(url=url, headers=header, data=json.dumps(data))
|
|
print(response.text)
|
|
|
|
def get_ini_config(botConfigPath, projectName):
|
|
config = ConfigParser()
|
|
if not (pathlib.Path(botConfigPath).exists()):
|
|
print("配置文件不存在！")
|
|
sys.exit(2)
|
|
config.read(botConfigPath, encoding="UTF-8")
|
|
return config[projectName]
|
|
|
|
def main(argv):
|
|
repoPath = ""
|
|
projectName = ""
|
|
botConfigPath = ""
|
|
|
|
try:
|
|
opts, args = getopt.getopt(argv, "hp:n:c:", ["help", "path=", "name=", "config="])
|
|
except getopt.GetoptError:
|
|
print("Error: DingTalkBot.py -p <repoPath> -n <projectName> -c <botConfigPath>")
|
|
print(" or: DingTalkBot.py --path=<repoPath> --name=<projectName> --config=<botConfigPath>")
|
|
sys.exit(2)
|
|
|
|
for opt, arg in opts:
|
|
if opt in ("-h", "--help"):
|
|
print("DingTalkBot.py -p <repoPath> -n <projectName> -c <botConfigPath>")
|
|
print("or: DingTalkBot.py --path=<repoPath> --name=<projectName> --config=<botConfigPath>")
|
|
sys.exit()
|
|
elif opt in ("-p", "--path"):
|
|
repoPath = arg
|
|
elif opt in ("-n", "--name"):
|
|
projectName = arg
|
|
elif opt in ("-c", "--config"):
|
|
botConfigPath = arg
|
|
|
|
if len(repoPath) == 0:
|
|
repoPath = pathlib.Path.cwd()
|
|
if len(projectName) == 0:
|
|
projectName = "default"
|
|
if len(botConfigPath) == 0:
|
|
botConfigPath = pathlib.Path(repoPath)
|
|
botConfigFilePath = pathlib.Path(botConfigPath).joinpath("DingTalkBot.ini")
|
|
print("项目路径：%s\n" %repoPath)
|
|
print("项目名称：%s\n" %projectName)
|
|
print("机器人配置文件路径：%s\n" %botConfigFilePath)
|
|
configInfo = get_ini_config(botConfigFilePath, projectName)
|
|
url = generate_url(configInfo['secret'], configInfo['webhook'])
|
|
commitInfo = get_last_commit_info(repoPath)
|
|
push_message(url, projectName, commitInfo)
|
|
|
|
if __name__ == "__main__":
|
|
main(sys.argv[1:]) |