第 13 章 · 项目后端实战

模块    函数名（蓝图.视图）              方法     路径
------  -----------------------------  -------  ------------------------------
用户    user.user_register             POST     /api/user/register
        user.user_login                POST     /api/user/login
        user.user_profile              GET      /api/user/profile          【需登录】
        user.user_profile_update       PUT      /api/user/profile          【需登录】
        user.user_avatar               POST     /api/user/avatar           【需登录】
        user.user_logout               POST     /api/user/logout           【需登录】
首页    home.banner_list               GET      /api/banner/list
        home.category_list             GET      /api/category/list
        home.article_recommend         GET      /api/article/recommend
作品    article.article_list           GET      /api/article/list
        article.article_detail         GET      /api/article/detail/<int:aid>
        article.article_like           POST     /api/article/like          【需登录】
书评    comment.comment_list           GET      /api/comment/list
        comment.comment_add            POST     /api/comment/add           【需登录】
        comment.comment_like           POST     /api/comment/like          【需登录】
        comment.comment_delete         DELETE   /api/comment/<int:cid>      【需登录】
收藏    favorite.favorite_toggle       POST     /api/favorite/toggle       【需登录】
        favorite.favorite_list         GET      /api/favorite/list         【需登录】
------  -----------------------------  -------  ------------------------------
合计 18 个接口；另有 1 个自己加的 GET /api/health（冒烟自测用）

hiflask/
├── app.py                  # 应用工厂 create_app() + 错误处理器 + 启动入口
├── config.py               # Config / DevConfig / ProdConfig / TestingConfig
├── extensions.py           # db / jwt / cors / migrate 实例（避免循环导入）
├── requirements.txt
├── .env.example            # 环境变量样例（真密码不进版本库）
├── models/                 # 数据层：一张表一个类
│   ├── __init__.py         # 统一导出，方便 from models import User
│   ├── user.py             # User
│   ├── article.py          # Category / Article / Banner
│   └── comment.py          # Comment / Favorite / UserLike
├── views/                  # 接口层：一个模块一个蓝图
│   ├── __init__.py         # register_blueprints(app)
│   ├── user.py             # user_bp    6 个接口
│   ├── home.py             # home_bp    3 个接口 + 健康检查
│   ├── article.py          # article_bp 3 个接口
│   ├── comment.py          # comment_bp 4 个接口
│   └── favorite.py         # favorite_bp 2 个接口
├── utils/
│   ├── response.py         # success() / fail()，统一 {code,msg,data}
│   └── auth.py             # @login_required 装饰器
└── tests/                  # 第 12 章的 pytest 用例
    ├── conftest.py
    ├── test_user_api.py
    └── test_article_api.py

from flask import Flask

from config import config_map
from extensions import cors, db, jwt, migrate
from utils.response import fail
from views import register_blueprints


def create_app(config_name="dev"):
    """应用工厂：按名字取配置，装配出一个完整的 app"""
    app = Flask(__name__)
    app.config.from_object(config_map[config_name])

    # ① 初始化扩展（顺序：db → jwt → cors → migrate）
    db.init_app(app)
    jwt.init_app(app)
    cors.init_app(app, resources={r"/api/*": {"origins": "*"}})
    migrate.init_app(app, db)

    # ② 注册蓝图：18 个接口在这里被挂到 app 上
    register_blueprints(app)

    # ③ 全局错误处理：一律返回统一结构，绝不把堆栈抛给前端
    register_error_handlers(app)
    return app


def register_error_handlers(app):
    # JWT 相关的异常交给 flask-jwt-extended 的回调，这里处理 404 / 405 / 500
    @app.errorhandler(404)
    def not_found(e):
        return fail("接口不存在", code=404)

    @app.errorhandler(405)
    def method_not_allowed(e):
        return fail("请求方法不被允许", code=405)

    @app.errorhandler(Exception)
    def server_error(e):
        app.logger.exception("未捕获异常")      # 堆栈写日志，不给前端
        return fail("服务器内部错误", code=500)


if __name__ == "__main__":
    create_app("dev").run(host="127.0.0.1", port=5000, debug=True)

