feat: enhance lxns provider

Author: TrueRou

docs/changelog.md

@@ -1,5 +1,15 @@
 # 更新日志
 
+## 1.3.6 (2025-08-22)
+
+Features:
+  - WechatProvider 现支持 获取好友 和 设置对手 的功能
+  - WechatProvider 现支持 IPlayerProvider 接口 -> [#36](https://github.com/TrueRou/maimai.py/pull/36)
+  - LXNSProvider 现支持 使用OAuth认证的个人API密钥 来访问接口
+
+Bugfixes:
+  - 修复 LXNSProvider 在获取单个成绩、最佳成绩时无法使用个人API密钥
+
 ## 1.3.5 (2025-08-05)
 
 Features:

docs/providers/lxns.md

@@ -14,10 +14,6 @@
 - 使用好友代码：`PlayerIdentifier(friend_code="123456789")`，将使用开发者API。
 - 使用QQ号：`PlayerIdentifier(qq="123456789")`，将使用开发者API。
 
-::: warning
-注意：获取单个成绩（`maimai.minfo(...)`）时，目前仅支持开发者API。
-:::
-
 ## 关于开发者API
 
 申请开发者Token：https://maimai.lxns.net/developer

docs/providers/wechat.md

@@ -2,7 +2,7 @@
 
 舞萌服务号的舞萌DX页面，通过HTML解析获取信息。
 
-实现：IScoreProvider, IPlayerIdentifierProvider
+实现：IScoreProvider, IPlayerIdentifierProvider, IPlayerProvider
 
 源站：https://maimai.wahlap.com/maimai-mobile/
 
@@ -22,6 +22,19 @@
 参考 [proxy_updater (示例项目)](../samples/proxy_updater.md) 部分，这是一个通过代理和微信 OAuth 认证更新查分器的示例。
 :::
 
+## 关于 Rival (对手)
+
+通过 NET 的隐藏接口，您可以修改到玩家的对手信息。
+
+```python
+wechat = WechatProvider()
+identifier = await maimai.wechat(r="r", t="t", code="code", state="state")
+friends = await wechat.get_friends(identifier, maimai) # 需要主动传入 MaimaiClient 实例
+await wechat.set_rival_on(identifier, friends[0], maimai) # 设置第一个好友为对手
+```
+
+鉴于接口原因，我们无法查询玩家已有的对手信息，只能开启和关闭对手。
+
 ## 已知问题
 
 - WechatProvider 可能出现轻微内存泄露问题。

maimai_py/providers/lxns.py

@@ -1,6 +1,7 @@
 import asyncio
 import dataclasses
 import hashlib
+import re
 from functools import reduce
 from json import JSONDecodeError
 from operator import concat
@@ -17,6 +18,8 @@
 if TYPE_CHECKING:
     from maimai_py.maimai import MaimaiClient, MaimaiSongs
 
+is_jwt = re.compile(r"^[a-zA-Z0-9-_]+\.[a-zA-Z0-9-_]+\.[a-zA-Z0-9-_]+$")
+
 
 class LXNSProvider(ISongProvider, IPlayerProvider, IScoreProvider, IScoreUpdateProvider, IAliasProvider, IItemListProvider):
     """The provider that fetches data from the LXNS.
@@ -61,7 +64,10 @@ async def _build_player_request(self, path: str, identifier: PlayerIdentifier, c
             # user-level API takes the precedence: If personal token provided, use it first
             assert isinstance(identifier.credentials, str)
             entrypoint = f"api/v0/user/maimai/player/{path}"
-            headers = {"X-User-Token": identifier.credentials}
+            if is_jwt.match(identifier.credentials) is not None:
+                headers = {"Authorization": f"Bearer {identifier.credentials}"}
+            else:
+                headers = {"X-User-Token": identifier.credentials}
         else:
             await self._ensure_friend_code(client, identifier)
             entrypoint = f"api/v0/maimai/player/{identifier.friend_code}/{path}"
@@ -235,44 +241,36 @@ async def get_scores_all(self, identifier: PlayerIdentifier, client: "MaimaiClie
 
     @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
     async def get_scores_best(self, identifier: PlayerIdentifier, client: "MaimaiClient") -> list[Score]:
-        if identifier.friend_code is None:
-            return await self.get_scores_all(identifier, client)
-        await self._ensure_friend_code(client, identifier)
-        entrypoint = f"api/v0/maimai/player/{identifier.friend_code}/bests"
-        resp = await client._client.get(self.base_url + entrypoint, headers=self.headers)
+        url, headers, _ = await self._build_player_request("bests", identifier, client)
+        resp = await client._client.get(url, headers=headers)
         resp_data = self._check_response_player(resp)["data"]
         return [s for score in resp_data["standard"] + resp_data["dx"] if (s := LXNSProvider._deser_score(score))]
 
     @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
     async def get_scores_one(self, identifier: PlayerIdentifier, song: Song, client: "MaimaiClient") -> list[Score]:
-        await self._ensure_friend_code(client, identifier)
-        request_tasks, create_task = [], lambda type: asyncio.create_task(
-            client._client.get(
-                self.base_url + f"api/v0/maimai/player/{identifier.friend_code}/bests",
-                params={"song_id": song.id if type != SongType.UTAGE else song.id + 100000, "song_type": type.value},
-                headers=self.headers,
-            )
-        )
+        async def get_scores_for_type(type: SongType) -> list[Score]:
+            url, headers, _ = await self._build_player_request("bests", identifier, client)
+            params = {"song_id": song.id if type != SongType.UTAGE else song.id + 100000, "song_type": type.value}
+            resp = await client._client.get(url, params=params, headers=headers)
+            resp_data = self._check_response_player(resp)["data"]
+            return [s for score in resp_data if (s := LXNSProvider._deser_score(score))]
+
+        results, _ = [], await self._ensure_friend_code(client, identifier)
         if len(song.difficulties.standard) > 0:
-            request_tasks.append(create_task(SongType.STANDARD))
+            results.append(asyncio.create_task(get_scores_for_type(SongType.STANDARD)))
         if len(song.difficulties.dx) > 0:
-            request_tasks.append(create_task(SongType.DX))
+            results.append(asyncio.create_task(get_scores_for_type(SongType.DX)))
         if len(song.difficulties.utage) > 0:
-            request_tasks.append(create_task(SongType.UTAGE))
-        resps = await asyncio.gather(*request_tasks)
-        resp_data = [self._check_response_player(resp)["data"] for resp in resps]
-        return [s for score in reduce(concat, resp_data) if (s := LXNSProvider._deser_score(score))]
+            results.append(asyncio.create_task(get_scores_for_type(SongType.UTAGE)))
+        return [score for score in reduce(concat, await asyncio.gather(*results))]
 
     @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
     async def update_scores(self, identifier: PlayerIdentifier, scores: Iterable[Score], client: "MaimaiClient") -> None:
         maimai_songs = await client.songs()
         url, headers, _ = await self._build_player_request("scores", identifier, client)
         scores_dict = {"scores": [json for score in scores if (json := await LXNSProvider._ser_score(score, maimai_songs))]}
         resp = await client._client.post(url, headers=headers, json=scores_dict)
-        resp.raise_for_status()
-        resp_json = resp.json()
-        if not resp_json["success"] and resp_json["code"] == 400:
-            raise ValueError(resp_json["message"])
+        self._check_response_player(resp)
 
     @retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(RequestError), reraise=True)
     async def get_aliases(self, client: "MaimaiClient") -> dict[int, list[str]]: