Python实例题:基于 Flask 的博客系统
目录
Python实例题
题目
要求:
解题思路:
代码实现:
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.nameclass 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.namepost_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.titleclass 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 += 1db.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.postsreturn 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_admindb.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 = namedb.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 = namedb.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)