maimai_py.providers

11class IProvider:
12    @abstractmethod
13    def _hash(self) -> str: ...

73class IScoreUpdateProvider(IProvider):
74    """The provider that updates scores to a specific source.
75
76    Available providers: `DivingFishProvider`, `LXNSProvider`
77    """
78
79    @abstractmethod
80    async def update_scores(self, identifier: PlayerIdentifier, scores: Iterable[Score], client: "MaimaiClient") -> None:
81        """@private"""
82        raise NotImplementedError()

28class IAliasProvider(IProvider):
29    """The provider that fetches song aliases from a specific source.
30
31    Available providers: `YuzuProvider`, `LXNSProvider`
32    """
33
34    @abstractmethod
35    async def get_aliases(self, client: "MaimaiClient") -> dict[int, list[str]]:
36        """@private"""
37        raise NotImplementedError()

146class IAreaProvider(IProvider):
147    """The provider that fetches area data from a specific source.
148
149    Available providers: `LocalProvider`
150    """
151
152    @abstractmethod
153    async def get_areas(self, lang: str, client: "MaimaiClient") -> dict[str, Area]:
154        """@private"""
155        raise NotImplementedError()

85class ICurveProvider(IProvider):
86    """The provider that fetches statistics curves from a specific source.
87
88    Available providers: `DivingFishProvider`
89    """
90
91    @abstractmethod
92    async def get_curves(self, client: "MaimaiClient") -> dict[tuple[int, SongType], list[CurveObject]]:
93        """@private"""
94        raise NotImplementedError()

109class IItemListProvider(IProvider):
110    """The provider that fetches player item list data from a specific source.
111
112    Available providers: `LXNSProvider`, `LocalProvider`
113    """
114
115    @abstractmethod
116    async def get_icons(self, client: "MaimaiClient") -> dict[int, PlayerIcon]:
117        """@private"""
118        raise NotImplementedError()
119
120    @abstractmethod
121    async def get_nameplates(self, client: "MaimaiClient") -> dict[int, PlayerNamePlate]:
122        """@private"""
123        raise NotImplementedError()
124
125    @abstractmethod
126    async def get_frames(self, client: "MaimaiClient") -> dict[int, PlayerFrame]:
127        """@private"""
128        raise NotImplementedError()
129
130    @abstractmethod
131    async def get_partners(self, client: "MaimaiClient") -> dict[int, PlayerPartner]:
132        """@private"""
133        raise NotImplementedError()
134
135    @abstractmethod
136    async def get_charas(self, client: "MaimaiClient") -> dict[int, PlayerChara]:
137        """@private"""
138        raise NotImplementedError()
139
140    @abstractmethod
141    async def get_trophies(self, client: "MaimaiClient") -> dict[int, PlayerTrophy]:
142        """@private"""
143        raise NotImplementedError()

