chore(merge): merge branch 'main' into feat/multiplayer-api

app/models/mods.py

@@ -105,3 +105,65 @@ def mods_to_int(mods: list[APIMod]) -> int:
     for mod in mods:
         sum_ |= API_MOD_TO_LEGACY.get(mod["acronym"], 0)
     return sum_
+
+
+NO_CHECK = "DO_NO_CHECK"
+
+# FIXME: 这里为空表示了两种情况：mod 没有配置项；任何时候都可以获得 pp
+# 如果是后者，则 mod 更新的时候可能会误判。
+COMMON_CONFIG: dict[str, dict] = {
+    "EZ": {"retries": 2},
+    "NF": {},
+    "HT": {"speed_change": 0.75, "adjust_pitch": NO_CHECK},
+    "DC": {"speed_change": 0.75},
+    "HR": {},
+    "SD": {},
+    "PF": {},
+    "HD": {},
+    "DT": {"speed_change": 1.5, "adjust_pitch": NO_CHECK},
+    "NC": {"speed_change": 1.5},
+    "FL": {"size_multiplier": 1.0, "combo_based_size": True},
+    "AC": {},
+    "MU": {},
+    "TD": {},
+}
+
+RANKED_MODS: dict[int, dict[str, dict]] = {
+    0: COMMON_CONFIG,
+    1: COMMON_CONFIG,
+    2: COMMON_CONFIG,
+    3: COMMON_CONFIG,
+}
+# osu
+RANKED_MODS[0]["HD"]["only_fade_approach_circles"] = False
+RANKED_MODS[0]["FL"]["follow_delay"] = 1.0
+RANKED_MODS[0]["BL"] = {}
+RANKED_MODS[0]["NS"] = {}
+RANKED_MODS[0]["SO"] = {}
+RANKED_MODS[0]["TC"] = {}
+# taiko
+del RANKED_MODS[1]["EZ"]["retries"]
+# catch
+RANKED_MODS[2]["NS"] = {}
+# mania
+del RANKED_MODS[3]["HR"]
+RANKED_MODS[3]["FL"]["combo_based_size"] = False
+RANKED_MODS[3]["MR"] = {}
+for i in range(4, 10):
+    RANKED_MODS[3][f"{i}K"] = {}
+
+
+def mods_can_get_pp(ruleset_id: int, mods: list[APIMod]) -> bool:
+    ranked_mods = RANKED_MODS[ruleset_id]
+    for mod in mods:
+        mod["settings"] = mod.get("settings", {})
+        if (settings := ranked_mods.get(mod["acronym"])) is None:
+            return False
+        if settings == {}:
+            continue
+        for setting, value in mod["settings"].items():
+            if (expected_value := settings.get(setting)) is None:
+                return False
+            if expected_value != NO_CHECK and value != expected_value:
+                return False
+    return True

app/models/score.py

@@ -44,6 +44,16 @@ class Rank(str, Enum):
     D = "D"
     F = "F"
 
+    @property
+    def in_statisctics(self):
+        return self in {
+            Rank.X,
+            Rank.XH,
+            Rank.S,
+            Rank.SH,
+            Rank.A,
+        }
+
 
 # https://github.com/ppy/osu/blob/master/osu.Game/Rulesets/Scoring/HitResult.cs
 class HitResult(str, Enum):

app/models/signalr.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import datetime
 from typing import Any, get_origin
 
-import msgpack
 from pydantic import (
     BaseModel,
     ConfigDict,
@@ -24,11 +23,11 @@ def serialize_to_list(value: BaseModel) -> list[Any]:
         elif anno and issubclass(anno, list):
             data.append(
                 TypeAdapter(
-                    info.annotation,
+                    info.annotation, config=ConfigDict(arbitrary_types_allowed=True)
                 ).dump_python(v)
             )
         elif isinstance(v, datetime.datetime):
-            data.append([msgpack.ext.Timestamp.from_datetime(v), 0])
+            data.append([v, 0])
         else:
             data.append(v)
     return data

app/models/spectator_hub.py

@@ -11,15 +11,8 @@ from .score import (
 )
 from .signalr import MessagePackArrayModel, UserState
 
-import msgpack
-from pydantic import BaseModel, Field, field_validator
-
-
-class APIMod(MessagePackArrayModel):
-    acronym: str
-    settings: dict[str, Any] | list = Field(
-        default_factory=dict
-    )  # FIXME: with settings
+from msgpack_lazer_api import APIMod
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 class SpectatedUserState(IntEnum):
@@ -32,6 +25,8 @@ class SpectatedUserState(IntEnum):
 
 
 class SpectatorState(MessagePackArrayModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
     beatmap_id: int | None = None
     ruleset_id: int | None = None  # 0,1,2,3
     mods: list[APIMod] = Field(default_factory=list)
@@ -58,6 +53,8 @@ class ScoreProcessorStatistics(MessagePackArrayModel):
 
 
 class FrameHeader(MessagePackArrayModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
     total_score: int
     acc: float
     combo: int
@@ -70,10 +67,8 @@ class FrameHeader(MessagePackArrayModel):
     @field_validator("received_time", mode="before")
     @classmethod
     def validate_timestamp(cls, v: Any) -> datetime.datetime:
-        if isinstance(v, msgpack.ext.Timestamp):
-            return v.to_datetime()
         if isinstance(v, list):
-            return v[0].to_datetime()
+            return v[0]
         if isinstance(v, datetime.datetime):
             return v
         if isinstance(v, int | float):
@@ -111,6 +106,8 @@ class APIUser(BaseModel):
 
 
 class ScoreInfo(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
     mods: list[APIMod]
     user: APIUser
     ruleset: int

app/models/user.py

@@ -2,16 +2,15 @@ from __future__ import annotations
 
 from datetime import datetime
 from enum import Enum
-
-from app.database import (
-    LazerUserAchievement,
-    Team as Team,
-)
+from typing import TYPE_CHECKING
 
 from .score import GameMode
 
 from pydantic import BaseModel
 
+if TYPE_CHECKING:
+    from app.database import LazerUserAchievement, Team
+
 
 class PlayStyle(str, Enum):
     MOUSE = "mouse"
@@ -83,7 +82,11 @@ class UserAchievement(BaseModel):
     achievement_id: int
 
     # 添加数据库模型转换方法
-    def to_db_model(self, user_id: int) -> LazerUserAchievement:
+    def to_db_model(self, user_id: int) -> "LazerUserAchievement":
+        from app.database import (
+            LazerUserAchievement,
+        )
+
         return LazerUserAchievement(
             user_id=user_id,
             achievement_id=self.achievement_id,
@@ -207,7 +210,7 @@ class User(BaseModel):
     rank_history: RankHistory | None = None
     rankHistory: RankHistory | None = None  # 兼容性别名
     replays_watched_counts: list[dict] = []
-    team: Team | None = None
+    team: "Team | None" = None
     user_achievements: list[UserAchievement] = []