Python Flask 项目实战

# coding: utf-8
from application import db

class User(db.Model):
    __tablename__ = 'user'
    id = db.Column(db.Integer, primary_key=True, info='主键ID')
    nickname = db.Column(db.String(30), info='昵称')
    login_name = db.Column(db.String(20), unique=True, info='登录用户名')
    login_pwd = db.Column(db.String(32), info='登录密码')
    login_salt = db.Column(db.String(32), info='登录密码随机码')
    status = db.Column(db.Integer, server_default=db.FetchedValue(), info='状态 0: 无效 1: 有效')
    updated_time = db.Column(db.DateTime, server_default=db.FetchedValue(), info='最后更新时间')
    created_time = db.Column(db.DateTime, server_default=db.FetchedValue(), info='插入时间')

生成后修改model
# coding: utf-8
from application import db
from common.models.base.BaseMixin import BaseMixin

class User(db.Model, BaseMixin):   加入 BaseMixin

from flask import Blueprint, render_template, jsonify,request

from common.models.user import User

# 创建蓝图对象
member_page = Blueprint("member_page", __name__)


# 注册路由
@member_page.route("/register")
def register():
    return render_template("member/register.html")

@member_page.route("/login")
def login():
    return render_template("member/login.html")

from controller.member import member_page
app.register_blueprint(member_page, url_prefix="/member")

from application import app


class UrlManager(object):

    @staticmethod
    def buildUrl(path):
        config_domain = app.config['DOMAIN']
        return config_domain + path

    @staticmethod
    def buildStaticUrl(path):
        path = app.config['STATIC_PATH'] + path
        return UrlManager.buildUrl(path)

"""
模板函数
"""
from common.libs.UrlManager import UrlManager
app.add_template_global(UrlManager.buildStaticUrl, "buildStaticUrl")
app.add_template_global(UrlManager.buildUrl, "buildUrl")

import datetime

def getCurrentTime(frm="%Y-%m-%d %H:%M:%S"):
    dt = datetime.datetime.now()
    return dt.strftime(frm)

from application import app,CURRENT_ENV
from common.libs.DataHelper import getCurrentTime

class UrlManager(object):
    
    @staticmethod
    def buildStaticUrl(path):
        path = app.config['STATIC_PATH'] + path +"?ver="+UrlManager.getReleaseVersion()
        return UrlManager.buildUrl(path)
    
    @staticmethod
    def getReleaseVersion():
        if CURRENT_ENV == 'development':
            return getCurrentTime("%Y-%m-%d %H:%M:%S:%f")
        if CURRENT_ENV == 'production':
            return app.config['RELEASE_VERSION']
        if  CURRENT_ENV == 'testing':
            return app.config['RELEASE_VERSION']
        return getCurrentTime("%Y-%m-%d %H:%M:%S:%f")

from flask import jsonify  # 导入 Flask 的 jsonify 工具，用于生成 JSON 响应


def json_result_ok(code=200, msg='操作成功', data=None):
    """
    返回一个包含状态码、消息和数据的 JSON 响应。

    参数:
        code (int): HTTP 状态码，默认为 200，表示请求成功。
        msg (str): 操作结果的描述信息，默认为“操作成功”。
        data (any): 要返回给客户端的数据，默认为 None。

    返回:
        Response: Flask 的 Response 对象，包含 JSON 格式的响应数据。
    """
    return jsonify(code=code, msg=msg, data=data)



def json_error(msg='操作失败', code=500):
    """
    返回一个表示错误的 JSON 响应，默认状态码为 500。

    参数:
        msg (str): 错误信息，默认为“操作失败”。
        code (int): HTTP 状态码，默认为 500，表示服务器内部错误。

    返回:
        Response: Flask 的 Response 对象，包含 JSON 格式的错误响应。
    """
    return jsonify(code=code, msg=msg)

# 导入必要的模块
import string  # 提供字符串常量（如字母、数字等）
import random  # 用于随机操作
import hashlib  # 提供哈希加密功能（如 MD5）
# 从项目中导入 User 模型和数据库实例
from common.models.user import User
from application import db