# ---------- models/user.py ----------
from datetime import datetime
from werkzeug.security import check_password_hash, generate_password_hash

from extensions import db


class User(db.Model):
    __tablename__ = "user"          # 表名是单数，别写成 users

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), nullable=False, unique=True)
    password = db.Column(db.String(255), nullable=False)   # 存哈希，不是明文
    nickname = db.Column(db.String(50))
    email = db.Column(db.String(100), unique=True)
    avatar = db.Column(db.String(255), default="")
    intro = db.Column(db.String(255))
    gender = db.Column(db.SmallInteger, default=0)
    status = db.Column(db.SmallInteger, nullable=False, default=1)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)
    updated_at = db.Column(db.DateTime, nullable=False,
                           default=datetime.now, onupdate=datetime.now)

    # 关系：一个用户有多条书评、多条收藏（lazy="dynamic" 便于再链式查询）
    comments = db.relationship("Comment", back_populates="user", lazy="dynamic")
    favorites = db.relationship("Favorite", back_populates="user", lazy="dynamic")

    def set_password(self, raw):
        self.password = generate_password_hash(raw)

    def check_password(self, raw):
        return check_password_hash(self.password, raw)

    def to_dict(self):
        """对外输出：绝不含 password 字段"""
        return {
            "id": self.id,
            "username": self.username,
            "nickname": self.nickname,
            "email": self.email,
            "avatar": self.avatar or "",
            "intro": self.intro,
            "gender": self.gender,
            "created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S"),
        }


# ---------- models/article.py ----------
class Article(db.Model):
    __tablename__ = "article"

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(150), nullable=False)
    author = db.Column(db.String(50))
    dynasty = db.Column(db.String(20))
    category_id = db.Column(db.Integer, db.ForeignKey("category.id"))
    cover = db.Column(db.String(255), default="")
    summary = db.Column(db.String(500))
    content = db.Column(db.Text)
    views = db.Column(db.Integer, nullable=False, default=0)
    likes = db.Column(db.Integer, nullable=False, default=0)
    is_recommend = db.Column(db.SmallInteger, nullable=False, default=0)
    status = db.Column(db.SmallInteger, nullable=False, default=1)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)
    updated_at = db.Column(db.DateTime, nullable=False,
                           default=datetime.now, onupdate=datetime.now)

    category = db.relationship("Category", back_populates="articles")
    comments = db.relationship("Comment", back_populates="article", lazy="dynamic")
    favorited_by = db.relationship("Favorite", back_populates="article", lazy="dynamic")

    def to_list_dict(self):
        """列表用：字段少一点，不含正文 content"""
        return {
            "id": self.id,
            "title": self.title,
            "author": self.author,
            "dynasty": self.dynasty,
            "category_name": self.category.name if self.category else "",
            "cover": self.cover or "",
            "summary": self.summary,
            "views": self.views,
            "likes": self.likes,
            "created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S"),
        }

class Comment(db.Model):
    __tablename__ = "comment"

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
    article_id = db.Column(db.Integer, db.ForeignKey("article.id"), nullable=False)
    parent_id = db.Column(db.Integer, db.ForeignKey("comment.id"))   # 自关联
    content = db.Column(db.String(1000), nullable=False)
    likes = db.Column(db.Integer, nullable=False, default=0)
    status = db.Column(db.SmallInteger, nullable=False, default=1)   # 1 正常 0 已删
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)

    user = db.relationship("User", back_populates="comments")
    article = db.relationship("Article", back_populates="comments")

    # 自关联：remote_side 必须指到「一」那一端（父评论的主键）
    parent = db.relationship("Comment", remote_side=[id], back_populates="children")
    children = db.relationship("Comment", back_populates="parent",
                               order_by="Comment.created_at")


class Favorite(db.Model):
    __tablename__ = "favorite"
    __table_args__ = (
        db.UniqueConstraint("user_id", "article_id", name="uk_fav_user_article"),
    )

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
    article_id = db.Column(db.Integer, db.ForeignKey("article.id"), nullable=False)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)

    user = db.relationship("User", back_populates="favorites")
    article = db.relationship("Article", back_populates="favorited_by")


