Python Flask 项目实战
Python Flask 项目实战
1.注册与登录
①.猫影前台功能
会员数据表设计
CREATE DATABASE `movie_cat` CHARACTER SET = `utf8mb4`;
use `movie_cat`;
CREATE TABLE `user` (`id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键ID',`nickname` varchar(30) DEFAULT NULL COMMENT '昵称',`login_name` varchar(20) DEFAULT NULL COMMENT '登录用户名',`login_pwd` varchar(32) DEFAULT NULL COMMENT '登录密码',`login_salt` varchar(32) DEFAULT NULL COMMENT '登录密码随机码',`status` tinyint(3) DEFAULT '1' COMMENT '状态 0: 无效 1: 有效',`updated_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最后更新时间',`created_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '插入时间',PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='会员表';ALTER TABLE `user` ADD UNIQUE INDEX `uk_login_name` (`login_name`);
执行生成文件
flask-sqlacodegen 'mysql://root:yanpeng2580@127.0.0.1/movie_cat' --tables user --outfile "common/models/user.py"
# coding: utf-8
from application import dbclass 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 BaseMixinclass User(db.Model, BaseMixin): 加入 BaseMixin
②.搭建登录注册页面
网页有什么构成
- 浏览器只能解析HTML
- CSS控制页面样式
- Javascript提供更好的交互体验
前端框架：Bootstrap
Bootstrap 入门 · Bootstrap v5 中文文档 v5.3 | Bootstrap 中文网
引入jquery和bootstrapv5
templates/common/layout.html
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>{% block title %}猫影影视{% endblock %}</title><link rel="stylesheet" href="/static/plugins/bootstrapv5/css/bootstrap.css"><style>html, body {height: 100%;margin: 0;font-size: 12px;}.page-container {display: flex;flex-direction: column;min-height: 100vh;}.content-wrapper {flex: 1; /* 占据除 header 和 footer 外的所有空间 */overflow-y: auto; /* 超出时显示滚动条 */}.footer {background-color: #212529;color: #fff;padding: 20px 0;text-align: center;}.footer a {color: #ccc;text-decoration: none;}.footer a:hover {color: #fff;}</style>
</head><body class="d-flex flex-column min-vh-100"><div class="page-container d-flex flex-column"><!-- 导航栏 --><nav class="navbar navbar-expand-lg bg-dark" data-bs-theme="dark"><div class="container-fluid"><a class="navbar-brand" href="#">猫影</a><button class="navbar-toggler" type="button" data-bs-toggle="collapse"data-bs-target="#navbarSupportedContent" aria-controls="navbarSupportedContent"aria-expanded="false" aria-label="Toggle navigation"><span class="navbar-toggler-icon"></span></button><div class="collapse navbar-collapse" id="navbarSupportedContent"><ul class="navbar-nav me-auto mb-2 mb-lg-0"><li class="nav-item"><a class="nav-link active" aria-current="page" href="#">影视</a></li></ul><ul class="navbar-nav ms-auto mb-2 mb-lg-0"><li class="nav-item"><a class="nav-link" href="/member/register">注册</a></li><li class="nav-item"><a class="nav-link" href="/member/login">登录</a></li></ul></div></div></nav><!-- 主体内容区域，超出则滚动 --><main class="content-wrapper"><div class="container mt-4">{% block main_content %}{% endblock %}</div></main><!-- 页脚 --><footer class="footer"><div class="container"><p>© {{ current_year }} 猫影. 版权所有.</p><p><a href="#">隐私政策</a> |<a href="#">服务条款</a> |<a href="#">联系我们</a></p></div></footer>
</div><script src="/static/plugins/bootstrapv5/js/bootstrap.js"></script>
<script src="/static/plugins/jquery.min.js"></script>
</body>
</html>
templates/index.html
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>{{name}}</title>
</head>
<body>
{% extends "common/layout.html" %}
</body>
</html>
templates/member/login.html
{% extends "common/layout.html" %}{% block title %}会员登录 - 猫影影视{% endblock %}{% block main_content %}
<div class="container mt-5 flex-grow-1 d-flex align-items-center justify-content-center"><div class="row justify-content-center w-100"><div class="col-md-4"> <!-- 从 col-md-6 改为 col-md-4 --><div class="card shadow-sm" style="max-width: 400px; width: 100%; margin: auto;"><div class="card-header bg-dark text-white text-center"><h3>会员登录</h3></div><div class="card-body"><form action="/member/login" method="post"><!-- 用户名 --><div class="mb-3"><label for="username" class="form-label">用户名</label><input type="text" class="form-control" id="username" name="username" placeholder="请输入登录用户名" required /></div><!-- 密码 --><div class="mb-3"><label for="password" class="form-label">密码</label><input type="password" class="form-control" id="password" name="password" placeholder="请输入登录密码" required /></div><!-- 记住我 + 忘记密码 --><div class="mb-3 form-check"><input type="checkbox" class="form-check-input" id="remember_me" name="remember_me"><label class="form-check-label" for="remember_me">记住我</label><a href="#" class="float-end">忘记密码？</a></div><!-- 登录按钮 --><div class="d-grid"><button type="submit" class="btn btn-primary btn-lg">登录</button></div></form></div><div class="card-footer text-muted text-center">还没有账号？<a href="/member/register">立即注册</a></div></div></div></div>
</div>
{% endblock %}
templates/member/register.html
{% extends "common/layout.html" %}{% block title %}会员注册 - 猫影影视{% endblock %}{% block main_content %}
<div class="container mt-5 flex-grow-1 d-flex align-items-center justify-content-center"><div class="row justify-content-center w-100"><div class="col-md-4 col-11"> <!-- 更窄的列：md占4/12，移动端自动适应 --><div class="card shadow-sm" style="max-width: 400px; width: 100%; margin: auto;"><div class="card-header bg-dark text-white text-center"><h3>会员注册</h3></div><div class="card-body"><form action="/member/register" method="post"><!-- 用户名 --><div class="mb-3"><label for="username" class="form-label">用户名</label><input type="text" class="form-control" id="username" name="username" placeholder="请输入登录用户名" required /></div><!-- 登录密码 --><div class="mb-3"><label for="password" class="form-label">密码</label><input type="password" class="form-control" id="password" name="password" placeholder="请输入登录密码" required /></div><!-- 确认登录密码 --><div class="mb-3"><label for="confirm_password" class="form-label">确认密码</label><input type="password" class="form-control" id="confirm_password" name="confirm_password" placeholder="请输入确认登录密码" required /></div><!-- 注册按钮 --><div class="d-grid"><button type="submit" class="btn btn-success btn-lg">确定</button></div></form></div><div class="card-footer text-muted text-center">已有账号？<a href="/member/login">立即登录</a></div></div></div></div>
</div>
{% endblock %}
controller/member.py
from flask import Blueprint, render_template, jsonify,requestfrom 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")
www.py添加
from controller.member import member_page
app.register_blueprint(member_page, url_prefix="/member")
全局静态方法
common/libs/UrlManager.py
from application import appclass UrlManager(object):@staticmethoddef buildUrl(path):config_domain = app.config['DOMAIN']return config_domain + path@staticmethoddef buildStaticUrl(path):path = app.config['STATIC_PATH'] + pathreturn UrlManager.buildUrl(path)
www.py
"""
模板函数
"""
from common.libs.UrlManager import UrlManager
app.add_template_global(UrlManager.buildStaticUrl, "buildStaticUrl")
app.add_template_global(UrlManager.buildUrl, "buildUrl")
模板中使用
<script src="{{ buildStaticUrl('/plugins/bootstrapv5/js/bootstrap.js') }}"></script>
<a class="nav-link" href="{{buildUrl('/member/register')}}">注册</a>
**版本管理 **: 防止浏览器缓存导致版本更新不灵敏
config/config.yml
RELEASE_VERSION: "2025-06-30 09:40:46:658971"
common/libs/DataHelper.py
import datetimedef getCurrentTime(frm="%Y-%m-%d %H:%M:%S"):dt = datetime.datetime.now()return dt.strftime(frm)
common/libs/UrlManager.py
from application import app,CURRENT_ENV
from common.libs.DataHelper import getCurrentTimeclass UrlManager(object):@staticmethoddef buildStaticUrl(path):path = app.config['STATIC_PATH'] + path +"?ver="+UrlManager.getReleaseVersion()return UrlManager.buildUrl(path)@staticmethoddef 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")统一返回值对象
common/libs/RepHelper.py
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)
JS全局参数配置
static/js/BaseConfig.js
;
const appConfig = {domain: 'http://localhost:5000' // 你的域名配置
};const UrlManager = {/*** 构建完整 URL* @param {string} path - 请求的路径* @returns {string} 完整的 URL 地址*/buildUrl(path) {return appConfig.domain + path;}
};window.UrlManager = UrlManager; // 挂载到 window 上方便全局使用
注册实现
static/js/member/register.js
;
var member_register_ops = {init: function () {this.eventBind();},eventBind: function () {var that = this;// 表单提交事件绑定$("form").on("submit", function (e) {e.preventDefault(); // 阻止默认提交行为var btn = $(this).find("button[type='submit']");var oldText = btn.text();if (btn.hasClass("disabled")) {alert("请勿重复提交");return false;}var login_name = $("#login_name").val().trim();var login_pwd = $("#login_pwd").val().trim();var login_pwd2 = $("#login_pwd2").val().trim();// 基本验证if (!login_name || !login_pwd || !login_pwd2) {alert("请输入完整的用户名、密码和确认密码！");return false;}if (login_pwd.length < 6) {alert("密码长度不能小于6位！");return false;}if (login_pwd !== login_pwd2) {alert("两次输入的密码不一致！");return false;}// 禁用按钮防止重复提交btn.addClass("disabled").text("提交中...");// 使用 jQuery 的 $.post 发送请求$.post($(this).attr("action"),{login_name: login_name,login_pwd: login_pwd,login_pwd2: login_pwd2},function (res) {btn.removeClass("disabled").text(oldText);console.log(res)if (res.code === 200) {alert("注册成功！");window.location.href = UrlManager.buildUrl("/member/login");} else {alert(res.msg);}},"json").fail(function () {btn.removeClass("disabled").text(oldText);alert("网络异常，请稍后再试");});});}
};
// 页面加载完成后初始化
$(function () {member_register_ops.init();
});
service/UserService.py
# 导入必要的模块
import string # 提供字符串常量（如字母、数字等）
import random # 用于随机操作
import hashlib # 提供哈希加密功能（如 MD5）
# 从项目中导入 User 模型和数据库实例
from common.models.user import User
from application import dbclass UserService(object):"""用户服务类，提供与用户相关的业务逻辑，例如密码加盐加密、注册等。"""@staticmethoddef geneSalt(length=16):"""生成指定长度的随机盐值（salt）。:param length: 盐值长度，默认为16位:return: 随机生成的 salt 字符串"""# 构建包含大小写字母和数字的字符列表keylist = list(string.ascii_letters + string.digits)# 打乱顺序以增加随机性random.shuffle(keylist)# 取前 length 个字符作为 saltreturn ''.join(keylist[:length])@staticmethoddef genePwd(pwd, salt):"""对明文密码进行加盐哈希处理（使用 MD5 算法）。:param pwd: 明文密码:param salt: 盐值:return: 哈希后的密码字符串"""md5 = hashlib.md5()# 将密码和盐拼接后进行 MD5 哈希md5.update((pwd + salt).encode('utf-8'))return md5.hexdigest()@staticmethoddef checkPwd(pwd, hashed_pwd, salt):"""验证明文密码是否与数据库中存储的哈希密码匹配。:param pwd: 明文密码（用户输入）:param hashed_pwd: 数据库中存储的哈希密码:param salt: 数据库中存储的盐值:return: True 如果密码正确，否则 False"""# 使用相同的方式重新生成哈希密码，并与数据库中的比较return UserService.genePwd(pwd, salt) == hashed_pwd@staticmethoddef 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()
controller/member.py
# 注册路由
@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)}")
登录功能实现
Session
static/js/member/login.js
var member_login_ops = {init: function () {this.eventBind();},eventBind: function () {var that = this;// 表单提交事件绑定$("form").on("submit", function (e) {e.preventDefault(); // 阻止默认提交行为var btn = $(this).find("button[type='submit']");var oldText = btn.text();if (btn.hasClass("disabled")) {alert("请勿重复提交");return false;}var login_name = $("#login_name").val().trim();var login_pwd = $("#login_pwd").val().trim();// 基本验证if (!login_name || !login_pwd) {alert("请输入完整的用户名和密码！");return false;}if (login_pwd.length < 6) {alert("密码长度不能小于6位！");return false;}// 禁用按钮防止重复提交btn.addClass("disabled").text("登录中...");// 使用 jQuery 的 $.post 发送请求$.post($(this).attr("action"),{login_name: login_name,login_pwd: login_pwd,remember_me: $('#remember_me').is(':checked') ? 1 : 0},function (res) {btn.removeClass("disabled").text(oldText);console.log(res);if (res.code === 200) {alert("登录成功！");window.location.href = UrlManager.buildUrl("/member/index");} else {alert(res.msg);}},"json").fail(function () {btn.removeClass("disabled").text(oldText);alert("网络异常，请稍后再试");});});}
};// 页面加载完成后初始化
$(function () {member_login_ops.init();
});
controller/member.py
@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.valueslogin_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="用户名或密码错误！")# 登录成功：设置 sessionsession['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()Flask 中Session如何实现的？
Cookie
controller/member.py
@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.valueslogin_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():# 如果你使用的是 sessionresponse=make_response(redirect(UrlManager.buildUrl('/')))response.delete_cookie('user')return response
service/UserService.py
@staticmethoddef 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@staticmethoddef 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, Noneuser_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
interceptors/Auth.py
import jsonfrom 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 = userInforeturn# 后拦截
@app.after_request
def after_request(response):app.logger.info("拦截器！-----------------after_request-----------------")return responsedef check_login():cookies = request.cookiesisAuth, userInfo = UserService.decodeAuthCode(cookies.get('user'))if not isAuth:return Nonereturn userInfocommon/libs/RepHelper.py
def ops_render(template, context={}):if 'current_user' in g:context['current_user'] = g.current_userreturn render_template(template, **context)
templates/common/layout.html
{% if current_user %}<ul class="navbar-nav ms-auto mb-2 mb-lg-0"><li class="nav-item"><a class="nav-link">{{current_user.nickname}}</a></li><li class="nav-item"><a class="nav-link" href="{{ buildUrl('/member/logout') }}">退出</a></li></ul>{% else %}<ul class="navbar-nav ms-auto mb-2 mb-lg-0"><li class="nav-item"><a class="nav-link" href="{{ buildUrl('/member/register') }}">注册</a></li><li class="nav-item"><a class="nav-link" href="{{ buildUrl('/member/login') }}">登录</a></li></ul>{% endif %}
Cookie和Session有什么区别？
- 存储位置：
- Cookie：数据存储在客户端（如用户的浏览器）。每次请求时都会将Cookie信息发送到服务器。
- Session：数据存储在服务器端。仅会话标识符（通常是一个特殊的ID）通过Cookie传递给客户端，实际的数据保存在服务器上。
- 安全性：
- Cookie：由于数据存放在客户端，容易受到XSS攻击。可以设置HttpOnly属性来防止JavaScript访问Cookie，从而减少XSS攻击的风险。另外，可以通过设置Secure属性确保Cookie仅在HTTPS连接上传输。
- Session：因为数据存储在服务器端，所以相对更安全。不过，仍然需要妥善管理会话ID，以防止CSRF攻击。
- 性能：
- Cookie：如果存储大量数据，会影响传输效率，因为每次请求都要携带Cookie数据。而且，浏览器对每个域名下的Cookie数量和大小有限制。
- Session：数据存储在服务器端，因此不会影响传输效率。但是，随着在线用户数的增加，服务器存储的需求也会相应增加。
- 有效期：
- Cookie：可以设置过期时间，甚至可以是永久性的。
- Session：通常在用户关闭浏览器后结束，也可以通过编程方式设置更长的有效期。
- 使用场景：
- Cookie：适用于需要在客户端持久化保存少量数据的情况，比如记住用户名、个性化设置等。
- Session：更适合用于存储临时的、敏感的数据，例如登录状态、购物车内容等。
Cookie 和 Session 的相同点
- 目的相同：两者都是用来跟踪和识别网站上的用户会话，以便提供个性化的用户体验，并支持诸如用户登录状态保持等功能。
- 依赖关系：在Web应用中，通常需要同时使用Cookie和Session。Session通常依赖于Cookie中的会话ID来标识不同的用户会话。
- 隐私问题：无论是Cookie还是Session，都需要考虑用户的隐私保护。滥用这些技术可能会侵犯用户隐私或导致安全漏洞，因此在使用时应遵循相关的最佳实践和法律规范。
使用 Flask-Session 存储 Session 到 Redis(共享session)
-
安装依赖： 除了
Flask-Session外，还需要安装redis库以便与Redis服务器交互。pip install Flask-Session redis -
配置 Flask 应用： 接下来，在你的Flask应用中配置
Flask-Session以使用Redis作为Session存储。from flask import Flask, session from flask_session import Sessionapp = Flask(__name__)# 设置应用配置 app.config["SESSION_TYPE"] = "redis" app.config["SESSION_PERMANENT"] = False app.config["SESSION_USE_SIGNER"] = True app.config["SECRET_KEY"] = 'your_secret_key' app.config["SESSION_REDIS"] = redis.from_url('redis://localhost:6379') # 修改为你的Redis服务器地址# 初始化应用与会话 Session(app)@app.route('/') def index():if 'view_count' not in session:session['view_count'] = 0session['view_count'] += 1return f"View count: {session['view_count']}"if __name__ == '__main__':app.run(debug=True)
2.通过定时器获取电影资料
获取影视资源的方案解析
- 人工找资源 并录入到数据库
- 自动找资源，自动录入
爬虫技术选型
- 方便嵌入Flask的定时任务框架
- Requests：Python Http客户端
- BeatifulSoup：HTML解析标签库
APScheduler和Flask-APScheduler
- APScheduler：Python任务调度模块
- Flask-APScheduler：Flask框架调度模块
爬虫核心功能
- 获取资源HTML
- 解析HTML的视频信息
- 将视频信息入库
安装APScheduler
pip install apscheduler
Demo:scheduler/ApSchedulerTest.py
from apscheduler.schedulers.blocking import BlockingScheduler
import datetimedef 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()
安装Flask-APScheduler
pip install Flask-APScheduler
application.py
from flask_apscheduler import APScheduler# 实例化 APScheduler
scheduler = APScheduler()
scheduler.init_app(app)
mananger.py
from application import app,scheduler
from www import *def apscheduler_test():import datetimeprint(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)
适合我们自己的Job框架
- 便于测试
- 便于管理
- 便于部署
3.打包部署
①.高并发部署方案
- Linux: nginx +uwsgi + flask
- windows: nginx +tornado + flask
云服务器的平台