阶段 3 · 后端开发

# 在 pyweb/code/hiflask/ 目录下执行
python -m venv .venv

# macOS / Linux 激活
source .venv/bin/activate
# Windows（PowerShell）：.venv\Scripts\Activate.ps1

pip install -r requirements.txt
pip list | grep -i flask

tree -L 2 --dirsfirst -I '.venv|__pycache__'

from flask import Flask

from config import config_map
from extensions import cors, db, jwt, migrate
from models import *  # noqa: F401,F403  —— 让 Flask-Migrate 能发现全部模型
from views import register_blueprints


def create_app(config_name: str = "dev") -> Flask:
    """创建并装配一个 Flask 应用。"""
    app = Flask(__name__)
    app.config.from_object(config_map[config_name])
    # 让 jsonify 直接输出中文，而不是 \uXXXX 转义（Flask 2.3+ 的写法）
    app.json.ensure_ascii = False

    # ① 初始化扩展（实例在 extensions.py 里创建，这里只做绑定）
    db.init_app(app)
    jwt.init_app(app)
    cors.init_app(app, resources={r"/api/*": {"origins": "*"}})
    migrate.init_app(app, db)

    # ② 挂上所有蓝图（18 个接口都在这一步注册）
    register_blueprints(app)

    # ③ 统一 JWT 异常与全局错误处理
    register_jwt_handlers()
    register_error_handlers(app)

    return app


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

import os
from urllib.parse import quote_plus

from dotenv import load_dotenv

BASE_DIR = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(BASE_DIR, ".env"))


def build_mysql_uri() -> str:
    """按环境变量拼出 MySQL 连接串。

    密码交给 quote_plus 处理：像 @ / # / % 这类字符不会把连接串拼坏。
    """
    user = os.getenv("DB_USER", "root")
    password = os.getenv("DB_PASSWORD", "123456")
    host = os.getenv("DB_HOST", "127.0.0.1")
    port = os.getenv("DB_PORT", "3306")
    name = os.getenv("DB_NAME", "hiwen")
    return (
        f"mysql+pymysql://{user}:{quote_plus(password)}@{host}:{port}/{name}"
        "?charset=utf8mb4"
    )


class Config:
    SECRET_KEY = os.getenv("SECRET_KEY", "dev-secret-key-change-me")
    JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY", "dev-jwt-secret-change-me")

    SQLALCHEMY_DATABASE_URI = build_mysql_uri()
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    # MySQL 连接空闲 8 小时会被服务端断开，pool_recycle 让连接提前回收
    SQLALCHEMY_ENGINE_OPTIONS = {"pool_recycle": 3600, "pool_pre_ping": True}

from datetime import datetime

from werkzeug.security import check_password_hash, generate_password_hash

from extensions import db


class User(db.Model):
    """用户。表名单数，字段与 hiwen.sql 逐字对应。"""

    __tablename__ = "user"

    id = db.Column(db.Integer, primary_key=True, comment="用户ID")
    username = db.Column(db.String(50), nullable=False, unique=True, comment="登录名")
    password = db.Column(db.String(255), nullable=False, comment="密码哈希（切勿明文）")
    nickname = db.Column(db.String(50), comment="昵称")
    email = db.Column(db.String(100), unique=True, comment="邮箱（唯一，可空）")
    avatar = db.Column(db.String(255), default="", comment="头像路径")
    intro = db.Column(db.String(255), comment="个人简介")
    gender = db.Column(db.SmallInteger, default=0, comment="0未知 1男 2女")
    status = db.Column(db.SmallInteger, nullable=False, default=1, comment="1正常 0禁用")
    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
    )

    # ---------- 密码：只存哈希，绝不存明文 ----------
    def set_password(self, raw: str) -> None:
        self.password = generate_password_hash(raw)  # 自动加盐

    def check_password(self, raw: str) -> bool:
        return check_password_hash(self.password, raw)

    # ---------- 序列化：白名单写法，password 永远不会出现在响应里 ----------
    def to_dict(self) -> dict:
        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"),
        }

    def to_brief(self) -> dict:
        """登录响应里的 user 字段，只要四个字段。"""
        return {
            "id": self.id,
            "username": self.username,
            "nickname": self.nickname,
            "avatar": self.avatar or "",
        }

from flask import jsonify


def success(data=None, msg: str = "success"):
    """成功响应。"""
    return jsonify(code=200, msg=msg, data=data)


def fail(msg: str = "参数错误", code: int = 400):
    """失败响应。msg 可直接展示给用户。"""
    return jsonify(code=code, msg=msg, data=None)


def page_data(items: list, total: int, page: int, page_size: int) -> dict:
    """分页响应的 data 结构（接口文档 1.5 节）。"""
    return {"list": items, "total": total, "page": page, "page_size": page_size}

from werkzeug.exceptions import HTTPException


