首页 > 编程札记 > 编程

Jenkins+SonarQube+Gitlab集成钉钉群消息自动通知(Python版)

阅读：评论：0

文章目录

- 一、前言
- 二、实现效果
- 三、预备知识
- - 1、钉钉自定义机器人
  - - 1.1、获取自定义机器人webhook
    - 1.2、使用自定义机器人
    - 1.3、消息类型及数据格式
    - 1.4、消息发送频率限制
  - 2、Python Jenkins API
  - - 2.1、连接Jenkins
    - 2.2、获取Jenkins job number
  - 3、SonarQube Web-API
  - 4、Python Requests
  - - 4.1、发送请求
    - 4.2、定制请求头
    - 4.3、更加复杂的 POST 请求
- 四、具体实现
- - 1、安装包
  - 2、Python 脚本
  - 3、Jenkins设置
- 五、结果验证

一、前言

在 Jenkins+SonarQube+Gitlab搭建自动化持续代码扫描质量平台一文中我们介绍了如何从 0 到 1 搭建一个自动化持续代码扫描质量平台，本文将在原有的基础上集成造钉群消息自动通知功能。

二、实现效果

在新代码扫描质量阀状态通过时候，推送通过消息及整体统计结果，如下图

在新代码扫描质量阀状态失败时候，推送失败消息及整体统计结果，如下图

三、预备知识

1、钉钉自定义机器人

API地址：
.htm?treeId=257&articleId=105735&docType=1#s1

1.1、获取自定义机器人webhook

在机器人管理页面选择“自定义”机器人，输入机器人名字并选择要发送消息的群。如果需要的话，可以为机器人设置一个头像。点击“完成添加”。

点击“复制”按钮，即可获得这个机器人对应的 Webhook 地址，其格式如下

=xxxxxxxx

1.2、使用自定义机器人

获取到 Webhook 地址后，用户可以使用任何方式向这个地址发起 HTTP POST 请求，即可实现给该群组发送消息。注意，发起POST请求时，必须将字符集编码设置成 UTF-8。
当前自定义机器人支持文本（text）、连接（link）、markdown（markdown）三种消息类型，大家可以根据自己的使用场景选择合适的消息类型，达到最好的展示样式。此我们选择连接（link）类型。
自定义机器人发送消息时，可以通过手机号码指定“被@人列表”。在“被@人列表”里面的人员，在收到该消息时，会有@消息提醒（免打扰会话仍然通知提醒，首屏出现“有人@你”）

1.3、消息类型及数据格式

link类型

{"msgtype": "link", "link": {"text":"群机器人是钉钉群的高级扩展功能。群机器人可以将第三方服务的信息聚合到群聊中，实现自动化的信息同步。例如：通过聚合GitHub，GitLab等源码管理服务，实现源码更新同步；通过聚合Trello，JIRA等项目协调服务，实现项目信息同步。不仅如此，群机器人支持Webhook协议的自定义接入，支持更多可能性，例如：你可将运维报警提醒通过自定义机器人聚合到钉钉群。", "title": "自定义机器人协议", "picUrl": "", "messageUrl": ".htm?spm=a219a.7629140.0.0.Rqyvqo&treeId=257&articleId=105735&docType=1"}
}

1.4、消息发送频率限制

每个机器人每分钟最多发送20条。
消息发送太频繁会严重影响群成员的使用体验，大量发消息的场景（譬如系统监控报警）可以将这些信息进行整合，通过 markdown 消息以摘要的形式发送到群里。

2、Python Jenkins API

Python Jenkins是 Jenkins REST API 的 python 包装器，旨在提供一种更传统的方式来控制 Jenkins 服务器。它提供了包含许多便捷功能的更高级API。

API地址：.html

2.1、连接Jenkins

这是一个示例，说明如何连接到 Jenkins实例。

import jenkinsserver = jenkins.Jenkins('localhost:8080', username='myuser', password='mypassword')
user = _whoami()
print('Hello %s from Jenkins %s' % (user['fullName']))