class UserService(object):
    """
    用户服务类，提供与用户相关的业务逻辑，例如密码加盐加密、注册等。
    """

    @staticmethod
    def geneSalt(length=16):
        """
        生成指定长度的随机盐值（salt）。

        :param length: 盐值长度，默认为16位
        :return: 随机生成的 salt 字符串
        """
        # 构建包含大小写字母和数字的字符列表
        keylist = list(string.ascii_letters + string.digits)
        # 打乱顺序以增加随机性
        random.shuffle(keylist)
        # 取前 length 个字符作为 salt
        return ''.join(keylist[:length])

    @staticmethod
    def genePwd(pwd, salt):
        """
        对明文密码进行加盐哈希处理（使用 MD5 算法）。

        :param pwd: 明文密码
        :param salt: 盐值
        :return: 哈希后的密码字符串
        """
        md5 = hashlib.md5()
        # 将密码和盐拼接后进行 MD5 哈希
        md5.update((pwd + salt).encode('utf-8'))
        return md5.hexdigest()

    @staticmethod
    def checkPwd(pwd, hashed_pwd, salt):
        """
        验证明文密码是否与数据库中存储的哈希密码匹配。

        :param pwd: 明文密码（用户输入）
        :param hashed_pwd: 数据库中存储的哈希密码
        :param salt: 数据库中存储的盐值
        :return: True 如果密码正确，否则 False
        """
        # 使用相同的方式重新生成哈希密码，并与数据库中的比较
        return UserService.genePwd(pwd, salt) == hashed_pwd

    @staticmethod
    def register(login_name, login_pwd):
        """
        注册新用户，并将用户信息保存到数据库中。

        :param login_name: 用户登录名
        :param login_pwd: 用户登录密码（明文）
        """
        # 创建 User 模型对象
        model_user = User()
        # 设置用户名
        model_user.login_name = login_name
        # 生成随机盐值
        model_user.login_salt = UserService.geneSalt()
        # 对密码进行哈希加密
        model_user.login_pwd = UserService.genePwd(login_pwd, model_user.login_salt)
        # 默认昵称设置为登录名
        model_user.nickname = login_name
        # 添加用户到数据库会话
        db.session.add(model_user)
        # 提交会话，保存数据到数据库
        db.session.commit()

# 注册路由
@member_page.route("/register", methods=["GET", "POST"])
def register():
    if request.method == "GET":
        # GET 请求：渲染注册页面
        return render_template("member/register.html")
    elif request.method == "POST":
        # POST 请求：处理注册逻辑
        # 获取所有请求参数
        req = request.values
        # 获取三个字段
        login_name = req.get("login_name", "").strip()
        login_pwd = req.get("login_pwd", "").strip()
        login_pwd2 = req.get("login_pwd2", "").strip()
        # 验证1：字段是否为空
        if not login_name or not login_pwd or not login_pwd2:
            return json_error(msg="请输入完整的用户名、密码和确认密码！")
        # 验证2：密码长度是否小于6位
        if len(login_pwd) < 6:
            return json_error(msg="密码长度不能小于6位！")
        # 验证3：两次密码是否一致
        if login_pwd != login_pwd2:
            return json_error(msg="两次输入的密码不一致！")

        userInfo = User.query.filter_by(login_name=login_name).first()
        if userInfo:
            return json_error(msg="用户已存在！")
        # 模拟注册逻辑（此处应替换为真实数据库操作）
        try:
            UserService.register(login_name, login_pwd)
            print(f"注册成功：{login_name}")
            return json_result_ok()  # 模拟成功响应
        except Exception as e:
            # 捕获异常并返回错误信息
            return json_error(msg=f"系统异常：{str(e)}")

@member_page.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "GET":
        return render_template("member/login.html")
    elif request.method == "POST":
        # 获取 POST 参数
        req = request.values
        login_name = req.get("login_name", "").strip()
        login_pwd = req.get("login_pwd", "").strip()
        remember_me = int(req.get("remember_me", 0))  # 0 或 1
        # 验证字段是否为空
        if not login_name or not login_pwd:
            return json_error(msg="请输入用户名和密码！")
        # 调用 UserService.check_login 方法
        user_info = UserService.check_login(login_name, login_pwd)
        if not user_info:
            return json_error(msg="用户名或密码错误！")
        # 登录成功：设置 session
        session['user'] = {
            'id': user_info.id,
            'login_name': user_info.login_name,
            'nickname': user_info.nickname,
        }
        # 可选：记住我功能（延长 session 过期时间）
        if remember_me:
            session.permanent = True  # 设置为长期有效（默认是关闭浏览器就失效）
        return json_result_ok()

