阿里云邮件推送服务 Python 发送示例

aliyun-dm-send-email.py

"""
https://dm.console.aliyun.com/
https://help.aliyun.com/document_detail/29444.html
https://www.jeesns.cn/article/detail/3422
https://text-compare.com/
"""
import time, datetime
import hmac
import base64
from urllib.parse import quote
import requests
from log import log


def percentEncode(str):
    res = quote(str.encode('utf8'), '')
    res = res.replace('+', '%20')
    res = res.replace('*', '%2A')
    res = res.replace('%7E', '~')
    return res


def get_signature(method, data, accessSecret):
    CanonicalizedQueryString = '&'.join(
        [
            f'{percentEncode(k)}={percentEncode(v)}'
            for k, v in sorted(data.items(), key=lambda x: x[0])
        ]
    )
    StringToSign = (
        method
        + "&"
        + percentEncode("/")
        + "&"
        + percentEncode(CanonicalizedQueryString)
    ).encode('utf8')

    key = f'{accessSecret}&'.encode('utf8')
    h = hmac.new(key, StringToSign, digestmod='sha1')
    rv = base64.b64encode(h.digest()).decode()
    return StringToSign.decode(), rv


def send_mail(
    html,
    subject='订单待支付',
    from_email='noreply@mail.example.com',
    to_emails='xxx@gmail.com',
    accessKeyId='xxx',
    accessSecret='xxx',
):
    headers = {
        'Accept': 'application/json',
    }

    data = {
        'Action': 'SingleSendMail',
        'AccountName': from_email,
        'AddressType': '1',
        'ReplyToAddress': 'true',
        'Subject': subject,
        'ToAddress': to_emails,
        'ClickTrace': '1',
        'HtmlBody': html,
        # sign
        'AccessKeyId': accessKeyId,
        'Version': '2015-11-23',
        'Format': 'JSON',
        'SignatureVersion': '1.0',
        'SignatureMethod': 'HMAC-SHA1',
        'SignatureNonce': str(int(time.time())),
        'Timestamp': datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ'),
    }

    str_to_sign, data['Signature'] = get_signature('POST', data, accessSecret)
    # print(data)
    try:
        response = requests.post('http://dm.aliyuncs.com/', headers=headers, data=data)
        # print(str_to_sign == response.json()['Message'].split(':', 1)[1].strip())
        return response.status_code == 200, response.json()
    except Exception as e:
        log(e)
        return False, e