企业微信添加机器人，并给机器人发送消息

标签：微信 alarm 机器人 robot content headers 添加 text post

找一个企业微信的群聊，点击右上角的"..."，添加群机器人。

创建一个机器人

填写机器人名字，添加机器人。

保存这里的webhook地址，后面给机器人发消息，就是给这个地址post消息。

具体的配置文档，可以点这里的“配置说明”去了解。

下面介绍，使用py给机器人发送消息

出于隐私的考虑，这里机器人的webhook地址，手机号是加了马赛克(xxx)的。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import requests
import json
headers={"Content-Type":"application/json","appType":"3","version":"5.1.0"}

#机器人地址（在申请成功机器人之后会有这个地址）
url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxx"

#要发送的消息
post_data={"msgtype":"text", "text":{"content": "hello python world.20230508"}}

post_data1={
    "msgtype": "text",
    "text": {
        "content": "hello world. i am lianghua robot.111",
        "mentioned_list":["[email protected]","@all"]
    }
}

post_data2={
    "msgtype": "text",
    "text": {
        "content": "您有告警信息，请注意查收",
        "mentioned_mobile_list": ["153xxxxxxxx", "135xxxxxxxx"]
    }
}

post_data3={
    "msgtype": "text",
    "text": {
        "content": "hello world. i am lianghua robot. 333",
        "mentioned_list":["[email protected]","@all"],
        "mentioned_mobile_list":["153xxxxxxx","135xxxxx"]
    }
}


post_markdown={
    "msgtype": "markdown",
    "markdown": {
        "content": """实时新增用户反馈 <font color="#800080">132例</font>，请相关同事注意。<@all>\n
        >类型:<font color="blue">用户反馈</font> 
        >普通用户反馈:<font color="red">117例</font> 
        >VIP用户反馈:<font color="green">15例</font>"""
    }
}



#json,data，2种方式都可以
#p_post = requests.post(url,headers=headers,data= json.dumps(post_data))
#p_post = requests.post(url,headers=headers,json=post_data)

# p_post = requests.post(url,headers=headers,json=post_data1)
# p_post = requests.post(url,headers=headers,json=post_data2)

#p_post = requests.post(url,headers=headers,json=post_markdown)
#p_post = requests.post(url,headers=headers,json=post_data2)

from algoproto.modules.robot_alarm import RobotAlarm
# 接收告警的用户手机号
user_mobile_list = ["135xxxxxxx"]
""" 添加机器人告警。如果计算的条数为0，就发出告警信息"""
robot_headers = {"Content-Type": "application/json", "appType": "3", "version": "5.1.0"}
# 机器人地址（在申请成功机器人之后会有这个地址）
robot_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxx"
i_error_count = 0
robot_helper = RobotAlarm(robot_headers, robot_url, user_mobile_list)

errorlist: list = []  # 添加监控逻辑，返回监控错误
alarm_title = """懂牛选股计算(<font color="#FF0000">test.py</font>)告警"""

errorlist.append(f"> aa<font color='red'>save_data failed, {6666}</font>")
errorlist.append(f"> bb<font color='red'>get_index_data failed, {7777}</font>")

sql = """SELECT IND_UNI_CODE as "symbol",TRADE_DATE as "dt",PRE_CLOSE_PRICE,CLOSE_PRICE,OPEN_PRICE,HIGH_PRICE,LOW_PRICE,TRADE_VOL,TRADE_AMUT from {}.IND_BASIC_MQ where ISVALID =1 and TRADE_DATE={} """.format('abcde', 20230612)
errorlist.append(f"> cc<font color='#FF0000'>process_index dtint:{20230614}</font>,sql查询结果为0. \nsql:{sql}")

err_ret = robot_helper.robot_alarm_markdown(20230614, alarm_title, errorlist)
i_error_count += err_ret

content = """aabbccde 
def \n
aaaaa \n
"""
err_ret = robot_helper.robot_alarm_text(user_mobile_list, content)
i_error_count += err_ret

if i_error_count > 0:
    robot_helper.msg_at_someone(user_mobile_list)

