Netease_url/music_api.py
2025-08-25 01:23:41 +08:00

674 lines
23 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""网易云音乐API模块
提供网易云音乐相关API接口的封装包括
- 音乐URL获取
- 歌曲详情获取
- 歌词获取
- 搜索功能
- 歌单和专辑详情
- 二维码登录
"""
import json
import urllib.parse
import time
from random import randrange
from typing import Dict, List, Optional, Tuple, Any
from hashlib import md5
from enum import Enum
import requests
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
class QualityLevel(Enum):
"""音质等级枚举"""
STANDARD = "standard" # 标准音质
EXHIGH = "exhigh" # 极高音质
LOSSLESS = "lossless" # 无损音质
HIRES = "hires" # Hi-Res音质
SKY = "sky" # 沉浸环绕声
JYEFFECT = "jyeffect" # 高清环绕声
JYMASTER = "jymaster" # 超清母带
# 常量定义
class APIConstants:
"""API相关常量"""
AES_KEY = b"e82ckenh8dichen8"
USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36 Chrome/91.0.4472.164 NeteaseMusicDesktop/2.10.2.200154'
REFERER = 'https://music.163.com/'
# API URLs
SONG_URL_V1 = "https://interface3.music.163.com/eapi/song/enhance/player/url/v1"
SONG_DETAIL_V3 = "https://interface3.music.163.com/api/v3/song/detail"
LYRIC_API = "https://interface3.music.163.com/api/song/lyric"
SEARCH_API = 'https://music.163.com/api/cloudsearch/pc'
PLAYLIST_DETAIL_API = 'https://music.163.com/api/v6/playlist/detail'
ALBUM_DETAIL_API = 'https://music.163.com/api/v1/album/'
QR_UNIKEY_API = 'https://interface3.music.163.com/eapi/login/qrcode/unikey'
QR_LOGIN_API = 'https://interface3.music.163.com/eapi/login/qrcode/client/login'
# 默认配置
DEFAULT_CONFIG = {
"os": "pc",
"appver": "",
"osver": "",
"deviceId": "pyncm!"
}
DEFAULT_COOKIES = {
"os": "pc",
"appver": "",
"osver": "",
"deviceId": "pyncm!"
}
class CryptoUtils:
"""加密工具类"""
@staticmethod
def hex_digest(data: bytes) -> str:
"""将字节数据转换为十六进制字符串"""
return "".join([hex(d)[2:].zfill(2) for d in data])
@staticmethod
def hash_digest(text: str) -> bytes:
"""计算MD5哈希值"""
return md5(text.encode("utf-8")).digest()
@staticmethod
def hash_hex_digest(text: str) -> str:
"""计算MD5哈希值并转换为十六进制字符串"""
return CryptoUtils.hex_digest(CryptoUtils.hash_digest(text))
@staticmethod
def encrypt_params(url: str, payload: Dict[str, Any]) -> str:
"""加密请求参数"""
url_path = urllib.parse.urlparse(url).path.replace("/eapi/", "/api/")
digest = CryptoUtils.hash_hex_digest(f"nobody{url_path}use{json.dumps(payload)}md5forencrypt")
params = f"{url_path}-36cd479b6b5-{json.dumps(payload)}-36cd479b6b5-{digest}"
# AES加密
padder = padding.PKCS7(algorithms.AES(APIConstants.AES_KEY).block_size).padder()
padded_data = padder.update(params.encode()) + padder.finalize()
cipher = Cipher(algorithms.AES(APIConstants.AES_KEY), modes.ECB())
encryptor = cipher.encryptor()
enc = encryptor.update(padded_data) + encryptor.finalize()
return CryptoUtils.hex_digest(enc)
class HTTPClient:
"""HTTP客户端类"""
@staticmethod
def post_request(url: str, params: str, cookies: Dict[str, str]) -> str:
"""发送POST请求并返回文本响应"""
headers = {
'User-Agent': APIConstants.USER_AGENT,
'Referer': APIConstants.REFERER,
}
request_cookies = APIConstants.DEFAULT_COOKIES.copy()
request_cookies.update(cookies)
try:
response = requests.post(url, headers=headers, cookies=request_cookies,
data={"params": params}, timeout=30)
response.raise_for_status()
return response.text
except requests.RequestException as e:
raise APIException(f"HTTP请求失败: {e}")
@staticmethod
def post_request_full(url: str, params: str, cookies: Dict[str, str]) -> requests.Response:
"""发送POST请求并返回完整响应对象"""
headers = {
'User-Agent': APIConstants.USER_AGENT,
'Referer': APIConstants.REFERER,
}
request_cookies = APIConstants.DEFAULT_COOKIES.copy()
request_cookies.update(cookies)
try:
response = requests.post(url, headers=headers, cookies=request_cookies,
data={"params": params}, timeout=30)
response.raise_for_status()
return response
except requests.RequestException as e:
raise APIException(f"HTTP请求失败: {e}")
class APIException(Exception):
"""API异常类"""
pass
class NeteaseAPI:
"""网易云音乐API主类"""
def __init__(self):
self.http_client = HTTPClient()
self.crypto_utils = CryptoUtils()
def get_song_url(self, song_id: int, quality: str, cookies: Dict[str, str]) -> Dict[str, Any]:
"""获取歌曲播放URL
Args:
song_id: 歌曲ID
quality: 音质等级 (standard, exhigh, lossless, hires, sky, jyeffect, jymaster)
cookies: 用户cookies
Returns:
包含歌曲URL信息的字典
Raises:
APIException: API调用失败时抛出
"""
try:
config = APIConstants.DEFAULT_CONFIG.copy()
config["requestId"] = str(randrange(20000000, 30000000))
payload = {
'ids': [song_id],
'level': quality,
'encodeType': 'flac',
'header': json.dumps(config),
}
if quality == 'sky':
payload['immerseType'] = 'c51'
params = self.crypto_utils.encrypt_params(APIConstants.SONG_URL_V1, payload)
response_text = self.http_client.post_request(APIConstants.SONG_URL_V1, params, cookies)
result = json.loads(response_text)
if result.get('code') != 200:
raise APIException(f"获取歌曲URL失败: {result.get('message', '未知错误')}")
return result
except (json.JSONDecodeError, KeyError) as e:
raise APIException(f"解析响应数据失败: {e}")
def get_song_detail(self, song_id: int) -> Dict[str, Any]:
"""获取歌曲详细信息
Args:
song_id: 歌曲ID
Returns:
包含歌曲详细信息的字典
Raises:
APIException: API调用失败时抛出
"""
try:
data = {'c': json.dumps([{"id": song_id, "v": 0}])}
response = requests.post(APIConstants.SONG_DETAIL_V3, data=data, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('code') != 200:
raise APIException(f"获取歌曲详情失败: {result.get('message', '未知错误')}")
return result
except requests.RequestException as e:
raise APIException(f"获取歌曲详情请求失败: {e}")
except json.JSONDecodeError as e:
raise APIException(f"解析歌曲详情响应失败: {e}")
def get_lyric(self, song_id: int, cookies: Dict[str, str]) -> Dict[str, Any]:
"""获取歌词信息
Args:
song_id: 歌曲ID
cookies: 用户cookies
Returns:
包含歌词信息的字典
Raises:
APIException: API调用失败时抛出
"""
try:
data = {
'id': song_id,
'cp': 'false',
'tv': '0',
'lv': '0',
'rv': '0',
'kv': '0',
'yv': '0',
'ytv': '0',
'yrv': '0'
}
headers = {
'User-Agent': APIConstants.USER_AGENT,
'Referer': APIConstants.REFERER
}
response = requests.post(APIConstants.LYRIC_API, data=data,
headers=headers, cookies=cookies, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('code') != 200:
raise APIException(f"获取歌词失败: {result.get('message', '未知错误')}")
return result
except requests.RequestException as e:
raise APIException(f"获取歌词请求失败: {e}")
except json.JSONDecodeError as e:
raise APIException(f"解析歌词响应失败: {e}")
def search_music(self, keywords: str, cookies: Dict[str, str], limit: int = 10) -> List[Dict[str, Any]]:
"""搜索音乐
Args:
keywords: 搜索关键词
cookies: 用户cookies
limit: 返回数量限制
Returns:
歌曲信息列表
Raises:
APIException: API调用失败时抛出
"""
try:
data = {'s': keywords, 'type': 1, 'limit': limit}
headers = {
'User-Agent': APIConstants.USER_AGENT,
'Referer': APIConstants.REFERER
}
response = requests.post(APIConstants.SEARCH_API, data=data,
headers=headers, cookies=cookies, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('code') != 200:
raise APIException(f"搜索失败: {result.get('message', '未知错误')}")
songs = []
for item in result.get('result', {}).get('songs', []):
song_info = {
'id': item['id'],
'name': item['name'],
'artists': '/'.join(artist['name'] for artist in item['ar']),
'album': item['al']['name'],
'picUrl': item['al']['picUrl']
}
songs.append(song_info)
return songs
except requests.RequestException as e:
raise APIException(f"搜索请求失败: {e}")
except (json.JSONDecodeError, KeyError) as e:
raise APIException(f"解析搜索响应失败: {e}")
def get_playlist_detail(self, playlist_id: int, cookies: Dict[str, str]) -> Dict[str, Any]:
"""获取歌单详情
Args:
playlist_id: 歌单ID
cookies: 用户cookies
Returns:
歌单详情信息
Raises:
APIException: API调用失败时抛出
"""
try:
data = {'id': playlist_id}
headers = {
'User-Agent': APIConstants.USER_AGENT,
'Referer': APIConstants.REFERER
}
response = requests.post(APIConstants.PLAYLIST_DETAIL_API, data=data,
headers=headers, cookies=cookies, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('code') != 200:
raise APIException(f"获取歌单详情失败: {result.get('message', '未知错误')}")
playlist = result.get('playlist', {})
info = {
'id': playlist.get('id'),
'name': playlist.get('name'),
'coverImgUrl': playlist.get('coverImgUrl'),
'creator': playlist.get('creator', {}).get('nickname', ''),
'trackCount': playlist.get('trackCount'),
'description': playlist.get('description', ''),
'tracks': []
}
# 获取所有trackIds并分批获取详细信息
track_ids = [str(t['id']) for t in playlist.get('trackIds', [])]
for i in range(0, len(track_ids), 100):
batch_ids = track_ids[i:i+100]
song_data = {'c': json.dumps([{'id': int(sid), 'v': 0} for sid in batch_ids])}
song_resp = requests.post(APIConstants.SONG_DETAIL_V3, data=song_data,
headers=headers, cookies=cookies, timeout=30)
song_resp.raise_for_status()
song_result = song_resp.json()
for song in song_result.get('songs', []):
info['tracks'].append({
'id': song['id'],
'name': song['name'],
'artists': '/'.join(artist['name'] for artist in song['ar']),
'album': song['al']['name'],
'picUrl': song['al']['picUrl']
})
return info
except requests.RequestException as e:
raise APIException(f"获取歌单详情请求失败: {e}")
except (json.JSONDecodeError, KeyError) as e:
raise APIException(f"解析歌单详情响应失败: {e}")
def get_album_detail(self, album_id: int, cookies: Dict[str, str]) -> Dict[str, Any]:
"""获取专辑详情
Args:
album_id: 专辑ID
cookies: 用户cookies
Returns:
专辑详情信息
Raises:
APIException: API调用失败时抛出
"""
try:
url = f'{APIConstants.ALBUM_DETAIL_API}{album_id}'
headers = {
'User-Agent': APIConstants.USER_AGENT,
'Referer': APIConstants.REFERER
}
response = requests.get(url, headers=headers, cookies=cookies, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('code') != 200:
raise APIException(f"获取专辑详情失败: {result.get('message', '未知错误')}")
album = result.get('album', {})
info = {
'id': album.get('id'),
'name': album.get('name'),
'coverImgUrl': self.get_pic_url(album.get('pic')),
'artist': album.get('artist', {}).get('name', ''),
'publishTime': album.get('publishTime'),
'description': album.get('description', ''),
'songs': []
}
for song in result.get('songs', []):
info['songs'].append({
'id': song['id'],
'name': song['name'],
'artists': '/'.join(artist['name'] for artist in song['ar']),
'album': song['al']['name'],
'picUrl': self.get_pic_url(song['al'].get('pic'))
})
return info
except requests.RequestException as e:
raise APIException(f"获取专辑详情请求失败: {e}")
except (json.JSONDecodeError, KeyError) as e:
raise APIException(f"解析专辑详情响应失败: {e}")
def netease_encrypt_id(self, id_str: str) -> str:
"""网易云加密图片ID算法
Args:
id_str: 图片ID字符串
Returns:
加密后的字符串
"""
import base64
import hashlib
magic = list('3go8&$8*3*3h0k(2)2')
song_id = list(id_str)
for i in range(len(song_id)):
song_id[i] = chr(ord(song_id[i]) ^ ord(magic[i % len(magic)]))
m = ''.join(song_id)
md5_bytes = hashlib.md5(m.encode('utf-8')).digest()
result = base64.b64encode(md5_bytes).decode('utf-8')
result = result.replace('/', '_').replace('+', '-')
return result
def get_pic_url(self, pic_id: Optional[int], size: int = 300) -> str:
"""获取网易云加密歌曲/专辑封面直链
Args:
pic_id: 封面ID
size: 图片尺寸
Returns:
图片URL
"""
if pic_id is None:
return ''
enc_id = self.netease_encrypt_id(str(pic_id))
return f'https://p3.music.126.net/{enc_id}/{pic_id}.jpg?param={size}y{size}'
class QRLoginManager:
"""二维码登录管理器"""
def __init__(self):
self.http_client = HTTPClient()
self.crypto_utils = CryptoUtils()
def generate_qr_key(self) -> Optional[str]:
"""生成二维码的key
Returns:
成功返回unikey失败返回None
Raises:
APIException: API调用失败时抛出
"""
try:
config = APIConstants.DEFAULT_CONFIG.copy()
config["requestId"] = str(randrange(20000000, 30000000))
payload = {
'type': 1,
'header': json.dumps(config)
}
params = self.crypto_utils.encrypt_params(APIConstants.QR_UNIKEY_API, payload)
response = self.http_client.post_request_full(APIConstants.QR_UNIKEY_API, params, {})
result = json.loads(response.text)
if result.get('code') == 200:
return result.get('unikey')
else:
raise APIException(f"生成二维码key失败: {result.get('message', '未知错误')}")
except (json.JSONDecodeError, KeyError) as e:
raise APIException(f"解析二维码key响应失败: {e}")
def create_qr_login(self) -> Optional[str]:
"""创建登录二维码并在控制台显示
Returns:
成功返回unikey失败返回None
"""
try:
import qrcode
unikey = self.generate_qr_key()
if not unikey:
print("生成二维码key失败")
return None
# 创建二维码
qr = qrcode.QRCode()
qr.add_data(f'https://music.163.com/login?codekey={unikey}')
qr.make(fit=True)
# 在控制台显示二维码
qr.print_ascii(tty=True)
print("\n请使用网易云音乐APP扫描上方二维码登录")
return unikey
except ImportError:
print("请安装qrcode库: pip install qrcode")
return None
except Exception as e:
print(f"创建二维码失败: {e}")
return None
def check_qr_login(self, unikey: str) -> Tuple[int, Dict[str, str]]:
"""检查二维码登录状态
Args:
unikey: 二维码key
Returns:
(登录状态码, cookie字典)
Raises:
APIException: API调用失败时抛出
"""
try:
config = APIConstants.DEFAULT_CONFIG.copy()
config["requestId"] = str(randrange(20000000, 30000000))
payload = {
'key': unikey,
'type': 1,
'header': json.dumps(config)
}
params = self.crypto_utils.encrypt_params(APIConstants.QR_LOGIN_API, payload)
response = self.http_client.post_request_full(APIConstants.QR_LOGIN_API, params, {})
result = json.loads(response.text)
cookie_dict = {}
if result.get('code') == 803:
# 登录成功提取cookie
all_cookies = response.headers.get('Set-Cookie', '').split(', ')
for cookie_str in all_cookies:
if 'MUSIC_U=' in cookie_str:
cookie_dict['MUSIC_U'] = cookie_str.split('MUSIC_U=')[1].split(';')[0]
return result.get('code', -1), cookie_dict
except (json.JSONDecodeError, KeyError) as e:
raise APIException(f"解析登录状态响应失败: {e}")
def qr_login(self) -> Optional[str]:
"""完整的二维码登录流程
Returns:
成功返回cookie字符串失败返回None
"""
try:
unikey = self.create_qr_login()
if not unikey:
return None
while True:
code, cookies = self.check_qr_login(unikey)
if code == 803:
print("\n登录成功!")
return f"MUSIC_U={cookies['MUSIC_U']};os=pc;appver=8.9.70;"
elif code == 801:
print("\r等待扫码...", end='')
elif code == 802:
print("\r扫码成功,请在手机上确认登录...", end='')
else:
print(f"\n登录失败,错误码:{code}")
return None
time.sleep(2)
except KeyboardInterrupt:
print("\n用户取消登录")
return None
except Exception as e:
print(f"\n登录过程中发生错误: {e}")
return None
# 向后兼容的函数接口
def url_v1(song_id: int, level: str, cookies: Dict[str, str]) -> Dict[str, Any]:
"""获取歌曲URL向后兼容"""
api = NeteaseAPI()
return api.get_song_url(song_id, level, cookies)
def name_v1(song_id: int) -> Dict[str, Any]:
"""获取歌曲详情(向后兼容)"""
api = NeteaseAPI()
return api.get_song_detail(song_id)
def lyric_v1(song_id: int, cookies: Dict[str, str]) -> Dict[str, Any]:
"""获取歌词(向后兼容)"""
api = NeteaseAPI()
return api.get_lyric(song_id, cookies)
def search_music(keywords: str, cookies: Dict[str, str], limit: int = 10) -> List[Dict[str, Any]]:
"""搜索音乐(向后兼容)"""
api = NeteaseAPI()
return api.search_music(keywords, cookies, limit)
def playlist_detail(playlist_id: int, cookies: Dict[str, str]) -> Dict[str, Any]:
"""获取歌单详情(向后兼容)"""
api = NeteaseAPI()
return api.get_playlist_detail(playlist_id, cookies)
def album_detail(album_id: int, cookies: Dict[str, str]) -> Dict[str, Any]:
"""获取专辑详情(向后兼容)"""
api = NeteaseAPI()
return api.get_album_detail(album_id, cookies)
def get_pic_url(pic_id: Optional[int], size: int = 300) -> str:
"""获取图片URL向后兼容"""
api = NeteaseAPI()
return api.get_pic_url(pic_id, size)
def qr_login() -> Optional[str]:
"""二维码登录(向后兼容)"""
manager = QRLoginManager()
return manager.qr_login()
if __name__ == "__main__":
# 测试代码
print("网易云音乐API模块")
print("支持的功能:")
print("- 歌曲URL获取")
print("- 歌曲详情获取")
print("- 歌词获取")
print("- 音乐搜索")
print("- 歌单详情")
print("- 专辑详情")
print("- 二维码登录")