def register_error_handlers(app: Flask) -> None:
    """兜底异常处理：绝不把堆栈抛给前端。"""

    @app.errorhandler(404)
    def not_found(error):
        return fail("接口不存在", code=404)

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

    @app.errorhandler(Exception)
    def internal_error(error):
        # 其余 HTTP 异常（400/415 等）保持原本的语义，不要一律当 500
        if isinstance(error, HTTPException):
            return error
        db.session.rollback()  # 出异常后必须回滚，否则会话一直处于失败状态
        app.logger.exception("未捕获的服务器异常")
        return fail("服务器内部错误", code=500)

import re

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

from extensions import db
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():
    """注册：POST /api/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("用户名已存在")

    nickname = (data.get("nickname") or "").strip() or username
    user = User(username=username, nickname=nickname)
    user.set_password(password)  # 只存哈希
    db.session.add(user)
    db.session.commit()
    return success({"user_id": user.id}, "注册成功")


@user_bp.post("/login")
def user_login():
    """登录：POST /api/user/login → 返回 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": user.to_brief()}, "登录成功")

from functools import wraps

from flask import g
from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request

from extensions import db
from models.user import User
from utils.response import fail


def login_required(view_func):
    """需要登录的视图加上它。

    流程：校验 token → 按 identity 取出用户 → 检查状态 → 挂到 g.current_user。
    token 缺失 / 无效 / 过期时，由 app.py 里注册的 JWT 回调统一返回 401。
    """

    @wraps(view_func)
    def wrapper(*args, **kwargs):
        verify_jwt_in_request()

        identity = get_jwt_identity()
        try:
            user_id = int(identity)
        except (TypeError, ValueError):
            return fail("登录状态无效，请重新登录", code=401)

        user = db.session.get(User, user_id)
        if user is None:
            return fail("请先登录", code=401)
        if user.status != 1:
            return fail("账号已被禁用", code=403)

        g.current_user = user
        return view_func(*args, **kwargs)

    return wrapper


def optional_login():
    """可选登录：带了合法 token 就返回 user_id，没带或失效返回 None。

    给「公开接口但要回填当前用户状态」的场景用，比如作品详情里的
    is_favorite / is_liked —— 没登录也能看，只是状态都是 false。
    """
    try:
        verify_jwt_in_request(optional=True)
    except Exception:
        return None
    identity = get_jwt_identity()
    if not identity:
        return None
    try:
        return int(identity)
    except (TypeError, ValueError):
        return None

"""蓝图集中注册。app.py 只调用 register_blueprints(app)，不关心有几个模块。"""

from views.article import article_bp
from views.comment import comment_bp
from views.favorite import favorite_bp
from views.home import home_bp
from views.user import user_bp

ALL_BLUEPRINTS = (home_bp, user_bp, article_bp, comment_bp, favorite_bp)


def register_blueprints(app) -> None:
    for bp in ALL_BLUEPRINTS:
        app.register_blueprint(bp)

from flask import Blueprint, request

from extensions import db
from models.article import Article
from utils.response import page_data, 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 / category_id / dynasty / keyword / order_by"""
    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)

    # 条件逐段叠加到同一个 query 上，有几段加几段
    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())

    page_obj = db.paginate(query, page=page, per_page=page_size, error_out=False)
    return success(
        page_data(
            [a.to_list_dict() for a in page_obj.items],
            page_obj.total,
            page,
            page_size,
        )
    )

// ① 环境变量（三个）
//    base_url    = http://127.0.0.1:5000/api
//    token       =            （留空，登录后自动写入）
//    article_id  = 1

// ② 在【登录】请求的 Tests 标签里写断言：既验证结果，又把 token 存下来
pm.test("登录成功并拿到 token", function () {
    const body = pm.response.json();
    pm.expect(body.code).to.eql(200);
    pm.expect(body.data.token).to.be.a("string");
});

// ③ 写入环境变量，后面的请求用 {{token}} 自动带上
pm.environment.set("token", pm.response.json().data.token);

// ④ 需要登录的请求，在 Headers 里加一行：
//    Authorization : Bearer {{token}}      ← Bearer 后面必须有一个空格

BASE=http://127.0.0.1:5000/api

# ① 注册（用户名重复会返回 400，换一个即可）
curl -s -X POST $BASE/user/register \
     -H "Content-Type: application/json" \
     -d '{"username": "zhangsan", "password": "123456", "nickname": "张三"}' \
     | python -m json.tool

# ② 登录，把 token 存进 shell 变量
TOKEN=$(curl -s -X POST $BASE/user/login \
        -H "Content-Type: application/json" \
        -d '{"username": "zhangsan", "password": "123456"}' \
        | python -c "import sys, json; print(json.load(sys.stdin)['data']['token'])")

# ③ 带着 token 查个人信息
curl -s $BASE/user/profile -H "Authorization: Bearer $TOKEN" | python -m json.tool

# ④ 反证：不带 token 应该被挡回来
curl -s $BASE/user/profile | python -m json.tool