这是写的一个机器人类，把若干方法封装了一下。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# author:henry
# desc:企业微信机器人告警
# Date:20230614

import os
import sys
import copy
import math
import pandas as pd
import time
import requests
import json
from algoproto.common import logger


class RobotAlarm(object):
    def __init__(self, robot_headers:str,robot_url:str,user_mobile_list:[]):
        self.__robot_headers = robot_headers
        self.__robot_url = robot_url
        self.__user_mobile_list = user_mobile_list

    def set_robot_info(self,robot_headers:str,robot_url:str,user_mobile_list:[]):
        self.__robot_headers = robot_headers
        self.__robot_url = robot_url
        self.__user_mobile_list = user_mobile_list


    # markdown类型的告警信息,成功发送消息返回1.
    def robot_alarm_markdown(self,trade_day:int,alarm_title:str,errlist:list):
        """
        类似这样的告警信息

        懂牛选股计算(DN_XGB_FACTOR_2.py)告警，请相关同事注意。
        条数：3条
        日期：20230515
        指标ID:1118,CRXG5,创5日新高计算结果为0
        指标ID:1118,CRXG20,创20日新高计算结果为0
        指标ID:1118,CRXG60,创60日新高计算结果为0
        """
        start_tm1 = time.perf_counter()
        count_err= len(errlist)
        if 0 == count_err:
            return 0
        alarm_count=f'条数：<font color=\'warning\'>{count_err}条</font>'
        alarm_trade_day=f"交易日期：{trade_day}"

        alarm_sub_content=""
        for value in errlist:
            alarm_sub_content += value
            alarm_sub_content += "\n"

        alarm_content = f"{alarm_title}，请相关同事注意。\n{alarm_count}\n{alarm_trade_day}\n{alarm_sub_content}\n<@所有人>"
        post_markdown={
            "msgtype": "markdown",
            "markdown":
                {"content": alarm_content}
        }
        p_post = requests.post(self.__robot_url,headers=self.__robot_headers,json=post_markdown)
        logger.info(f" post error msg1:{json.loads(p_post.text)}")
        logger.info(f"robot_alarm_markdown post_markdown:{post_markdown}")

        # '{"errcode":0,"errmsg":"ok"}'
        ret_dict = json.loads(p_post.text)
        if 0==ret_dict["errcode"]:
            logger.info(f"企业微信告警消息，发送正常。")
        else:
            logger.info(f"企业微信告警消息，发送异常。")
            alarm_sub_content2 = alarm_sub_content[0:2000]
            alarm_sub_content2 +="\n.......\n"
            alarm_sub_content2 +="<font color='red'>错误消息内容超过4096，部分告警消息发送不出来，请排查问题。</font>"
            alarm_content2 = f"{alarm_title}，请相关同事注意。\n{alarm_count}\n{alarm_trade_day}\n{alarm_sub_content2}\n<@all>"
            post_markdown2 = {"msgtype": "markdown","markdown": {"content": alarm_content2}}
            p_post2 = requests.post(self.__robot_url,headers=self.__robot_headers,json=post_markdown2)
            logger.info(f" post error msg2:{json.loads(p_post2.text)}")

        start_tm2 = time.perf_counter()
        logger.info(f"post error msg cost:{round(start_tm2 - start_tm1, 4)}")
        return 1


    # text类型的告警信息...
    def robot_alarm_text(self,user_mobile_list:[],content:str):
        start_tm1 = time.perf_counter()
        if len(user_mobile_list)==0:
            user_mobile_list.append("@all")
        post_text = {
            "msgtype": "text",
            "text": {
                "content": content,
                "mentioned_mobile_list":user_mobile_list
            }
        }
        p_post = requests.post(self.__robot_url,headers=self.__robot_headers, json=post_text)
        ret_dict = json.loads(p_post.text)
        logger.info(f"post error msg1:{json.loads(p_post.text)}")
        logger.info(f"robot_alarm_text post_text:{post_text}")

        # '{"errcode":0,"errmsg":"ok"}'
        if 0 == ret_dict["errcode"]:
            logger.info(f"企业微信告警消息，发送正常。")
        else:
            logger.info(f"企业微信告警消息，发送异常。")
            alarm_content2 = content[0:1500]
            alarm_content2 += "\n.......\n"
            alarm_content2 += "错误消息内容超过2048，部分告警消息已截断，请排查问题"
            post_text2 = {
                "msgtype": "text",
                "text": {
                    "content": alarm_content2,
                    "mentioned_mobile_list": user_mobile_list
                }
            }
            p_post2 = requests.post(self.__robot_url, headers=self.__robot_headers, json=post_text2)
            logger.info(f" post error msg1:{json.loads(p_post2.text)}")

        start_tm2 = time.perf_counter()
        logger.info(f"post error msg cost:{round(start_tm2 - start_tm1, 4)}")
        return 1

    # 这个方法主要是用来 @指定成员 的，markdown暂时还没研究出来如何 @指定成员
    def msg_at_someone(self,user_mobile_list:[]):
        if len(user_mobile_list)==0:
            user_mobile_list.append("@all")
        post_text = {
            "msgtype": "text",
            "text": {
                "content": "您有告警信息，请注意查收",
                "mentioned_mobile_list":user_mobile_list
            }
        }
        p_post = requests.post(self.__robot_url,headers=self.__robot_headers, json=post_text)
        ret_dict = json.loads(p_post.text)
        logger.info(f" message ret:{ret_dict}")
        return 0

