python 钉钉发邮件报警设置

1.异常信息内容编辑

# 异常信息发送至钉钉报警
if res.get("msg")=="查询失败":
    print("查询失败")
    XiaoTian.send_msg_to_dingding(f'hobby策略{platforms_Chinese[plat]}异常', name='hobby_check')

2.机器人需要设置关键字与接送方法

"""
@Time: 2022/12/10 10:54
@Desc: 
"""
import json
import requests
from retry import retry


class DingDingMSG(object):
    """
    机器人需要设置关键字：任务提醒
    """
    _ACCESS_TOKEN = ''

    @classmethod
    @retry(tries=5, delay=2)
    def send_msg_to_dingding(cls, content, name=None, logger=None, at_all=False):
        assert cls._ACCESS_TOKEN, 'access_token 不能为空'

        url = f'https://oapi.dingtalk.com/robot/send?access_token={cls._ACCESS_TOKEN}'
        headers = {
            "Content-Type": "application/json",
            "Charset": "UTF-8"
        }
        message = {
            "at": {
                "isAtAll": "true" if at_all else "false",
            },
            "msgtype": "text",
            # content 需要发送报警的内容
            "text": {
                "content": f"{name or ''} 任务提醒:\n{content or ''}"
            }
        }
        message_json = json.dumps(message)
        info = requests.post(url=url, data=message_json, headers=headers).json()
        if info['errcode'] == 0:
            return True
        else:
            if logger:
                logger.warning(f'发送给钉钉失败 {info}')
            else:
                print(info)

class XiaoTian(DingDingMSG):
    _ACCESS_TOKEN = '1d7f090a060550e283be0955772c1c6932073125fb6ee694288525be6fcf84b1'

if __name__ == '__main__':
    XiaoTian.send_msg_to_dingding('test123678', name='hobby_check', at_all=True)