feat(beatmap,beatmapset): get beatmap & beatmapset information by id

2025-07-25 17:01:01 +08:00
parent f8abc7067f
commit 376d98ad5a
12 changed files with 452 additions and 9 deletions
--- a/app/auth.py
+++ b/app/auth.py
@@ -159,6 +159,13 @@ def store_token(
    for token in old_tokens:
        db.delete(token)

+    # 检查是否有重复的 access_token
+    duplicate_token = db.exec(
+        select(OAuthToken).where(OAuthToken.access_token == access_token)
+    ).first()
+    if duplicate_token:
+        db.delete(duplicate_token)
+
    # 创建新令牌记录
    token_record = OAuthToken(
        user_id=user_id,
--- a/app/database/init.py
+++ b/app/database/init.py
@@ -1,6 +1,14 @@
 from __future__ import annotations

 from .auth import OAuthToken
+from .beatmap import (
+    Beatmap as Beatmap,
+    BeatmapResp as BeatmapResp,
+)
+from .beatmapset import (
+    Beatmapset as Beatmapset,
+    BeatmapsetResp as BeatmapsetResp,
+)
 from .legacy import LegacyOAuthToken, LegacyUserStatistics
 from .team import Team, TeamMember
 from .user import (
@@ -23,7 +31,13 @@ from .user import (
    UserAvatar,
 )

+BeatmapsetResp.model_rebuild()
+BeatmapResp.model_rebuild()
 __all__ = [
+    "Beatmap",
+    "BeatmapResp",
+    "Beatmapset",
+    "BeatmapsetResp",
    "DailyChallengeStats",
    "LazerUserAchievement",
    "LazerUserBadge",
--- a/app/database/beatmap.py
+++ b/app/database/beatmap.py
@@ -0,0 +1,94 @@
+# ruff: noqa: I002
+from datetime import datetime
+
+from app.models.beatmap import BeatmapRankStatus
+from app.models.score import MODE_TO_INT, GameMode
+
+from .beatmapset import Beatmapset, BeatmapsetResp
+
+from sqlalchemy import DECIMAL, Column, DateTime
+from sqlmodel import VARCHAR, Field, Relationship, SQLModel
+
+
+class BeatmapOwner(SQLModel):
+    id: int
+    username: str
+
+
+class BeatmapBase(SQLModel):
+    # Beatmap
+    url: str
+    mode: GameMode
+    beatmapset_id: int = Field(foreign_key="beatmapsets.id", index=True)
+    difficulty_rating: float = Field(
+        default=0.0, sa_column=Column(DECIMAL(precision=10, scale=6))
+    )
+    beatmap_status: BeatmapRankStatus
+    total_length: int
+    user_id: int
+    version: str
+
+    # optional
+    checksum: str = Field(sa_column=Column(VARCHAR(32), index=True))
+    current_user_playcount: int = Field(default=0)
+    max_combo: int = Field(default=0)
+    # TODO: failtimes, owners
+
+    # BeatmapExtended
+    ar: float = Field(default=0.0, sa_column=Column(DECIMAL(precision=10, scale=2)))
+    cs: float = Field(default=0.0, sa_column=Column(DECIMAL(precision=10, scale=2)))
+    drain: float = Field(
+        default=0.0,
+        sa_column=Column(DECIMAL(precision=10, scale=2)),
+    )  # hp
+    accuracy: float = Field(
+        default=0.0,
+        sa_column=Column(DECIMAL(precision=10, scale=2)),
+    )  # od
+    bpm: float = Field(default=0.0, sa_column=Column(DECIMAL(precision=10, scale=2)))
+    count_circles: int = Field(default=0)
+    count_sliders: int = Field(default=0)
+    count_spinners: int = Field(default=0)
+    deleted_at: datetime | None = Field(default=None, sa_column=Column(DateTime))
+    hit_length: int = Field(default=0)
+    last_updated: datetime = Field(sa_column=Column(DateTime))
+    passcount: int = Field(default=0)
+    playcount: int = Field(default=0)
+
+
+class Beatmap(BeatmapBase, table=True):
+    __tablename__ = "beatmaps"  # pyright: ignore[reportAssignmentType]
+    id: int | None = Field(default=None, primary_key=True, index=True)
+    beatmapset_id: int = Field(foreign_key="beatmapsets.id", index=True)
+    # optional
+    beatmapset: Beatmapset = Relationship(back_populates="beatmaps")
+
+
+class BeatmapResp(BeatmapBase):
+    id: int
+    beatmapset_id: int
+    beatmapset: BeatmapsetResp | None = None
+    convert: bool = False
+    is_scoreable: bool
+    status: str
+    mode_int: int
+    ranked: int
+    url: str = ""
+
+    @classmethod
+    def from_db(
+        cls,
+        beatmap: Beatmap,
+        query_mode: GameMode | None = None,
+        from_set: bool = False,
+    ) -> "BeatmapResp":
+        beatmap_ = beatmap.model_dump()
+        if query_mode is not None and beatmap.mode != query_mode:
+            beatmap_["convert"] = True
+        beatmap_["is_scoreable"] = beatmap.beatmap_status > BeatmapRankStatus.PENDING
+        beatmap_["status"] = beatmap.beatmap_status.name.lower()
+        beatmap_["ranked"] = beatmap.beatmap_status.value
+        beatmap_["mode_int"] = MODE_TO_INT[beatmap.mode]
+        if not from_set:
+            beatmap_["beatmapset"] = BeatmapsetResp.from_db(beatmap.beatmapset)
+        return cls.model_validate(beatmap_)
--- a/app/database/beatmapset.py
+++ b/app/database/beatmapset.py
@@ -0,0 +1,205 @@
+# ruff: noqa: I002
+
+from datetime import datetime
+from typing import TYPE_CHECKING, cast
+
+from app.models.beatmap import BeatmapRankStatus, Genre, Language
+from app.models.score import GameMode
+
+from pydantic import BaseModel, model_serializer
+from sqlalchemy import DECIMAL, JSON, Column, DateTime, Text
+from sqlmodel import Field, Relationship, SQLModel
+
+if TYPE_CHECKING:
+    from .beatmap import Beatmap, BeatmapResp
+
+
+class BeatmapCovers(SQLModel):
+    cover: str
+    card: str
+    list: str
+    slimcover: str
+    cover_2_x: str | None = Field(default=None, alias="cover@2x")
+    card_2_x: str | None = Field(default=None, alias="card@2x")
+    list_2_x: str | None = Field(default=None, alias="list@2x")
+    slimcover_2_x: str | None = Field(default=None, alias="slimcover@2x")
+
+    @model_serializer
+    def _(self) -> dict[str, str | None]:
+        self = cast(dict[str, str | None] | BeatmapCovers, self)
+        if isinstance(self, dict):
+            return {
+                "cover": self["cover"],
+                "card": self["card"],
+                "list": self["list"],
+                "slimcover": self["slimcover"],
+                "cover@2x": self.get("cover@2x"),
+                "card@2x": self.get("card@2x"),
+                "list@2x": self.get("list@2x"),
+                "slimcover@2x": self.get("slimcover@2x"),
+            }
+        else:
+            return {
+                "cover": self.cover,
+                "card": self.card,
+                "list": self.list,
+                "slimcover": self.slimcover,
+                "cover@2x": self.cover_2_x,
+                "card@2x": self.card_2_x,
+                "list@2x": self.list_2_x,
+                "slimcover@2x": self.slimcover_2_x,
+            }
+
+
+class BeatmapHype(BaseModel):
+    current: int
+    required: int
+
+
+class BeatmapAvailability(BaseModel):
+    more_information: str | None = Field(default=None)
+    download_disabled: bool | None = Field(default=None)
+
+
+class BeatmapNominations(SQLModel):
+    current: int | None = Field(default=None)
+    required: int | None = Field(default=None)
+
+
+class BeatmapNomination(SQLModel):
+    beatmapset_id: int
+    reset: bool
+    user_id: int
+    rulesets: list[GameMode] | None = None
+
+
+class BeatmapDescription(SQLModel):
+    bbcode: str | None = None
+    description: str | None = None
+
+
+class BeatmapTranslationText(BaseModel):
+    name: str
+    id: int | None = None
+
+
+class BeatmapsetBase(SQLModel):
+    # Beatmapset
+    artist: str = Field(index=True)
+    artist_unicode: str = Field(index=True)
+    covers: BeatmapCovers = Field(sa_column=Column(JSON))
+    creator: str
+    favourite_count: int
+    nsfw: bool = Field(default=False)
+    play_count: int
+    preview_url: str
+    source: str = Field(default="")
+
+    spotlight: bool = Field(default=False)
+    title: str
+    title_unicode: str
+    user_id: int
+    video: bool
+
+    # optional
+    # converts: list[Beatmap] = Relationship(back_populates="beatmapset")
+    current_nominations: list[BeatmapNomination] | None = Field(
+        None, sa_column=Column(JSON)
+    )
+    description: BeatmapDescription | None = Field(default=None, sa_column=Column(JSON))
+    # TODO: discussions: list[BeatmapsetDiscussion] = None
+    # TODO: current_user_attributes: Optional[CurrentUserAttributes] = None
+    # TODO: events: Optional[list[BeatmapsetEvent]] = None
+
+    pack_tags: list[str] = Field(default=[], sa_column=Column(JSON))
+    ratings: list[int] = Field(default=None, sa_column=Column(JSON))
+    # TODO: recent_favourites: Optional[list[User]] = None
+    # TODO: related_users: Optional[list[User]] = None
+    # TODO: user: Optional[User] = Field(default=None)
+    track_id: int | None = Field(default=None)  # feature artist?
+    # TODO: has_favourited
+
+    # BeatmapsetExtended
+    bpm: float = Field(default=0.0, sa_column=Column(DECIMAL(10, 2)))
+    can_be_hyped: bool = Field(default=False)
+    discussion_locked: bool = Field(default=False)
+    last_updated: datetime = Field(sa_column=Column(DateTime))
+    ranked_date: datetime | None = Field(default=None, sa_column=Column(DateTime))
+    storyboard: bool = Field(default=False)
+    submitted_date: datetime = Field(sa_column=Column(DateTime))
+    tags: str = Field(default="", sa_column=Column(Text))
+
+
+class Beatmapset(BeatmapsetBase, table=True):
+    __tablename__ = "beatmapsets"  # pyright: ignore[reportAssignmentType]
+
+    id: int | None = Field(default=None, primary_key=True, index=True)
+    # Beatmapset
+    beatmap_status: BeatmapRankStatus = Field(
+        default=BeatmapRankStatus.GRAVEYARD,
+    )
+
+    # optional
+    beatmaps: list["Beatmap"] = Relationship(back_populates="beatmapset")
+    beatmap_genre: Genre = Field(default=Genre.UNSPECIFIED)
+    beatmap_language: Language = Field(default=Language.UNSPECIFIED)
+    nominations_required: int = Field(default=0)
+    nominations_current: int = Field(default=0)
+
+    # BeatmapExtended
+    hype_current: int = Field(default=0)
+    hype_required: int = Field(default=0)
+    availability_info: str | None = Field(default=None)
+    download_disabled: bool = Field(default=False)
+
+
+class BeatmapsetResp(BeatmapsetBase):
+    id: int
+    beatmaps: list["BeatmapResp"]
+    discussion_enabled: bool = True
+    status: str
+    ranked: int
+    legacy_thread_url: str = ""
+    is_scoreable: bool
+    hype: BeatmapHype
+    availability: BeatmapAvailability
+    genre: BeatmapTranslationText
+    language: BeatmapTranslationText
+    nominations: BeatmapNominations
+
+    @classmethod
+    def from_db(cls, beatmapset: Beatmapset) -> "BeatmapsetResp":
+        from .beatmap import BeatmapResp
+
+        beatmaps = [
+            BeatmapResp.from_db(beatmap, from_set=True)
+            for beatmap in beatmapset.beatmaps
+        ]
+        return cls.model_validate(
+            {
+                "beatmaps": beatmaps,
+                "hype": BeatmapHype(
+                    current=beatmapset.hype_current, required=beatmapset.hype_required
+                ),
+                "availability": BeatmapAvailability(
+                    more_information=beatmapset.availability_info,
+                    download_disabled=beatmapset.download_disabled,
+                ),
+                "genre": BeatmapTranslationText(
+                    name=beatmapset.beatmap_genre.name,
+                    id=beatmapset.beatmap_genre.value,
+                ),
+                "language": BeatmapTranslationText(
+                    name=beatmapset.beatmap_language.name,
+                    id=beatmapset.beatmap_language.value,
+                ),
+                "nominations": BeatmapNominations(
+                    required=beatmapset.nominations_required,
+                    current=beatmapset.nominations_current,
+                ),
+                "status": beatmapset.beatmap_status.name.lower(),
+                "ranked": beatmapset.beatmap_status.value,
+                "is_scoreable": beatmapset.beatmap_status > BeatmapRankStatus.PENDING,
+                **beatmapset.model_dump(),
+            }
+        )
--- a/app/database/user.py
+++ b/app/database/user.py
@@ -8,7 +8,7 @@ from .team import TeamMember

 from sqlalchemy import DECIMAL, JSON, Column, Date, DateTime, Text
 from sqlalchemy.dialects.mysql import VARCHAR
-from sqlmodel import Field, Relationship, SQLModel
+from sqlmodel import BigInteger, Field, Relationship, SQLModel


 class User(SQLModel, table=True):
@@ -263,10 +263,10 @@ class LazerUserStatistics(SQLModel, table=True):
    # PP 和分数
    pp: float = Field(default=0.00, sa_column=Column(DECIMAL(10, 2)))
    pp_exp: float = Field(default=0.00, sa_column=Column(DECIMAL(10, 2)))
-    ranked_score: int = Field(default=0)
+    ranked_score: int = Field(default=0, sa_column=Column(BigInteger))
    hit_accuracy: float = Field(default=0.00, sa_column=Column(DECIMAL(5, 2)))
-    total_score: int = Field(default=0)
-    total_hits: int = Field(default=0)
+    total_score: int = Field(default=0, sa_column=Column(BigInteger))
+    total_hits: int = Field(default=0, sa_column=Column(BigInteger))
    maximum_combo: int = Field(default=0)

    # 游戏统计
@@ -406,6 +406,7 @@ class UserAchievement:
    achieved_at: datetime
    achievement_id: int

+
 class DailyChallengeStats(SQLModel, table=True):
    __tablename__ = "daily_challenge_stats"  # pyright: ignore[reportAssignmentType]

@@ -441,7 +442,6 @@ class RankHistory(SQLModel, table=True):
    user: "User" = Relationship(back_populates="rank_history")


-
 class UserAvatar(SQLModel, table=True):
    __tablename__ = "user_avatars"  # pyright: ignore[reportAssignmentType]

--- a/app/models/beatmap.py
+++ b/app/models/beatmap.py
@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from enum import IntEnum
+
+
+class BeatmapRankStatus(IntEnum):
+    GRAVEYARD = -2
+    WIP = -1
+    PENDING = 0
+    RANKED = 1
+    APPROVED = 2
+    QUALIFIED = 3
+    LOVED = 4
+
+
+class Genre(IntEnum):
+    ANY = 0
+    UNSPECIFIED = 1
+    VIDEO_GAME = 2
+    ANIME = 3
+    ROCK = 4
+    POP = 5
+    OTHER = 6
+    NOVELTY = 7
+    HIP_HOP = 9
+    ELECTRONIC = 10
+    METAL = 11
+    CLASSICAL = 12
+    FOLK = 13
+    JAZZ = 14
+
+
+class Language(IntEnum):
+    ANY = 0
+    UNSPECIFIED = 1
+    ENGLISH = 2
+    JAPANESE = 3
+    CHINESE = 4
+    INSTRUMENTAL = 5
+    KOREAN = 6
+    FRENCH = 7
+    GERMAN = 8
+    ITALIAN = 9
+    SPANISH = 10
+    RUSSIAN = 11
+    POLISH = 12
+    OTHER = 13
--- a/app/models/score.py
+++ b/app/models/score.py
@@ -1,17 +1,31 @@
+from __future__ import annotations
+
 from enum import Enum, IntEnum
 from typing import Any
+
 from pydantic import BaseModel

+
 class GameMode(str, Enum):
    OSU = "osu"
    TAIKO = "taiko"
    FRUITS = "fruits"
    MANIA = "mania"

+
+MODE_TO_INT = {
+    GameMode.OSU: 0,
+    GameMode.TAIKO: 1,
+    GameMode.FRUITS: 2,
+    GameMode.MANIA: 3,
+}
+
+
 class APIMod(BaseModel):
    acronym: str
    settings: dict[str, Any] = {}

+
 # https://github.com/ppy/osu/blob/master/osu.Game/Rulesets/Scoring/HitResult.cs
 class HitResult(IntEnum):
    PERFECT = 0  # [Order(0)]
--- a/app/router/init.py
+++ b/app/router/init.py
@@ -1,6 +1,12 @@
 from __future__ import annotations

-from . import me  # pyright: ignore[reportUnusedImport]  # noqa: F401
+from . import (  # pyright: ignore[reportUnusedImport]  # noqa: F401
+    beatmap,
+    beatmapset,
+    me,
+)
 from .api_router import router as api_router
 from .auth import router as auth_router
 from .signalr import signalr_router as signalr_router
+
+__all__ = ["api_router", "auth_router", "signalr_router"]
--- a/app/router/beatmap.py
+++ b/app/router/beatmap.py
@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from app.database import (
+    Beatmap,
+    BeatmapResp,
+    User as DBUser,
+)
+from app.dependencies.database import get_db
+from app.dependencies.user import get_current_user
+
+from .api_router import router
+
+from fastapi import Depends, HTTPException
+from sqlmodel import Session, select
+
+
+@router.get("/beatmaps/{bid}", tags=["beatmap"], response_model=BeatmapResp)
+async def get_beatmap(
+    bid: int,
+    current_user: DBUser = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    beatmap = db.exec(select(Beatmap).where(Beatmap.id == bid)).first()
+    if not beatmap:
+        raise HTTPException(status_code=404, detail="Beatmap not found")
+    return BeatmapResp.from_db(beatmap)
--- a/app/router/beatmapset.py
+++ b/app/router/beatmapset.py
@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from app.database import (
+    Beatmapset,
+    BeatmapsetResp,
+    User as DBUser,
+)
+from app.dependencies.database import get_db
+from app.dependencies.user import get_current_user
+
+from .api_router import router
+
+from fastapi import Depends, HTTPException
+from sqlmodel import Session, select
+
+
+@router.get("/beatmapsets/{sid}", tags=["beatmapset"], response_model=BeatmapsetResp)
+async def get_beatmapset(
+    sid: int,
+    current_user: DBUser = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    beatmapset = db.exec(select(Beatmapset).where(Beatmapset.id == sid)).first()
+    if not beatmapset:
+        raise HTTPException(status_code=404, detail="Beatmapset not found")
+    return BeatmapsetResp.from_db(beatmapset)
--- a/main.py
+++ b/main.py
@@ -3,9 +3,11 @@ from __future__ import annotations
 from datetime import datetime

 from app.config import settings
+from app.dependencies.database import engine
 from app.router import api_router, auth_router, signalr_router

 from fastapi import FastAPI
+from sqlmodel import SQLModel

 # 注意: 表结构现在通过 migrations 管理，不再自动创建
 # 如需创建表，请运行: python quick_sync.py
@@ -15,6 +17,8 @@ app.include_router(api_router, prefix="/api/v2")
 app.include_router(signalr_router, prefix="/signalr")
 app.include_router(auth_router)

+SQLModel.metadata.create_all(bind=engine)
+

@app.get("/")
 async def root():
--- a/test_api.py
+++ b/test_api.py
@@ -132,8 +132,8 @@ def main():
        print("身份验证失败，请检查用户名和密码")
        return

-    print(f"访问令牌: {token_data['access_token'][:50]}...")
-    print(f"刷新令牌: {token_data['refresh_token'][:30]}...")
+    print(f"访问令牌: {token_data['access_token']}")
+    print(f"刷新令牌: {token_data['refresh_token']}")
    print(f"令牌有效期: {token_data['expires_in']} 秒")

    # 4. 获取用户数据
@@ -153,7 +153,7 @@ def main():
    print("\n5. 测试令牌刷新...")
    new_token_data = refresh_token(token_data["refresh_token"])
    if new_token_data:
-        print(f"新访问令牌: {new_token_data['access_token'][:50]}...")
+        print(f"新访问令牌: {new_token_data['access_token']}")

        # 使用新令牌获取用户数据
        print("\n6. 使用新令牌获取用户数据...")