forked from xiaohei/taiji-AI-PAD
223 lines
7.9 KiB
Python
223 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
阿里云 DirectMail API 邮件发送测试脚本
|
||
使用 SingleSendMail API 发送邮件（无需 SMTP 密码）
|
||
|
||
文档参考: https://www.alibabacloud.com/help/en/direct-mail/singlesendmail
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
|
||
# ==================== 配置区域 ====================
|
||
# 阿里云 AccessKey（可以在阿里云控制台 -> AccessKey 管理中获取）
|
||
ACCESS_KEY_ID = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID", "your_access_key_id")
|
||
ACCESS_KEY_SECRET = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET", "your_access_key_secret")
|
||
|
||
# DirectMail 配置
|
||
# 注意：使用新加坡区域的 endpoint
|
||
REGION_ID = "ap-southeast-1" # 新加坡
|
||
ENDPOINT = "dm.ap-southeast-1.aliyuncs.com"
|
||
|
||
# 发件人配置（必须是在 DirectMail 控制台已验证的发信地址）
|
||
ACCOUNT_NAME = "supportagnet@taijiaicloud.com"
|
||
|
||
# 收件人
|
||
TO_ADDRESS = "zsbgnw@gmail.com"
|
||
|
||
# ==================== 测试代码 ====================
|
||
|
||
def check_sdk():
|
||
"""检查并安装 SDK"""
|
||
try:
|
||
from alibabacloud_dm20151123.client import Client
|
||
from alibabacloud_dm20151123 import models as dm_models
|
||
from alibabacloud_tea_openapi import models as open_api_models
|
||
print("✓ 阿里云 DirectMail SDK 已安装")
|
||
return True
|
||
except ImportError:
|
||
print("✗ 缺少阿里云 SDK，请运行以下命令安装：")
|
||
print(" pip install alibabacloud-dm20151123")
|
||
return False
|
||
|
||
|
||
def send_email_via_api():
|
||
"""使用 SingleSendMail API 发送邮件"""
|
||
from alibabacloud_dm20151123.client import Client
|
||
from alibabacloud_dm20151123 import models as dm_models
|
||
from alibabacloud_tea_openapi import models as open_api_models
|
||
from alibabacloud_tea_util import models as util_models
|
||
|
||
print(f"\n{'='*50}")
|
||
print("使用 SingleSendMail API 发送邮件")
|
||
print(f"{'='*50}")
|
||
print(f"发件人: {ACCOUNT_NAME}")
|
||
print(f"收件人: {TO_ADDRESS}")
|
||
print(f"区域: {REGION_ID}")
|
||
print(f"{'='*50}\n")
|
||
|
||
# 检查 AccessKey
|
||
if ACCESS_KEY_ID == "your_access_key_id" or ACCESS_KEY_SECRET == "your_access_key_secret":
|
||
print("✗ 错误: 请配置 AccessKey")
|
||
print("\n方式1: 设置环境变量")
|
||
print(" export ALIBABA_CLOUD_ACCESS_KEY_ID='你的AccessKeyId'")
|
||
print(" export ALIBABA_CLOUD_ACCESS_KEY_SECRET='你的AccessKeySecret'")
|
||
print("\n方式2: 直接修改脚本中的配置")
|
||
return False
|
||
|
||
try:
|
||
print("1. 创建 DirectMail 客户端...")
|
||
config = open_api_models.Config(
|
||
access_key_id=ACCESS_KEY_ID,
|
||
access_key_secret=ACCESS_KEY_SECRET,
|
||
endpoint=ENDPOINT
|
||
)
|
||
client = Client(config)
|
||
print(" ✓ 客户端创建成功")
|
||
|
||
print("2. 构建邮件请求...")
|
||
request = dm_models.SingleSendMailRequest(
|
||
account_name=ACCOUNT_NAME,
|
||
address_type=1, # 1: 发信地址
|
||
reply_to_address="false", # 不使用回复地址
|
||
subject="Taiji AI-PAD 测试邮件 (API)",
|
||
to_address=TO_ADDRESS,
|
||
html_body="""
|
||
<html>
|
||
<body style="font-family: Arial, sans-serif; padding: 20px;">
|
||
<h2 style="color: #333;">Taiji AI-PAD 邮件测试</h2>
|
||
<p>这是一封通过 <strong>阿里云 DirectMail API</strong> 发送的测试邮件。</p>
|
||
<p>如果您收到此邮件，说明 API 配置正确！</p>
|
||
<hr style="border: none; border-top: 1px solid #eee; margin: 20px 0;">
|
||
<p style="color: #666; font-size: 12px;">
|
||
--- Taiji AI-PAD 团队
|
||
</p>
|
||
</body>
|
||
</html>
|
||
""",
|
||
tag_name="test" # 可选：邮件标签
|
||
)
|
||
print(" ✓ 请求构建完成")
|
||
|
||
print("3. 发送邮件...")
|
||
runtime = util_models.RuntimeOptions()
|
||
response = client.single_send_mail_with_options(request, runtime)
|
||
|
||
print(" ✓ 邮件发送成功!")
|
||
print(f"\n响应详情:")
|
||
print(f" RequestId: {response.body.request_id}")
|
||
print(f" EnvId: {response.body.env_id}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f" ✗ 发送失败: {e}")
|
||
|
||
# 解析常见错误
|
||
error_msg = str(e)
|
||
if "InvalidAccessKeyId" in error_msg:
|
||
print("\n可能原因: AccessKeyId 无效")
|
||
elif "SignatureDoesNotMatch" in error_msg:
|
||
print("\n可能原因: AccessKeySecret 错误")
|
||
elif "InvalidMailAddress.NotFound" in error_msg:
|
||
print("\n可能原因: 发信地址未在 DirectMail 控制台配置或未验证")
|
||
elif "InvalidIP.NotFound" in error_msg:
|
||
print("\n可能原因: 如果启用了 IP 保护，当前 IP 不在白名单中")
|
||
|
||
return False
|
||
|
||
|
||
def send_email_simple():
|
||
"""简化版发送（直接使用 HTTP 请求，不依赖 SDK）"""
|
||
import hmac
|
||
import hashlib
|
||
import base64
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
import datetime
|
||
import json
|
||
|
||
print(f"\n{'='*50}")
|
||
print("使用 HTTP 请求直接调用 API")
|
||
print(f"{'='*50}\n")
|
||
|
||
if ACCESS_KEY_ID == "your_access_key_id":
|
||
print("✗ 请先配置 AccessKey")
|
||
return False
|
||
|
||
# 构建请求参数
|
||
params = {
|
||
"Action": "SingleSendMail",
|
||
"AccountName": ACCOUNT_NAME,
|
||
"AddressType": "1",
|
||
"ReplyToAddress": "false",
|
||
"Subject": "Taiji AI-PAD 测试邮件 (HTTP)",
|
||
"ToAddress": TO_ADDRESS,
|
||
"TextBody": "这是一封测试邮件。如果您收到此邮件，说明配置正确。\n\n--- Taiji AI-PAD 团队",
|
||
"Format": "JSON",
|
||
"Version": "2015-11-23",
|
||
"AccessKeyId": ACCESS_KEY_ID,
|
||
"SignatureMethod": "HMAC-SHA1",
|
||
"Timestamp": datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||
"SignatureVersion": "1.0",
|
||
"SignatureNonce": str(uuid.uuid4()),
|
||
}
|
||
|
||
# 计算签名
|
||
def compute_signature(params, access_key_secret):
|
||
sorted_params = sorted(params.items())
|
||
query_string = urllib.parse.urlencode(sorted_params, quote_via=urllib.parse.quote)
|
||
string_to_sign = "GET&%2F&" + urllib.parse.quote(query_string, safe="")
|
||
key = (access_key_secret + "&").encode("utf-8")
|
||
signature = base64.b64encode(
|
||
hmac.new(key, string_to_sign.encode("utf-8"), hashlib.sha1).digest()
|
||
).decode("utf-8")
|
||
return signature
|
||
|
||
params["Signature"] = compute_signature(params, ACCESS_KEY_SECRET)
|
||
|
||
# 发送请求
|
||
url = f"https://{ENDPOINT}/?" + urllib.parse.urlencode(params)
|
||
|
||
try:
|
||
print("发送请求...")
|
||
req = urllib.request.Request(url)
|
||
with urllib.request.urlopen(req, timeout=30) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
print("✓ 发送成功!")
|
||
print(f" RequestId: {result.get('RequestId')}")
|
||
print(f" EnvId: {result.get('EnvId')}")
|
||
return True
|
||
except urllib.error.HTTPError as e:
|
||
error_body = e.read().decode("utf-8")
|
||
print(f"✗ 发送失败: {e.code}")
|
||
print(f" 错误信息: {error_body}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"✗ 发送失败: {e}")
|
||
return False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("\n" + "="*60)
|
||
print(" 阿里云 DirectMail API 邮件发送测试")
|
||
print("="*60)
|
||
|
||
# 检查 SDK
|
||
sdk_available = check_sdk()
|
||
|
||
if sdk_available:
|
||
# 使用 SDK 发送
|
||
send_email_via_api()
|
||
else:
|
||
print("\n如果不想安装 SDK，可以使用简化版（HTTP 直接请求）")
|
||
choice = input("是否使用简化版发送? (y/n): ").strip().lower()
|
||
if choice == 'y':
|
||
send_email_simple()
|
||
|
||
print("\n" + "="*60)
|
||
print("测试完成")
|
||
print("="*60 + "\n")
|
||
|