目录
Python实例题
题目
基于 Flask 的博客系统
要求:
- 使用 Flask 框架构建一个博客系统,支持以下功能:
- 用户认证(注册、登录、注销)
- 博客文章的创建、编辑、删除
- 文章分类和标签管理
- 评论系统
- 文章搜索和归档
- 管理员后台管理
- 使用 SQLite 数据库存储用户、文章、评论等信息。
- 添加美观的前端界面,支持响应式设计。
解题思路:
- 使用 Flask 蓝图组织代码结构。
- 通过 Flask-Login 处理用户认证。
- 使用 Flask-SQLAlchemy 管理数据库操作。
- 结合 Bootstrap 美化前端界面。
代码实现:
from flask import Flask, render_template, request, redirect, url_for, flash, Blueprint
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime
import os
# 初始化 Flask 应用
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key-here'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(app.root_path, 'blog.db')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 初始化数据库
db = SQLAlchemy(app)
# 初始化 Flask-Login
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
# 数据模型
class User(UserMixin, db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password_hash = db.Column(db.String(120), nullable=False)
is_admin = db.Column(db.Boolean, default=False)
posts = db.relationship('Post', backref='author', lazy=True)
comments = db.relationship('Comment', backref='author', lazy=True)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
class Category(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True, nullable=False)
posts = db.relationship('Post', backref='category', lazy=True)
def __repr__(self):
return self.name
class Tag(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True, nullable=False)
posts = db.relationship('Post', secondary='post_tag', backref='tags', lazy=True)
def __repr__(self):
return self.name
post_tag = db.Table('post_tag',
db.Column('post_id', db.Integer, db.ForeignKey('post.id'), primary_key=True),
db.Column('tag_id', db.Integer, db.ForeignKey('tag.id'), primary_key=True)
)
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100), nullable=False)
content = db.Column(db.Text, nullable=False)
summary = db.Column(db.Text)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
category_id = db.Column(db.Integer, db.ForeignKey('category.id'), nullable=False)
comments = db.relationship('Comment', backref='post', lazy=True)
views = db.Column(db.Integer, default=0)
def __repr__(self):
return self.title
class Comment(db.Model):
id = db.Column(db.Integer, primary_key=True)
content = db.Column(db.Text, nullable=False)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
post_id = db.Column(db.Integer, db.ForeignKey('post.id'), nullable=False)
def __repr__(self):
return f"Comment by {self.author.username} on {self.post.title}"
# 用户加载回调
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
# 认证蓝图
auth = Blueprint('auth', __name__)
@auth.route('/register', methods=['GET', 'POST'])
def register():
if request.method == 'POST':
username = request.form['username']
password = request.form['password']
if User.query.filter_by(username=username).first():
flash('用户名已存在', 'error')
return redirect(url_for('auth.register'))
user = User(username=username)
user.set_password(password)
db.session.add(user)
db.session.commit()
flash('注册成功,请登录', 'success')
return redirect(url_for('auth.login'))
return render_template('register.html')
@auth.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form['username']
password = request.form['password']
user = User.query.filter_by(username=username).first()
if user and user.check_password(password):
login_user(user)
flash('登录成功', 'success')
return redirect(url_for('main.index'))
else:
flash('用户名或密码错误', 'error')
return render_template('login.html')
@auth.route('/logout')
@login_required
def logout():
logout_user()
flash('已注销', 'success')
return redirect(url_for('main.index'))
# 主蓝图
main = Blueprint('main', __name__)
@main.route('/')
def index():
page = request.args.get('page', 1, type=int)
search = request.args.get('search', '')
category = request.args.get('category', '')
tag = request.args.get('tag', '')
query = Post.query.order_by(Post.created_at.desc())
if search:
query = query.filter(Post.title.ilike(f'%{search}%') | Post.content.ilike(f'%{search}%'))
if category:
category_obj = Category.query.filter_by(name=category).first()
if category_obj:
query = query.filter_by(category_id=category_obj.id)
if tag:
tag_obj = Tag.query.filter_by(name=tag).first()
if tag_obj:
query = query.filter(Post.tags.contains(tag_obj))
posts = query.paginate(page=page, per_page=5, error_out=False)
categories = Category.query.all()
tags = Tag.query.all()
return render_template('index.html', posts=posts, categories=categories, tags=tags,
search=search, category=category, tag=tag)
@main.route('/post/<int:post_id>')
def post_detail(post_id):
post = Post.query.get_or_404(post_id)
post.views += 1
db.session.commit()
comments = Comment.query.filter_by(post_id=post_id).order_by(Comment.created_at.asc()).all()
return render_template('post_detail.html', post=post, comments=comments)
@main.route('/post/<int:post_id>/comment', methods=['POST'])
@login_required
def add_comment(post_id):
post = Post.query.get_or_404(post_id)
content = request.form['content']
if not content:
flash('评论内容不能为空', 'error')
return redirect(url_for('main.post_detail', post_id=post_id))
comment = Comment(
content=content,
author=current_user,
post=post
)
db.session.add(comment)
db.session.commit()
flash('评论已提交', 'success')
return redirect(url_for('main.post_detail', post_id=post_id))
@main.route('/category/<string:category_name>')
def category_posts(category_name):
category = Category.query.filter_by(name=category_name).first_or_404()
posts = Post.query.filter_by(category=category).order_by(Post.created_at.desc()).all()
return render_template('category_posts.html', category=category, posts=posts)
@main.route('/tag/<string:tag_name>')
def tag_posts(tag_name):
tag = Tag.query.filter_by(name=tag_name).first_or_404()
posts = tag.posts
return render_template('tag_posts.html', tag=tag, posts=posts)
@main.route('/archive')
def archive():
posts = Post.query.order_by(Post.created_at.desc()).all()
# 按年月归档
archive_dict = {}
for post in posts:
year_month = post.created_at.strftime('%Y-%m')
if year_month not in archive_dict:
archive_dict[year_month] = []
archive_dict[year_month].append(post)
# 按年月排序
sorted_archive = sorted(archive_dict.items(), reverse=True)
return render_template('archive.html', archive=sorted_archive)
@main.route('/about')
def about():
return render_template('about.html')
# 用户蓝图
user = Blueprint('user', __name__)
@user.route('/profile')
@login_required
def profile():
return render_template('profile.html')
@user.route('/posts')
@login_required
def user_posts():
posts = Post.query.filter_by(user_id=current_user.id).order_by(Post.created_at.desc()).all()
return render_template('user_posts.html', posts=posts)
@user.route('/post/new', methods=['GET', 'POST'])
@login_required
def new_post():
categories = Category.query.all()
tags = Tag.query.all()
if request.method == 'POST':
title = request.form['title']
content = request.form['content']
summary = request.form.get('summary', '')
category_id = request.form['category']
tag_ids = request.form.getlist('tags')
if not title or not content:
flash('标题和内容不能为空', 'error')
return redirect(url_for('user.new_post'))
post = Post(
title=title,
content=content,
summary=summary,
author=current_user,
category_id=category_id
)
# 添加标签
for tag_id in tag_ids:
tag = Tag.query.get(tag_id)
if tag:
post.tags.append(tag)
db.session.add(post)
db.session.commit()
flash('文章已发布', 'success')
return redirect(url_for('user.user_posts'))
return render_template('post_form.html', categories=categories, tags=tags, is_new=True)
@user.route('/post/<int:post_id>/edit', methods=['GET', 'POST'])
@login_required
def edit_post(post_id):
post = Post.query.get_or_404(post_id)
# 确保只有作者可以编辑文章
if post.author != current_user and not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.post_detail', post_id=post_id))
categories = Category.query.all()
tags = Tag.query.all()
if request.method == 'POST':
post.title = request.form['title']
post.content = request.form['content']
post.summary = request.form.get('summary', '')
post.category_id = request.form['category']
# 清除原有标签
post.tags = []
# 添加新标签
tag_ids = request.form.getlist('tags')
for tag_id in tag_ids:
tag = Tag.query.get(tag_id)
if tag:
post.tags.append(tag)
db.session.commit()
flash('文章已更新', 'success')
return redirect(url_for('main.post_detail', post_id=post_id))
return render_template('post_form.html', post=post, categories=categories, tags=tags, is_new=False)
@user.route('/post/<int:post_id>/delete', methods=['POST'])
@login_required
def delete_post(post_id):
post = Post.query.get_or_404(post_id)
# 确保只有作者或管理员可以删除文章
if post.author != current_user and not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.post_detail', post_id=post_id))
db.session.delete(post)
db.session.commit()
flash('文章已删除', 'success')
return redirect(url_for('user.user_posts'))
# 管理蓝图
admin = Blueprint('admin', __name__)
@admin.route('/')
@login_required
def dashboard():
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
users_count = User.query.count()
posts_count = Post.query.count()
comments_count = Comment.query.count()
recent_posts = Post.query.order_by(Post.created_at.desc()).limit(5).all()
recent_comments = Comment.query.order_by(Comment.created_at.desc()).limit(5).all()
return render_template('admin/dashboard.html',
users_count=users_count,
posts_count=posts_count,
comments_count=comments_count,
recent_posts=recent_posts,
recent_comments=recent_comments)
@admin.route('/users')
@login_required
def users():
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
users = User.query.all()
return render_template('admin/users.html', users=users)
@admin.route('/user/<int:user_id>/toggle_admin', methods=['POST'])
@login_required
def toggle_admin(user_id):
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
user = User.query.get_or_404(user_id)
# 不能修改自己的管理员权限
if user == current_user:
flash('不能修改自己的管理员权限', 'error')
return redirect(url_for('admin.users'))
user.is_admin = not user.is_admin
db.session.commit()
flash(f"用户 {user.username} 的管理员权限已更新", 'success')
return redirect(url_for('admin.users'))
@admin.route('/categories')
@login_required
def categories():
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
categories = Category.query.all()
return render_template('admin/categories.html', categories=categories)
@admin.route('/category/new', methods=['POST'])
@login_required
def new_category():
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
name = request.form['name']
if not name:
flash('分类名称不能为空', 'error')
return redirect(url_for('admin.categories'))
if Category.query.filter_by(name=name).first():
flash('分类名称已存在', 'error')
return redirect(url_for('admin.categories'))
category = Category(name=name)
db.session.add(category)
db.session.commit()
flash('分类已创建', 'success')
return redirect(url_for('admin.categories'))
@admin.route('/category/<int:category_id>/edit', methods=['POST'])
@login_required
def edit_category(category_id):
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
category = Category.query.get_or_404(category_id)
name = request.form['name']
if not name:
flash('分类名称不能为空', 'error')
return redirect(url_for('admin.categories'))
if Category.query.filter_by(name=name).first() and name != category.name:
flash('分类名称已存在', 'error')
return redirect(url_for('admin.categories'))
category.name = name
db.session.commit()
flash('分类已更新', 'success')
return redirect(url_for('admin.categories'))
@admin.route('/category/<int:category_id>/delete', methods=['POST'])
@login_required
def delete_category(category_id):
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
category = Category.query.get_or_404(category_id)
# 检查是否有文章属于该分类
if category.posts:
flash('该分类下有文章,不能删除', 'error')
return redirect(url_for('admin.categories'))
db.session.delete(category)
db.session.commit()
flash('分类已删除', 'success')
return redirect(url_for('admin.categories'))
@admin.route('/tags')
@login_required
def tags():
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
tags = Tag.query.all()
return render_template('admin/tags.html', tags=tags)
@admin.route('/tag/new', methods=['POST'])
@login_required
def new_tag():
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
name = request.form['name']
if not name:
flash('标签名称不能为空', 'error')
return redirect(url_for('admin.tags'))
if Tag.query.filter_by(name=name).first():
flash('标签名称已存在', 'error')
return redirect(url_for('admin.tags'))
tag = Tag(name=name)
db.session.add(tag)
db.session.commit()
flash('标签已创建', 'success')
return redirect(url_for('admin.tags'))
@admin.route('/tag/<int:tag_id>/edit', methods=['POST'])
@login_required
def edit_tag(tag_id):
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
tag = Tag.query.get_or_404(tag_id)
name = request.form['name']
if not name:
flash('标签名称不能为空', 'error')
return redirect(url_for('admin.tags'))
if Tag.query.filter_by(name=name).first() and name != tag.name:
flash('标签名称已存在', 'error')
return redirect(url_for('admin.tags'))
tag.name = name
db.session.commit()
flash('标签已更新', 'success')
return redirect(url_for('admin.tags'))
@admin.route('/tag/<int:tag_id>/delete', methods=['POST'])
@login_required
def delete_tag(tag_id):
if not current_user.is_admin:
flash('权限不足', 'error')
return redirect(url_for('main.index'))
tag = Tag.query.get_or_404(tag_id)
# 清除标签与文章的关联
tag.posts = []
db.session.delete(tag)
db.session.commit()
flash('标签已删除', 'success')
return redirect(url_for('admin.tags'))
# 注册蓝图
app.register_blueprint(main)
app.register_blueprint(auth)
app.register_blueprint(user, url_prefix='/user')
app.register_blueprint(admin, url_prefix='/admin')
# 创建管理员用户
def create_admin(username, password):
with app.app_context():
admin = User.query.filter_by(username=username).first()
if not admin:
admin = User(username=username, is_admin=True)
admin.set_password(password)
db.session.add(admin)
db.session.commit()
print(f"管理员用户 '{username}' 已创建")
else:
print(f"管理员用户 '{username}' 已存在")
# 创建数据库表
with app.app_context():
db.create_all()
# 创建默认管理员用户 (用户名: admin, 密码: admin123)
create_admin('admin', 'admin123')
if __name__ == '__main__':
app.run(debug=True)