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

高楼Zee

发布于 2019-07-17 16:15:11

3.2K0

发布于 2019-07-17 16:15:11

文章被收录于专栏：7DGroup

前言

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

实现效果

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

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

预备知识

钉钉自定义机器人

API地址：https://open-doc.dingtalk.com/docs/doc.htm?treeId=257&articleId=105735&docType=1#s1

获取自定义机器人webhook

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

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

https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxx

使用自定义机器人

获取到 Webhook 地址后，用户可以使用任何方式向这个地址发起 HTTP POST 请求，即可实现给该群组发送消息。注意，发起POST请求时，必须将字符集编码设置成 UTF-8。

当前自定义机器人支持文本（text）、连接（link）、markdown（markdown）三种消息类型，大家可以根据自己的使用场景选择合适的消息类型，达到最好的展示样式。此我们选择连接（link）类型。

自定义机器人发送消息时，可以通过手机号码指定“被@人列表”。在“被@人列表”里面的人员，在收到该消息时，会有@消息提醒（免打扰会话仍然通知提醒，首屏出现“有人@你”）

消息类型及数据格式

link类型

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

消息发送频率限制

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

Python Jenkins API

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

API地址：https://python-jenkins.readthedocs.io/en/latest/index.html

连接Jenkins

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

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

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

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

获取Jenkins job number

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

last_build_number = server.get_job_info('api-test')['lastCompletedBuild']['number']build_info = server.get_build_info('api-test', last_build_number)print build_info

SonarQube Web-API

API地址：http://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"    }  ]}

Python Requests

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

API地址： http://docs.python-requests.org/zh_CN/latest/user/quickstart.html

发送请求

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

>>> import requests

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

>>> r = requests.get('https://api.github.com/events')

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

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

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

定制请求头

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

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

>>> url = 'https://api.github.com/some/endpoint'>>> headers = {'user-agent': 'my-app/0.0.1'}
>>> r = requests.get(url, headers=headers)

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

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

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

更加复杂的 POST 请求

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

>>> payload = {'key1': 'value1', 'key2': 'value2'}
>>> r = requests.post("http://httpbin.org/post", data=payload)>>> print(r.text){  ...  "form": {    "key2": "value2",    "key1": "value1"  },  ...}

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

>>> payload = (('key1', 'value1'), ('key1', 'value2'))>>> r = requests.post('http://httpbin.org/post', data=payload)>>> print(r.text){  ...  "form": {    "key1": [      "value1",      "value2"    ]  },  ...}

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

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

>>> import json
>>> url = 'https://api.github.com/some/endpoint'>>> payload = {'some': 'data'}
>>> r = requests.post(url, data=json.dumps(payload))

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

>>> url = 'https://api.github.com/some/endpoint'>>> payload = {'some': 'data'}
>>> r = requests.post(url, json=payload)

具体实现

安装包

pip install requestspip install python-jenkinspip install json262

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_token    pagrem = {        "msgtype": "link",        "link": {            'title':title,            "text": content,            'picUrl': messageUrl,            'messageUrl':'http://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 API    sonar_Url = 'http://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 = requests.get(sonar_Url).text    # 转换成josn    result = json.loads(resopnse)    bug = 0    leak = 0    code_smell = 0    coverage = 0    density = 0    status = ''    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 = 'http://www.iconsdb.com/icons/preview/soylent-red/x-mark-3-xxl.png'        statusStr = '失败'    elif status == 'OK':        statusStr = '成功'        # 正确图片        messageUrl = 'http://icons.iconarchive.com/icons/paomedia/small-n-flat/1024/sign-check-icon.png'
    # 消息内容。如果太长只会部分展示    code_reslut=  "Bug数:" + bug + "个，" + \                  "漏洞数:" + leak + "个，" + \                  "可能存在问题代码："+ code_smell + "行，" + \                  "覆盖率:" + coverage + "%，" + \                  "重复率:" + density + "%"    print("静态代码扫描统计："+"状态:"+ status +","+code_reslut)
    # 连接jenkins    server=jenkins.Jenkins(url="http://172.16.14.251:8080",username='user',password="password")
    # 获取指定项目最后编译number    get_number = server.get_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()

Jenkins设置

由于我的 Jenkins 跑在 window 平台，所以执行脚本需要加到 window 批处理最后一行。

结果验证

查看控制台日志输出

钉钉群消息通知

SonarQube扫描结果

脚本地址：

https://github.com/7DGroup/Jenkins-CI/tree/master/SonarQube-dingding-notifications-python

参考资料：

https://open-doc.dingtalk.com/docs/doc.htm?treeId=257&articleId=105735&docType=1#s1 https://python-jenkins.readthedocs.io/en/latest/index.html http://docs.python-requests.org/zh_CN/latest/user/quickstart.html

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2019-03-19，如有侵权请联系 cloudcommunity@tencent.com 删除

jenkins