maimai_py.providers

20class IAliasProvider:
21    """The provider that fetches song aliases from a specific source.
22
23    Available providers: `YuzuProvider`, `LXNSProvider`
24    """
25
26    @abstractmethod
27    async def get_aliases(self, client: AsyncClient) -> list[SongAlias]:
28        """@private"""
29        raise NotImplementedError()

32class IPlayerProvider:
33    """The provider that fetches players from a specific source.
34
35    Available providers: `DivingFishProvider`, `LXNSProvider`
36    """
37
38    @abstractmethod
39    async def get_player(self, identifier: PlayerIdentifier, client: AsyncClient) -> Player:
40        """@private"""
41        raise NotImplementedError()

 8class ISongProvider:
 9    """The provider that fetches songs from a specific source.
10
11    Available providers: `DivingFishProvider`, `LXNSProvider`
12    """
13
14    @abstractmethod
15    async def get_songs(self, client: AsyncClient) -> list[Song]:
16        """@private"""
17        raise NotImplementedError()

44class IScoreProvider:
45    """The provider that fetches scores from a specific source.
46
47    Available providers: `DivingFishProvider`, `LXNSProvider`, `WechatProvider`
48    """
49
50    @abstractmethod
51    async def get_scores_best(self, identifier: PlayerIdentifier, client: AsyncClient) -> tuple[list[Score] | None, list[Score] | None]:
52        """@private"""
53        # Return (None, None) will call the main client to handle this, which will then fetch all scores instead
54        return None, None
55
56    @abstractmethod
57    async def get_scores_all(self, identifier: PlayerIdentifier, client: AsyncClient) -> list[Score]:
58        """@private"""
59        raise NotImplementedError()
60
61    @abstractmethod
62    async def update_scores(self, identifier: PlayerIdentifier, scores: list[Score], client: AsyncClient) -> None:
63        """@private"""
64        raise NotImplementedError()

67class ICurveProvider:
68    """The provider that fetches statistics curves from a specific source.
69
70    Available providers: `DivingFishProvider`
71    """
72
73    @abstractmethod
74    async def get_curves(self, client: AsyncClient) -> dict[str, list[CurveObject | None]]:
75        """@private"""
76        raise NotImplementedError()

 91class IItemListProvider:
 92    """The provider that fetches player item list data from a specific source.
 93
 94    Available providers: `LXNSProvider`, `LocalProvider`
 95    """
 96
 97    @abstractmethod
 98    async def get_icons(self, client: AsyncClient) -> dict[int, PlayerIcon]:
 99        """@private"""
100        raise NotImplementedError()
101
102    @abstractmethod
103    async def get_nameplates(self, client: AsyncClient) -> dict[int, PlayerNamePlate]:
104        """@private"""
105        raise NotImplementedError()
106
107    @abstractmethod
108    async def get_frames(self, client: AsyncClient) -> dict[int, PlayerFrame]:
109        """@private"""
110        raise NotImplementedError()
111
112    @abstractmethod
113    async def get_partners(self, client: AsyncClient) -> dict[int, PlayerPartner]:
114        """@private"""
115        raise NotImplementedError()
116
117    @abstractmethod
118    async def get_charas(self, client: AsyncClient) -> dict[int, PlayerChara]:
119        """@private"""
120        raise NotImplementedError()
121
122    @abstractmethod
123    async def get_trophies(self, client: AsyncClient) -> dict[int, PlayerTrophy]:
124        """@private"""
125        raise NotImplementedError()

128class IAreaProvider:
129    """The provider that fetches area data from a specific source.
130
131    Available providers: `LocalProvider`
132    """
133
134    @abstractmethod
135    async def get_areas(self, lang: str, client: AsyncClient) -> dict[str, Area]:
136        """@private"""
137        raise NotImplementedError()

79class IRegionProvider:
80    """The provider that fetches player regions from a specific source.
81
82    Available providers: `ArcadeProvider`
83    """
84
85    @abstractmethod
86    async def get_regions(self, identifier: PlayerIdentifier, client: AsyncClient) -> list[PlayerRegion]:
87        """@private"""
88        raise NotImplementedError()

