107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
|
# _*_ coding : UTF-8 _*_
|
|||
|
|
# @Time : 2025/01/26 21:51
|
|||
|
|
# @UpdateTime : 2025/01/26 21:51
|
|||
|
|
# @Author : sonder
|
|||
|
|
# @File : mail.py
|
|||
|
|
# @Software : PyCharm
|
|||
|
|
# @Comment : 本程序
|
|||
|
|
import random
|
|||
|
|
from datetime import timedelta
|
|||
|
|
from email.message import EmailMessage
|
|||
|
|
from email.utils import formataddr
|
|||
|
|
|
|||
|
|
from aiosmtplib import send
|
|||
|
|
from fastapi import Request
|
|||
|
|
from jinja2 import Environment, FileSystemLoader
|
|||
|
|
|
|||
|
|
from config.constant import RedisKeyConfig
|
|||
|
|
from config.env import AppConfig, EmailConfig
|
|||
|
|
from utils.log import logger
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Email:
|
|||
|
|
"""
|
|||
|
|
邮件发送类，用于发送邮件。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
async def generate_verification_code(cls, length: int = 4) -> str:
|
|||
|
|
"""
|
|||
|
|
随机生成数字验证码
|
|||
|
|
:param length: 验证码长度
|
|||
|
|
:return:
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
return ''.join(str(random.randint(0, 9)) for _ in range(length))
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
async def send_email(cls, request: Request, username: str, title: str = "注册", mail: str = "") -> bool:
|
|||
|
|
"""
|
|||
|
|
发送邮件
|
|||
|
|
:param request: 请求对象
|
|||
|
|
:param username: 用户账号
|
|||
|
|
:param title: 邮件标题
|
|||
|
|
:param mail: 邮箱地址
|
|||
|
|
"""
|
|||
|
|
code = await cls.generate_verification_code(4)
|
|||
|
|
codeStr = ""
|
|||
|
|
for i in code:
|
|||
|
|
codeStr += f"""<button class="button">{i}</button>"""
|
|||
|
|
env = Environment(loader=FileSystemLoader('templates'))
|
|||
|
|
template = env.get_template('mail_en.html')
|
|||
|
|
content = template.render(
|
|||
|
|
TITLE=title,
|
|||
|
|
CODE=codeStr,
|
|||
|
|
PROJECTNAME=AppConfig.app_name,
|
|||
|
|
)
|
|||
|
|
subject = f"{AppConfig.app_name}-{title} Verification Code"
|
|||
|
|
sendName = AppConfig.app_name
|
|||
|
|
hostname = EmailConfig.email_host
|
|||
|
|
port = EmailConfig.email_port
|
|||
|
|
message = EmailMessage()
|
|||
|
|
message["From"] = formataddr((sendName, EmailConfig.email_username))
|
|||
|
|
message["To"] = mail
|
|||
|
|
message["Subject"] = subject
|
|||
|
|
message.set_content(content, subtype="html")
|
|||
|
|
try:
|
|||
|
|
await send(
|
|||
|
|
message,
|
|||
|
|
hostname=hostname,
|
|||
|
|
port=port,
|
|||
|
|
username=EmailConfig.email_username,
|
|||
|
|
password=EmailConfig.email_password
|
|||
|
|
)
|
|||
|
|
await request.app.state.redis.set(f"{RedisKeyConfig.EMAIL_CODES.key}:{mail}-{username}", code,
|
|||
|
|
ex=timedelta(minutes=2))
|
|||
|
|
logger.info(f"发送邮件至{mail}成功，验证码：{code}")
|
|||
|
|
return True
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(e)
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
async def verify_code(cls, request: Request, username: str, mail: str, code: str) -> dict:
|
|||
|
|
"""
|
|||
|
|
验证验证码
|
|||
|
|
:param request: 请求对象
|
|||
|
|
:param username: 用户账号
|
|||
|
|
:param mail: 邮箱地址
|
|||
|
|
:param code: 验证码
|
|||
|
|
"""
|
|||
|
|
redis_code = await request.app.state.redis.get(f"{RedisKeyConfig.EMAIL_CODES.key}:{mail}-{username}")
|
|||
|
|
if redis_code is None:
|
|||
|
|
return {
|
|||
|
|
"status": False,
|
|||
|
|
"msg": "验证码已过期"
|
|||
|
|
}
|
|||
|
|
if str(redis_code).lower() == code.lower():
|
|||
|
|
await request.app.state.redis.delete(f"{RedisKeyConfig.EMAIL_CODES.key}:{mail}-{username}")
|
|||
|
|
return {
|
|||
|
|
"status": True,
|
|||
|
|
"msg": "验证码正确"
|
|||
|
|
}
|
|||
|
|
return {
|
|||
|
|
"status": False,
|
|||
|
|
"msg": "验证码错误"
|
|||
|
|
}
|