40class IPlayerProvider(IProvider):
41    """The provider that fetches players from a specific source.
42
43    Available providers: `DivingFishProvider`, `LXNSProvider`
44    """
45
46    @abstractmethod
47    async def get_player(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> Player:
48        """@private"""
49        raise NotImplementedError()

 97class IRegionProvider(IProvider):
 98    """The provider that fetches player regions from a specific source.
 99
100    Available providers: `ArcadeProvider`
101    """
102
103    @abstractmethod
104    async def get_regions(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> list[PlayerRegion]:
105        """@private"""
106        raise NotImplementedError()

52class IScoreProvider(IProvider):
53    """The provider that fetches scores from a specific source.
54
55    Available providers: `DivingFishProvider`, `LXNSProvider`, `WechatProvider`
56    """
57
58    async def get_scores_one(self, identifier: PlayerIdentifier, song: Song, client: "MaimaiClient") -> list[Score]:
59        """@private"""
60        scores = await self.get_scores_all(identifier, client)
61        return [score for score in scores if score.id == song.id]
62
63    async def get_scores_best(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> list[Score]:
64        """@private"""
65        return await self.get_scores_all(identifier, client)
66
67    @abstractmethod
68    async def get_scores_all(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> list[Score]:
69        """@private"""
70        raise NotImplementedError()

16class ISongProvider(IProvider):
17    """The provider that fetches songs from a specific source.
18
19    Available providers: `DivingFishProvider`, `LXNSProvider`
20    """
21
22    @abstractmethod
23    async def get_songs(self, client: "MaimaiClient") -> list[Song]:
24        """@private"""
25        raise NotImplementedError()

158class IPlayerIdentifierProvider(IProvider):
159    """The provider that fetches player identifiers from a specific source.
160
161    Available providers: `ArcadeProvider`
162    """
163
164    @abstractmethod
165    async def get_identifier(self, code: Union[str, dict[str, str]], client: "MaimaiClient") -> PlayerIdentifier:
166        """@private"""
167        raise NotImplementedError()

 19class ArcadeProvider(IPlayerProvider, IScoreProvider, IRegionProvider, IPlayerIdentifierProvider):
 20    """The provider that fetches data from the wahlap maimai arcade.
 21
 22    This part of the maimai.py is not open-source, we distribute the compiled version of this part of the code as maimai_ffi.
 23
 24    Feel free to ask us to solve if your platform or architecture is not supported.
 25
 26    maimai.ffi: https://pypi.org/project/maimai-ffi
 27    """
 28
 29    _http_proxy: Optional[str] = None
 30
 31    def __init__(self, http_proxy: Optional[str] = None):
 32        self._http_proxy = http_proxy
 33
 34    def _hash(self) -> str:
 35        return hashlib.md5(b"arcade").hexdigest()
 36
 37    @staticmethod
 38    async def _deser_score(score: dict, songs: "MaimaiSongs") -> Optional[Score]:
 39        song_type = SongType._from_id(score["musicId"])
 40        level_index = LevelIndex(score["level"]) if song_type != SongType.UTAGE else None
 41        achievement = float(score["achievement"]) / 10000
 42        if song := await songs.by_id(score["musicId"] % 10000):
 43            if level_index and (diff := song.get_difficulty(song_type, level_index)):
 44                fs_type = FSType(score["syncStatus"]) if 0 < score["syncStatus"] < 5 else None
 45                fs_type = FSType.SYNC if score["syncStatus"] == 5 else fs_type
 46                return Score(
 47                    id=song.id,
 48                    level=diff.level,
 49                    level_index=diff.level_index,
 50                    achievements=achievement,
 51                    fc=FCType(4 - score["comboStatus"]) if score["comboStatus"] != 0 else None,
 52                    fs=fs_type,
 53                    dx_score=score["deluxscoreMax"],
 54                    dx_rating=ScoreCoefficient(achievement).ra(diff.level_value),
 55                    play_count=score["playCount"],
 56                    rate=RateType._from_achievement(achievement),
 57                    type=song_type,
 58                )
 59
 60    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(TitleServerNetworkError), reraise=True)
 61    async def get_player(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> ArcadePlayer:
 62        maimai_icons = await client.items(PlayerIcon)
 63        maimai_trophies = await client.items(PlayerTrophy)
 64        maimai_nameplates = await client.items(PlayerNamePlate)
 65        if identifier.credentials and isinstance(identifier.credentials, str):
 66            resp_dict = await arcade.get_user_preview(identifier.credentials.encode(), http_proxy=self._http_proxy)
 67            return ArcadePlayer(
 68                name=resp_dict["userName"],
 69                rating=resp_dict["playerRating"],
 70                is_login=resp_dict["isLogin"],
 71                name_plate=await maimai_nameplates.by_id(resp_dict["nameplateId"]),
 72                icon=await maimai_icons.by_id(resp_dict["iconId"]),
 73                trophy=await maimai_trophies.by_id(resp_dict["trophyId"]),
 74            )
 75        raise InvalidPlayerIdentifierError("Player identifier credentials should be provided.")
 76
 77    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(TitleServerNetworkError), reraise=True)
 78    async def get_scores_all(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> list[Score]:
 79        maimai_songs = await client.songs()
 80        if identifier.credentials and isinstance(identifier.credentials, str):
 81            resp_list = await arcade.get_user_scores(identifier.credentials.encode(), http_proxy=self._http_proxy)
 82            return [s for score in resp_list if (s := await ArcadeProvider._deser_score(score, maimai_songs))]
 83        raise InvalidPlayerIdentifierError("Player identifier credentials should be provided.")
 84
 85    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(TitleServerNetworkError), reraise=True)
 86    async def get_regions(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> list[PlayerRegion]:
 87        if identifier.credentials and isinstance(identifier.credentials, str):
 88            resp_dict = await arcade.get_user_region(identifier.credentials.encode(), http_proxy=self._http_proxy)
 89            return [
 90                PlayerRegion(
 91                    region_id=region["regionId"],
 92                    region_name=region["regionName"],
 93                    play_count=region["playCount"],
 94                    created_at=datetime.strptime(region["created"], "%Y-%m-%d %H:%M:%S"),
 95                )
 96                for region in resp_dict["userRegionList"]
 97            ]
 98        raise InvalidPlayerIdentifierError("Player identifier credentials should be provided.")
 99
100    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type((TimeoutException, NetworkError)), reraise=True)
101    async def get_identifier(self, code: Union[str, dict[str, str]], client: "MaimaiClient") -> PlayerIdentifier:
102        resp_bytes: bytes = await arcade.get_uid_encrypted(str(code), http_proxy=self._http_proxy)
103        return PlayerIdentifier(credentials=resp_bytes.decode())

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

 15class LocalProvider(IItemListProvider, IAreaProvider):
 16    """The provider that fetches data from the local storage.
 17
 18    Most of the data are stored in JSON files in the same directory as this file.
 19    """
 20
 21    def _hash(self) -> str:
 22        current_folder = Path(__file__).resolve().parent
 23        json_files = sorted(current_folder.glob("*.json"))
 24
 25        if not json_files:
 26            return hashlib.md5(b"local").hexdigest()
 27
 28        combined_content = b""
 29        for file_path in json_files:
 30            with open(file_path, "rb") as f:
 31                combined_content += f.read()
 32
 33        return hashlib.md5(combined_content).hexdigest()
 34
 35    def _read_file(self, file_name: str) -> Any:
 36        current_folder = Path(__file__).resolve().parent
 37        path = current_folder / f"{file_name}.json"
 38        if not path.exists():
 39            raise FileNotFoundError(f"File {path} not found.")
 40        with open(path, "r", encoding="utf-8") as f:
 41            return json.load(f)
 42
 43    def _read_file_dict(self, file_name: str) -> dict:
 44        obj = self._read_file(file_name)
 45        if isinstance(obj, dict):
 46            return obj["data"]
 47        else:
 48            raise ValueError(f"File {file_name} is not a dictionary.")
 49
 50    def _read_file_list(self, file_name: str) -> list:
 51        obj = self._read_file(file_name)
 52        if isinstance(obj, list):
 53            return obj
 54        else:
 55            raise ValueError(f"File {file_name} is not a list.")
 56
 57    async def get_icons(self, client: "MaimaiClient") -> dict[int, PlayerIcon]:
 58        return {int(k): PlayerIcon(id=int(k), name=v, description=None, genre=None) for k, v in self._read_file_dict("icons").items()}
 59
 60    async def get_nameplates(self, client: "MaimaiClient") -> dict[int, PlayerNamePlate]:
 61        return {int(k): PlayerNamePlate(id=int(k), name=v, description=None, genre=None) for k, v in self._read_file_dict("nameplates").items()}
 62
 63    async def get_frames(self, client: "MaimaiClient") -> dict[int, PlayerFrame]:
 64        return {int(k): PlayerFrame(id=int(k), name=v, description=None, genre=None) for k, v in self._read_file_dict("frames").items()}
 65
 66    async def get_partners(self, client: "MaimaiClient") -> dict[int, PlayerPartner]:
 67        return {int(k): PlayerPartner(id=int(k), name=v) for k, v in self._read_file_dict("partners").items()}
 68
 69    async def get_charas(self, client: "MaimaiClient") -> dict[int, PlayerChara]:
 70        return {int(k): PlayerChara(id=int(k), name=v) for k, v in self._read_file_dict("charas").items()}
 71
 72    async def get_trophies(self, client: "MaimaiClient") -> dict[int, PlayerTrophy]:
 73        return {int(k): PlayerTrophy(id=int(k), name=v["title"], color=v["rareType"]) for k, v in self._read_file_dict("trophies").items()}
 74
 75    async def get_areas(self, lang: str, client: "MaimaiClient") -> dict[str, Area]:
 76        maimai_songs = await client.songs()
 77        areas = {
 78            item["id"]: Area(
 79                id=item["id"],
 80                name=item["name"],
 81                comment=item["comment"],
 82                description=item["description"],
 83                video_id=item["video_id"],
 84                characters=[
 85                    AreaCharacter(
 86                        name=char["name"],
 87                        illustrator=char["illustrator"],
 88                        description1=char["description1"],
 89                        description2=char["description2"],
 90                        team=char["team"],
 91                        props=char["props"],
 92                    )
 93                    for char in item["characters"]
 94                ],
 95                songs=[
 96                    AreaSong(
 97                        id=None,
 98                        title=song["title"],
 99                        artist=song["artist"],
100                        description=song["description"],
101                        illustrator=song["illustrator"],
102                        movie=song["movie"],
103                    )
104                    for song in item["songs"]
105                ],
106            )
107            for item in self._read_file_list(f"areas_{lang}")
108        }
109        for area in areas.values():
110            for song in area.songs:
111                maimai_song = await maimai_songs.by_title(song.title)
112                song.id = maimai_song.id if maimai_song else None
113        return areas

 22class LXNSProvider(ISongProvider, IPlayerProvider, IScoreProvider, IScoreUpdateProvider, IAliasProvider, IItemListProvider):
 23    """The provider that fetches data from the LXNS.
 24
 25    LXNS: https://maimai.lxns.net/
 26    """
 27
 28    developer_token: Optional[str]
 29    """The developer token used to access the LXNS API."""
 30    base_url = "https://maimai.lxns.net/"
 31    """The base URL for the LXNS API."""
 32
 33    @property
 34    def headers(self):
 35        """@private"""
 36        if not self.developer_token:
 37            raise InvalidDeveloperTokenError("Developer token is not provided.")
 38        return {"Authorization": self.developer_token}
 39
 40    def __init__(self, developer_token: Optional[str] = None):
 41        """Initializes the LXNSProvider.
 42
 43        Args:
 44            developer_token: The developer token used to access the LXNS API.
 45        """
 46        self.developer_token = developer_token
 47
 48    def _hash(self) -> str:
 49        return hashlib.md5(b"lxns").hexdigest()
 50
 51    async def _ensure_friend_code(self, client: "MaimaiClient", identifier: PlayerIdentifier) -> None:
 52        if identifier.friend_code is None:
 53            if identifier.qq is not None:
 54                resp = await client._client.get(self.base_url + f"api/v0/maimai/player/qq/{identifier.qq}", headers=self.headers)
 55                if not resp.json()["success"]:
 56                    raise InvalidPlayerIdentifierError(resp.json()["message"])
 57                identifier.friend_code = resp.json()["data"]["friend_code"]
 58
 59    async def _build_player_request(self, path: str, identifier: PlayerIdentifier, client: "MaimaiClient") -> tuple[str, dict[str, str], bool]:
 60        use_user_api = identifier.credentials is not None and isinstance(identifier.credentials, str)
 61        if use_user_api:
 62            # user-level API takes the precedence: If personal token provided, use it first
 63            assert isinstance(identifier.credentials, str)
 64            entrypoint = f"api/v0/user/maimai/player/{path}"
 65            headers = {"X-User-Token": identifier.credentials}
 66        else:
 67            await self._ensure_friend_code(client, identifier)
 68            entrypoint = f"api/v0/maimai/player/{identifier.friend_code}/{path}"
 69            headers = self.headers
 70        entrypoint = entrypoint.removesuffix("/")
 71        return self.base_url + entrypoint, headers, use_user_api
 72
 73    @staticmethod
 74    def _deser_note(diff: dict, key: str) -> int:
 75        if "notes" in diff:
 76            if "is_buddy" in diff and diff["is_buddy"]:
 77                return diff["notes"]["left"][key] + diff["notes"]["right"][key]
 78            return diff["notes"][key]
 79        return 0
 80
 81    @staticmethod
 82    def _deser_item(item: dict, cls: type) -> Any:
 83        return cls(
 84            id=item["id"],
 85            name=item["name"],
 86            description=item["description"] if "description" in item else None,
 87            genre=item["genre"] if "genre" in item else None,
 88        )
 89
 90    @staticmethod
 91    def _deser_song(song: dict) -> Song:
 92        return Song(
 93            id=song["id"],
 94            title=song["title"],
 95            artist=song["artist"],
 96            genre=name_to_genre[song["genre"]],
 97            bpm=song["bpm"],
 98            aliases=song["aliases"] if "aliases" in song else None,
 99            map=song["map"] if "map" in song else None,
100            version=song["version"],
101            rights=song["rights"] if "rights" in song else None,
102            disabled=song["disabled"] if "disabled" in song else False,
103            difficulties=SongDifficulties(standard=[], dx=[], utage=[]),
104        )
105
106    @staticmethod
107    def _deser_diff(difficulty: dict) -> SongDifficulty:
108        return SongDifficulty(
109            type=SongType[difficulty["type"].upper()],
110            level=difficulty["level"],
111            level_value=difficulty["level_value"],
112            level_index=LevelIndex(difficulty["difficulty"]),
113            note_designer=difficulty["note_designer"],
114            version=difficulty["version"],
115            tap_num=LXNSProvider._deser_note(difficulty, "tap"),
116            hold_num=LXNSProvider._deser_note(difficulty, "hold"),
117            slide_num=LXNSProvider._deser_note(difficulty, "slide"),
118            touch_num=LXNSProvider._deser_note(difficulty, "touch"),
119            break_num=LXNSProvider._deser_note(difficulty, "break"),
120            curve=None,
121        )
122
123    @staticmethod
124    def _deser_diff_utage(difficulty: dict) -> SongDifficultyUtage:
125        return SongDifficultyUtage(
126            **dataclasses.asdict(LXNSProvider._deser_diff(difficulty)),
127            kanji=difficulty["kanji"],
128            description=difficulty["description"],
129            is_buddy=difficulty["is_buddy"],
130        )
131
132    @staticmethod
133    def _deser_score(score: dict) -> Score:
134        return Score(
135            id=score["id"],
136            level=score["level"],
137            level_index=LevelIndex(score["level_index"]),
138            achievements=score["achievements"] if "achievements" in score else None,
139            fc=FCType[score["fc"].upper()] if score["fc"] else None,
140            fs=FSType[score["fs"].upper()] if score["fs"] else None,
141            dx_score=score["dx_score"] if "dx_score" in score else None,
142            dx_rating=int(score["dx_rating"]) if "dx_rating" in score else None,
143            play_count=None,
144            rate=RateType[score["rate"].upper()],
145            type=SongType[score["type"].upper()],
146        )
147
148    @staticmethod
149    async def _ser_score(score: Score, songs: "MaimaiSongs") -> Optional[dict]:
150        song_title = song.title if (song := await songs.by_id(score.id)) else None
151        if song_title is not None:
152            return {
153                "id": score.id,
154                "song_name": song_title,
155                "level": score.level,
156                "level_index": score.level_index.value,
157                "achievements": score.achievements,
158                "fc": score.fc.name.lower() if score.fc else None,
159                "fs": score.fs.name.lower() if score.fs else None,
160                "dx_score": score.dx_score,
161                "dx_rating": score.dx_rating,
162                "rate": score.rate.name.lower(),
163                "type": score.type.name.lower(),
164            }
165
166    def _check_response_player(self, resp: Response) -> dict:
167        try:
168            resp_json = resp.json()
169            if not resp_json["success"]:
170                if resp_json["code"] in [400, 404]:
171                    raise InvalidPlayerIdentifierError(resp_json["message"])
172                elif resp_json["code"] in [403]:
173                    raise PrivacyLimitationError(resp_json["message"])
174                elif resp_json["code"] in [401]:
175                    raise InvalidDeveloperTokenError(resp_json["message"])
176                elif resp.status_code in [400, 401]:
177                    raise InvalidPlayerIdentifierError(resp_json["message"])
178                elif not resp.is_success:
179                    resp.raise_for_status()
180            return resp_json
181        except JSONDecodeError as exc:
182            raise InvalidJsonError(resp.text) from exc
183        except HTTPStatusError as exc:
184            raise MaimaiPyError(exc) from exc
185
186    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
187    async def get_songs(self, client: "MaimaiClient") -> list[Song]:
188        resp = await client._client.get(self.base_url + "api/v0/maimai/song/list?notes=true")
189        resp.raise_for_status()
190        resp_json = resp.json()
191        unique_songs: dict[int, Song] = {}
192        for song in resp_json["songs"]:
193            unique_key = int(song["id"]) % 10000
194            if unique_key not in unique_songs:
195                unique_songs[unique_key] = LXNSProvider._deser_song(song)
196            difficulties = unique_songs[unique_key].difficulties
197            difficulties.standard.extend(LXNSProvider._deser_diff(difficulty) for difficulty in song["difficulties"].get("standard", []))
198            difficulties.dx.extend(LXNSProvider._deser_diff(difficulty) for difficulty in song["difficulties"].get("dx", []))
199            difficulties.utage.extend(LXNSProvider._deser_diff_utage(difficulty) for difficulty in song["difficulties"].get("utage", []))
200        return list(unique_songs.values())
201
202    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
203    async def get_player(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> LXNSPlayer:
204        maimai_frames = await client.items(PlayerFrame)
205        maimai_icons = await client.items(PlayerIcon)
206        maimai_trophies = await client.items(PlayerTrophy)
207        maimai_nameplates = await client.items(PlayerNamePlate)
208        url, headers, _ = await self._build_player_request("", identifier, client)
209        resp = await client._client.get(url, headers=headers)
210        resp_data = self._check_response_player(resp)["data"]
211        return LXNSPlayer(
212            name=resp_data["name"],
213            rating=resp_data["rating"],
214            friend_code=resp_data["friend_code"],
215            course_rank=resp_data["course_rank"],
216            class_rank=resp_data["class_rank"],
217            star=resp_data["star"],
218            frame=await maimai_frames.by_id(resp_data["frame"]["id"]) if "frame" in resp_data else None,
219            icon=await maimai_icons.by_id(resp_data["icon"]["id"]) if "icon" in resp_data else None,
220            trophy=await maimai_trophies.by_id(resp_data["trophy"]["id"]) if "trophy" in resp_data else None,
221            name_plate=await maimai_nameplates.by_id(resp_data["name_plate"]["id"]) if "name_plate" in resp_data else None,
222            upload_time=resp_data["upload_time"],
223        )
224
225    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
226    async def get_scores_all(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> list[Score]:
227        url, headers, use_user_api = await self._build_player_request("scores", identifier, client)
228        resp = await client._client.get(url, headers=headers)
229        resp_data = self._check_response_player(resp)["data"]
230        scores = [s for score in resp_data if (s := LXNSProvider._deser_score(score))]
231        if not use_user_api:
232            # LXNSProvider's developer-level API scores are incomplete, which doesn't contain dx_rating and achievements, leading to sorting difficulties.
233            # In this case, we should always fetch the b35 and b15 scores for LXNSProvider.
234            scores.extend(await self.get_scores_best(identifier, client))
235        return scores
236
237    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
238    async def get_scores_best(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> list[Score]:
239        if identifier.friend_code is None:
240            return await self.get_scores_all(identifier, client)
241        await self._ensure_friend_code(client, identifier)
242        entrypoint = f"api/v0/maimai/player/{identifier.friend_code}/bests"
243        resp = await client._client.get(self.base_url + entrypoint, headers=self.headers)
244        resp_data = self._check_response_player(resp)["data"]
245        return [s for score in resp_data["standard"] + resp_data["dx"] if (s := LXNSProvider._deser_score(score))]
246
247    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
248    async def get_scores_one(self, identifier: PlayerIdentifier, song: Song, client: "MaimaiClient") -> list[Score]:
249        await self._ensure_friend_code(client, identifier)
250        request_tasks, create_task = [], lambda type: asyncio.create_task(
251            client._client.get(
252                self.base_url + f"api/v0/maimai/player/{identifier.friend_code}/bests",
253                params={"song_id": song.id if type != SongType.UTAGE else song.id + 100000, "song_type": type.value},
254                headers=self.headers,
255            )
256        )
257        if len(song.difficulties.standard) > 0:
258            request_tasks.append(create_task(SongType.STANDARD))
259        if len(song.difficulties.dx) > 0:
260            request_tasks.append(create_task(SongType.DX))
261        if len(song.difficulties.utage) > 0:
262            request_tasks.append(create_task(SongType.UTAGE))
263        resps = await asyncio.gather(*request_tasks)
264        resp_data = [self._check_response_player(resp)["data"] for resp in resps]
265        return [s for score in reduce(concat, resp_data) if (s := LXNSProvider._deser_score(score))]
266
267    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
268    async def update_scores(self, identifier: PlayerIdentifier, scores: Iterable[Score], client: "MaimaiClient") -> None:
269        maimai_songs = await client.songs()
270        url, headers, _ = await self._build_player_request("scores", identifier, client)
271        scores_dict = {"scores": [json for score in scores if (json := await LXNSProvider._ser_score(score, maimai_songs))]}
272        resp = await client._client.post(url, headers=headers, json=scores_dict)
273        resp.raise_for_status()
274        resp_json = resp.json()
275        if not resp_json["success"] and resp_json["code"] == 400:
276            raise ValueError(resp_json["message"])
277
278    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
279    async def get_aliases(self, client: "MaimaiClient") -> dict[int, list[str]]:
280        resp = await client._client.get(self.base_url + "api/v0/maimai/alias/list")
281        resp.raise_for_status()
282        return {item["song_id"]: item["aliases"] for item in resp.json()["aliases"]}
283
284    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
285    async def get_icons(self, client: "MaimaiClient") -> dict[int, PlayerIcon]:
286        resp = await client._client.get(self.base_url + "api/v0/maimai/icon/list")
287        resp.raise_for_status()
288        return {item["id"]: LXNSProvider._deser_item(item, PlayerIcon) for item in resp.json()["icons"]}
289
290    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
291    async def get_nameplates(self, client: "MaimaiClient") -> dict[int, PlayerNamePlate]:
292        resp = await client._client.get(self.base_url + "api/v0/maimai/plate/list")
293        resp.raise_for_status()
294        return {item["id"]: LXNSProvider._deser_item(item, PlayerNamePlate) for item in resp.json()["plates"]}
295
296    @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
297    async def get_frames(self, client: "MaimaiClient") -> dict[int, PlayerFrame]:
298        resp = await client._client.get(self.base_url + "api/v0/maimai/frame/list")
299        resp.raise_for_status()
300        return {item["id"]: LXNSProvider._deser_item(item, PlayerFrame) for item in resp.json()["frames"]}