1.下载redis的服务器端,现在本地运行,因为没云服务器

windows版本,官网没有win版

参考文章Window下Redis的安装和部署详细图文教程(Redis的安装和可视化工具的使用)_redis安装-CSDN博客2. 在项目环境下下载redis客户端,pip install redis

3.先建一个文件夹utils用来保存工具,记得建立一个__init__.py文件,导包的时候python会将其加入搜索路径,在目录下建立redis_config.py文件,用来设置redis的配置信息,建立了redis客户端redis_client

#配置redis服务器的基本参数,用于验证码存储和定期销毁
#pip install redis安装redis客户端
import redis

# 创建Redis连接池,用于持续连接,每次连接不需要断开
redis_pool = redis.ConnectionPool(
    host='localhost',  # Redis服务器地址
    port=6379,         # Redis服务器端口
    db=0,              # 使用的数据库编号
    decode_responses=True  # 返回字符串而非字节
)

# 创建Redis连接对象
redis_client = redis.Redis(connection_pool=redis_pool)

4.之后在utils目录下建立captcha.py文件,里面书写保存和验证验证码的操作函数

# Redis 验证码工具类
from utils.redis_config import redis_client
import time

# 验证码默认有效期(5分钟)
CAPTCHA_EXPIRE_SECONDS = 5 * 60


def save_captcha(email: str, captcha: str):
    """保存验证码到Redis"""
    # 使用邮箱作为key,验证码作为value
    redis_client.setex(email, CAPTCHA_EXPIRE_SECONDS, captcha)


def verify_captcha(email: str, user_input: str) -> bool:
    """验证验证码是否有效"""
    # 获取存储的验证码
    stored_captcha = redis_client.get(email)

    # 检查验证码是否存在
    if not stored_captcha:
        return False

    # 验证验证码是否匹配
    is_valid = stored_captcha == user_input

    # 如果验证通过,删除验证码(一次性使用)
    if is_valid:
        redis_client.delete(email)

    return is_valid


def delete_captcha(email: str):
    """删除指定邮箱的验证码"""
    redis_client.delete(email)


def clear_expired_captchas():
    """清理所有过期的验证码(Redis会自动处理)"""
    # Redis会自动删除过期的key,不需要手动清理
    pass

5.在auth.py文件中更改用户点击获取验证码之后的操作逻辑,用户点击后会在后端产生验证码,并用redis暂存验证码,具体逻辑在captcha/email中

#登录功能
#从flask中导入蓝图对象
import random

from flask import Blueprint

#渲染登录页面模板
from flask import render_template

#导入格式化邮件信息的包
from flask_mail import Message

#导入邮件发送器
from exts import mail,db

#查询字符串获取参数
from flask import request

#原始可供提取验证码的字符串
import string

#从原始字符串中提取出任意几位的验证码
import random

#将字典或者列表转换为json响应
from flask import jsonify


#导入数据库对象
from models import EmailCaptchaModel

#从表单验证模块forms.py中导入表单验证功能
from .forms  import RegisterForm


#从models导入数据库写入库
from models import UserModel


#导入页面跳转包
from flask import redirect


#导入redis中存储验证码的函数
from utils.captcha_redis import save_captcha


bp = Blueprint("auth",__name__,url_prefix="/auth")

#/auth/login
@bp.route("/login")
def login():
    return "what"

#GET获取页面
#POST提交注册数据默认是GET
@bp.route("/register",methods=['GET','POST'])
def register():
    if request.method == 'GET':
        return render_template("register.html")
    else:
        #提交注册数据后,1.验证   2.验证通过转向登录界面,不通过返回注册界面
        #表单验证 flask_wtf wtforms
        #从forms.py中导入表单验证的部分验证数据是否正确
        form = RegisterForm(request.form)
        #如果验证通过就存储
        if form.validate():
            email = form.email.data
            username = form.username.data
            password = form.password.data
            #将注册用户数据存储进数据库
            user = UserModel(email=email,username=username,password=password)
            db.session.add(user)
            db.session.commit()
            #转向登录界面,通过视图函数找到url跳转
            return redirect("/auth/login")#第一种跳转方法,路径跳转法
            # return redirect(url_for("auth.login"))#蓝图跳转法
        else:#表单验证失败,重点位到注册页面

            print(form.errors)
            return "失败"



#当用户点击获取验证码的button时候,会在register.js中触发这个事件,然后执行下面的函数,发送验证码,并存储在数据库中
@bp.route("/captcha/email")
def get_email_captcha():
    #变量传参: /captcha/email/<email>
    #查询字符串:/captcha/email?email=xxx@qq.com
    email = request.args.get("email")
    if not email:
        return jsonify({"code": 400, "message": "邮箱不能为空"}), 400

    #验证码格式是4到6位的数字字母组合
    source = string.digits*4#0123456789012345678901234567890123456789
    captcha = random.sample(source,4)#从source中随机提取出4位字符,组成一个list
    captcha = "".join(captcha)#以“”为分隔符合并列表
    message = Message(subject="知乎注册验证码",recipients=[email],body=f"您的验证码是{captcha}")#写邮件,注意获取查询字符串的方式使用中括号
    mail.send(message)#发送邮件


    #使用数据库存储验证码用于验证,有memcached/redis两种暂存方法,这里使用数据库存储
    # email_captcha = EmailCaptchaModel(email=email,captcha=captcha)#赋值,一条记录,id不用写出来
    # db.session.add(email_captcha)
    # db.session.commit() #在terminal输入三个步骤进行数据库更新,1.flask db init (只执行一次) 2.flask db migrate 3.flask db upgrade


    #使用redis存储验证码1.from captcha_redis import save_captcha  2. save_captcha(email, captcha)
    save_captcha(email,captcha)#redis使用内存存储
    return jsonify({"code": 200,"message":"","data": None})