10class LocalProvider(IItemListProvider, IAreaProvider):
11    """The provider that fetches data from the local storage.
12
13    Most of the data are stored in JSON files in the same directory as this file.
14    """
15
16    def __eq__(self, value):
17        return isinstance(value, LocalProvider)
18
19    def _read_file(self, file_name: str) -> Any:
20        current_folder = Path(__file__).resolve().parent
21        path = current_folder / f"{file_name}.json"
22        if not path.exists():
23            raise FileNotFoundError(f"File {path} not found.")
24        with open(path, "r", encoding="utf-8") as f:
25            return json.load(f)
26
27    def _read_file_dict(self, file_name: str) -> dict:
28        obj = self._read_file(file_name)
29        if isinstance(obj, dict):
30            return obj["data"]
31        else:
32            raise ValueError(f"File {file_name} is not a dictionary.")
33
34    def _read_file_list(self, file_name: str) -> list:
35        obj = self._read_file(file_name)
36        if isinstance(obj, list):
37            return obj
38        else:
39            raise ValueError(f"File {file_name} is not a list.")
40
41    async def get_icons(self, client: AsyncClient) -> dict[int, PlayerIcon]:
42        return {int(k): PlayerIcon(id=int(k), name=v) for k, v in self._read_file_dict("icons").items()}
43
44    async def get_nameplates(self, client: AsyncClient) -> dict[int, PlayerNamePlate]:
45        return {int(k): PlayerNamePlate(id=int(k), name=v) for k, v in self._read_file_dict("nameplates").items()}
46
47    async def get_frames(self, client: AsyncClient) -> dict[int, PlayerFrame]:
48        return {int(k): PlayerFrame(id=int(k), name=v) for k, v in self._read_file_dict("frames").items()}
49
50    async def get_partners(self, client: AsyncClient) -> dict[int, PlayerPartner]:
51        return {int(k): PlayerPartner(id=int(k), name=v) for k, v in self._read_file_dict("partners").items()}
52
53    async def get_charas(self, client: AsyncClient) -> dict[int, PlayerChara]:
54        return {int(k): PlayerChara(id=int(k), name=v) for k, v in self._read_file_dict("charas").items()}
55
56    async def get_trophies(self, client: AsyncClient) -> dict[int, PlayerTrophy]:
57        return {int(k): PlayerTrophy(id=int(k), name=v["title"], color=v["rareType"]) for k, v in self._read_file_dict("trophies").items()}
58
59    async def get_areas(self, lang: str, client: AsyncClient) -> dict[str, Area]:
60        songs = await MaimaiSongs._get_or_fetch(client)
61        return {
62            item["id"]: Area(
63                id=item["id"],
64                name=item["name"],
65                comment=item["comment"],
66                description=item["description"],
67                video_id=item["video_id"],
68                characters=[
69                    AreaCharacter(
70                        name=char["name"],
71                        illustrator=char["illustrator"],
72                        description1=char["description1"],
73                        description2=char["description2"],
74                        team=char["team"],
75                        props=char["props"],
76                    )
77                    for char in item["characters"]
78                ],
79                songs=[
80                    AreaSong(
81                        id=s.id if (s := songs.by_title(song["title"])) else -1,
82                        title=song["title"],
83                        artist=song["artist"],
84                        description=song["description"],
85                        illustrator=song["illustrator"],
86                        movie=song["movie"],
87                    )
88                    for song in item["songs"]
89                ],
90            )
91            for item in self._read_file_list(f"areas_{lang}")
92        }

 12class DivingFishProvider(ISongProvider, IPlayerProvider, IScoreProvider, ICurveProvider):
 13    """The provider that fetches data from the Diving Fish.
 14
 15    DivingFish: https://www.diving-fish.com/maimaidx/prober/
 16    """
 17
 18    developer_token: str | None
 19    """The developer token used to access the Diving Fish API."""
 20    base_url = "https://www.diving-fish.com/api/maimaidxprober/"
 21    """The base URL for the Diving Fish API."""
 22
 23    @property
 24    def headers(self):
 25        """@private"""
 26        if not self.developer_token:
 27            raise InvalidDeveloperTokenError()
 28        return {"developer-token": self.developer_token}
 29
 30    def __init__(self, developer_token: str | None = None):
 31        """Initializes the DivingFishProvider.
 32
 33        Args:
 34            developer_token: The developer token used to access the Diving Fish API.
 35        """
 36        self.developer_token = developer_token
 37
 38    def __eq__(self, value):
 39        return isinstance(value, DivingFishProvider) and value.developer_token == self.developer_token
 40
 41    @staticmethod
 42    def _deser_song(song: dict) -> Song:
 43        return Song(
 44            id=int(song["id"]) % 10000,
 45            title=song["basic_info"]["title"] if int(song["id"]) != 383 else "Link",
 46            artist=song["basic_info"]["artist"],
 47            genre=name_to_genre[song["basic_info"]["genre"]],
 48            bpm=song["basic_info"]["bpm"],
 49            map=None,
 50            rights=None,
 51            aliases=None,
 52            version=divingfish_to_version[song["basic_info"]["from"]].value,
 53            disabled=False,
 54            difficulties=SongDifficulties(standard=[], dx=[], utage=[]),
 55        )
 56
 57    @staticmethod
 58    def _deser_diffs(song: dict) -> Generator[SongDifficulty, None, None]:
 59        song_type = SongType._from_id(song["id"])
 60        for idx, chart in enumerate(song["charts"]):
 61            song_diff = SongDifficulty(
 62                type=song_type,
 63                level=song["level"][idx],
 64                level_value=song["ds"][idx],
 65                level_index=LevelIndex(idx),
 66                note_designer=chart["charter"],
 67                version=divingfish_to_version[song["basic_info"]["from"]].value,
 68                tap_num=chart["notes"][0],
 69                hold_num=chart["notes"][1],
 70                slide_num=chart["notes"][2],
 71                touch_num=chart["notes"][3] if song_type == SongType.DX else 0,
 72                break_num=chart["notes"][4] if song_type == SongType.DX else chart["notes"][3],
 73                curve=None,
 74            )
 75            if song_type == SongType.UTAGE:
 76                song_diff = SongDifficultyUtage(
 77                    **dataclasses.asdict(song_diff),
 78                    kanji=song["basic_info"]["title"][1:2],
 79                    description="LET'S PARTY!",
 80                    is_buddy=False,
 81                )
 82            yield song_diff
 83
 84    @staticmethod
 85    def _deser_score(score: dict) -> Score:
 86        return Score(
 87            id=score["song_id"] % 10000,
 88            song_name=score["title"] if score["song_id"] != 383 else "Link(CoF)",
 89            level=score["level"],
 90            level_index=LevelIndex(score["level_index"]),
 91            achievements=score["achievements"],
 92            fc=FCType[score["fc"].upper()] if score["fc"] else None,
 93            fs=FSType[score["fs"].upper()] if score["fs"] else None,
 94            dx_score=score["dxScore"],
 95            dx_rating=score["ra"],
 96            rate=RateType[score["rate"].upper()],
 97            type=SongType._from_id(score["song_id"]),
 98        )
 99