上面的代码打印用户的 fullName 属性。

从 Jenkins 版本 1.426 开始，您可以在针对 Jenkins 实例对用户进行身份验证时指定 API 令牌而不是您的真实密码。有关如何生成API令牌的详细信息，请参阅 Jenkins 身份验证 Wiki。获得 API 令牌后，您可以在创建 Jenkins 实例时传递 API令牌而不是真实密码。

2.2、获取Jenkins job number

这是一个示例，说明如何获取Jenkins job number及 build_info 。

last_build_number = _job_info('api-test')['lastCompletedBuild']['number']
build_info = _build_info('api-test', last_build_number)
print build_info

3、SonarQube Web-API

API地址：lyourSonarQube/web_api/api/measures?deprecated=true&internal=true

api/measures：获取具有指定度量的组件或子组件。GET api/measures/search
搜索按项目名称排序的项目度量。
最多可提供100个项目。
返回具有“浏览”权限的项目

参数响应示例：

{"measures": [{"metric": "complexity","value": "12","component": "MY_PROJECT_1"},{"metric": "complexity","value": "35","periods": [{"index": 1,"value": "0"}],"component": "MY_PROJECT_2"},{"metric": "complexity","value": "42","component": "MY_PROJECT_3"},{"metric": "ncloc","value": "114","component": "MY_PROJECT_1"},{"metric": "ncloc","value": "217","periods": [{"index": 1,"value": "0"}],"component": "MY_PROJECT_2"},{"metric": "ncloc","value": "1984","component": "MY_PROJECT_3"},{"metric": "new_violations","periods": [{"index": 1,"value": "25"}],"component": "MY_PROJECT_1"},{"metric": "new_violations","periods": [{"index": 1,"value": "25"}],"component": "MY_PROJECT_2"},{"metric": "new_violations","periods": [{"index": 1,"value": "255"}],"component": "MY_PROJECT_3"}]
}

4、Python Requests

Requests 是一个优雅而简单的 Python HTTP 库，广泛使用于 Python API 测试。

API地址：
.html

4.1、发送请求

使用 Requests 发送网络请求非常简单。
一开始要导入 Requests 模块：

>>> import requests

然后，尝试获取某个网页。本例子中，我们来获取 Github 的公共时间线：

>>> r = ('')

现在，我们有一个名为 r 的 Response 对象。我们可以从这个对象中获取所有我们想要的信息。

Requests 简便的 API 意味着所有 HTTP 请求类型都是显而易见的。例如，你可以这样发送一个 HTTP POST 请求：

>>> r = requests.post('', data = {'key':'value'})

4.2、定制请求头

Requests 中有一个内置的 JSON 解码器，助你处理 JSON 数据：
如果你想为请求添加 HTTP 头部，只要简单地传递一个 dict 给 headers 参数就可以了。

例如，在前一个示例中我们没有指定 content-type:

>>> url = ''
>>> headers = {'user-agent': 'my-app/0.0.1'}>>> r = (url, headers=headers)

注意: 定制 header 的优先级低于某些特定的信息源，例如：

如果在 rc 中设置了用户认证信息，使用 headers= 设置的授权就不会生效。而如果设置了 auth= 参数，rc 的设置就无效了。
如果被重定向到别的主机，授权 header 就会被删除。
代理授权 header 会被 URL 中提供的代理身份覆盖掉。
在我们能判断内容长度的情况下，header 的 Content-Length 会被改写。
更进一步讲，Requests 不会基于定制 header 的具体情况改变自己的行为。只不过在最后的请求中，所有的 header 信息都会被传递进去。

注意: 所有的 header 值必须是 string、bytestring 或者 unicode。尽管传递 unicode header 也是允许的，但不建议这样做。

4.3、更加复杂的 POST 请求

通常，你想要发送一些编码为表单形式的数据——非常像一个 HTML 表单。要实现这个，只需简单地传递一个字典给 data 参数。你的数据字典在发出请求时会自动编码为表单形式：

