252 lines
8.2 KiB
Python
252 lines
8.2 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
邮件服务工具类，使用腾讯云SES服务发送邮件
|
||
"""
|
||
|
||
import logging
|
||
import random
|
||
import string
|
||
import json
|
||
import time
|
||
from typing import Dict, Any, Optional
|
||
import requests
|
||
from tencentcloud.common import credential
|
||
from tencentcloud.common.profile.client_profile import ClientProfile
|
||
from tencentcloud.common.profile.http_profile import HttpProfile
|
||
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
|
||
from tencentcloud.ses.v20201002 import ses_client, models
|
||
|
||
from cryptoai.utils.config_loader import ConfigLoader
|
||
|
||
# 配置日志
|
||
logger = logging.getLogger('email_service')
|
||
|
||
# 验证码缓存，格式: {email: {'code': '123456', 'expire_time': timestamp}}
|
||
verification_codes = {}
|
||
|
||
class EmailService:
|
||
"""邮件服务类，用于发送验证码邮件"""
|
||
|
||
def __init__(self):
|
||
"""初始化邮件服务"""
|
||
# 加载腾讯云SES配置
|
||
config_loader = ConfigLoader()
|
||
self.ses_config = config_loader.get_ses_config()
|
||
|
||
# 初始化SES客户端
|
||
self.client = None
|
||
self._init_client()
|
||
|
||
def _init_client(self) -> None:
|
||
"""初始化腾讯云SES客户端"""
|
||
try:
|
||
# 创建认证对象
|
||
cred = credential.Credential(
|
||
self.ses_config.get('secret_id'),
|
||
self.ses_config.get('secret_key')
|
||
)
|
||
|
||
# 创建HTTP配置
|
||
http_profile = HttpProfile()
|
||
http_profile.endpoint = "ses.tencentcloudapi.com"
|
||
|
||
# 创建客户端配置
|
||
client_profile = ClientProfile()
|
||
client_profile.httpProfile = http_profile
|
||
|
||
# 创建SES客户端
|
||
self.client = ses_client.SesClient(cred, self.ses_config.get('region'), client_profile)
|
||
|
||
logger.info("成功初始化腾讯云SES客户端")
|
||
|
||
except TencentCloudSDKException as e:
|
||
logger.error(f"初始化腾讯云SES客户端失败: {e}")
|
||
self.client = None
|
||
|
||
def generate_verification_code(self, length: int = 6) -> str:
|
||
"""
|
||
生成数字验证码
|
||
|
||
Args:
|
||
length: 验证码长度，默认6位
|
||
|
||
Returns:
|
||
生成的验证码
|
||
"""
|
||
return ''.join(random.choices(string.digits, k=length))
|
||
|
||
def save_verification_code(self, email: str, code: str, expire_minutes: int = 10) -> None:
|
||
"""
|
||
保存验证码到缓存
|
||
|
||
Args:
|
||
email: 邮箱地址
|
||
code: 验证码
|
||
expire_minutes: 过期时间（分钟），默认10分钟
|
||
"""
|
||
expire_time = time.time() + expire_minutes * 60
|
||
verification_codes[email] = {
|
||
'code': code,
|
||
'expire_time': expire_time
|
||
}
|
||
|
||
def verify_code(self, email: str, code: str) -> bool:
|
||
"""
|
||
验证邮箱验证码
|
||
|
||
Args:
|
||
email: 邮箱地址
|
||
code: 验证码
|
||
|
||
Returns:
|
||
验证是否成功
|
||
"""
|
||
# 检查验证码是否存在
|
||
if email not in verification_codes:
|
||
return False
|
||
|
||
# 获取验证码信息
|
||
code_info = verification_codes[email]
|
||
|
||
# 检查验证码是否过期
|
||
if time.time() > code_info['expire_time']:
|
||
# 删除过期验证码
|
||
del verification_codes[email]
|
||
return False
|
||
|
||
# 验证验证码
|
||
if code_info['code'] == code:
|
||
# 验证成功后删除验证码
|
||
del verification_codes[email]
|
||
return True
|
||
|
||
return False
|
||
|
||
def send_verification_email(self, email: str) -> Dict[str, Any]:
|
||
"""
|
||
发送验证码邮件
|
||
|
||
Args:
|
||
email: 收件人邮箱
|
||
|
||
Returns:
|
||
发送结果
|
||
"""
|
||
if not self.client:
|
||
try:
|
||
self._init_client()
|
||
except Exception as e:
|
||
logger.error(f"重新初始化SES客户端失败: {e}")
|
||
return {
|
||
'success': False,
|
||
'message': '邮件服务初始化失败'
|
||
}
|
||
|
||
try:
|
||
# 生成验证码
|
||
code = self.generate_verification_code()
|
||
|
||
# 保存验证码
|
||
self.save_verification_code(email, code)
|
||
|
||
# 构建邮件内容
|
||
subject = "CryptoAI - 您的验证码"
|
||
html_content = f"""
|
||
<html>
|
||
<head>
|
||
<style>
|
||
body {{ font-family: Arial, sans-serif; line-height: 1.6; }}
|
||
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
|
||
.header {{ background-color: #4A90E2; color: white; padding: 10px; text-align: center; }}
|
||
.content {{ padding: 20px; }}
|
||
.code {{ font-size: 24px; font-weight: bold; text-align: center;
|
||
color: #4A90E2; padding: 10px; margin: 20px 0; }}
|
||
.footer {{ font-size: 12px; color: #999; text-align: center; margin-top: 30px; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
<div class="header">
|
||
<h2>CryptoAI 验证码</h2>
|
||
</div>
|
||
<div class="content">
|
||
<p>您好，</p>
|
||
<p>您正在注册 CryptoAI 账号，请使用以下验证码完成注册：</p>
|
||
<div class="code">{code}</div>
|
||
<p>验证码有效期为10分钟，请勿将验证码泄露给他人。</p>
|
||
<p>如果您没有进行此操作，请忽略此邮件。</p>
|
||
</div>
|
||
<div class="footer">
|
||
<p>此邮件由系统自动发送，请勿回复。</p>
|
||
<p>© {time.strftime('%Y')} CryptoAI. 保留所有权利。</p>
|
||
</div>
|
||
</div>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
# 创建发送邮件请求
|
||
req = models.SendEmailRequest()
|
||
req.FromEmailAddress = self.ses_config.get('from_email')
|
||
req.Destination = [email]
|
||
req.Subject = subject
|
||
req.Template = {
|
||
"TemplateID": self.ses_config.get('template_id', 1), # 使用默认模板ID或配置的模板ID
|
||
"TemplateData": json.dumps({
|
||
"code": code
|
||
})
|
||
}
|
||
|
||
# 如果没有配置模板ID，则使用HTML内容
|
||
if not self.ses_config.get('template_id'):
|
||
req.Template = None
|
||
req.Simple = {
|
||
"Html": html_content
|
||
}
|
||
|
||
# 发送邮件
|
||
resp = self.client.SendEmail(req)
|
||
|
||
logger.info(f"成功发送验证码邮件到 {email}")
|
||
|
||
return {
|
||
'success': True,
|
||
'message': '验证码已发送',
|
||
'request_id': resp.RequestId
|
||
}
|
||
|
||
except TencentCloudSDKException as e:
|
||
logger.error(f"发送验证码邮件失败: {e}")
|
||
return {
|
||
'success': False,
|
||
'message': f'发送验证码失败: {e}'
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"发送验证码邮件出现未知错误: {e}")
|
||
return {
|
||
'success': False,
|
||
'message': f'发送验证码失败: {str(e)}'
|
||
}
|
||
|
||
# 单例模式
|
||
_email_service_instance = None
|
||
|
||
def get_email_service() -> EmailService:
|
||
"""
|
||
获取邮件服务实例（单例模式）
|
||
|
||
Returns:
|
||
邮件服务实例
|
||
"""
|
||
global _email_service_instance
|
||
|
||
# 如果已经初始化过，直接返回
|
||
if _email_service_instance is not None:
|
||
return _email_service_instance
|
||
|
||
# 创建实例
|
||
_email_service_instance = EmailService()
|
||
|
||
return _email_service_instance |