feature: new fetch api for retry

* NCMLyrics also can stores track ID, but need set value manually
* Convert NCM*.fromApi() to @classmethod, and pass response directly
This commit is contained in:
Puqns67 2024-10-13 18:11:48 +08:00
parent db90c5b693
commit 5492aee0d6
Signed by: Puqns67
GPG Key ID: 9669DF042554F536
2 changed files with 99 additions and 63 deletions

View File

@ -1,12 +1,19 @@
from dataclasses import dataclass from dataclasses import dataclass
from http.cookiejar import LoadError, MozillaCookieJar from http.cookiejar import LoadError, MozillaCookieJar
from json import dumps as dumpJson from json import dumps as dumpJson, JSONDecodeError
from typing import Any, Self from typing import Any, Iterable, Self
from httpx import Client as HttpClient from httpx import Client as HttpXClient
from httpx import Request as HttpXRequest
from httpx import Response as HttpXResponse
from .constant import CONFIG_API_DETAIL_TRACK_PER_REQUEST, NCM_API_BASE_URL, PLATFORM from .constant import CONFIG_API_DETAIL_TRACK_PER_REQUEST, NCM_API_BASE_URL, PLATFORM
from .error import NCMApiParseError, NCMApiRequestError, UnsupportedPureMusicTrackError from .error import (
NCMApiResponseParseError,
NCMApiRequestError,
NCMApiRetryLimitExceededError,
UnsupportedPureMusicTrackError,
)
from .lrc import Lrc, LrcType from .lrc import Lrc, LrcType
REQUEST_HEADERS = { REQUEST_HEADERS = {
@ -23,30 +30,37 @@ class NCMTrack:
name: str name: str
artists: list[str] artists: list[str]
def fromApi(data: dict) -> list[Self]: @classmethod
def fromApi(cls, response: HttpXResponse) -> list[Self]:
try:
data: dict = response.json()
except JSONDecodeError:
raise NCMApiResponseParseError("无法以预期的 Json 格式解析响应")
if data.get("code") != 200: if data.get("code") != 200:
raise NCMApiParseError(f"响应码不为 200: {data["code"]}") raise NCMApiResponseParseError(f"响应码不为 200: {data["code"]}")
data = data.get("songs") data = data.get("songs")
if data is None: if data is None:
raise NCMApiParseError("不存在 Track 对应的结构") raise NCMApiResponseParseError("不存在单曲对应的结构")
result: list[NCMTrack] = [] result = []
for track in data: for track in data:
result.append(NCMTrack.fromData(track)) result.append(cls.fromData(track))
return result return result
def fromData(data: dict) -> Self: @classmethod
def fromData(cls, data: dict) -> Self:
try: try:
return NCMTrack( return cls(
id=data["id"], id=data["id"],
name=data["name"], name=data["name"],
artists=[artist["name"] for artist in data["ar"]], artists=[artist["name"] for artist in data["ar"]],
) )
except KeyError as e: except KeyError as e:
raise NCMApiParseError(f"需要的键不存在: {e}") raise NCMApiResponseParseError(f"需要的键不存在: {e}")
def link(self) -> str: def link(self) -> str:
return f"https://music.163.com/song?id={self.id}" return f"https://music.163.com/song?id={self.id}"
@ -58,22 +72,28 @@ class NCMAlbum:
name: str name: str
tracks: list[NCMTrack] tracks: list[NCMTrack]
def fromApi(data: dict) -> Self: @classmethod
def fromApi(cls, response: HttpXResponse) -> Self:
try:
data: dict = response.json()
except JSONDecodeError:
raise NCMApiResponseParseError("无法以预期的 Json 格式解析响应")
if data.get("code") != 200: if data.get("code") != 200:
raise NCMApiParseError(f"响应码不为 200: {data["code"]}") raise NCMApiResponseParseError(f"响应码不为 200: {data["code"]}")
album = data.get("album") album = data.get("album")
if album is None: if album is None:
raise NCMApiParseError("不存在 Album 对应的结构") raise NCMApiResponseParseError("不存在专辑对应的结构")
try: try:
return NCMAlbum( return cls(
id=album["id"], id=album["id"],
name=album["name"], name=album["name"],
tracks=[NCMTrack.fromData(track) for track in data["songs"]], tracks=[NCMTrack.fromData(track) for track in data["songs"]],
) )
except KeyError as e: except KeyError as e:
raise NCMApiParseError(f"需要的键不存在: {e}") raise NCMApiResponseParseError(f"需要的键不存在: {e}")
def link(self) -> str: def link(self) -> str:
return f"https://music.163.com/album?id={self.id}" return f"https://music.163.com/album?id={self.id}"
@ -86,31 +106,37 @@ class NCMPlaylist:
tracks: list[NCMTrack] tracks: list[NCMTrack]
trackIds: list[int] trackIds: list[int]
def fromApi(data: dict) -> Self: @classmethod
def fromApi(cls, response: HttpXResponse) -> Self:
try:
data: dict = response.json()
except JSONDecodeError:
raise NCMApiResponseParseError("无法以预期的 Json 格式解析响应")
if data.get("code") != 200: if data.get("code") != 200:
raise NCMApiParseError(f"响应码不为 200: {data["code"]}") raise NCMApiResponseParseError(f"响应码不为 200: {data["code"]}")
playlist = data.get("playlist") playlist = data.get("playlist")
if playlist is None: if playlist is None:
raise NCMApiParseError("不存在 Playlist 对应的结构") raise NCMApiResponseParseError("不存在歌单对应的结构")
try: try:
tracks: list[NCMTrack] = [] tracks: list[NCMTrack] = []
trackIds: list[int] = [track["id"] for track in playlist["trackIds"]] trackIds: list[int] = [track["id"] for track in playlist["trackIds"]]
for track in playlist["tracks"]: for track in playlist["tracks"]:
parsedTrack: NCMTrack = NCMTrack.fromData(track) parsedTrack = NCMTrack.fromData(track)
trackIds.remove(parsedTrack.id) trackIds.remove(parsedTrack.id)
tracks.append(parsedTrack) tracks.append(parsedTrack)
return NCMPlaylist( return cls(
id=playlist["id"], id=playlist["id"],
name=playlist["name"], name=playlist["name"],
tracks=tracks, tracks=tracks,
trackIds=trackIds, trackIds=trackIds,
) )
except KeyError as e: except KeyError as e:
raise NCMApiParseError(f"需要的键不存在: {e}") raise NCMApiResponseParseError(f"需要的键不存在: {e}")
def link(self) -> str: def link(self) -> str:
return f"https://music.163.com/playlist?id={self.id}" return f"https://music.163.com/playlist?id={self.id}"
@ -122,17 +148,28 @@ class NCMPlaylist:
@dataclass @dataclass
class NCMLyrics: class NCMLyrics:
id: int | None
isPureMusic: bool isPureMusic: bool
data: Any | None data: Any | None
def fromApi(data: dict) -> Self: @classmethod
def fromApi(cls, response: HttpXResponse) -> Self:
try:
data: dict = response.json()
except JSONDecodeError:
raise NCMApiResponseParseError("无法以预期的 Json 格式解析响应")
if data.get("code") != 200: if data.get("code") != 200:
raise NCMApiParseError(f"响应码不为 200: {data["code"]}") raise NCMApiResponseParseError(f"响应码不为 200: {data["code"]}")
if data.get("pureMusic") is True: if data.get("pureMusic") is True:
return NCMLyrics(isPureMusic=True, data=None) return cls(id=None, isPureMusic=True, data=None)
return NCMLyrics(isPureMusic=False, data=data) return cls(id=None, isPureMusic=False, data=data)
def withId(self, id: int) -> Self:
self.id = id
return self
def lrc(self) -> Lrc: def lrc(self) -> Lrc:
if self.isPureMusic: if self.isPureMusic:
@ -153,9 +190,6 @@ class NCMLyrics:
class NCMApi: class NCMApi:
_cookieJar: MozillaCookieJar
_httpClient: HttpClient
def __init__(self, http2: bool = True) -> None: def __init__(self, http2: bool = True) -> None:
self._cookieJar = MozillaCookieJar() self._cookieJar = MozillaCookieJar()
@ -164,25 +198,38 @@ class NCMApi:
except FileNotFoundError | LoadError: except FileNotFoundError | LoadError:
pass pass
self._httpClient = HttpClient( self._httpClient = HttpXClient(
base_url=NCM_API_BASE_URL, base_url=NCM_API_BASE_URL,
cookies=self._cookieJar, cookies=self._cookieJar,
headers=REQUEST_HEADERS, headers=REQUEST_HEADERS,
http2=http2, http2=http2,
) )
def _fetch(self, request: HttpXRequest, retry: int | None = 4) -> HttpXResponse:
if retry: # None => Disable retry
if retry < 0:
retry = 0
while retry < 0:
try:
return self._httpClient.send(request)
except Exception:
retry -= 1
raise NCMApiRetryLimitExceededError
else:
try:
return self._httpClient.send(request)
except Exception:
raise NCMApiRequestError
def saveCookies(self) -> None: def saveCookies(self) -> None:
self._cookieJar.save(str(PLATFORM.user_config_path / "cookies.txt")) self._cookieJar.save(str(PLATFORM.user_config_path / "cookies.txt"))
def getDetailsForTrack(self, trackId: int) -> NCMTrack: def getDetailsForTrack(self, trackId: int) -> NCMTrack:
params = {"c": f"[{{'id':{trackId}}}]"} request = self._httpClient.build_request("GET", "/v3/song/detail", params={"c": f"[{{'id':{trackId}}}]"})
return NCMTrack.fromApi(self._fetch(request)).pop()
try:
response = self._httpClient.request("GET", "/v3/song/detail", params=params)
except BaseException:
raise NCMApiRequestError
return NCMTrack.fromApi(response.json()).pop()
def getDetailsForTracks(self, trackIds: list[int]) -> list[NCMTrack]: def getDetailsForTracks(self, trackIds: list[int]) -> list[NCMTrack]:
result: list[NCMTrack] = [] result: list[NCMTrack] = []
@ -201,34 +248,21 @@ class NCMApi:
) )
} }
try: request = self._httpClient.build_request("GET", "/v3/song/detail", params=params)
response = self._httpClient.request("GET", "/v3/song/detail", params=params)
except BaseException:
raise NCMApiRequestError
result.extend(NCMTrack.fromApi(response.json())) result.extend(NCMTrack.fromApi(self._fetch(request)))
seek += CONFIG_API_DETAIL_TRACK_PER_REQUEST seek += CONFIG_API_DETAIL_TRACK_PER_REQUEST
return result return result
def getDetailsForAlbum(self, albumId: int) -> NCMAlbum: def getDetailsForAlbum(self, albumId: int) -> NCMAlbum:
try: request = self._httpClient.build_request("GET", f"/v1/album/{albumId}")
response = self._httpClient.request("GET", f"/v1/album/{albumId}") return NCMAlbum.fromApi(self._fetch(request))
except BaseException:
raise NCMApiRequestError
return NCMAlbum.fromApi(response.json())
def getDetailsForPlaylist(self, playlistId: int) -> NCMPlaylist: def getDetailsForPlaylist(self, playlistId: int) -> NCMPlaylist:
params = {"id": playlistId} request = self._httpClient.build_request("GET", "/v6/playlist/detail", params={"id": playlistId})
return NCMPlaylist.fromApi(self._fetch(request))
try:
response = self._httpClient.request("GET", "/v6/playlist/detail", params=params)
except BaseException:
raise NCMApiRequestError
return NCMPlaylist.fromApi(response.json())
def getLyricsByTrack(self, trackId: int) -> NCMLyrics: def getLyricsByTrack(self, trackId: int) -> NCMLyrics:
params = { params = {
@ -243,9 +277,7 @@ class NCMApi:
"yrv": 0, "yrv": 0,
} }
try: request = self._httpClient.build_request("GET", "/song/lyric/v1", params=params)
response = self._httpClient.request("GET", "/song/lyric/v1", params=params) return NCMLyrics.fromApi(self._fetch(request)).withId(trackId)
except BaseException:
raise NCMApiRequestError
return NCMLyrics.fromApi(response.json()) return NCMLyrics.fromApi(response.json())

View File

@ -13,10 +13,14 @@ class NCMApiRequestError(NCMApiError, RequestError):
"""请求网易云音乐 API 时出现错误""" """请求网易云音乐 API 时出现错误"""
class NCMApiParseError(NCMApiError): class NCMApiResponseParseError(NCMApiError):
"""解析网易云音乐 API 返回的数据时出现错误""" """解析网易云音乐 API 返回的数据时出现错误"""
class NCMApiRetryLimitExceededError(NCMApiError):
"""请求网易云音乐 API 时错误次数超过重试次数上限"""
class ParseLinkError(NCMLyricsAppError): class ParseLinkError(NCMLyricsAppError):
"""无法解析此分享链接""" """无法解析此分享链接"""