>>> payload = {'key1': 'value1', 'key2': 'value2'}>>> r = requests.post("", data=payload)
>>> )
{..."form": {"key2": "value2","key1": "value1"},...
}

你还可以为 data 参数传入一个元组列表。在表单中多个元素使用同一 key 的时候，这种方式尤其有效：

>>> payload = (('key1', 'value1'), ('key1', 'value2'))
>>> r = requests.post('', data=payload)
>>> )
{..."form": {"key1": ["value1","value2"]},...
}

很多时候你想要发送的数据并非编码为表单形式的。如果你传递一个 string 而不是一个 dict，那么数据会被直接发布出去。

例如，Github API v3 接受编码为 JSON 的 POST/PATCH 数据：

>>> import json>>> url = ''
>>> payload = {'some': 'data'}>>> r = requests.post(url, data=json.dumps(payload))

此处除了可以自行对 dict 进行编码，你还可以使用 json 参数直接传递，然后它就会被自动编码。这是 2.4.2 版的新加功能：

>>> url = ''
>>> payload = {'some': 'data'}>>> r = requests.post(url, json=payload)

四、具体实现

1、安装包

pip install requests
pip install python-jenkins
pip install json262

2、Python 脚本

# coding=utf-8'''
@author: zuozewei
@file: SonarQubeDingtalk.py
@time: 2019/3/5 18:00
'''
import  requests,json,jenkins,os,time# 接受jenkins当前JOB_NAME参数
projectName = os.getenv("JOB_NAME")def sendding(Dingtalk_access_token,content,title,messageUrl):url = Dingtalk_access_tokenpagrem = {"msgtype": "link","link": {'title':title,"text": content,'picUrl': messageUrl,'messageUrl':'172.16.14.251:9088/dashboard?id='+ projectName}}headers = {'Content-Type': 'application/json'}# 发送消息requests.post(url, data=json.dumps(pagrem), headers=headers)def notification():# 钉钉hook地址Dingtalk_access_token = 'your Dingtalk_access_token'# sonar APIsonar_Url = '172.16.14.251:9088/api/measures/search?projectKeys='+ projectName +'&metricKeys=alert_status%2Cbugs%2Creliability_rating%2Cvulnerabilities%2Csecurity_rating%2Ccode_smells%2Csqale_rating%2Cduplicated_lines_density%2Ccoverage%2Cncloc%2Cncloc_language_distribution'# 获取sonar指定项目结果resopnse = (sonar_Url).text# 转换成josnresult = json.loads(resopnse)bug = 0leak = 0code_smell = 0coverage = 0density = 0status = ''statusStr = ''# 解析sonar json结果for item in result['measures']:if item['metric']=="bugs":bug = item['value']elif item['metric']=="vulnerabilities":leak = item['value']elif item['metric']=='code_smells':code_smell = item['value']elif item['metric']=='coverage':coverage = item['value']elif item['metric']=='duplicated_lines_density':density = item['value']elif item['metric']=='alert_status':status = item['value']else:pass# 判断新代码质量阀状态if status == 'ERROR':# 错误图片messageUrl = '.png'statusStr = '失败'elif status == 'OK':statusStr = '成功'# 正确图片messageUrl = '.png'# 消息内容。如果太长只会部分展示code_reslut=  "Bug数:" + bug + "个，" + "漏洞数:" + leak + "个，" + "可能存在问题代码："+ code_smell + "行，" + "覆盖率:" + coverage + "%，" + "重复率:" + density + "%"print("静态代码扫描统计："+"状态:"+ status +","+code_reslut)# 连接jenkinsserver=jenkins.Jenkins(url="172.16.14.251:8080",username='user',password="password")# 获取指定项目最后编译numberget_number = _job_info(projectName)['lastBuild']['number']print("BUILD_NUMBER："+ str(get_number))sendding(Dingtalk_access_token, content=code_reslut, title=projectName+"#"+str(get_number)+"新代码扫描" + statusStr,messageUrl=messageUrl)if __name__=="__main__":# 等待10秒,确保SonarQube刷新结果time.sleep(10)notification()