给机器人发消息，就是一个Post请求。把要发送的内容放在json参数字段，Url是机器人地址。

requests.post(self.__robot_url,headers=self.__robot_headers,json=post_markdown) 再来一个robot的使用示例：

import requests
import json

def robot_alarm(trade_day:int,alarm_title:str,errlist:list):
    """ 添加机器人告警。如果计算的条数为0，就发出告警信息"""
    headers={"Content-Type":"application/json","appType":"3","version":"5.1.0"}
    #机器人地址（在申请成功机器人之后会有这个地址）
    url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

    errlist.append(f">指标521-XG_KPDRise3Days,<font color='red'>庄家控盘_控盘3日递增，计算结果为0</font>")
    errlist.append(f">指标521-XG_YellowDKP,<font color='red'>庄家控盘_黄柱低控盘，计算结果为0</font>")
    errlist.append(f">指标521-XG_RedZKP,<font color='red'>庄家控盘_红柱中控盘，计算结果为0</font>")


    str_count= len(errlist)
    alarm_count=f'条数：<font color=\'warning\'>{str_count}条</font>'
    alarm_trade_day=f"日期：{trade_day}"

    alarm_sub_content=""
    for value in errlist:
        alarm_sub_content += value
        alarm_sub_content += "\n"

    alarm_content = f"{alarm_title}，请相关同事注意。\n{alarm_count}\n{alarm_trade_day}\n{alarm_sub_content}\n<@所有人>"
    post_markdown={
    "msgtype": "markdown",
    "markdown": {
        "content": alarm_content
        }
    }
    print("content is:",len(post_markdown["markdown"]["content"]))
    p_post = requests.post(url,headers=headers,json=post_markdown)
    print(p_post.text)
    return

result_map={}
"""
result_map["521-XG_KPDRise3Days"]="庄家控盘_控盘3日递增，计算结果为0"
result_map["521-XG_YellowDKP"]="庄家控盘_黄柱低控盘，计算结果为0"
result_map["521-XG_RedZKP"]="庄家控盘_红柱中控盘，计算结果为0"
"""

errlist22=[]
alarm_title = "懂牛选股计算(DN_XGB_FACTOR_2.py)告警"

robot_alarm(20230516,alarm_title,errlist22)

print("errlist22:",errlist22)

从示例中可以看到，最核心的就是一个 Post 函数调用。

标签：微信,alarm,机器人,robot,content,headers,添加,text,post
From： https://www.cnblogs.com/music-liang/p/17507796.html

企业微信添加机器人，并给机器人发送消息

相关文章

赞助商

阅读排行