实战项目 - 博客系统¶

# app/models.py
from datetime import datetime, timezone
from app import db
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash

class User(UserMixin, db.Model):
    """用户模型 - 继承UserMixin提供Flask-Login所需的认证方法"""
    __tablename__ = 'users'

    # === 基本字段 ===
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(20), unique=True, nullable=False, index=True)  # 用户名，建立索引加速查询
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)    # 邮箱，唯一约束防止重复注册
    password_hash = db.Column(db.String(128))  # 存储哈希后的密码，不保存明文
    bio = db.Column(db.Text)        # 个人简介
    avatar = db.Column(db.String(200))  # 头像URL
    is_admin = db.Column(db.Boolean, default=False)   # 管理员标记，用于权限控制
    is_active = db.Column(db.Boolean, default=True)    # 账号激活状态，可用于禁用账号
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))  # 注册时间，使用UTC时区

    # === 关系定义 ===
    # lazy='dynamic' 返回查询对象而非直接加载，适合数据量大的一对多关系
    # cascade='all, delete-orphan' 删除用户时级联删除其所有文章和评论
    posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
    comments = db.relationship('Comment', backref='author', lazy='dynamic', cascade='all, delete-orphan')

    def set_password(self, password):
        """使用werkzeug生成密码哈希，单向加密不可逆"""
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        """验证密码是否匹配哈希值"""
        return check_password_hash(self.password_hash, password)

    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'bio': self.bio,
            'avatar': self.avatar,
            'created_at': self.created_at.isoformat()
        }

class Post(db.Model):
    """文章模型 - 博客系统核心实体"""
    __tablename__ = 'posts'

    # === 基本字段 ===
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)   # 文章标题
    content = db.Column(db.Text, nullable=False)         # 文章正文，使用Text类型存储长文本
    slug = db.Column(db.String(120), unique=True, nullable=False, index=True)  # URL友好的文章标识（如：my-first-post）
    summary = db.Column(db.String(200))     # 文章摘要，用于列表页展示
    cover_image = db.Column(db.String(200)) # 封面图片路径
    is_published = db.Column(db.Boolean, default=False)  # 发布状态，默认为草稿
    view_count = db.Column(db.Integer, default=0)        # 浏览计数器
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), index=True)  # 创建时间，建立索引用于排序
    updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))  # 更新时间，onupdate自动更新

    # === 外键 - 建立与用户和分类的关联 ===
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)   # 作者ID，不可为空
    category_id = db.Column(db.Integer, db.ForeignKey('categories.id'))          # 分类ID，可为空

    # === 关系定义 ===
    comments = db.relationship('Comment', backref='post', lazy='dynamic', cascade='all, delete-orphan')
    # 多对多关系：通过post_tags中间表关联标签
    tags = db.relationship('Tag', secondary='post_tags', backref=db.backref('posts', lazy='dynamic'))

    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'content': self.content,
            'slug': self.slug,
            'summary': self.summary,
            'cover_image': self.cover_image,
            'is_published': self.is_published,
            'view_count': self.view_count,
            'created_at': self.created_at.isoformat(),
            'author': self.author.to_dict() if self.author else None,
            'category': self.category.to_dict() if self.category else None
        }

class Comment(db.Model):
    """评论模型"""
    __tablename__ = 'comments'

    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))  # lambda匿名函数：简洁的单行函数

    # 外键
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    post_id = db.Column(db.Integer, db.ForeignKey('posts.id'), nullable=False)

    def to_dict(self):
        return {
            'id': self.id,
            'content': self.content,
            'created_at': self.created_at.isoformat(),
            'author': self.author.to_dict() if self.author else None
        }

class Category(db.Model):
    """分类模型"""
    __tablename__ = 'categories'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True, nullable=False)
    description = db.Column(db.Text)
    slug = db.Column(db.String(60), unique=True, nullable=False)

    # 关系
    posts = db.relationship('Post', backref='category', lazy='dynamic')

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'description': self.description,
            'slug': self.slug
        }

