Bennett/g0v0-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add grade hot cache

Browse Source
This commit is contained in:
咕谷酱
2025-08-21 23:35:25 +08:00
parent 7c193937d1
commit 822d7c6377
13 changed files with 973 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
app/router/v1/user.py
View File

@@ -1,12 +1,16 @@
from __future__ import annotations
import asyncio
from datetime import datetime
from typing import Literal
from app.config import settings
from app.database.lazer_user import User
from app.database.statistics import UserStatistics, UserStatisticsResp
from app.dependencies.database import Database
from app.dependencies.database import Database, get_redis
from app.log import logger
from app.models.score import GameMode
from app.service.user_cache_service import get_user_cache_service
from .router import AllStrModel, router
@@ -38,10 +42,21 @@ class V1User(AllStrModel):
pp_country_rank: int
events: list[dict]
@classmethod
def _get_cache_key(cls, user_id: int, ruleset: GameMode | None = None) -> str:
"""生成 V1 用户缓存键"""
if ruleset:
return f"v1_user:{user_id}:ruleset:{ruleset}"
return f"v1_user:{user_id}"
@classmethod
async def from_db(
cls, session: Database, db_user: User, ruleset: GameMode | None = None
) -> "V1User":
# 确保 user_id 不为 None
if db_user.id is None:
raise ValueError("User ID cannot be None")
ruleset = ruleset or db_user.playmode
current_statistics: UserStatistics | None = None
for i in await db_user.awaitable_attrs.statistics:
@@ -101,24 +116,55 @@ async def get_user(
default=1, ge=1, le=31, description="从现在起所有事件的最大天数"
),
):
redis = get_redis()
cache_service = get_user_cache_service(redis)
# 确定查询方式和用户ID
is_id_query = type == "id" or user.isdigit()
# 解析 ruleset
ruleset = GameMode.from_int_extra(ruleset_id) if ruleset_id else None
# 如果是 ID 查询，先尝试从缓存获取
cached_v1_user = None
user_id_for_cache = None
if is_id_query:
try:
user_id_for_cache = int(user)
cached_v1_user = await cache_service.get_v1_user_from_cache(user_id_for_cache, ruleset)
if cached_v1_user:
return [V1User(**cached_v1_user)]
except (ValueError, TypeError):
pass # 不是有效的用户ID，继续数据库查询
# 从数据库查询用户
db_user = (
await session.exec(
select(User).where(
User.id == user
if type == "id" or user.isdigit()
else User.username == user,
User.id == user if is_id_query else User.username == user,
)
)
).first()
if not db_user:
return []
try:
return [
await V1User.from_db(
session,
db_user,
GameMode.from_int_extra(ruleset_id) if ruleset_id else None,
# 生成用户数据
v1_user = await V1User.from_db(session, db_user, ruleset)
# 异步缓存结果（如果有用户ID）
if db_user.id is not None:
user_data = v1_user.model_dump()
asyncio.create_task(
cache_service.cache_v1_user(user_data, db_user.id, ruleset)
)
]
return [v1_user]
except KeyError:
raise HTTPException(400, "Invalid request")
except ValueError as e:
logger.error(f"Error processing V1 user data: {e}")
raise HTTPException(500, "Internal server error")

144
app/router/v2/cache.py Normal file
View File

@@ -0,0 +1,144 @@
"""
缓存管理和监控接口
提供缓存统计、清理和预热功能
"""
from __future__ import annotations
from app.dependencies.database import get_redis
from app.dependencies.user import get_current_user
from app.service.user_cache_service import get_user_cache_service
from .router import router
from fastapi import Depends, HTTPException, Security
from pydantic import BaseModel
from redis.asyncio import Redis
class CacheStatsResponse(BaseModel):
user_cache: dict
redis_info: dict
@router.get(
"/cache/stats",
response_model=CacheStatsResponse,
name="获取缓存统计信息",
description="获取用户缓存和Redis的统计信息，需要管理员权限。",
tags=["缓存管理"],
)
async def get_cache_stats(
redis: Redis = Depends(get_redis),
# current_user: User = Security(get_current_user, scopes=["admin"]), # 暂时注释，可根据需要启用
):
try:
cache_service = get_user_cache_service(redis)
user_cache_stats = await cache_service.get_cache_stats()
# 获取 Redis 基本信息
redis_info = await redis.info()
redis_stats = {
"connected_clients": redis_info.get("connected_clients", 0),
"used_memory_human": redis_info.get("used_memory_human", "0B"),
"used_memory_peak_human": redis_info.get("used_memory_peak_human", "0B"),
"total_commands_processed": redis_info.get("total_commands_processed", 0),
"keyspace_hits": redis_info.get("keyspace_hits", 0),
"keyspace_misses": redis_info.get("keyspace_misses", 0),
"evicted_keys": redis_info.get("evicted_keys", 0),
"expired_keys": redis_info.get("expired_keys", 0),
}
# 计算缓存命中率
hits = redis_stats["keyspace_hits"]
misses = redis_stats["keyspace_misses"]
hit_rate = hits / (hits + misses) * 100 if (hits + misses) > 0 else 0
redis_stats["cache_hit_rate_percent"] = round(hit_rate, 2)
return CacheStatsResponse(
user_cache=user_cache_stats,
redis_info=redis_stats
)
except Exception as e:
raise HTTPException(500, f"Failed to get cache stats: {str(e)}")
@router.post(
"/cache/invalidate/{user_id}",
name="清除指定用户缓存",
description="清除指定用户的所有缓存数据，需要管理员权限。",
tags=["缓存管理"],
)
async def invalidate_user_cache(
user_id: int,
redis: Redis = Depends(get_redis),
# current_user: User = Security(get_current_user, scopes=["admin"]), # 暂时注释
):
try:
cache_service = get_user_cache_service(redis)
await cache_service.invalidate_user_cache(user_id)
await cache_service.invalidate_v1_user_cache(user_id)
return {"message": f"Cache invalidated for user {user_id}"}
except Exception as e:
raise HTTPException(500, f"Failed to invalidate cache: {str(e)}")
@router.post(
"/cache/clear",
name="清除所有用户缓存",
description="清除所有用户相关的缓存数据，需要管理员权限。谨慎使用！",
tags=["缓存管理"],
)
async def clear_all_user_cache(
redis: Redis = Depends(get_redis),
# current_user: User = Security(get_current_user, scopes=["admin"]), # 暂时注释
):
try:
# 获取所有用户相关的缓存键
user_keys = await redis.keys("user:*")
v1_user_keys = await redis.keys("v1_user:*")
all_keys = user_keys + v1_user_keys
if all_keys:
await redis.delete(*all_keys)
return {"message": f"Cleared {len(all_keys)} cache entries"}
else:
return {"message": "No cache entries found"}
except Exception as e:
raise HTTPException(500, f"Failed to clear cache: {str(e)}")
class CacheWarmupRequest(BaseModel):
user_ids: list[int] | None = None
limit: int = 100
@router.post(
"/cache/warmup",
name="缓存预热",
description="对指定用户或活跃用户进行缓存预热，需要管理员权限。",
tags=["缓存管理"],
)
async def warmup_cache(
request: CacheWarmupRequest,
redis: Redis = Depends(get_redis),
# current_user: User = Security(get_current_user, scopes=["admin"]), # 暂时注释
):
try:
cache_service = get_user_cache_service(redis)
if request.user_ids:
# 预热指定用户
from app.dependencies.database import with_db
async with with_db() as session:
await cache_service.preload_user_cache(session, request.user_ids)
return {"message": f"Warmed up cache for {len(request.user_ids)} users"}
else:
# 预热活跃用户
from app.scheduler.user_cache_scheduler import schedule_user_cache_preload_task
await schedule_user_cache_preload_task()
return {"message": f"Warmed up cache for top {request.limit} active users"}
except Exception as e:
raise HTTPException(500, f"Failed to warmup cache: {str(e)}")

12
app/router/v2/score.py
View File

@@ -47,6 +47,7 @@ from app.models.score import (
Rank,
SoloScoreSubmissionInfo,
)
from app.service.user_cache_service import get_user_cache_service
from app.storage.base import StorageService
from app.storage.local import LocalStorageService
@@ -182,6 +183,17 @@ async def submit_score(
}
db.add(rank_event)
await db.commit()
# 成绩提交后刷新用户缓存
try:
user_cache_service = get_user_cache_service(redis)
if current_user.id is not None:
await user_cache_service.refresh_user_cache_on_score_submit(
db, current_user.id, score.gamemode
)
except Exception as e:
logger.error(f"Failed to refresh user cache after score submit: {e}")
background_task.add_task(process_user_achievement, resp.id)
return resp

143
app/router/v2/user.py
View File

@@ -1,8 +1,10 @@
from __future__ import annotations
import asyncio
from datetime import UTC, datetime, timedelta
from typing import Literal
from app.config import settings
from app.const import BANCHOBOT_ID
from app.database import (
BeatmapPlaycounts,
@@ -15,10 +17,11 @@ from app.database.events import EventResp
from app.database.lazer_user import SEARCH_INCLUDED
from app.database.pp_best_score import PPBestScore
from app.database.score import Score, ScoreResp
from app.dependencies.database import Database
from app.dependencies.database import Database, get_redis
from app.dependencies.user import get_current_user
from app.models.score import GameMode
from app.models.user import BeatmapsetType
from app.service.user_cache_service import get_user_cache_service
from .router import router
@@ -51,23 +54,55 @@ async def get_users(
default=False, description="是否包含各模式的统计信息"
), # TODO: future use
):
redis = get_redis()
cache_service = get_user_cache_service(redis)
if user_ids:
searched_users = (
await session.exec(select(User).limit(50).where(col(User.id).in_(user_ids)))
).all()
# 先尝试从缓存获取
cached_users = []
uncached_user_ids = []
for user_id in user_ids[:50]: # 限制50个
cached_user = await cache_service.get_user_from_cache(user_id)
if cached_user:
cached_users.append(cached_user)
else:
uncached_user_ids.append(user_id)
# 查询未缓存的用户
if uncached_user_ids:
searched_users = (
await session.exec(select(User).where(col(User.id).in_(uncached_user_ids)))
).all()
# 将查询到的用户添加到缓存并返回
for searched_user in searched_users:
if searched_user.id != BANCHOBOT_ID:
user_resp = await UserResp.from_db(
searched_user,
session,
include=SEARCH_INCLUDED,
)
cached_users.append(user_resp)
# 异步缓存，不阻塞响应
asyncio.create_task(cache_service.cache_user(user_resp))
return BatchUserResponse(users=cached_users)
else:
searched_users = (await session.exec(select(User).limit(50))).all()
return BatchUserResponse(
users=[
await UserResp.from_db(
searched_user,
session,
include=SEARCH_INCLUDED,
)
for searched_user in searched_users
if searched_user.id != BANCHOBOT_ID
]
)
users = []
for searched_user in searched_users:
if searched_user.id != BANCHOBOT_ID:
user_resp = await UserResp.from_db(
searched_user,
session,
include=SEARCH_INCLUDED,
)
users.append(user_resp)
# 异步缓存
asyncio.create_task(cache_service.cache_user(user_resp))
return BatchUserResponse(users=users)
@router.get(
@@ -83,6 +118,16 @@ async def get_user_info_ruleset(
ruleset: GameMode | None = Path(description="指定 ruleset"),
# current_user: User = Security(get_current_user, scopes=["public"]),
):
redis = get_redis()
cache_service = get_user_cache_service(redis)
# 如果是数字ID，先尝试从缓存获取
if user_id.isdigit():
user_id_int = int(user_id)
cached_user = await cache_service.get_user_from_cache(user_id_int, ruleset)
if cached_user:
return cached_user
searched_user = (
await session.exec(
select(User).where(
@@ -94,12 +139,18 @@ async def get_user_info_ruleset(
).first()
if not searched_user or searched_user.id == BANCHOBOT_ID:
raise HTTPException(404, detail="User not found")
return await UserResp.from_db(
user_resp = await UserResp.from_db(
searched_user,
session,
include=SEARCH_INCLUDED,
ruleset=ruleset,
)
# 异步缓存结果
asyncio.create_task(cache_service.cache_user(user_resp, ruleset))
return user_resp
@router.get("/users/{user_id}/", response_model=UserResp, include_in_schema=False)
@@ -115,6 +166,16 @@ async def get_user_info(
user_id: str = Path(description="用户 ID 或用户名"),
# current_user: User = Security(get_current_user, scopes=["public"]),
):
redis = get_redis()
cache_service = get_user_cache_service(redis)
# 如果是数字ID，先尝试从缓存获取
if user_id.isdigit():
user_id_int = int(user_id)
cached_user = await cache_service.get_user_from_cache(user_id_int)
if cached_user:
return cached_user
searched_user = (
await session.exec(
select(User).where(
@@ -126,11 +187,17 @@ async def get_user_info(
).first()
if not searched_user or searched_user.id == BANCHOBOT_ID:
raise HTTPException(404, detail="User not found")
return await UserResp.from_db(
user_resp = await UserResp.from_db(
searched_user,
session,
include=SEARCH_INCLUDED,
)
# 异步缓存结果
asyncio.create_task(cache_service.cache_user(user_resp))
return user_resp
@router.get(
@@ -148,6 +215,20 @@ async def get_user_beatmapsets(
limit: int = Query(100, ge=1, le=1000, description="返回条数 (1-1000)"),
offset: int = Query(0, ge=0, description="偏移量"),
):
redis = get_redis()
cache_service = get_user_cache_service(redis)
# 先尝试从缓存获取
cached_result = await cache_service.get_user_beatmapsets_from_cache(
user_id, type.value, limit, offset
)
if cached_result is not None:
# 根据类型恢复对象
if type == BeatmapsetType.MOST_PLAYED:
return [BeatmapPlaycountsResp(**item) for item in cached_result]
else:
return [BeatmapsetResp(**item) for item in cached_result]
user = await session.get(User, user_id)
if not user or user.id == BANCHOBOT_ID:
raise HTTPException(404, detail="User not found")
@@ -190,6 +271,11 @@ async def get_user_beatmapsets(
else:
raise HTTPException(400, detail="Invalid beatmapset type")
# 异步缓存结果
asyncio.create_task(
cache_service.cache_user_beatmapsets(user_id, type.value, resp, limit, offset)
)
return resp
@@ -218,6 +304,17 @@ async def get_user_scores(
offset: int = Query(0, ge=0, description="偏移量"),
current_user: User = Security(get_current_user, scopes=["public"]),
):
redis = get_redis()
cache_service = get_user_cache_service(redis)
# 先尝试从缓存获取（对于recent类型使用较短的缓存时间）
cache_expire = 30 if type == "recent" else settings.user_scores_cache_expire_seconds
cached_scores = await cache_service.get_user_scores_from_cache(
user_id, type, mode, limit, offset
)
if cached_scores is not None:
return cached_scores
db_user = await session.get(User, user_id)
if not db_user or db_user.id == BANCHOBOT_ID:
raise HTTPException(404, detail="User not found")
@@ -253,13 +350,23 @@ async def get_user_scores(
).all()
if not scores:
return []
return [
score_responses = [
await ScoreResp.from_db(
session,
score,
)
for score in scores
]
# 异步缓存结果
asyncio.create_task(
cache_service.cache_user_scores(
user_id, type, score_responses, mode, limit, offset, cache_expire
)
)
return score_responses
@router.get(