100    @staticmethod
101    def _ser_score(score: Score) -> dict:
102        return {
103            "song_id": score.type._to_id(score.id),
104            "title": score.song_name if score.id != 383 else "Link(CoF)",
105            "level": score.level,
106            "level_index": score.level_index.value,
107            "achievements": score.achievements,
108            "fc": score.fc.name.lower() if score.fc else None,
109            "fs": score.fs.name.lower() if score.fs else None,
110            "dxScore": score.dx_score,
111            "ra": score.dx_rating,
112            "rate": score.rate.name.lower(),
113            "type": score.type._to_abbr(),
114        }
115
116    @staticmethod
117    def _deser_curve(chart: dict) -> CurveObject:
118        return CurveObject(
119            sample_size=int(chart["cnt"]),
120            fit_level_value=chart["fit_diff"],
121            avg_achievements=chart["avg"],
122            stdev_achievements=chart["std_dev"],
123            avg_dx_score=chart["avg_dx"],
124            rate_sample_size={v: chart["dist"][13 - i] for i, v in enumerate(RateType)},
125            fc_sample_size={v: chart["dist"][4 - i] for i, v in enumerate(FCType)},
126        )
127
128    def _check_response_player(self, resp: Response) -> dict:
129        resp.raise_for_status()
130        resp_json = resp.json()
131        if "msg" in resp_json and resp_json["msg"] in ["请先联系水鱼申请开发者token", "开发者token有误", "开发者token被禁用"]:
132            raise InvalidDeveloperTokenError(resp_json["msg"])
133        elif "message" in resp_json and resp_json["message"] in ["导入token有误", "尚未登录", "会话过期"]:
134            raise InvalidPlayerIdentifierError(resp_json["message"])
135        elif resp.status_code in [400, 401]:
136            raise InvalidPlayerIdentifierError(resp_json["message"])
137        elif resp.status_code == 403:
138            raise PrivacyLimitationError(resp_json["message"])
139        return resp_json
140
141    async def get_songs(self, client: AsyncClient) -> list[Song]:
142        resp = await client.get(self.base_url + "music_data")
143        resp.raise_for_status()
144        resp_json = resp.json()
145        unique_songs: dict[int, Song] = {}
146        for song in resp_json:
147            unique_key = int(song["id"]) % 10000
148            song_type: SongType = SongType._from_id(song["id"])
149            if unique_key not in unique_songs:
150                unique_songs[unique_key] = DivingFishProvider._deser_song(song)
151            difficulties: list[SongDifficulty] = unique_songs[unique_key].difficulties.__getattribute__(song_type.value)
152            difficulties.extend(DivingFishProvider._deser_diffs(song))
153        return list(unique_songs.values())
154
155    async def get_player(self, identifier: PlayerIdentifier, client: AsyncClient) -> DivingFishPlayer:
156        resp = await client.post(self.base_url + "query/player", json=identifier._as_diving_fish())
157        resp_json = self._check_response_player(resp)
158        return DivingFishPlayer(
159            name=resp_json["username"],
160            rating=resp_json["rating"],
161            nickname=resp_json["nickname"],
162            plate=resp_json["plate"],
163            additional_rating=resp_json["additional_rating"],
164        )
165
166    async def get_scores_best(self, identifier: PlayerIdentifier, client: AsyncClient) -> tuple[list[Score], list[Score]]:
167        req_json = identifier._as_diving_fish()
168        req_json["b50"] = True
169        resp = await client.post(self.base_url + "query/player", json=req_json)
170        resp_json = self._check_response_player(resp)
171        return (
172            [DivingFishProvider._deser_score(score) for score in resp_json["charts"]["sd"]],
173            [DivingFishProvider._deser_score(score) for score in resp_json["charts"]["dx"]],
174        )
175
176    async def get_scores_all(self, identifier: PlayerIdentifier, client: AsyncClient) -> list[Score]:
177        resp = await client.get(self.base_url + "dev/player/records", params=identifier._as_diving_fish(), headers=self.headers)
178        resp_json = self._check_response_player(resp)
179        return [s for score in resp_json["records"] if (s := DivingFishProvider._deser_score(score))]
180
181    async def update_scores(self, identifier: PlayerIdentifier, scores: list[Score], client: AsyncClient) -> None:
182        headers, cookies = None, None
183        if identifier.username and identifier.credentials:
184            login_json = {"username": identifier.username, "password": identifier.credentials}
185            resp1 = await client.post("https://www.diving-fish.com/api/maimaidxprober/login", json=login_json)
186            self._check_response_player(resp1)
187            cookies = resp1.cookies
188        elif not identifier.username and identifier.credentials and isinstance(identifier.credentials, str):
189            headers = {"Import-Token": identifier.credentials}
190        else:
191            raise InvalidPlayerIdentifierError("Either username and password or import token is required to deliver scores")
192        scores_json = [DivingFishProvider._ser_score(score) for score in scores]
193        resp2 = await client.post(self.base_url + "player/update_records", cookies=cookies, headers=headers, json=scores_json)
194        self._check_response_player(resp2)
195
196    async def get_curves(self, client: AsyncClient) -> dict[str, list[CurveObject | None]]:
197        resp = await client.get(self.base_url + "chart_stats")
198        resp.raise_for_status()
199        return {idx: ([DivingFishProvider._deser_curve(chart) for chart in charts if chart != {}]) for idx, charts in (resp.json())["charts"].items()}

 11class LXNSProvider(ISongProvider, IPlayerProvider, IScoreProvider, IAliasProvider, IItemListProvider):
 12    """The provider that fetches data from the LXNS.
 13
 14    LXNS: https://maimai.lxns.net/
 15    """
 16
 17    developer_token: str | None
 18    """The developer token used to access the LXNS API."""
 19    base_url = "https://maimai.lxns.net/"
 20    """The base URL for the LXNS API."""
 21
 22    @property
 23    def headers(self):
 24        """@private"""
 25        if not self.developer_token:
 26            raise InvalidDeveloperTokenError()
 27        return {"Authorization": self.developer_token}
 28
 29    def __init__(self, developer_token: str | None = None):
 30        """Initializes the LXNSProvider.
 31
 32        Args:
 33            developer_token: The developer token used to access the LXNS API.
 34        """
 35        self.developer_token = developer_token
 36
 37    def __eq__(self, value):
 38        return isinstance(value, LXNSProvider) and value.developer_token == self.developer_token
 39
 40    async def _ensure_friend_code(self, client: AsyncClient, identifier: PlayerIdentifier) -> None:
 41        if identifier.friend_code is None:
 42            if identifier.qq is not None:
 43                resp = await client.get(self.base_url + f"api/v0/maimai/player/qq/{identifier.qq}", headers=self.headers)
 44                if not resp.json()["success"]:
 45                    raise InvalidPlayerIdentifierError(resp.json()["message"])
 46                identifier.friend_code = resp.json()["data"]["friend_code"]
 47
 48    @staticmethod
 49    def _deser_note(diff: dict, key: str) -> int:
 50        if "notes" in diff:
 51            if "is_buddy" in diff and diff["is_buddy"]:
 52                return diff["notes"]["left"][key] + diff["notes"]["right"][key]
 53            return diff["notes"][key]
 54        return 0
 55
 56    @staticmethod
 57    def _deser_item(item: dict, cls: type) -> Any:
 58        return cls(
 59            id=item["id"],
 60            name=item["name"],
 61            description=item["description"] if "description" in item else None,
 62            genre=item["genre"] if "genre" in item else None,
 63        )
 64
 65    @staticmethod
 66    def _deser_song(song: dict) -> Song:
 67        return Song(
 68            id=song["id"],
 69            title=song["title"],
 70            artist=song["artist"],
 71            genre=name_to_genre[song["genre"]],
 72            bpm=song["bpm"],
 73            aliases=song["aliases"] if "aliases" in song else None,
 74            map=song["map"] if "map" in song else None,
 75            version=song["version"],
 76            rights=song["rights"] if "rights" in song else None,
 77            disabled=song["disabled"] if "disabled" in song else False,
 78            difficulties=SongDifficulties(standard=[], dx=[], utage=[]),
 79        )
 80
 81    @staticmethod
 82    def _deser_diff(difficulty: dict) -> SongDifficulty:
 83        return SongDifficulty(
 84            type=SongType[difficulty["type"].upper()],
 85            level=difficulty["level"],
 86            level_value=difficulty["level_value"],
 87            level_index=LevelIndex(difficulty["difficulty"]),
 88            note_designer=difficulty["note_designer"],
 89            version=difficulty["version"],
 90            tap_num=LXNSProvider._deser_note(difficulty, "tap"),
 91            hold_num=LXNSProvider._deser_note(difficulty, "hold"),
 92            slide_num=LXNSProvider._deser_note(difficulty, "slide"),
 93            touch_num=LXNSProvider._deser_note(difficulty, "touch"),
 94            break_num=LXNSProvider._deser_note(difficulty, "break"),
 95            curve=None,
 96        )
 97
 98    @staticmethod
 99    def _deser_diff_utage(difficulty: dict) -> SongDifficultyUtage:
100        return SongDifficultyUtage(
101            **dataclasses.asdict(LXNSProvider._deser_diff(difficulty)),
102            kanji=difficulty["kanji"],
103            description=difficulty["description"],
104            is_buddy=difficulty["is_buddy"],
105        )
106
107    @staticmethod
108    def _deser_score(score: dict) -> Score:
109        return Score(
110            id=score["id"],
111            song_name=score["song_name"],
112            level=score["level"],
113            level_index=LevelIndex(score["level_index"]),
114            achievements=score["achievements"] if "achievements" in score else None,
115            fc=FCType[score["fc"].upper()] if score["fc"] else None,
116            fs=FSType[score["fs"].upper()] if score["fs"] else None,
117            dx_score=score["dx_score"] if "dx_score" in score else None,
118            dx_rating=score["dx_rating"] if "dx_rating" in score else None,
119            rate=RateType[score["rate"].upper()],
120            type=SongType[score["type"].upper()],
121        )
122
123    @staticmethod
124    def _ser_score(score: Score) -> dict:
125        return {
126            "id": score.id,
127            "song_name": score.song_name,
128            "level": score.level,
129            "level_index": score.level_index.value,
130            "achievements": score.achievements,
131            "fc": score.fc.name.lower() if score.fc else None,
132            "fs": score.fs.name.lower() if score.fs else None,
133            "dx_score": score.dx_score,
134            "dx_rating": score.dx_rating,
135            "rate": score.rate.name.lower(),
136            "type": score.type.name.lower(),
137        }
138
139    def _check_response_player(self, resp: Response) -> dict:
140        resp.raise_for_status()
141        resp_json = resp.json()
142        if not resp_json["success"]:
143            if resp_json["code"] in [400, 404]:
144                raise InvalidPlayerIdentifierError(resp_json["message"])
145            elif resp_json["code"] in [403]:
146                raise PrivacyLimitationError(resp_json["message"])
147            elif resp_json["code"] in [401]:
148                raise InvalidDeveloperTokenError(resp_json["message"])
149        return resp_json
150
151    async def get_song(self, id: int, client: AsyncClient) -> Song:
152        # Fetch single detailed song from LXNS with cache support, due to get_songs not providing detailed notes data
153        if "lxns_detailed_songs" not in default_caches._caches:
154            default_caches._caches["lxns_detailed_songs"] = {}
155        if str(id) in default_caches._caches["lxns_detailed_songs"]:
156            return default_caches._caches["lxns_detailed_songs"][str(id)]
157        resp = await client.get(self.base_url + f"api/v0/maimai/song/{id}")
158        resp.raise_for_status()
159        resp_json = resp.json()
160        song: Song = LXNSProvider._deser_song(resp_json)
161        difficulties = song.difficulties
162        difficulties.standard.extend(LXNSProvider._deser_diff(difficulty) for difficulty in resp_json["difficulties"].get("standard", []))
163        difficulties.dx.extend(LXNSProvider._deser_diff(difficulty) for difficulty in resp_json["difficulties"].get("dx", []))
164        maimai_songs = await MaimaiSongs._get_or_fetch(client)
165        if (new_song := maimai_songs.by_id(id)) and new_song.difficulties.utage:
166            # Fetch utage difficulties separately, if the song has utage difficulties
167            resp1 = await client.get(self.base_url + f"api/v0/maimai/song/{id + 100000}")
168            resp1.raise_for_status()
169            resp_json1 = resp1.json()
170            difficulties.utage.extend(LXNSProvider._deser_diff_utage(difficulty) for difficulty in resp_json1["difficulties"].get("utage", []))
171        default_caches._caches["lxns_detailed_songs"][str(id)] = song
172        return song
173
174    async def get_songs(self, client: AsyncClient) -> list[Song]:
175        resp = await client.get(self.base_url + "api/v0/maimai/song/list")
176        resp.raise_for_status()
177        resp_json = resp.json()
178        unique_songs: dict[int, Song] = {}
179        for song in resp_json["songs"]:
180            unique_key = int(song["id"]) % 10000
181            if unique_key not in unique_songs:
182                unique_songs[unique_key] = LXNSProvider._deser_song(song)
183            difficulties = unique_songs[unique_key].difficulties
184            difficulties.standard.extend(LXNSProvider._deser_diff(difficulty) for difficulty in song["difficulties"].get("standard", []))
185            difficulties.dx.extend(LXNSProvider._deser_diff(difficulty) for difficulty in song["difficulties"].get("dx", []))
186            difficulties.utage.extend(LXNSProvider._deser_diff_utage(difficulty) for difficulty in song["difficulties"].get("utage", []))
187        return list(unique_songs.values())
188
189    async def get_player(self, identifier: PlayerIdentifier, client: AsyncClient) -> LXNSPlayer:
190        resp = await client.get(self.base_url + f"api/v0/maimai/player/{identifier._as_lxns()}", headers=self.headers)
191        resp_data = self._check_response_player(resp)["data"]
192        return LXNSPlayer(
193            name=resp_data["name"],
194            rating=resp_data["rating"],
195            friend_code=resp_data["friend_code"],
196            trophy=PlayerTrophy(id=resp_data["trophy"]["id"], name=resp_data["trophy"]["name"], color=resp_data["trophy"]["color"]),
197            course_rank=resp_data["course_rank"],
198            class_rank=resp_data["class_rank"],
199            star=resp_data["star"],
200            icon=(
201                PlayerIcon(id=resp_data["icon"]["id"], name=resp_data["icon"]["name"], genre=resp_data["icon"]["genre"])
202                if "icon" in resp_data
203                else None
204            ),
205            name_plate=PlayerNamePlate(id=resp_data["name_plate"]["id"], name=resp_data["name_plate"]["name"]) if "name_plate" in resp_data else None,
206            frame=PlayerFrame(id=resp_data["frame"]["id"], name=resp_data["frame"]["name"]) if "frame" in resp_data else None,
207            upload_time=resp_data["upload_time"],
208        )
209
210    async def get_scores_best(self, identifier: PlayerIdentifier, client: AsyncClient) -> tuple[list[Score], list[Score]]:
211        await self._ensure_friend_code(client, identifier)
212        entrypoint = f"api/v0/maimai/player/{identifier.friend_code}/bests"
213        resp = await client.get(self.base_url + entrypoint, headers=self.headers)
214        resp_data = self._check_response_player(resp)["data"]
215        return (
216            [s for score in resp_data["standard"] if (s := LXNSProvider._deser_score(score))],
217            [s for score in resp_data["dx"] if (s := LXNSProvider._deser_score(score))],
218        )
219
220    async def get_scores_all(self, identifier: PlayerIdentifier, client: AsyncClient) -> list[Score]:
221        await self._ensure_friend_code(client, identifier)
222        entrypoint = f"api/v0/maimai/player/{identifier.friend_code}/scores"
223        resp = await client.get(self.base_url + entrypoint, headers=self.headers)
224        resp_data = self._check_response_player(resp)["data"]
225        return [s for score in resp_data if (s := LXNSProvider._deser_score(score))]
226
227    async def update_scores(self, identifier: PlayerIdentifier, scores: list[Score], client: AsyncClient) -> None:
228        await self._ensure_friend_code(client, identifier)
229        entrypoint = f"api/v0/maimai/player/{identifier.friend_code}/scores"
230        use_headers = self.headers
231        if identifier.credentials and isinstance(identifier.credentials, str):
232            # If the player has a personal token, use it to update the scores
233            use_headers["X-User-Token"] = identifier.credentials
234            entrypoint = f"api/v0/user/maimai/player/scores"
235        scores_dict = {"scores": [LXNSProvider._ser_score(score) for score in scores]}
236        resp = await client.post(self.base_url + entrypoint, headers=use_headers, json=scores_dict)
237        resp.raise_for_status()
238        resp_json = resp.json()
239        if not resp_json["success"] and resp_json["code"] == 400:
240            raise ValueError(resp_json["message"])
241
242    async def get_aliases(self, client: AsyncClient) -> list[SongAlias]:
243        resp = await client.get(self.base_url + "api/v0/maimai/alias/list")
244        resp.raise_for_status()
245        return [SongAlias(song_id=item["song_id"], aliases=item["aliases"]) for item in resp.json()["aliases"]]
246
247    async def get_icons(self, client: AsyncClient) -> dict[int, PlayerIcon]:
248        resp = await client.get(self.base_url + "api/v0/maimai/icon/list")
249        resp.raise_for_status()
250        return {item["id"]: LXNSProvider._deser_item(item, PlayerIcon) for item in resp.json()["icons"]}
251
252    async def get_nameplates(self, client: AsyncClient) -> dict[int, PlayerNamePlate]:
253        resp = await client.get(self.base_url + "api/v0/maimai/plate/list")
254        resp.raise_for_status()
255        return {item["id"]: LXNSProvider._deser_item(item, PlayerNamePlate) for item in resp.json()["plates"]}
256
257    async def get_frames(self, client: AsyncClient) -> dict[int, PlayerFrame]:
258        resp = await client.get(self.base_url + "api/v0/maimai/frame/list")
259        resp.raise_for_status()
260        return {item["id"]: LXNSProvider._deser_item(item, PlayerFrame) for item in resp.json()["frames"]}

 8class YuzuProvider(IAliasProvider):
 9    """The provider that fetches song aliases from the Yuzu.
10
11    Yuzu is a bot API that provides song aliases for maimai DX.
12
13    Yuzu: https://bot.yuzuchan.moe/
14    """
15
16    base_url = "https://api.yuzuchan.moe/"
17    """The base URL for the Yuzu API."""
18
19    def __eq__(self, value):
20        return isinstance(value, YuzuProvider)
21
22    async def get_aliases(self, client: AsyncClient) -> list[SongAlias]:
23        resp = await client.get(self.base_url + "maimaidx/maimaidxalias")
24        resp.raise_for_status()
25        return [SongAlias(song_id=item["SongID"] % 10000, aliases=item["Alias"]) for item in resp.json()["content"]]