class UserLike(db.Model):
    __tablename__ = "user_like"
    __table_args__ = (
        db.UniqueConstraint("user_id", "target_type", "target_id", name="uk_like"),
    )

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, nullable=False)
    # 通用点赞：通过 target_type 区分点赞的是作品还是书评
    target_type = db.Column(db.Enum("article", "comment", name="like_target_type"),
                            nullable=False)
    target_id = db.Column(db.Integer, nullable=False)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)

import re

from flask import Blueprint, g, request
from flask_jwt_extended import create_access_token

from extensions import db
from models.comment import Comment, Favorite
from models.user import User
from utils.auth import login_required
from utils.response import fail, success

user_bp = Blueprint("user", __name__, url_prefix="/api/user")

USERNAME_RE = re.compile(r"^[A-Za-z0-9_]{3,20}$")


@user_bp.post("/register")
def user_register():
    """① 注册：参数校验前置，密码只存哈希"""
    data = request.get_json(silent=True) or {}
    username = (data.get("username") or "").strip()
    password = data.get("password") or ""

    if not USERNAME_RE.match(username):
        return fail("用户名需为 3-20 位字母、数字或下划线")
    if not (6 <= len(password) <= 20):
        return fail("密码长度需为 6-20 位")
    if User.query.filter_by(username=username).first():
        return fail("用户名已存在")

    email = (data.get("email") or "").strip() or None
    if email and User.query.filter_by(email=email).first():
        return fail("邮箱已被注册")

    # 昵称不填就用用户名兜底——对应接口文档里 nickname「不填则默认取 username」
    nickname = (data.get("nickname") or "").strip() or username

    user = User(username=username, nickname=nickname, email=email)
    user.set_password(password)
    db.session.add(user)
    db.session.commit()
    return success({"user_id": user.id}, "注册成功")


@user_bp.post("/login")
def user_login():
    """② 登录：签发 JWT，前端后续用 Authorization: Bearer <token>"""
    data = request.get_json(silent=True) or {}
    username = (data.get("username") or "").strip()
    password = data.get("password") or ""

    if not username or not password:
        return fail("缺少参数：username / password")

    user = User.query.filter_by(username=username).first()
    # 用户不存在 / 密码错 / 被禁用，统一返回同一句提示，避免暴露账号是否存在
    if user is None or not user.check_password(password):
        return fail("用户名或密码错误")
    if user.status != 1:
        return fail("账号已被禁用", code=403)

    # flask-jwt-extended 4.x 要求 identity 是字符串
    token = create_access_token(identity=str(user.id))
    return success({
        "token": token,
        "user": {
            "id": user.id,
            "username": user.username,
            "nickname": user.nickname,
            "avatar": user.avatar or "",
        },
    }, "登录成功")

@user_bp.get("/profile")
@login_required
def user_profile():
    """③ 获取个人信息：@login_required 校验通过后，g.current_user 就是当前用户"""
    user = g.current_user

    data = user.to_dict()
    # stat 是给个人中心首页用的聚合统计，省掉前端两次额外的请求
    data["stat"] = {
        "comment_count": Comment.query.filter_by(user_id=user.id, status=1).count(),
        "favorite_count": Favorite.query.filter_by(user_id=user.id).count(),
    }
    return success(data)

from flask import Blueprint, request

from extensions import db
from models.article import Article, Banner, Category
from utils.response import success

home_bp = Blueprint("home", __name__, url_prefix="/api")


@home_bp.get("/health")
def health():
    """健康检查：自己加的，用来判断服务和数据库是否都通"""
    db.session.execute(db.text("SELECT 1"))   # 顺带探一下数据库
    return success({"service": "hiflask", "status": "up"})


@home_bp.get("/banner/list")
def banner_list():
    """① 轮播图：按 sort 升序，sort 相同的按 id"""
    rows = (Banner.query
            .filter_by(status=1)
            .order_by(Banner.sort.asc(), Banner.id.asc())
            .all())
    return success([
        {"id": b.id, "title": b.title, "image": b.image, "link": b.link or ""}
        for b in rows
    ])


@home_bp.get("/category/list")
def category_list():
    """② 分类：同样是 sort 升序，前端拿它渲染筛选下拉框"""
    rows = (Category.query
            .filter_by(status=1)
            .order_by(Category.sort.asc(), Category.id.asc())
            .all())
    return success([
        {"id": c.id, "name": c.name, "description": c.description or ""}
        for c in rows
    ])


@home_bp.get("/article/recommend")
def article_recommend():
    """③ 文学推荐：is_recommend = 1，按浏览量倒序，limit 可控"""
    limit = request.args.get("limit", 6, type=int)
    limit = min(max(limit, 1), 20)          # 兜住非法值，最多 20 条

    rows = (Article.query
            .filter_by(status=1, is_recommend=1)
            .order_by(Article.views.desc(), Article.id.desc())
            .limit(limit)
            .all())
    return success([a.to_list_dict() for a in rows])

from flask import Blueprint, g, request

from extensions import db
from models.article import Article
from models.comment import Favorite, UserLike
from utils.auth import login_required
from utils.response import fail, success

article_bp = Blueprint("article", __name__, url_prefix="/api/article")


def _parse_page():
    """分页参数统一夹取，免得 page=0 或 page_size=99999 打崩数据库"""
    page = request.args.get("page", 1, type=int)
    page_size = request.args.get("page_size", 10, type=int)
    return max(page, 1), min(max(page_size, 1), 50)


@article_bp.get("/list")
def article_list():
    """① 作品列表：分页 / 分类筛选 / 朝代筛选 / 关键词搜索 / 三种排序"""
    page, page_size = _parse_page()
    category_id = request.args.get("category_id", type=int)
    dynasty = request.args.get("dynasty")
    keyword = request.args.get("keyword")
    order_by = request.args.get("order_by", "latest")

    # ① 基础条件：只查已发布的
    query = Article.query.filter(Article.status == 1)

    # ② 按需叠加筛选条件（参数没传就不加）
    if category_id:
        query = query.filter(Article.category_id == category_id)
    if dynasty:
        query = query.filter(Article.dynasty == dynasty)
    if keyword:
        like = f"%{keyword}%"
        query = query.filter(db.or_(Article.title.like(like),
                                   Article.summary.like(like)))

    # ③ 排序：都要带次关键字 id，保证结果稳定，翻页不重不漏
    if order_by == "hot":
        query = query.order_by(Article.views.desc(), Article.id.desc())
    elif order_by == "likes":
        query = query.order_by(Article.likes.desc(), Article.id.desc())
    else:
        query = query.order_by(Article.created_at.desc(), Article.id.desc())

    # ④ 分页（Flask-SQLAlchemy 3.x 的 db.paginate）
    page_obj = db.paginate(query, page=page, per_page=page_size, error_out=False)
    return success({
        "list": [a.to_list_dict() for a in page_obj.items],
        "total": page_obj.total,
        "page": page,
        "page_size": page_size,
    })

from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request
from models.comment import Comment        # 算评论数要用


def _current_user_id():
    """可选登录：带了合法 token 就返回用户 id，没带或失效就返回 None"""
    try:
        verify_jwt_in_request(optional=True)
        identity = get_jwt_identity()
        return int(identity) if identity else None
    except Exception:
        return None


@article_bp.get("/detail/<int:aid>")
def article_detail(aid):
    """② 作品详情：浏览量 +1，并回填「我有没有收藏 / 点赞」"""
    article = db.session.get(Article, aid)
    if article is None or article.status != 1:
        return fail("作品不存在", code=404)

    # 浏览量自增：这是唯一允许「读接口写库」的地方，业务上必须这么做
    article.views = (article.views or 0) + 1
    db.session.commit()

    uid = _current_user_id()
    is_favorite = is_liked = False
    if uid:
        is_favorite = Favorite.query.filter_by(user_id=uid, article_id=aid).first() is not None
        is_liked = UserLike.query.filter_by(user_id=uid, target_type="article",
                                           target_id=aid).first() is not None

    data = {
        "id": article.id,
        "title": article.title,
        "author": article.author,
        "dynasty": article.dynasty,
        "category_name": article.category.name if article.category else "",
        "content": article.content,
        "views": article.views,
        "likes": article.likes,
        "is_favorite": is_favorite,
        "is_liked": is_liked,
        "comment_count": Comment.query.filter_by(article_id=aid, status=1).count(),
    }
    return success(data)

from flask import Blueprint, g, request

from extensions import db
from models.article import Article
from models.comment import Comment, UserLike
from utils.auth import login_required
from utils.response import fail, success

comment_bp = Blueprint("comment", __name__, url_prefix="/api/comment")


def _comment_to_dict(comment, with_children=False):
    """把一条书评转成前端要的结构；with_children 控制是否带上回复"""
    data = {
        "id": comment.id,
        "content": comment.content,
        "likes": comment.likes,
        "user": {
            "id": comment.user.id,
            "nickname": comment.user.nickname,
            "avatar": comment.user.avatar or "",
        },
        "created_at": comment.created_at.strftime("%Y-%m-%d %H:%M:%S"),
    }
    if with_children:
        data["children"] = [
            _comment_to_dict(child)
            for child in comment.children if child.status == 1
        ]
    return data


@comment_bp.get("/list")
def comment_list():
    """① 书评列表：分页只算顶层，回复嵌在 children 里"""
    article_id = request.args.get("article_id", type=int)
    if not article_id:
        return fail("缺少参数：article_id")

    page = max(request.args.get("page", 1, type=int), 1)
    page_size = min(max(request.args.get("page_size", 10, type=int), 1), 50)

    query = (Comment.query
             .filter_by(article_id=article_id, status=1, parent_id=None)
             .order_by(Comment.created_at.desc(), Comment.id.desc()))
    page_obj = db.paginate(query, page=page, per_page=page_size, error_out=False)

    return success({
        "list": [_comment_to_dict(c, with_children=True) for c in page_obj.items],
        "total": page_obj.total,
        "page": page,
        "page_size": page_size,
    })

@comment_bp.post("/add")
@login_required
def comment_add():
    """② 发表书评：parent_id 不为空就是「回复某条书评」"""
    data = request.get_json(silent=True) or {}
    article_id = data.get("article_id")
    content = (data.get("content") or "").strip()
    parent_id = data.get("parent_id")

    if not article_id:
        return fail("缺少参数：article_id")
    if not content:
        return fail("书评内容不能为空")
    if len(content) > 1000:
        return fail("书评内容不能超过 1000 字")

    article = db.session.get(Article, article_id)
    if article is None or article.status != 1:
        return fail("作品不存在", code=404)

    # 回复时必须确认父书评存在，而且要属于同一篇作品（防止跨作品乱回复）
    if parent_id:
        parent = db.session.get(Comment, parent_id)
        if parent is None or parent.article_id != article_id:
            return fail("被回复的书评不存在", code=404)

    comment = Comment(user_id=g.current_user.id,
                      article_id=article_id,
                      parent_id=parent_id,
                      content=content)
    db.session.add(comment)
    db.session.commit()
    return success({"comment_id": comment.id}, "发表成功")


@comment_bp.delete("/<int:cid>")
@login_required
def comment_delete(cid):
    """④ 删除书评：只能删自己的，别人的返回 403"""
    comment = db.session.get(Comment, cid)
    if comment is None or comment.status != 1:
        return fail("书评不存在", code=404)
    if comment.user_id != g.current_user.id:
        return fail("无权删除他人书评", code=403)

    comment.status = 0        # 软删除，保留数据便于追溯
    db.session.commit()
    return success(None, "删除成功")

from flask import Blueprint, g, request

from extensions import db
from models.article import Article
from models.comment import Favorite
from utils.auth import login_required
from utils.response import fail, success

favorite_bp = Blueprint("favorite", __name__, url_prefix="/api/favorite")


@favorite_bp.post("/toggle")
@login_required
def favorite_toggle():
    """① 收藏 / 取消收藏：同一个接口，后端判断该加还是该减"""
    data = request.get_json(silent=True) or {}
    article_id = data.get("article_id")
    if not article_id:
        return fail("缺少参数：article_id")
    if db.session.get(Article, article_id) is None:
        return fail("作品不存在", code=404)

    uid = g.current_user.id

    # 第 ① 步：查「这个用户对这篇作品」有没有记录
    row = Favorite.query.filter_by(user_id=uid, article_id=article_id).first()

    if row:
        # 第 ② 步（分支 A）：已有 → 取消收藏
        db.session.delete(row)
        favorited = False
    else:
        # 第 ② 步（分支 B）：没有 → 新建收藏
        db.session.add(Favorite(user_id=uid, article_id=article_id))
        favorited = True

    # 第 ③ 步：提交，并把最新状态回给前端，前端据此更新按钮
    db.session.commit()
    return success({"favorited": favorited}, "操作成功")


@favorite_bp.get("/list")
@login_required
def favorite_list():
    """② 我的收藏：收藏表 JOIN 作品表，才能拿到标题、作者、分类名"""
    page = max(request.args.get("page", 1, type=int), 1)
    page_size = min(max(request.args.get("page_size", 10, type=int), 1), 50)

    # 以 favorite 为主表：它决定「收藏时间」和分页总数
    query = (Favorite.query
             .filter_by(user_id=g.current_user.id)
             .order_by(Favorite.created_at.desc(), Favorite.id.desc()))
    page_obj = db.paginate(query, page=page, per_page=page_size, error_out=False)

    items = []
    for fav in page_obj.items:
        article = fav.article
        if article is None or article.status != 1:
            continue          # 作品被下架，就不出现在收藏列表里
        items.append({
            "article_id": article.id,
            "title": article.title,
            "author": article.author,
            "category_name": article.category.name if article.category else "",
            "created_at": fav.created_at.strftime("%Y-%m-%d %H:%M:%S"),
        })

    return success({
        "list": items,
        "total": page_obj.total,
        "page": page,
        "page_size": page_size,
    })

// M3 链路（在集合里按这个顺序建请求，Runner 会依次执行）
// 1. POST {{base_url}}/user/register     注册
// 2. POST {{base_url}}/user/login        登录   → 写 token
// 3. GET  {{base_url}}/user/profile      个人信息
// 4. POST {{base_url}}/comment/add       发表书评 → 写 comment_id
// 5. POST {{base_url}}/article/like      点赞作品
// 6. POST {{base_url}}/favorite/toggle   收藏
// 7. GET  {{base_url}}/favorite/list     我的收藏

// ---------- 请求 1「注册」的 Tests ----------
// 每次跑都换一个用户名，避免第二次「用户名已存在」导致整条链断掉
const suffix = Date.now().toString().slice(-6);
pm.environment.set("username", "stu" + suffix);
pm.test("注册成功", function () {
    pm.expect(pm.response.json().code).to.eql(200);
});

// ---------- 请求 2「登录」的 Tests ----------
const loginBody = pm.response.json();
pm.test("登录成功并拿到 token", function () {
    pm.expect(loginBody.code).to.eql(200);
    pm.expect(loginBody.data.token).to.be.a("string");
});
pm.environment.set("token", loginBody.data.token);

// ---------- 请求 4「发表书评」的 Tests ----------
// 把新书评 id 存下来，供后续删除用例复用
const commentBody = pm.response.json();
pm.test("发表成功", function () {
    pm.expect(commentBody.code).to.eql(200);
});
pm.environment.set("comment_id", commentBody.data.comment_id);

// ---------- 请求 5「点赞作品」的 Tests ----------
pm.test("点赞状态被切换", function () {
    const d = pm.response.json().data;
    pm.expect(d).to.have.property("liked");
    pm.expect(d).to.have.property("likes");
});

// ---------- 请求 7「我的收藏」的 Tests ----------
// 链条闭环的判定：刚才收藏的作品，必须出现在我的收藏里
pm.test("收藏的作品出现在我的收藏里", function () {
    const data = pm.response.json().data;
    const ids = data.list.map(function (item) { return item.article_id; });
    pm.expect(ids).to.include(Number(pm.environment.get("article_id")));
});