目的：接口自动化测试定时执行推送钉钉提醒

接入：jmeter接口自动化测试提醒

python：使用python把自动化测试结果推送到钉钉群，方便提醒和查看

python钉钉提醒脚本并结合jmeter自动化执行脚本shell

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#author xqc
import time
import hmac
import hashlib
import base64
import urllib
import json
import requests
import logging
import re,sys

try:
    JSONDecodeError = json.decoder.JSONDecodeError
except AttributeError:
    JSONDecodeError = ValueError


def is_not_null_and_blank_str(content):
    if content and content.strip():
        return True
    else:
        return False


class DingtalkRobot(object):
    def __init__(self, webhook, sign=None):
        super(DingtalkRobot, self).__init__()
        self.webhook = webhook
        self.sign = sign
        self.headers = {'Content-Type': 'application/json; charset=utf-8'}
        self.times = 0
        self.start_time = time.time()

    # 加密签名
    def __spliceUrl(self):
        timestamp = int(round(time.time() * 1000))
        secret = self.sign
        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))
        url = f"{self.webhook}&timestamp={timestamp}&sign={sign}"
        return url

    def send_text(self, msg, is_at_all=False, at_mobiles=[]):
        data = {"msgtype": "text", "at": {}}
        if is_not_null_and_blank_str(msg):
            data["text"] = {"content": msg}
        else:
            logging.error("text类型，消息内容不能为空！")
            raise ValueError("text类型，消息内容不能为空！")

        if is_at_all:
            data["at"]["isAtAll"] = is_at_all

        if at_mobiles:
            at_mobiles = list(map(str, at_mobiles))
            data["at"]["atMobiles"] = at_mobiles

        logging.debug('text类型：%s' % data)
        return self.__post(data)

    def __post(self, data):
        """
        发送消息（内容UTF-8编码）
        :param data: 消息数据（字典）
        :return: 返回发送结果
        """
        self.times += 1
        if self.times > 20:
            if time.time() - self.start_time < 60:
                logging.debug('钉钉官方限制每个机器人每分钟最多发送20条，当前消息发送频率已达到限制条件，休眠一分钟')
                time.sleep(60)
            self.start_time = time.time()

        post_data = json.dumps(data)
        try:
            response = requests.post(self.__spliceUrl(), headers=self.headers, data=post_data)
        except requests.exceptions.HTTPError as exc:
            logging.error("消息发送失败， HTTP error: %d, reason: %s" % (exc.response.status_code, exc.response.reason))
            raise
        except requests.exceptions.ConnectionError:
            logging.error("消息发送失败，HTTP connection error!")
            raise
        except requests.exceptions.Timeout:
            logging.error("消息发送失败，Timeout error!")
            raise
        except requests.exceptions.RequestException:
            logging.error("消息发送失败, Request Exception!")
            raise
        else:
            try:
                result = response.json()
            except JSONDecodeError:
                logging.error("服务器响应异常，状态码：%s，响应内容：%s" % (response.status_code, response.text))
                return {'errcode': 500, 'errmsg': '服务器响应异常'}
            else:
                logging.debug('发送结果：%s' % result)
                if result['errcode']:
                    error_data = {"msgtype": "text", "text": {"content": "钉钉机器人消息发送失败，原因：%s" % result['errmsg']},
                                  "at": {"isAtAll": True}}
                    logging.error("消息发送失败，自动通知：%s" % error_data)
                    requests.post(self.webhook, headers=self.headers, data=json.dumps(error_data))
                return result


if __name__ == '__main__':
    URL = "https://oapi.dingtalk.com/robot/send?access_token=6cbc31d837ee6025cdb117a555a4a71e0563ce673d1049efdfad42a3309dcb4a"
    SIGN = "SECfaca9f9f6d5c630b8d857eec69cef2bdee11519257d80b1ab9500da7ad018c91"
    ding = DingtalkRobot(URL, SIGN)
    dff = re.search(
        '<td align="center">?\d+</td><td align="center">?\d+</td><td align="center">?.*</td><td align="center">?\d+ ms</td>',
        '<td align="center">22</td><td align="center">444</td><td align="center">244</td><td align="center">553 ms</td>')
    reportname = sys.argv[1]
    reportpath = sys.argv[2]
    reportname1 = sys.argv[3]
        #"/opt/apache-jmeter-5.4.1/bin/testrun/Interface_test/report/workbench/20220311154220/index.html"
    pattern1 = '<td align="center">?\d+</td><td align="center">?\d+</td><td align="center">?.*</td><td align="center">?\d+ ms</td>'
    # pattern= re.compile(pattern1)
    with open(reportpath, 'r', encoding='utf-8') as fr:
        for line in fr:
            match = re.search(pattern1, line)
            if match is not None:
                break

    con = match.group()
    count = re.findall(
        '<td align="center">?\d+</td><td align="center">?\d+</td><td align="center">?.*</td><td align="center">?\d+ ms</td>',
        con)
    countv = re.findall('\d+', count[0])
    t0=countv[0]
    t1 = countv[1]
    t2 = countv[2]
    t3 = countv[3]
    t4 = countv[4]



    print(ding.send_text("********"+reportname+"接口自动化测试报告********\n"+
                        "接口用例数量:"+t0+"\n"+
                        "用例失败数量:"+t1+"\n"+
                        "用例执行成功率:"+t2+"."+t3+"%\n"+
                        "详情测试报告地址:http://*.*.*/report/jenkins-report/"+reportname+"\n"
                         ))