225 lines
9.3 KiB
Python
225 lines
9.3 KiB
Python
"""Resolve NetEase song IDs to short-lived playable audio URLs."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
import urllib.parse
|
||
from typing import Any
|
||
|
||
import aiohttp
|
||
|
||
|
||
class NeteaseResolver:
|
||
def __init__(
|
||
self,
|
||
logger: logging.Logger,
|
||
*,
|
||
api_base: str = "https://music.163.com",
|
||
music_u: str = "",
|
||
):
|
||
self.logger = logger
|
||
self.api_base = api_base.rstrip("/")
|
||
self.music_u = str(music_u or "").strip()
|
||
self.last_error_code = ""
|
||
|
||
def update_api_base(self, api_base: str):
|
||
self.api_base = str(api_base or "https://music.163.com").rstrip("/")
|
||
|
||
def update_auth(self, music_u: str):
|
||
self.music_u = str(music_u or "").strip()
|
||
|
||
def _headers(self) -> dict[str, str]:
|
||
cookie = "os=pc; appver=2.9.8;"
|
||
if self.music_u:
|
||
cookie += f" MUSIC_U={self.music_u};"
|
||
return {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||
"Referer": "https://music.163.com/",
|
||
"Cookie": cookie,
|
||
"X-Real-IP": "218.75.111.114",
|
||
"X-Forwarded-For": "218.75.111.114",
|
||
}
|
||
|
||
async def _probe_url(self, url: str) -> str:
|
||
timeout = aiohttp.ClientTimeout(total=12)
|
||
async with aiohttp.ClientSession(headers=self._headers(), timeout=timeout) as session:
|
||
try:
|
||
async with session.get(url, allow_redirects=True, headers={"Range": "bytes=0-1"}) as resp:
|
||
content_type = str(resp.headers.get("Content-Type", "")).lower()
|
||
if resp.status in (200, 206) and ("audio" in content_type or "octet-stream" in content_type):
|
||
return str(resp.url)
|
||
except Exception:
|
||
return ""
|
||
return ""
|
||
|
||
async def _get_json(self, url: str) -> dict[str, Any] | None:
|
||
timeout = aiohttp.ClientTimeout(total=12)
|
||
try:
|
||
async with aiohttp.ClientSession(headers=self._headers(), timeout=timeout) as session:
|
||
async with session.get(url, allow_redirects=True) as resp:
|
||
if resp.status != 200:
|
||
return None
|
||
payload = await resp.json(content_type=None)
|
||
return payload if isinstance(payload, dict) else None
|
||
except Exception:
|
||
return None
|
||
|
||
async def account_status(self) -> dict[str, Any]:
|
||
if not self.music_u:
|
||
return {"authenticated": False}
|
||
payload = await self._get_json(f"{self.api_base}/api/nuser/account/get")
|
||
if not payload:
|
||
return {"authenticated": False}
|
||
account = payload.get("account")
|
||
profile = payload.get("profile")
|
||
if not isinstance(account, dict) or not account.get("id"):
|
||
return {"authenticated": False}
|
||
profile = profile if isinstance(profile, dict) else {}
|
||
return {
|
||
"authenticated": True,
|
||
"user_id": str(account.get("id") or ""),
|
||
"nickname": str(profile.get("nickname") or "网易云用户"),
|
||
"vip_type": int(account.get("vipType") or profile.get("vipType") or 0),
|
||
}
|
||
|
||
async def _fetch_player_entry(self, song_id: str) -> dict[str, Any] | None:
|
||
url = (
|
||
f"{self.api_base}/api/song/enhance/player/url"
|
||
f"?ids=%5B{urllib.parse.quote(song_id)}%5D&br=320000"
|
||
)
|
||
payload = await self._get_json(url)
|
||
entries = payload.get("data") if payload else None
|
||
if not isinstance(entries, list) or not entries or not isinstance(entries[0], dict):
|
||
return None
|
||
return entries[0]
|
||
|
||
@staticmethod
|
||
def _is_trial_entry(entry: dict[str, Any]) -> bool:
|
||
if entry.get("freeTrialInfo"):
|
||
return True
|
||
privilege = entry.get("freeTrialPrivilege")
|
||
if not isinstance(privilege, dict):
|
||
return False
|
||
return any(bool(privilege.get(key)) for key in ("resConsumable", "userConsumable", "listenType"))
|
||
|
||
@staticmethod
|
||
def playlist_id(value: str | int) -> str:
|
||
match = re.search(r"(?:playlist\?id=|\bid=)?(\d{5,})", str(value or ""))
|
||
return match.group(1) if match else ""
|
||
|
||
async def fetch_playlist(self, playlist: str | int) -> dict[str, Any] | None:
|
||
playlist_id = self.playlist_id(playlist)
|
||
if not playlist_id:
|
||
return None
|
||
timeout = aiohttp.ClientTimeout(total=15)
|
||
url = f"{self.api_base}/api/playlist/detail?id={urllib.parse.quote(playlist_id)}"
|
||
try:
|
||
async with aiohttp.ClientSession(headers=self._headers(), timeout=timeout) as session:
|
||
async with session.get(url, allow_redirects=True) as resp:
|
||
if resp.status != 200:
|
||
return None
|
||
payload = await resp.json(content_type=None)
|
||
except Exception as exc:
|
||
self.logger.warning(f"[mpv] 获取网易云歌单失败: {exc}")
|
||
return None
|
||
result = payload.get("result") if isinstance(payload, dict) else None
|
||
if not isinstance(result, dict):
|
||
return None
|
||
tracks = result.get("tracks")
|
||
if not isinstance(tracks, list):
|
||
return None
|
||
songs: list[dict[str, Any]] = []
|
||
for track in tracks:
|
||
if not isinstance(track, dict):
|
||
continue
|
||
song_id = self.playlist_id(track.get("id", ""))
|
||
if not song_id:
|
||
continue
|
||
artists_raw = track.get("artists") or track.get("ar") or []
|
||
artists = "/".join(
|
||
str(item.get("name") or "").strip()
|
||
for item in artists_raw
|
||
if isinstance(item, dict) and str(item.get("name") or "").strip()
|
||
)
|
||
album = track.get("album") or track.get("al") or {}
|
||
duration_ms = track.get("duration") or track.get("dt") or 0
|
||
try:
|
||
duration_sec = max(0, int(duration_ms) // 1000)
|
||
except (TypeError, ValueError):
|
||
duration_sec = 0
|
||
songs.append({
|
||
"id": song_id,
|
||
"name": str(track.get("name") or f"歌曲{song_id}"),
|
||
"artist": artists or "未知歌手",
|
||
"duration_sec": duration_sec,
|
||
"cover": str(album.get("picUrl") or "") if isinstance(album, dict) else "",
|
||
"source": "background_playlist",
|
||
"playlist_id": playlist_id,
|
||
})
|
||
if not songs:
|
||
return None
|
||
return {
|
||
"id": playlist_id,
|
||
"name": str(result.get("name") or f"歌单{playlist_id}"),
|
||
"songs": songs,
|
||
}
|
||
|
||
async def fetch_song_detail(self, song_id: str | int) -> dict[str, Any] | None:
|
||
clean_id = self.playlist_id(song_id)
|
||
if not clean_id:
|
||
return None
|
||
timeout = aiohttp.ClientTimeout(total=12)
|
||
url = f"{self.api_base}/api/song/detail/?id={clean_id}&ids=[{clean_id}]"
|
||
try:
|
||
async with aiohttp.ClientSession(headers=self._headers(), timeout=timeout) as session:
|
||
async with session.get(url, allow_redirects=True) as resp:
|
||
payload = await resp.json(content_type=None)
|
||
except Exception:
|
||
return None
|
||
songs = payload.get("songs") if isinstance(payload, dict) else None
|
||
if not isinstance(songs, list) or not songs or not isinstance(songs[0], dict):
|
||
return None
|
||
track = songs[0]
|
||
album = track.get("album") or track.get("al") or {}
|
||
return {
|
||
"cover": str(album.get("picUrl") or album.get("blurPicUrl") or "") if isinstance(album, dict) else "",
|
||
}
|
||
|
||
async def resolve(self, song: dict[str, Any]) -> dict[str, Any] | None:
|
||
self.last_error_code = ""
|
||
song_id = "".join(ch for ch in str(song.get("id", "")) if ch.isdigit())
|
||
if not song_id:
|
||
self.last_error_code = "invalid_song_id"
|
||
return None
|
||
entry = await self._fetch_player_entry(song_id)
|
||
if entry:
|
||
if self._is_trial_entry(entry):
|
||
reason = "登录已失效或账号没有完整播放权益" if self.music_u else "未配置网易云登录"
|
||
self.logger.warning(
|
||
f"[mpv] 拒绝播放试听片段: {song.get('name')} ({song_id})，{reason}"
|
||
)
|
||
self.last_error_code = "preview_only"
|
||
return None
|
||
player_url = str(entry.get("url") or "").strip()
|
||
if player_url:
|
||
resolved = await self._probe_url(player_url)
|
||
if resolved:
|
||
return {
|
||
"url": resolved,
|
||
"source": "netease.player.auth" if self.music_u else "netease.player",
|
||
"song_id": song_id,
|
||
}
|
||
# NetEase's public outer URL provides a short-lived CDN redirect for songs available to the current region/account tier.
|
||
outer = f"{self.api_base}/song/media/outer/url?id={urllib.parse.quote(song_id)}.mp3"
|
||
resolved = await self._probe_url(outer)
|
||
if not resolved:
|
||
self.last_error_code = "unavailable"
|
||
self.logger.warning(f"[mpv] 无法获取可播放地址: {song.get('name')} ({song_id})")
|
||
return None
|
||
return {
|
||
"url": resolved,
|
||
"source": "netease.outer",
|
||
"song_id": song_id,
|
||
}
|