python调用企业微信api实现发送群机器人消息实例

Posted 2020-12-13 wenm1128

 背景：自动化测试代码运行结果失败的地方需要通知相关同学去维护自己的代码，以及总的运行时长和用例数形成一个简易的通知。

 

import sys
import requests
import argparse
import re

#映射关系
user_info={}

class SendMsg2QYRot():
    """
    1、发送消息到企业微信机器人;
    2、因为@功能只能用text消息，所有需要先发markdown格式的消息，再发一条msg@失败的人；
    3、企业微信api：https://work.weixin.qq.com/help?person_id=1&doc_id=13376#markdown%E7%B1%BB%E5%9E%8B/markdown%E7%B1%BB%E5%9E%8B
    """
    def __init__(self, url=None):
        self.url=url
        self.session = requests.Session()

    def send_fail_msg(self, floder_name="wanghuiwu",duration=None,testcase=None,report_url=None):
        notice_msg={
            "msgtype": "markdown",
            "markdown": {
                    "content": """<font color="info">{}</font>的测试报告
>测试结果：<font color="warning">失败~</font>
>运行时长：{}
>用例数(success/fail):{}({}/<font color="warning">{}</font>)
>[点击查看详情]({})""".format(user_info[floder_name][0],duration,self.split_str(testcase)[0] ,self.split_str(testcase)[1],self.split_str(testcase)[2],report_url),
                        }
            }
        at_msg={"msgtype": "text","text": {"content": "","mentioned_mobile_list": [user_info[floder_name][1]]}}
        res = self.session.post(url=self.url, json=notice_msg, verify=False)
        res1 = self.session.post(url=self.url, json=at_msg, verify=False)
        if not (res.status_code == 200 and res1.status_code == 200):
            sys.exit(1)



    def send_report_msg(self,total,success,fail_count,durations):
        """
        :param durations: 运行总时长
        :param testcases: 用例总数
        :return:
        """
        notice_msg={
            "msgtype": "markdown",
            "markdown": {
                    "content": """>用例总数为：<font color="info">{}</font>(<font color="info">{}</font>/<font color="warning">{}</font>)
>运行总时长：<font color="info">{} senconds</font>""".format(total,success,fail_count,durations)
                        }
            }
        at_msg = {"msgtype":"text","text":{"content":"","mentioned_list":["@all"]}}
        r = self.session.post(url=self.url,json=notice_msg,verify=False)
        r1 = self.session.post(url=self.url, json=at_msg, verify=False)
        if not (r.status_code == 200 and r1.status_code == 200):
            sys.exit(1)

    def split_str(self,testcase):
        """
        获取单个报告的用例数并切割成一个list
        :param str:
        :return:
        """
        self.new_str = re.split(r‘[ (/)]+‘, str(testcase))
        return self.new_str


if __name__ == ‘__main__‘:
    parser = argparse.ArgumentParser(description="给企业微信推送消息")
    parser.add_argument(‘-U‘, ‘--url‘, default="", help="企业机器人webhook")
    parser.add_argument(‘-F‘, ‘--floder‘, default=‘common‘, help="用例目录")
    parser.add_argument(‘-R‘, ‘--report‘, default=‘http://140.143.140.118:8000/22-20200514181951-wanghuiwu.html‘, help="测试报告路径")
    parser.add_argument(‘-D‘,‘--duration‘,default=‘80s‘,help="运行时长")
    parser.add_argument(‘-T‘,‘--testcase‘,help=‘用例数‘)
    parser.add_argument(‘-t‘,‘--total‘,help=‘用例总数‘)
    parser.add_argument(‘-s‘,‘--success‘,help=‘成功用例总数‘)
    parser.add_argument(‘-f‘,‘--fail_count‘,help=‘失败用例总数‘)
    parser.add_argument(‘-d‘,‘--durations‘,help=‘运行总时长‘)
    args = parser.parse_args()

    # if len(sys.argv) == 1:
    #     # no argument passed
    #     parser.print_help()
    #     sys.exit(0)


    sm = SendMsg2QYRot(url=args.url)


#根据业务判断发哪那部分消息
    for i in sys.argv[3::2]:
        #print("**********",i)
        if i in [‘-F‘]:
            sm.send_fail_msg(report_url=args.report, duration=args.duration, testcase=args.testcase,floder_name=args.floder)
        elif i in [‘-t‘]:
            sm.send_report_msg(total=args.total,success=args.success,fail_count=args.fail_count, durations=args.durations)
        else:
            #parser.print_help()
            sys.exit(0)

 

运行传参：

E:pytest202004>py -3 testreport.py -U https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=11e5dcd2-1ed7-4e94-bc6c-772d12612f8d -F common -R http://140.143.140.118:8000/22-20200514181951-wanghuiwu.html -D 100s -T 32(32/0)