Python实现钉钉通报和邮件通知

#python 3.8
import time
import hmac
import hashlib
import base64
import urllib.parse

timestamp = str(round(time.time() * 1000))
secret = 'this is secret'
secret_enc = secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))

url = f"{webhook}&timestamp={timestamp}&sign={sign}"
# 钉钉文本消息格式
data = {
      "at": {
                "isAtAll": at_all
            },
            "text": {
                "content": "测试通知"
            },
            "msgtype": "text"
  }

response = requests.post(url=url, data=json.dumps(data), headers=headers).textp)
print(response)

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr

my_sender = 'test@foxmail.com'  # 发件人邮箱账号
my_pass = 'XXXXXXXXXXX'  # 发件人邮箱密码,若使用腾讯邮箱，则填对应的SMTP服务授权码
my_user = 'test@foxmail.com'  # 收件人邮箱账号，我这边发送给自己


def mail():
    ret = True
    try:
        msg = MIMEText('数据采集正常', 'plain', 'utf-8')
        msg['From'] = formataddr(["通知助手", my_sender])  # 括号里的对应发件人邮箱昵称、发件人邮箱账号
        msg['To'] = formataddr(["FK", my_user])  # 括号里的对应收件人邮箱昵称、收件人邮箱账号
        msg['Subject'] = "[任务执行成功] -------"  # 邮件的主题，也可以说是标题

        server = smtplib.SMTP_SSL("smtp.qq.com", 465)  # 发件人邮箱中的SMTP服务器，端口是25
        server.login(my_sender, my_pass)  # 括号中对应的是发件人邮箱账号、邮箱密码
        server.sendmail(my_sender, [my_user, ], msg.as_string())  # 括号中对应的是发件人邮箱账号、收件人邮箱账号、发送邮件
        server.quit()  # 关闭连接
    except Exception as e:  # 如果 try 中的语句没有执行，则会执行下面的 ret=False
        ret = False
        print(e)
    return ret


ret = mail()
if ret:
    print("邮件发送成功")
else:
    print("邮件发送失败")

import json
import time
import hmac
import hashlib
import base64
import urllib.parse
import requests

def create_noticer(enable_mail=False, enable_dingtalk=True):
    noticer = Noticer()
    if enable_mail:
        noticer.add_handler(MailHandler())
    if enable_dingtalk:
        noticer.add_handler(DingTalkHandler())
    return noticer

class Handler:
    """通知处理程序"""
    def send(self, msg):
        """作为接口"""
        pass

class DingTalkHandler(Handler):
    """钉钉通知处理程序"""
    def __init__(self):
        self.init_variables()
    def init_variables(self):
        # 密钥和webhook可抽出到配置文件中
        self.secret = "XXX"
        self.webhook = 'https://oapi.dingtalk.com/robot/send?access_token=XXX'

    def send(self, msg, at_all=False):
        """
        发送消息，当前只实现文本类型消息发送
        :param msg:
        :param at_all:
        :return:
        """
        headers = {'Content-Type': 'application/json', "Charset": "UTF-8"}
        #  webhook 地址

        timestamp = str(round(time.time() * 1000))
        # 加签密钥
        secret_enc = self.secret.encode("utf-8")
        string_to_sign = '{}\n{}'.format(timestamp, self.secret)
        string_to_sign_enc = string_to_sign.encode("utf-8")
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        url = f"{self.webhook}&timestamp={timestamp}&sign={sign}"
        # 钉钉消息格式，其中 msg 就是我们要发送的具体内容
        data = {
            "at": {
                "isAtAll": at_all
            },
            "text": {
                "content": msg
            },
            "msgtype": "text"
        }

        return requests.post(url=url, data=json.dumps(data), headers=headers).text



class MailHandler(Handler):
    """邮件通知处理程序"""
    pass

class Noticer:
    """
    消息通知器
    """
    def __init__(self):
        self.handlers = []

    def notice(self, msg):
        # 对msg进行一定的封装、定制、格式化，此处省略
        # ...
        self.handle(msg)

    def handle(self, msg):
        for handler in self.handlers:
            handler.send(msg)

    def add_handler(self, hdlr):
        """
        添加处理程序
        WARNING: 如果Handler使用了文件或数据库，则该方法线程不安全，需针对数据同步问题进行优化
        :param hdlr:
        :return:
        """
        if not (hdlr in self.handlers):
            self.handlers.append(hdlr)

    def remove_handler(self, hdlr):
        """
        移除处理程序
        WARNING: 如果Handler使用了文件或数据库，则该方法线程不安全

        :param hdlr:
        :return:
        """
        if hdlr in self.handlers:
            self.handlers.remove(hdlr)