class Tag(db.Model):
    """标签模型"""
    __tablename__ = 'tags'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(30), unique=True, nullable=False)
    slug = db.Column(db.String(30), unique=True, nullable=False)

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'slug': self.slug
        }

# 多对多关系表 - 文章与标签的关联表
# 使用联合主键（post_id + tag_id）确保同一文章不会重复关联同一标签
post_tags = db.Table('post_tags',
    db.Column('post_id', db.Integer, db.ForeignKey('posts.id'), primary_key=True),  # 关联文章
    db.Column('tag_id', db.Integer, db.ForeignKey('tags.id'), primary_key=True)     # 关联标签
)

# app/auth/routes.py
from flask import render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, current_user
from app.auth.forms import LoginForm, RegistrationForm
from app.models import User
from app import db

@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
    # 已登录用户直接跳转首页，避免重复登录
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))

    form = LoginForm()

    # validate_on_submit() 同时检查是否为POST请求和表单验证是否通过
    if form.validate_on_submit():
        # 根据用户名查询用户记录
        user = User.query.filter_by(username=form.username.data).first()

        if user and user.check_password(form.password.data):
            # 检查账号是否被禁用
            if not user.is_active:
                flash('账号已被禁用', 'error')
                return render_template('auth/login.html', form=form)

            # Flask-Login的login_user：将用户ID写入session
            # remember参数控制是否生成"记住我"的持久cookie
            login_user(user, remember=form.remember_me.data)
            flash('登录成功！', 'success')

            # 安全重定向：获取登录前的目标页面
            # 检查next参数是否以'/'开头，防止开放重定向攻击
            next_page = request.args.get('next')
            if not next_page or not next_page.startswith('/'):
                next_page = url_for('main.index')

            return redirect(next_page)
        else:
            flash('用户名或密码错误', 'error')

    return render_template('auth/login.html', form=form)

@auth_bp.route('/register', methods=['GET', 'POST'])
def register():
    # 已登录用户无需再注册
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))

    form = RegistrationForm()

    if form.validate_on_submit():
        # 创建新用户实例
        user = User(
            username=form.username.data,
            email=form.email.data
        )
        # 对密码进行哈希处理后存储，确保安全
        user.set_password(form.password.data)

        # 将新用户添加到数据库会话并提交
        db.session.add(user)
        db.session.commit()

        flash('注册成功！请登录', 'success')
        return redirect(url_for('auth.login'))

    return render_template('auth/register.html', form=form)

@auth_bp.route('/logout')
@login_required
def logout():
    logout_user()
    flash('您已退出登录', 'info')
    return redirect(url_for('main.index'))

# app/post/routes.py
from flask import render_template, redirect, url_for, flash, request, abort
from flask_login import login_required, current_user
from app.post.forms import PostForm, CommentForm
from app.models import Post, Comment
from app import db
from app.utils import save_upload_file, generate_slug

@post_bp.route('/create', methods=['GET', 'POST'])
@login_required  # 装饰器确保只有登录用户才能创建文章
def create_post():
    form = PostForm()

    if form.validate_on_submit():
        # 创建文章实例，generate_slug将标题转换为URL友好格式
        post = Post(
            title=form.title.data,
            content=form.content.data,
            slug=generate_slug(form.title.data),
            summary=form.summary.data,
            author=current_user,          # 自动关联当前登录用户为作者
            is_published=form.is_published.data
        )

        # 处理封面图片
        if form.cover_image.data:
            filename = save_upload_file(form.cover_image.data, 'covers')
            if filename:
                post.cover_image = filename

        # 处理标签
        tags = form.tags.data.split(',')
        for tag_name in tags:
            tag_name = tag_name.strip()
            if tag_name:
                tag = Tag.query.filter_by(name=tag_name).first()
                if not tag:
                    tag = Tag(name=tag_name, slug=generate_slug(tag_name))
                    db.session.add(tag)
                post.tags.append(tag)

        db.session.add(post)
        db.session.commit()

        flash('文章创建成功！', 'success')
        return redirect(url_for('post.detail', slug=post.slug))

    return render_template('post/create.html', form=form)

