# _*_ 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"""""" 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": "验证码错误" }