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()