@post_bp.route('/<slug>')
def detail(slug):
    post = Post.query.filter_by(slug=slug).first_or_404()

    # 增加浏览次数
    post.view_count += 1
    db.session.commit()

    form = CommentForm()

    return render_template('post/detail.html', post=post, form=form)

@post_bp.route('/<slug>/edit', methods=['GET', 'POST'])
@login_required
def edit_post(slug):
    # first_or_404(): 查找文章，不存在时自动返回404响应
    post = Post.query.filter_by(slug=slug).first_or_404()

    # 权限检查：只有文章作者或管理员才能编辑
    if post.author != current_user and not current_user.is_admin:
        abort(403)  # 返回403禁止访问

    # obj=post 将现有文章数据填充到表单中
    form = PostForm(obj=post)

    if form.validate_on_submit():
        post.title = form.title.data
        post.content = form.content.data
        post.summary = form.summary.data
        post.is_published = form.is_published.data

        # 处理封面图片
        if form.cover_image.data:
            filename = save_upload_file(form.cover_image.data, 'covers')
            if filename:
                post.cover_image = filename

        # 更新标签
        post.tags.clear()
        tags = form.tags.data.split(',')
        for tag_name in tags:
            tag_name = tag_name.strip()
            if tag_name:
                tag = Tag.query.filter_by(name=tag_name).first()
                if not tag:
                    tag = Tag(name=tag_name, slug=generate_slug(tag_name))
                    db.session.add(tag)
                post.tags.append(tag)

        db.session.commit()

        flash('文章更新成功！', 'success')
        return redirect(url_for('post.detail', slug=post.slug))

    return render_template('post/edit.html', form=form, post=post)

@post_bp.route('/<slug>/comment', methods=['POST'])
@login_required
def add_comment(slug):
    """添加评论 - 仅接受POST请求"""
    post = Post.query.filter_by(slug=slug).first_or_404()
    form = CommentForm()

    if form.validate_on_submit():
        # 创建评论并关联当前用户和目标文章
        comment = Comment(
            content=form.content.data,
            author=current_user,  # 评论作者
            post=post             # 所属文章
        )

        db.session.add(comment)
        db.session.commit()

        flash('评论发布成功！', 'success')

    return redirect(url_for('post.detail', slug=slug))

# app/api/routes.py
from flask import jsonify, request
from app.models import Post, User, Comment
from app import db

@api_bp.route('/posts', methods=['GET'])
def get_posts():
    """获取文章列表 - 支持分页、分类和标签过滤"""
    # 从URL查询参数中获取分页和过滤条件，type=int自动转换类型
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    category = request.args.get('category')
    tag = request.args.get('tag')

    # 基础查询：只返回已发布的文章
    query = Post.query.filter_by(is_published=True)

    # 链式查询构建：根据条件动态添加过滤器
    if category:
        query = query.join(Category).filter(Category.slug == category)  # 按分类过滤

    if tag:
        query = query.join(Tag).filter(Tag.slug == tag)  # 按标签过滤

    # 分页查询：按创建时间倒序排列，error_out=False防止页码越界时抛异常
    pagination = query.order_by(Post.created_at.desc())\
        .paginate(page=page, per_page=per_page, error_out=False)

    # 将ORM对象列表转换为字典列表，便于JSON序列化
    posts = [post.to_dict() for post in pagination.items]

    # 返回文章数据和分页元信息
    return jsonify({
        'posts': posts,
        'total': pagination.total,       # 总文章数
        'pages': pagination.pages,       # 总页数
        'current_page': page             # 当前页码
    })

@api_bp.route('/posts/<slug>', methods=['GET'])
def get_post(slug):
    """获取单篇文章"""
    post = Post.query.filter_by(slug=slug, is_published=True).first_or_404()
    return jsonify(post.to_dict())

@api_bp.route('/posts/<slug>/comments', methods=['GET'])
def get_post_comments(slug):
    """获取文章评论"""
    post = Post.query.filter_by(slug=slug, is_published=True).first_or_404()
    comments = [comment.to_dict() for comment in post.comments.order_by(Comment.created_at.desc()).all()]

    return jsonify({'comments': comments})