0933b936f9
Some checks failed
continuous-integration/drone/push Build is failing
- 后端: 新增 user_levels, exp_history, badge_definitions, user_badges, level_configs 表 - 后端: 新增 LevelService 和 BadgeService 服务 - 后端: 新增等级/奖章/签到/排行榜 API 端点 - 后端: 考试/练习/陪练完成时触发经验值和奖章检查 - 前端: 新增 LevelBadge, ExpProgress, BadgeCard, LevelUpDialog 组件 - 前端: 新增排行榜页面 - 前端: 成长路径页面集成真实等级数据 - 数据库: 包含迁移脚本和初始数据
278 lines
7.5 KiB
Python
278 lines
7.5 KiB
Python
"""
|
||
等级与奖章 API
|
||
|
||
提供等级查询、奖章查询、排行榜、签到等接口
|
||
"""
|
||
|
||
from typing import Optional
|
||
from fastapi import APIRouter, Depends, Query
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.deps import get_db, get_current_user
|
||
from app.schemas.base import ResponseModel
|
||
from app.services.level_service import LevelService
|
||
from app.services.badge_service import BadgeService
|
||
from app.models.user import User
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
# ============================================
|
||
# 等级相关接口
|
||
# ============================================
|
||
|
||
@router.get("/me", response_model=ResponseModel)
|
||
async def get_my_level(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取当前用户等级信息
|
||
|
||
返回用户的等级、经验值、称号、连续登录天数等信息
|
||
"""
|
||
level_service = LevelService(db)
|
||
level_info = await level_service.get_user_level_info(current_user.id)
|
||
|
||
return ResponseModel(
|
||
message="获取成功",
|
||
data=level_info
|
||
)
|
||
|
||
|
||
@router.get("/user/{user_id}", response_model=ResponseModel)
|
||
async def get_user_level(
|
||
user_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取指定用户等级信息
|
||
|
||
Args:
|
||
user_id: 用户ID
|
||
"""
|
||
level_service = LevelService(db)
|
||
level_info = await level_service.get_user_level_info(user_id)
|
||
|
||
return ResponseModel(
|
||
message="获取成功",
|
||
data=level_info
|
||
)
|
||
|
||
|
||
@router.post("/checkin", response_model=ResponseModel)
|
||
async def daily_checkin(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
每日签到
|
||
|
||
每天首次签到获得经验值，连续签到有额外奖励
|
||
"""
|
||
level_service = LevelService(db)
|
||
badge_service = BadgeService(db)
|
||
|
||
# 执行签到
|
||
checkin_result = await level_service.daily_checkin(current_user.id)
|
||
|
||
# 检查是否解锁新奖章
|
||
new_badges = []
|
||
if checkin_result["success"]:
|
||
new_badges = await badge_service.check_and_award_badges(current_user.id)
|
||
await db.commit()
|
||
|
||
return ResponseModel(
|
||
message=checkin_result["message"],
|
||
data={
|
||
**checkin_result,
|
||
"new_badges": new_badges
|
||
}
|
||
)
|
||
|
||
|
||
@router.get("/exp-history", response_model=ResponseModel)
|
||
async def get_exp_history(
|
||
limit: int = Query(default=50, ge=1, le=100),
|
||
offset: int = Query(default=0, ge=0),
|
||
exp_type: Optional[str] = Query(default=None, description="经验值类型筛选"),
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取经验值变化历史
|
||
|
||
Args:
|
||
limit: 每页数量（默认50，最大100）
|
||
offset: 偏移量
|
||
exp_type: 类型筛选（exam/practice/training/task/login/badge/other）
|
||
"""
|
||
level_service = LevelService(db)
|
||
history, total = await level_service.get_exp_history(
|
||
user_id=current_user.id,
|
||
limit=limit,
|
||
offset=offset,
|
||
exp_type=exp_type
|
||
)
|
||
|
||
return ResponseModel(
|
||
message="获取成功",
|
||
data={
|
||
"items": history,
|
||
"total": total,
|
||
"limit": limit,
|
||
"offset": offset
|
||
}
|
||
)
|
||
|
||
|
||
@router.get("/leaderboard", response_model=ResponseModel)
|
||
async def get_leaderboard(
|
||
limit: int = Query(default=50, ge=1, le=100),
|
||
offset: int = Query(default=0, ge=0),
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取等级排行榜
|
||
|
||
Args:
|
||
limit: 每页数量（默认50，最大100）
|
||
offset: 偏移量
|
||
"""
|
||
level_service = LevelService(db)
|
||
|
||
# 获取排行榜
|
||
leaderboard, total = await level_service.get_leaderboard(limit=limit, offset=offset)
|
||
|
||
# 获取当前用户排名
|
||
my_rank = await level_service.get_user_rank(current_user.id)
|
||
|
||
# 获取当前用户等级信息
|
||
my_level_info = await level_service.get_user_level_info(current_user.id)
|
||
|
||
return ResponseModel(
|
||
message="获取成功",
|
||
data={
|
||
"items": leaderboard,
|
||
"total": total,
|
||
"limit": limit,
|
||
"offset": offset,
|
||
"my_rank": my_rank,
|
||
"my_level_info": my_level_info
|
||
}
|
||
)
|
||
|
||
|
||
# ============================================
|
||
# 奖章相关接口
|
||
# ============================================
|
||
|
||
@router.get("/badges/all", response_model=ResponseModel)
|
||
async def get_all_badges(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取所有奖章定义
|
||
|
||
返回所有可获得的奖章列表
|
||
"""
|
||
badge_service = BadgeService(db)
|
||
badges = await badge_service.get_all_badges()
|
||
|
||
return ResponseModel(
|
||
message="获取成功",
|
||
data=badges
|
||
)
|
||
|
||
|
||
@router.get("/badges/me", response_model=ResponseModel)
|
||
async def get_my_badges(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取当前用户的奖章（含解锁状态）
|
||
|
||
返回所有奖章及用户是否已解锁
|
||
"""
|
||
badge_service = BadgeService(db)
|
||
badges = await badge_service.get_user_badges_with_status(current_user.id)
|
||
|
||
# 统计已解锁数量
|
||
unlocked_count = sum(1 for b in badges if b["unlocked"])
|
||
|
||
return ResponseModel(
|
||
message="获取成功",
|
||
data={
|
||
"badges": badges,
|
||
"total": len(badges),
|
||
"unlocked_count": unlocked_count
|
||
}
|
||
)
|
||
|
||
|
||
@router.get("/badges/unnotified", response_model=ResponseModel)
|
||
async def get_unnotified_badges(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取未通知的新奖章
|
||
|
||
用于前端显示新获得奖章的弹窗提示
|
||
"""
|
||
badge_service = BadgeService(db)
|
||
badges = await badge_service.get_unnotified_badges(current_user.id)
|
||
|
||
return ResponseModel(
|
||
message="获取成功",
|
||
data=badges
|
||
)
|
||
|
||
|
||
@router.post("/badges/mark-notified", response_model=ResponseModel)
|
||
async def mark_badges_notified(
|
||
badge_ids: Optional[list[int]] = None,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
标记奖章为已通知
|
||
|
||
Args:
|
||
badge_ids: 要标记的奖章ID列表（为空则标记全部）
|
||
"""
|
||
badge_service = BadgeService(db)
|
||
await badge_service.mark_badges_notified(current_user.id, badge_ids)
|
||
await db.commit()
|
||
|
||
return ResponseModel(
|
||
message="标记成功"
|
||
)
|
||
|
||
|
||
@router.post("/check-badges", response_model=ResponseModel)
|
||
async def check_and_award_badges(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
检查并授予符合条件的奖章
|
||
|
||
手动触发奖章检查，返回新获得的奖章
|
||
"""
|
||
badge_service = BadgeService(db)
|
||
new_badges = await badge_service.check_and_award_badges(current_user.id)
|
||
await db.commit()
|
||
|
||
return ResponseModel(
|
||
message="检查完成",
|
||
data={
|
||
"new_badges": new_badges,
|
||
"count": len(new_badges)
|
||
}
|
||
)
|