#测试用途,没有参与项目
@bp.route("/mail")
def mail_test():
    message = Message(subject="邮箱测试",recipients=["3125988872@qq.com"],body="这是一条测试信息")
    mail.send(message)
    return "邮件发送成功"



6.在forms.py文件中更改验证码的验证逻辑,将用户提交的验证码和redis中暂存的验证码进行比对

#表单验证
#包:pip install flask_wtf,会自动安装wtforms
import wtforms
#从wtforms.validators中导入检查邮件格式,验证码格式,的包
from wtforms.validators import Email,Length,EqualTo

#导入数据表模型,查看邮箱是否已经注册
from models import UserModel

#导入数据表,查询临时存储的邮箱和验证码的映射是否正确
from models import EmailCaptchaModel

#导入数据库
from exts import db

#导入时间库
from datetime import datetime,timedelta



#导入redis的处理函数from captcha_redis import verify_captcha, delete_captcha
from utils.captcha_redis import verify_captcha, delete_captcha

#Form: 用于验证前端提交的注册数据是否正确
#1.验证邮箱格式 ,验证码,密码,确认密码格式 2.邮箱是否已经注册 3。邮箱和验证码必须双正确
class RegisterForm(wtforms.Form):
    #
    email = wtforms.StringField(validators=[Email(message="邮件格式错误")])
    captcha = wtforms.StringField(validators=[Length(min=4,max=4,message="验证码格式错误!")])
    username = wtforms.StringField(validators=[Length(min=3,max=20,message="用户名格式错误")])
    password = wtforms.StringField(validators=[Length(min=6,max=20,message="密码格式错误")])
    password_confirm = wtforms.StringField(validators=[EqualTo("password")])

    #自定义验证
    #1.邮箱是否已经注册
    def validate_email(self, field):
        email = field.data#field指代上面的email对象,field.data获得属性
        # 注意:这里应该查询UserModel而不是EmailCaptchaModel
        user = UserModel.query.filter_by(email=email).first()#根据email这个值去查询表,看是否有记录相等
        if user:#存在,重复,不能重复注册
            raise wtforms.ValidationError(message="该邮箱已经被注册")
        #不存在就不做处理

    #2.验证验证码是否正确,邮箱必须和验证码一一对应
    def validate_captcha(self,field):
        captcha = field.data
        email = self.email.data
        # captcha_model = EmailCaptchaModel.query.filter_by(email=email,captcha=captcha),不加first是查询对象,不能被修改
        # 执行查询获取实际的模型实例
        # captcha_model = EmailCaptchaModel.query.filter_by(email=email, captcha=captcha).first()
        # if not captcha_model:#如果查不到记录,则错误
        #     raise wtforms.ValidationError(message="邮箱或者验证码错误")
        # else:#正确,删除验证码记录
        #     # todo : 可以删除captcha_model
        #     db.session.delete(captcha_model)
        #     db.session.commit()

        #查到验证码,验证码不能过期
        # now = datetime.now()
        # expire_time = captcha_model.create_time + timedelta(minutes=5)#总经历的时间
        #
        # if now > expire_time:
            #验证码过期,同时删除当前验证码记录
            # EmailCaptchaModel.query.filter_by(email=email,captcha=captcha).delete()
            # db.session.commit()
            # raise wtforms.ValidationError(message="验证码已过期,请重新获取")

        #验证码有效,删除该邮箱对应的所有验证码记录
        # try:
        #     EmailCaptchaModel.query.filter_by(email=email).delete()
        #     db.session.commit()
        # except Exception as e:
        #     db.session.rollback()
        #     raise wtforms.ValidationError(message="验证码处理失败,请重试")


        #使用redis验证验证码
        #1.导入redis的处理函数from captcha_redis import verify_captcha, delete_captcha
        #2.验证验证码是否正确且未过期
        # 使用Redis验证验证码
        if not verify_captcha(email, captcha):
            raise wtforms.ValidationError(message="邮箱或者验证码错误")

        # 验证通过后删除验证码
        delete_captcha(email)








总结:redis服务器->redis客户端(pip install redis)->redis服务器配置(redis.config)->建立redis客户端(redis_client)->前端请求验证码->后端产生验证码并暂存在redis中->前端收到验证码填写发到后端->后端在redis中比对结果,正确删除,错误报错,redis定期删除,本例是5*60,单位是秒

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