music_api.py

import time
import json
import requests
import urllib.parse
from enum import Enum
from hashlib import md5
from random import randrange
from typing import Dict, List, Optional, Tuple, Any
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"
    SKY = "sky"
    JYEFFECT = "jyeffect"
    JYMASTER = "jymaster"
    DOLBY = "dolby"
class APIConstants:
    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/'
    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'
    REQUEST_TIMEOUT = 5
    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:
        return md5(text.encode("utf-8")).digest()
    @staticmethod
    def hash_hex_digest(text: str) -> str:
        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}"
        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:
    @staticmethod
    def post_request(url: str, params: str, cookies: Dict[str, str]) -> str:
        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=APIConstants.REQUEST_TIMEOUT)
            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:
        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=APIConstants.REQUEST_TIMEOUT)
            response.raise_for_status()
            return response
        except requests.RequestException as e:
            raise APIException(f"HTTP请求失败: {e}")
class APIException(Exception):
    pass
class NeteaseAPI:
    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]:
        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]:
        try:
            data = {'c': json.dumps([{"id": song_id, "v": 0}])}
            response = requests.post(APIConstants.SONG_DETAIL_V3, data=data, timeout=APIConstants.REQUEST_TIMEOUT)
            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]:
        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=APIConstants.REQUEST_TIMEOUT)
            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]]:
        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=APIConstants.REQUEST_TIMEOUT)
            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]:
        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=APIConstants.REQUEST_TIMEOUT)
            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': []
            }
            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=APIConstants.REQUEST_TIMEOUT)
                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]:
        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=APIConstants.REQUEST_TIMEOUT)
            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:
        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:
        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]:
        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]:
        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]]:
        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:
                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]:
        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]:
    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:
    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("- 二维码登录")