@member_page.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "GET":
        return ops_render("member/login.html")
    elif request.method == "POST":
        # 获取 POST 参数
        req = request.values
        login_name = req.get("login_name", "").strip()
        login_pwd = req.get("login_pwd", "").strip()
        # 验证字段是否为空
        if not login_name or not login_pwd:
            return json_error(msg="请输入用户名和密码！")
        # 调用 UserService.check_login 方法
        user_info = UserService.check_login(login_name, login_pwd)
        if not user_info:
            return json_error(msg="用户名或密码错误！")
        response = make_response(json_result_ok(200, "登录成功！"))
        response.set_cookie('user', UserService.geneAuthCode(user_info))
        return response


@member_page.route("/logout", methods=["GET"])
def logout():
    # 如果你使用的是 session
    response=make_response(redirect(UrlManager.buildUrl('/')))
    response.delete_cookie('user')
    return response

    @staticmethod
    def geneAuthCode(user_info):
        """
        生成用户认证码，包含 MD5 签名。

        :param user_info: 用户信息对象（需包含 id、nickname）
        :return: 生成的认证码字符串（格式：user_id,nickname,timestamp,nonce,signature）
        """
        time_stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        nonce = str(random.randint(0, 10000))  # 随机数（nonce）

        # 构造签名原文（注意使用统一格式拼接）
        raw_signature = f"{user_info.id}{user_info.nickname}{time_stamp}{nonce}"+app.config['SECRET_KEY']
        signature = hashlib.md5(raw_signature.encode()).hexdigest()

        # 返回拼接好的认证码（含签名）
        auth_code = f"{user_info.id},{user_info.nickname},{time_stamp},{nonce},{signature}"
        return auth_code

    @staticmethod
    def decodeAuthCode(auth_code):
        """
        解码并验证认证码的合法性。

        :param auth_code: 认证书符串（格式：user_id,nickname,timestamp,nonce,signature）
        :return: tuple (is_valid, data)，其中 data 包含 user_id、nickname 和时间戳
        """
        try:
            parts = auth_code.split(',')
            if len(parts) != 5:
                return False, None

            user_id_str, nickname, timestamp_str, nonce, received_signature = parts

            # 转换类型
            user_id = int(user_id_str)
            timestamp = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S")

            # 重新计算签名
            raw_signature = f"{user_id}{nickname}{timestamp_str}{nonce}"+app.config['SECRET_KEY']
            expected_signature = hashlib.md5(raw_signature.encode()).hexdigest()

            # 校验签名是否一致
            if expected_signature != received_signature:
                return False, None

            # 可选：校验有效期（比如5分钟）
            expiration = timestamp + timedelta(minutes=5)
            if datetime.now() > expiration:
                return False, None  # 已过期

            return True, {
                "user_id": user_id,
                "nickname": nickname,
                "timestamp": timestamp_str
            }

        except Exception as e:
            print("解码失败:", str(e))
            return False, None

import json

from flask import request, g
from application import app
from service.UserService import UserService

"""
请求拦截处理器
"""


# 先拦截
@app.before_request
def before_request():
    app.logger.info("拦截器！-----------------before_request-----------------")
    userInfo = check_login()
    app.logger.info(userInfo)
    if userInfo:
        g.current_user = userInfo
    return


# 后拦截
@app.after_request
def after_request(response):
    app.logger.info("拦截器！-----------------after_request-----------------")
    return response


def check_login():
    cookies = request.cookies
    isAuth, userInfo = UserService.decodeAuthCode(cookies.get('user'))
    if not isAuth:
        return None
    return userInfo

def ops_render(template, context={}):
    if 'current_user' in g:
        context['current_user'] = g.current_user
    return render_template(template, **context)

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

def apscheduler_test():
    print(datetime. datetime. now(). strftime("%Y-%m-%d %H:%M:%S"))

scheduler = BlockingScheduler()
scheduler.add_job(apscheduler_test, 'cron', second="*/5")
scheduler.start()

from flask_apscheduler import APScheduler

# 实例化 APScheduler
scheduler = APScheduler()
scheduler.init_app(app)

from application import app,scheduler
from www import *


def apscheduler_test():
    import datetime
    print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

if __name__ == '__main__':
    app.apscheduler.add_job(func=apscheduler_test, trigger='cron', second="*/5",id="apscheduler_test")
    scheduler.start()
    app.run(
        host="0.0.0.0",
        port=5000
    )