feat(recalculate): add scheduled job to recalculate failed scores

app/calculator.py

@@ -14,6 +14,7 @@ from app.models.score import GameMode
 
 from osupyparser import HitObject, OsuFile
 from osupyparser.osu.objects import Slider
+from redis.asyncio import Redis
 from sqlmodel import col, exists, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -29,6 +30,7 @@ except ImportError:
 
 if TYPE_CHECKING:
     from app.database.score import Score
+    from app.fetcher import Fetcher
 
 
 def clamp[T: int | float](n: T, min_value: T, max_value: T) -> T:
@@ -123,28 +125,30 @@ async def calculate_pp(score: "Score", beatmap: str, session: AsyncSession) -> f
     return pp
 
 
-async def pre_fetch_and_calculate_pp(score: "Score", beatmap_id: int, session: AsyncSession, redis, fetcher) -> float:
+async def pre_fetch_and_calculate_pp(
+    score: "Score", session: AsyncSession, redis: Redis, fetcher: "Fetcher"
+) -> tuple[float, bool]:
     """
     优化版PP计算：预先获取beatmap文件并使用缓存
     """
-    import asyncio
-
     from app.database.beatmap import BannedBeatmaps
 
+    beatmap_id = score.beatmap_id
+
     # 快速检查是否被封禁
     if settings.suspicious_score_check:
         beatmap_banned = (
             await session.exec(select(exists()).where(col(BannedBeatmaps.beatmap_id) == beatmap_id))
         ).first()
         if beatmap_banned:
-            return 0
+            return 0, False
 
     # 异步获取beatmap原始文件，利用已有的Redis缓存机制
     try:
         beatmap_raw = await fetcher.get_or_fetch_beatmap_raw(redis, beatmap_id)
     except Exception as e:
         logger.error(f"Failed to fetch beatmap {beatmap_id}: {e}")
-        return 0
+        return 0, False
 
     # 在获取文件的同时，可以检查可疑beatmap
     if settings.suspicious_score_check:
@@ -158,12 +162,12 @@ async def pre_fetch_and_calculate_pp(score: "Score", beatmap_id: int, session: A
             if is_sus:
                 session.add(BannedBeatmaps(beatmap_id=beatmap_id))
                 logger.warning(f"Beatmap {beatmap_id} is suspicious, banned")
-                return 0
+                return 0, True
         except Exception:
             logger.exception(f"Error checking if beatmap {beatmap_id} is suspicious")
 
     # 调用已优化的PP计算函数
-    return await calculate_pp(score, beatmap_raw, session)
+    return await calculate_pp(score, beatmap_raw, session), True
 
 
 async def batch_calculate_pp(

app/database/score.py

@@ -11,6 +11,7 @@ from app.calculator import (
     calculate_weighted_acc,
     calculate_weighted_pp,
     clamp,
+    pre_fetch_and_calculate_pp,
 )
 from app.database.team import TeamMember
 from app.dependencies.database import get_redis
@@ -880,15 +881,16 @@ async def process_score(
         maximum_statistics=info.maximum_statistics,
         processed=True,
     )
+    successed = True
     if can_get_pp:
-        from app.calculator import pre_fetch_and_calculate_pp
+        pp, successed = await pre_fetch_and_calculate_pp(score, session, redis, fetcher)
-
-        pp = await pre_fetch_and_calculate_pp(score, beatmap_id, session, redis, fetcher)
         score.pp = pp
     session.add(score)
     user_id = user.id
     await session.commit()
     await session.refresh(score)
+    if not successed:
+        await redis.rpush("score:need_recalculate", score.id)  # pyright: ignore[reportGeneralTypeIssues]
     if can_get_pp and score.pp != 0:
         previous_pp_best = await get_user_best_pp_in_beatmap(session, beatmap_id, user_id, score.gamemode)
         if previous_pp_best is None or score.pp > previous_pp_best.pp:

app/service/__init__.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from .daily_challenge import create_daily_challenge_room
 from .recalculate_banned_beatmap import recalculate_banned_beatmap
+from .recalculate_failed_score import recalculate_failed_score
 from .room import create_playlist_room, create_playlist_room_from_api
 
 __all__ = [
@@ -9,4 +10,5 @@ __all__ = [
     "create_playlist_room",
     "create_playlist_room_from_api",
     "recalculate_banned_beatmap",
+    "recalculate_failed_score",
 ]

app/service/recalculate_failed_score.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from app.calculator import pre_fetch_and_calculate_pp
+from app.database.score import Score, calculate_user_pp
+from app.database.statistics import UserStatistics
+from app.dependencies.database import get_redis, with_db
+from app.dependencies.fetcher import get_fetcher
+from app.dependencies.scheduler import get_scheduler
+from app.log import logger
+
+from sqlmodel import select
+
+
+@get_scheduler().scheduled_job("interval", id="recalculate_banned_beatmap", minutes=5)
+async def recalculate_failed_score():
+    redis = get_redis()
+    fetcher = await get_fetcher()
+    need_add = set()
+    affected_user = set()
+    while True:
+        scores = await redis.lpop("score:need_recalculate", 100)  # pyright: ignore[reportGeneralTypeIssues]
+        if not scores:
+            break
+        if isinstance(scores, bytes):
+            scores = [scores]
+        async with with_db() as session:
+            for score_id in scores:
+                score_id = int(score_id)
+                score = await session.get(Score, score_id)
+                if score is None:
+                    continue
+                pp, successed = await pre_fetch_and_calculate_pp(score, session, redis, fetcher)
+                if not successed:
+                    need_add.add(score_id)
+                else:
+                    score.pp = pp
+                    logger.info(
+                        f"Recalculated PP for score {score.id} (user: {score.user_id}) at {score.ended_at}: {pp}"
+                    )
+                    affected_user.add((score.user_id, score.gamemode))
+            await session.commit()
+            for user_id, gamemode in affected_user:
+                stats = (
+                    await session.exec(
+                        select(UserStatistics).where(UserStatistics.user_id == user_id, UserStatistics.mode == gamemode)
+                    )
+                ).first()
+                if not stats:
+                    continue
+                stats.pp, stats.hit_accuracy = await calculate_user_pp(session, user_id, gamemode)
+            await session.commit()
+    if need_add:
+        await redis.rpush("score:need_recalculate", *need_add)  # pyright: ignore[reportGeneralTypeIssues]