15class WechatProvider(IScoreProvider):
16    """The provider that fetches data from the Wahlap Wechat OffiAccount.
17
18    PlayerIdentifier must have the `credentials` attribute, we suggest you to use the `maimai.wechat()` method to get the identifier.
19
20    PlayerIdentifier should not be cached or stored in the database, as the cookies may expire at any time.
21
22    Wahlap Wechat OffiAccount: https://maimai.wahlap.com/maimai-mobile/
23    """
24
25    def __eq__(self, value):
26        return isinstance(value, WechatProvider)
27
28    @staticmethod
29    def _deser_score(score: dict, songs: "MaimaiSongs") -> Score | None:
30        if song := songs.by_title(score["title"]):
31            is_utage = (len(song.difficulties.dx) + len(song.difficulties.standard)) == 0
32            song_type = SongType.STANDARD if score["type"] == "SD" else SongType.DX if score["type"] == "DX" and not is_utage else SongType.UTAGE
33            level_index = LevelIndex(score["level_index"])
34            if diff := song.get_difficulty(song_type, level_index):
35                rating = ScoreCoefficient(score["achievements"]).ra(diff.level_value)
36                return Score(
37                    id=song.id,
38                    song_name=song.title,
39                    level=score["level"],
40                    level_index=level_index,
41                    achievements=score["achievements"],
42                    fc=FCType[score["fc"].upper()] if score["fc"] else None,
43                    fs=FSType[score["fs"].upper().replace("FDX", "FSD")] if score["fs"] else None,
44                    dx_score=score["dxScore"],
45                    dx_rating=rating,
46                    rate=RateType[score["rate"].upper()],
47                    type=song_type,
48                )
49
50    async def _crawl_scores_diff(self, client: AsyncClient, diff: int, cookies: Cookies, songs: "MaimaiSongs") -> list[Score]:
51        await asyncio.sleep(random.randint(0, 300) / 1000)  # sleep for a random amount of time between 0 and 300ms
52        resp1 = await client.get(f"https://maimai.wahlap.com/maimai-mobile/record/musicGenre/search/?genre=99&diff={diff}", cookies=cookies)
53        # body = re.search(r"<html.*?>([\s\S]*?)</html>", resp1.text).group(1).replace(r"\s+", " ")
54        wm_json = wmdx_html2json(resp1.text)
55        return [parsed for score in wm_json if (parsed := WechatProvider._deser_score(score, songs))]
56
57    async def _crawl_scores(self, client: AsyncClient, cookies: Cookies, songs: "MaimaiSongs") -> Sequence[Score]:
58        tasks = [self._crawl_scores_diff(client, diff, cookies, songs) for diff in [0, 1, 2, 3, 4]]
59        results = await asyncio.gather(*tasks)
60        return functools.reduce(operator.concat, results, [])
61
62    async def get_scores_all(self, identifier: PlayerIdentifier, client: AsyncClient) -> list[Score]:
63        if not identifier.credentials or not isinstance(identifier.credentials, Cookies):
64            raise InvalidPlayerIdentifierError("Wahlap wechat cookies are required to fetch scores")
65        msongs = await MaimaiSongs._get_or_fetch(client)
66        scores = await self._crawl_scores(client, identifier.credentials, msongs)
67        return list(scores)