feat(score): support recalculate statistics

app/calculator.py

@@ -75,11 +75,16 @@ async def calculate_pp(score: "Score", beatmap: str, session: AsyncSession) -> f
         ).first()
         if beatmap_banned:
             return 0
-        is_suspicious = is_suspicious_beatmap(beatmap)
-        if is_suspicious:
-            session.add(BannedBeatmaps(beatmap_id=score.beatmap_id))
-            logger.warning(f"Beatmap {score.beatmap_id} is suspicious, banned")
-            return 0
+        try:
+            is_suspicious = is_suspicious_beatmap(beatmap)
+            if is_suspicious:
+                session.add(BannedBeatmaps(beatmap_id=score.beatmap_id))
+                logger.warning(f"Beatmap {score.beatmap_id} is suspicious, banned")
+                return 0
+        except Exception:
+            logger.exception(
+                f"Error checking if beatmap {score.beatmap_id} is suspicious"
+            )
 
     map = rosu.Beatmap(content=beatmap)
     mods = deepcopy(score.mods.copy())
@@ -108,7 +113,8 @@ async def calculate_pp(score: "Score", beatmap: str, session: AsyncSession) -> f
         (attrs.difficulty.stars > 25 and score.accuracy < 0.8) or pp > 2300
     ):
         logger.warning(
-            f"User {score.user_id} played {score.beatmap_id} with {pp=} "
+            f"User {score.user_id} played {score.beatmap_id} "
+            f"(star={attrs.difficulty.stars}) with {pp=} "
             f"acc={score.accuracy}. The score is suspicious and return 0pp"
             f"({score.id=})"
         )
@@ -335,7 +341,7 @@ def is_2b(hit_objects: list[HitObject]) -> bool:
 
 
 def is_suspicious_beatmap(content: str) -> bool:
-    osufile = OsuFile(content=content.encode("utf-8-sig")).parse_file()
+    osufile = OsuFile(content=content.encode("utf-8")).parse_file()
 
     if (
         osufile.hit_objects[-1].start_time - osufile.hit_objects[0].start_time

app/database/score.py

@@ -21,7 +21,7 @@ from app.models.model import (
     RespWithCursor,
     UTCBaseModel,
 )
-from app.models.mods import APIMod, mods_can_get_pp
+from app.models.mods import APIMod, mod_to_save, mods_can_get_pp
 from app.models.score import (
     GameMode,
     HitResult,
@@ -443,7 +443,6 @@ async def get_user_best_score_in_beatmap(
     ).first()
 
 
-# FIXME
 async def get_user_best_score_with_mod_in_beatmap(
     session: AsyncSession,
     beatmap: int,
@@ -458,7 +457,7 @@ async def get_user_best_score_with_mod_in_beatmap(
                 BestScore.gamemode == mode if mode is not None else True,
                 BestScore.beatmap_id == beatmap,
                 BestScore.user_id == user,
-                # BestScore.mods == mod,
+                BestScore.mods == mod,
             )
             .order_by(col(BestScore.total_score).desc())
         )
@@ -508,7 +507,7 @@ async def process_user(
 ):
     assert user.id
     assert score.id
-    mod_for_save = list({mod["acronym"] for mod in score.mods})
+    mod_for_save = mod_to_save(score.mods)
     previous_score_best = await get_user_best_score_in_beatmap(
         session, score.beatmap_id, user.id, score.gamemode
     )

app/models/mods.py

@@ -201,3 +201,9 @@ def parse_enum_to_str(ruleset_id: int, mods: list[APIMod]):
             for setting in mod.get("settings", {}):
                 if setting in ENUM_TO_STR[ruleset_id][mod["acronym"]]:
                     mod["settings"][setting] = str(mod["settings"][setting])  # pyright: ignore[reportTypedDictNotRequiredAccess]
+
+
+def mod_to_save(mods: list[APIMod]) -> list[str]:
+    s = list({mod["acronym"] for mod in mods})
+    s.sort()
+    return s

app/service/pp_recalculate.py

@@ -1,123 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-import math
-
-from app.calculator import (
-    calculate_pp,
-    calculate_weighted_acc,
-    calculate_weighted_pp,
-    clamp,
-)
-from app.config import settings
-from app.database import UserStatistics
-from app.database.beatmap import Beatmap
-from app.database.pp_best_score import PPBestScore
-from app.database.score import Score
-from app.dependencies.database import engine, get_redis
-from app.dependencies.fetcher import get_fetcher
-from app.fetcher import Fetcher
-from app.log import logger
-from app.models.mods import mods_can_get_pp
-from app.models.score import GameMode
-
-from httpx import HTTPError
-from redis.asyncio import Redis
-from sqlmodel import col, delete, select
-from sqlmodel.ext.asyncio.session import AsyncSession
-
-
-async def recalculate_all_players_pp():
-    async with AsyncSession(engine, autoflush=False) as session:
-        fetcher = await get_fetcher()
-        redis = get_redis()
-        for mode in GameMode:
-            await session.execute(
-                delete(PPBestScore).where(col(PPBestScore.gamemode) == mode)
-            )
-            logger.info(f"Recalculating PP for mode: {mode}")
-            statistics_list = (
-                await session.exec(
-                    select(UserStatistics).where(UserStatistics.mode == mode)
-                )
-            ).all()
-            await asyncio.gather(
-                *[
-                    _recalculate_pp(statistics, session, fetcher, redis)
-                    for statistics in statistics_list
-                ]
-            )
-            await session.commit()
-            logger.success(
-                f"Recalculated PP for mode: {mode}, total: {len(statistics_list)}"
-            )
-
-
-async def _recalculate_pp(
-    statistics: UserStatistics, session: AsyncSession, fetcher: Fetcher, redis: Redis
-):
-    scores = (
-        await session.exec(
-            select(Score).where(
-                Score.user_id == statistics.user_id,
-                Score.gamemode == statistics.mode,
-                col(Score.passed).is_(True),
-            )
-        )
-    ).all()
-    score_list: list[tuple[float, float]] = []
-    prev: dict[int, PPBestScore] = {}
-    for score in scores:
-        time = 10
-        beatmap_id = score.beatmap_id
-        while time > 0:
-            try:
-                db_beatmap = await Beatmap.get_or_fetch(
-                    session, fetcher, bid=beatmap_id
-                )
-            except HTTPError:
-                time -= 1
-                await asyncio.sleep(2)
-                continue
-            ranked = db_beatmap.beatmap_status.has_pp() | settings.enable_all_beatmap_pp
-            if not ranked or not mods_can_get_pp(int(score.gamemode), score.mods):
-                score.pp = 0
-                break
-            try:
-                beatmap_raw = await fetcher.get_or_fetch_beatmap_raw(redis, beatmap_id)
-                pp = await calculate_pp(score, beatmap_raw, session)
-                score.pp = pp
-                if score.beatmap_id not in prev or prev[score.beatmap_id].pp < pp:
-                    best_score = PPBestScore(
-                        user_id=statistics.user_id,
-                        beatmap_id=beatmap_id,
-                        acc=score.accuracy,
-                        score_id=score.id,
-                        pp=pp,
-                        gamemode=score.gamemode,
-                    )
-                    prev[score.beatmap_id] = best_score
-                score_list.append((score.pp, score.accuracy))
-                break
-            except HTTPError:
-                time -= 1
-                await asyncio.sleep(2)
-                continue
-        if time <= 0:
-            logger.error(f"Failed to fetch beatmap {beatmap_id} after 10 attempts")
-            score.pp = 0
-    # according to pp desc
-    score_list.sort(key=lambda x: x[0], reverse=True)
-    pp_sum = 0
-    acc_sum = 0
-    for i, s in enumerate(score_list):
-        pp_sum += calculate_weighted_pp(s[0], i)
-        acc_sum += calculate_weighted_acc(s[1], i)
-    if len(score_list):
-        # https://github.com/ppy/osu-queue-score-statistics/blob/c538ae/osu.Server.Queues.ScoreStatisticsProcessor/Helpers/UserTotalPerformanceAggregateHelper.cs#L41-L45
-        acc_sum *= 100 / (20 * (1 - math.pow(0.95, len(score_list))))
-    acc_sum = clamp(acc_sum, 0.0, 100.0)
-    statistics.pp = pp_sum
-    statistics.hit_accuracy = acc_sum
-    for best_score in prev.values():
-        session.add(best_score)

app/service/recalculate.py

@@ -0,0 +1,287 @@
+from __future__ import annotations
+
+import asyncio
+import math
+
+from app.calculator import (
+    calculate_pp,
+    calculate_score_to_level,
+    calculate_weighted_acc,
+    calculate_weighted_pp,
+    clamp,
+)
+from app.config import settings
+from app.database import BestScore, UserStatistics
+from app.database.beatmap import Beatmap
+from app.database.pp_best_score import PPBestScore
+from app.database.score import Score, get_user_best_pp
+from app.dependencies.database import engine, get_redis
+from app.dependencies.fetcher import get_fetcher
+from app.fetcher import Fetcher
+from app.log import logger
+from app.models.mods import mod_to_save, mods_can_get_pp
+from app.models.score import GameMode, Rank
+
+from httpx import HTTPError
+from redis.asyncio import Redis
+from sqlalchemy.orm import joinedload
+from sqlmodel import col, delete, select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+
+async def recalculate():
+    async with AsyncSession(engine, autoflush=False) as session:
+        fetcher = await get_fetcher()
+        redis = get_redis()
+        for mode in GameMode:
+            await session.execute(
+                delete(PPBestScore).where(col(PPBestScore.gamemode) == mode)
+            )
+            await session.execute(
+                delete(BestScore).where(col(BestScore.gamemode) == mode)
+            )
+            await session.commit()
+            logger.info(f"Recalculating for mode: {mode}")
+            statistics_list = (
+                await session.exec(
+                    select(UserStatistics).where(UserStatistics.mode == mode)
+                )
+            ).all()
+            await asyncio.gather(
+                *[
+                    _recalculate_pp(
+                        statistics.user_id, statistics.mode, session, fetcher, redis
+                    )
+                    for statistics in statistics_list
+                ]
+            )
+            await asyncio.gather(
+                *[
+                    _recalculate_best_score(
+                        statistics.user_id, statistics.mode, session
+                    )
+                    for statistics in statistics_list
+                ]
+            )
+            await session.commit()
+            await asyncio.gather(
+                *[
+                    _recalculate_statistics(statistics, session)
+                    for statistics in statistics_list
+                ]
+            )
+
+            await session.commit()
+            logger.success(
+                f"Recalculated for mode: {mode}, total: {len(statistics_list)}"
+            )
+
+
+async def _recalculate_pp(
+    user_id: int,
+    gamemode: GameMode,
+    session: AsyncSession,
+    fetcher: Fetcher,
+    redis: Redis,
+):
+    scores = (
+        await session.exec(
+            select(Score).where(
+                Score.user_id == user_id,
+                Score.gamemode == gamemode,
+                col(Score.passed).is_(True),
+            )
+        )
+    ).all()
+    prev: dict[int, PPBestScore] = {}
+    for score in scores:
+        time = 10
+        beatmap_id = score.beatmap_id
+        while time > 0:
+            try:
+                db_beatmap = await Beatmap.get_or_fetch(
+                    session, fetcher, bid=beatmap_id
+                )
+            except HTTPError:
+                time -= 1
+                await asyncio.sleep(2)
+                continue
+            ranked = db_beatmap.beatmap_status.has_pp() | settings.enable_all_beatmap_pp
+            if not ranked or not mods_can_get_pp(int(score.gamemode), score.mods):
+                score.pp = 0
+                break
+            try:
+                beatmap_raw = await fetcher.get_or_fetch_beatmap_raw(redis, beatmap_id)
+                pp = await calculate_pp(score, beatmap_raw, session)
+                score.pp = pp
+                if score.beatmap_id not in prev or prev[score.beatmap_id].pp < pp:
+                    best_score = PPBestScore(
+                        user_id=user_id,
+                        beatmap_id=beatmap_id,
+                        acc=score.accuracy,
+                        score_id=score.id,
+                        pp=pp,
+                        gamemode=score.gamemode,
+                    )
+                    prev[score.beatmap_id] = best_score
+                break
+            except HTTPError:
+                time -= 1
+                await asyncio.sleep(2)
+                continue
+            except Exception:
+                logger.exception(
+                    f"Error calculating pp for score {score.id} on beatmap {beatmap_id}"
+                )
+                break
+        if time <= 0:
+            logger.error(f"Failed to fetch beatmap {beatmap_id} after 10 attempts")
+            score.pp = 0
+
+    for best_score in prev.values():
+        session.add(best_score)
+
+
+async def _recalculate_best_score(
+    user_id: int,
+    gamemode: GameMode,
+    session: AsyncSession,
+):
+    beatmap_best_score: dict[int, list[BestScore]] = {}
+    scores = (
+        await session.exec(
+            select(Score).where(
+                Score.gamemode == gamemode,
+                col(Score.passed).is_(True),
+                Score.user_id == user_id,
+            )
+        )
+    ).all()
+    for score in scores:
+        if not (
+            (await score.awaitable_attrs.beatmap).beatmap_status.has_leaderboard()
+            | settings.enable_all_beatmap_leaderboard
+        ):
+            continue
+        mod_for_save = mod_to_save(score.mods)
+        bs = BestScore(
+            user_id=score.user_id,
+            score_id=score.id,
+            beatmap_id=score.beatmap_id,
+            gamemode=score.gamemode,
+            total_score=score.total_score,
+            mods=mod_for_save,
+            rank=score.rank,
+        )
+        if score.beatmap_id not in beatmap_best_score:
+            beatmap_best_score[score.beatmap_id] = [bs]
+        else:
+            b = next(
+                (
+                    s
+                    for s in beatmap_best_score[score.beatmap_id]
+                    if s.mods == mod_for_save and s.beatmap_id == score.beatmap_id
+                ),
+                None,
+            )
+            if b is None:
+                beatmap_best_score[score.beatmap_id].append(bs)
+            elif score.total_score > b.total_score:
+                beatmap_best_score[score.beatmap_id].remove(b)
+                beatmap_best_score[score.beatmap_id].append(bs)
+
+    for best_score_in_beatmap in beatmap_best_score.values():
+        for score in best_score_in_beatmap:
+            session.add(score)
+
+
+async def _recalculate_statistics(statistics: UserStatistics, session: AsyncSession):
+    await session.refresh(statistics)
+    pp_sum = 0
+    acc_sum = 0
+    bps = await get_user_best_pp(session, statistics.user_id, statistics.mode)
+    for i, s in enumerate(bps):
+        pp_sum += calculate_weighted_pp(s.pp, i)
+        acc_sum += calculate_weighted_acc(s.acc, i)
+    if len(bps):
+        # https://github.com/ppy/osu-queue-score-statistics/blob/c538ae/osu.Server.Queues.ScoreStatisticsProcessor/Helpers/UserTotalPerformanceAggregateHelper.cs#L41-L45
+        acc_sum *= 100 / (20 * (1 - math.pow(0.95, len(bps))))
+    acc_sum = clamp(acc_sum, 0.0, 100.0)
+    statistics.pp = pp_sum
+    statistics.hit_accuracy = acc_sum
+
+    statistics.play_count = 0
+    statistics.total_score = 0
+    statistics.maximum_combo = 0
+    statistics.play_time = 0
+    statistics.total_hits = 0
+    statistics.count_100 = 0
+    statistics.count_300 = 0
+    statistics.count_50 = 0
+    statistics.count_miss = 0
+    statistics.ranked_score = 0
+    statistics.grade_ss = 0
+    statistics.grade_ssh = 0
+    statistics.grade_s = 0
+    statistics.grade_sh = 0
+    statistics.grade_a = 0
+
+    scores = (
+        await session.exec(
+            select(Score)
+            .where(
+                Score.user_id == statistics.user_id,
+                Score.gamemode == statistics.mode,
+            )
+            .options(joinedload(Score.beatmap))
+        )
+    ).all()
+
+    cached_beatmap_best: dict[int, Score] = {}
+
+    for score in scores:
+        beatmap: Beatmap = score.beatmap
+        ranked = beatmap.beatmap_status.has_pp() | settings.enable_all_mods_pp
+
+        statistics.play_count += 1
+        statistics.total_score += score.total_score
+        statistics.play_time += beatmap.hit_length
+        statistics.count_300 += score.n300 + score.ngeki
+        statistics.count_100 += score.n100 + score.nkatu
+        statistics.count_50 += score.n50
+        statistics.count_miss += score.nmiss
+        statistics.total_hits += (
+            score.n300 + score.ngeki + score.n100 + score.nkatu + score.n50
+        )
+
+        if ranked and score.passed:
+            statistics.maximum_combo = max(statistics.maximum_combo, score.max_combo)
+            previous = cached_beatmap_best.get(score.beatmap_id)
+            difference = score.total_score - (previous.total_score if previous else 0)
+            if difference > 0:
+                cached_beatmap_best[score.beatmap_id] = score
+                statistics.ranked_score += difference
+                match score.rank:
+                    case Rank.X:
+                        statistics.grade_ss += 1
+                    case Rank.XH:
+                        statistics.grade_ssh += 1
+                    case Rank.S:
+                        statistics.grade_s += 1
+                    case Rank.SH:
+                        statistics.grade_sh += 1
+                    case Rank.A:
+                        statistics.grade_a += 1
+                if previous is not None:
+                    match previous.rank:
+                        case Rank.X:
+                            statistics.grade_ss -= 1
+                        case Rank.XH:
+                            statistics.grade_ssh -= 1
+                        case Rank.S:
+                            statistics.grade_s -= 1
+                        case Rank.SH:
+                            statistics.grade_sh -= 1
+                        case Rank.A:
+                            statistics.grade_a -= 1
+    statistics.level_current = calculate_score_to_level(statistics.total_score)

main.py

@@ -25,7 +25,7 @@ from app.service.calculate_all_user_rank import calculate_user_rank
 from app.service.create_banchobot import create_banchobot
 from app.service.daily_challenge import daily_challenge_job
 from app.service.osu_rx_statistics import create_rx_statistics
-from app.service.pp_recalculate import recalculate_all_players_pp
+from app.service.recalculate import recalculate
 
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.exceptions import RequestValidationError
@@ -38,8 +38,8 @@ import sentry_sdk
 async def lifespan(app: FastAPI):
     # on startup
     await get_fetcher()  # 初始化 fetcher
-    if os.environ.get("RECALCULATE_PP", "false").lower() == "true":
-        await recalculate_all_players_pp()
+    if os.environ.get("RECALCULATE", "false").lower() == "true":
+        await recalculate()
     await create_rx_statistics()
     await calculate_user_rank(True)
     init_scheduler()