Python 企微机器人发送 MarkDown消息

python发送企微md消息:

class WXRobot:
    def robot_send(self):
        import requests

        headers = {"Content-Type": "text/plain"}

        send_url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx-xxx-xxxxxxxxaaa123'

        send_data = {
            "msgtype": "markdown",  # 消息类型,此时固定为markdown
            "markdown": {
                "content": "# **【测试】**<font color="warning">**xxx例**</font>
" +  # 标题 (支持1至6级标题,注意#与文字中间要有空格)
                           "#### **请注意**
" +  # 加粗:**需要加粗的字**
                           "> 类型:<font color="info">XXX</font> 
" +  # 引用:> 需要引用的文字
                           "> 1XXX:<font color="warning">xxx例</font> 
" +  # 字体颜色(只支持3种内置颜色)
                           "> 2XXX:<font color="warning">xxx例</font>"  # 绿色:info、灰色:comment、橙红:warning
            }
        }

        res = requests.post(url = send_url, headers = headers, json = send_data)

        print(res.text)

if __name__ == '__main__':
    wx_robot = WXRobot()
    wx_robot.robot_send()

参考加入数据库数据校验:

class WXRobot:
    def get_doris_prd_engine(self):
        engine = create_engine(
            "mysql+pymysql://{}:{}@{}:{}/{}?charset={}".format('user_name', 'password', 'ip_addr', 'port', 'database_name', 'utf8'))
        return engine

    def robot_send(self):
        import requests

        headers = {"Content-Type": "text/plain"}

        send_url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx-xxx-xxxxxxxxaaa123'

        send_data = {
            "msgtype": "markdown",  # 消息类型,此时固定为markdown
            "markdown": {
                "content": "# **【测试】**<font color="warning">**xxx例**</font>
" +  # 标题 (支持1至6级标题,注意#与文字中间要有空格)
                           "#### **请注意**
" +  # 加粗:**需要加粗的字**
                           "> 类型:<font color="info">XXX</font> 
" +  # 引用:> 需要引用的文字
                           "> 1XXX:<font color="warning">xxx例</font> 
" +  # 字体颜色(只支持3种内置颜色)
                           "> 2XXX:<font color="warning">xxx例</font>"  # 绿色:info、灰色:comment、橙红:warning
            }
        }

        res = requests.post(url = send_url, headers = headers, json = send_data)

        print(res.text)

if __name__ == '__main__':
    engine =wx_robot.get_doris_prd_engine()
    result = pd.read_sql('show databases', engine)
    num_rows = result.shape[0]
    # print(result)
    print(num_rows) #todo 在此处添加SQL查询结果行数校验
    wx_robot = WXRobot()
    wx_robot.robot_send()