第15章 · 看图猜成语小程序后台

import sqlite3, json

conn = sqlite3.connect("idiom.db")
c = conn.cursor()

# ============ 用户表 ============
c.execute("DROP TABLE IF EXISTS users")
c.execute("""
    CREATE TABLE users (
        id        INTEGER PRIMARY KEY AUTOINCREMENT,
        username  TEXT UNIQUE NOT NULL,
        nickname  TEXT,
        level     INTEGER DEFAULT 1   -- 当前关卡，从 1 开始
    )
""")

# ============ 题目表 ============
# answer = "杯弓蛇影"   options = "杯弓蛇影山月日云风火..."（含正确答案的备选字）
c.execute("DROP TABLE IF EXISTS games")
c.execute("""
    CREATE TABLE games (
        level    INTEGER PRIMARY KEY,    -- 关卡号
        image    TEXT NOT NULL,           -- 图片 URL（emoji 代替）
        answer   TEXT NOT NULL,           -- 正确答案
        options  TEXT NOT NULL            -- 18 个备选字（一串）
    )
""")

# 灌一组测试题
games = [
    (1, "🐍🌿",    "杯弓蛇影", "杯弓蛇影山月日云风火水土木金"),
    (2, "🐎🚀",    "一马当先", "一马当先二三四五六七八九十百千"),
    (3, "🌕🌳",    "花好月圆", "花好月圆春夏秋冬山水风云东西南北"),
    (4, "🐯🐲",    "龙腾虎跃", "龙腾虎跃日月星辰天地玄黄宇宙洪荒"),
]
for g in games:
    c.execute("INSERT INTO games VALUES (?, ?, ?, ?)", g)

# 默认建一个测试用户
c.execute("INSERT INTO users (username, nickname) VALUES (?, ?)",
          ("alice", "小爱"))

conn.commit(); conn.close()
print("✅ 数据库就绪")

import sqlite3
DB = "idiom.db"


def _conn():
    c = sqlite3.connect(DB)
    c.row_factory = sqlite3.Row
    return c


def get_or_create_user(username: str):
    # 没注册就建一个，返回用户行
    with _conn() as c:
        u = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
        if u: return u
        c.execute("INSERT INTO users (username, nickname) VALUES (?, ?)",
                  (username, username))
        return c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()


def get_game(level: int):
    with _conn() as c:
        return c.execute("SELECT * FROM games WHERE level=?", (level,)).fetchone()


def level_up(username: str):
    # 答对了 → 关卡 +1
    with _conn() as c:
        c.execute("UPDATE users SET level=level+1 WHERE username=?", (username,))


def rank_top10():
    with _conn() as c:
        return c.execute(
            "SELECT nickname, level FROM users ORDER BY level DESC LIMIT 10"
        ).fetchall()

# ============ JWT 鉴权工具 ============
from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from fastapi import Header, HTTPException


# 上线项目要从环境变量读，且要足够复杂的随机串
SECRET_KEY = "idiom-2026-classroom-demo-key"
ALGORITHM = "HS256"          # HMAC + SHA256，最常用的 JWT 算法
EXPIRE_DAYS = 7


def create_token(username: str) -> str:
    # payload 是要塞进 token 里的数据
    # sub 是 JWT 标准字段——subject（主体），用来放用户标识
    # exp 是过期时间
    payload = {
        "sub": username,
        "exp": datetime.now(timezone.utc) + timedelta(days=EXPIRE_DAYS)
    }
    # jwt.encode：用 SECRET_KEY 给 payload 签名 → 字符串 token
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)


def get_current_user(authorization: str = Header(None)) -> str:
    # 这是 FastAPI 的"依赖项"——在每个需要登录的接口参数里 Depends 它
    # Header(None) 表示从 HTTP 头取 Authorization

    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="缺少认证 token")

    # 标准格式："Bearer eyJhbGciOiJIUzI1..."
    token = authorization[7:]   # 去掉 "Bearer " 前缀

    try:
        # jwt.decode：用 SECRET_KEY 验签 + 自动检查过期
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if not username:
            raise HTTPException(status_code=401, detail="token 内容不正确")
        return username
    except JWTError:
        raise HTTPException(status_code=401, detail="token 无效或已过期")

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
import db
from auth import create_token, get_current_user

app = FastAPI(title="看图猜成语 API")


# =========== 请求 / 响应模型 ===========
class LoginRequest(BaseModel):
    username: str


class LoginResponse(BaseModel):
    token: str
    nickname: str
    level: int


# =========== ① 登录接口 ===========
@app.post("/login", response_model=LoginResponse)
def login(req: LoginRequest):
    # 在小程序对接时，这里用 wx.login 拿到的 openid 替代 username
    user = db.get_or_create_user(req.username)
    return LoginResponse(
        token=create_token(req.username),
        nickname=user["nickname"],
        level=user["level"],
    )

# =========== 业务模型 ===========
class GameInfo(BaseModel):
    level: int
    image: str
    options: str          # 18 个字组成的字符串


class AnswerRequest(BaseModel):
    answer: str           # 用户提交的 4 字答案


class AnswerResponse(BaseModel):
    correct: bool
    next_level: int      # 答对则进入的下一关；答错保持原关
    finished: bool        # 是否通关


# =========== ② 当前关卡题目 ===========
# 注意：参数 user = Depends(get_current_user)
# Depends：FastAPI 自动调用 get_current_user 把返回值塞进来
# 没有合法 token 的请求会被它拦下来 → 返回 401，根本进不到函数体
@app.get("/game/current", response_model=GameInfo)
def current_game(user: str = Depends(get_current_user)):
    u = db.get_or_create_user(user)
    game = db.get_game(u["level"])
    if not game:
        raise HTTPException(404, "你已经通关啦！")
    # 注意：不能把 answer 字段返回给前端——前端拿到答案就作弊了
    return GameInfo(
        level=game["level"],
        image=game["image"],
        options=game["options"],
    )


# =========== ③ 提交答案 ===========
@app.post("/game/answer", response_model=AnswerResponse)
def submit_answer(req: AnswerRequest,
                   user: str = Depends(get_current_user)):
    u = db.get_or_create_user(user)
    game = db.get_game(u["level"])

    if not game:
        return AnswerResponse(correct=False,
                              next_level=u["level"],
                              finished=True)

    # 比较答案。strip 防止前端误传空格
    if req.answer.strip() == game["answer"]:
        db.level_up(user)         # 关卡 +1
        new_level = u["level"] + 1
        finished = db.get_game(new_level) is None   # 没下一题 = 通关
        return AnswerResponse(correct=True,
                              next_level=new_level,
                              finished=finished)
    else:
        # 错了不升级，让前端提示用户重答
        return AnswerResponse(correct=False,
                              next_level=u["level"],
                              finished=False)

from typing import List


class RankItem(BaseModel):
    rank: int
    nickname: str
    level: int


# =========== ④ 排行榜 ===========
# 不加 Depends 就是公开接口，任何人都能访问
@app.get("/rank", response_model=List[RankItem])
def rank():
    rows = db.rank_top10()
    # enumerate(rows, start=1) → (1, rowA), (2, rowB), ...
    return [
        RankItem(rank=i, nickname=r["nickname"], level=r["level"])
        for i, r in enumerate(rows, start=1)
    ]