NeteaseUrl/music_api/api.py

# -*- coding: utf-8 -*-
"""网易云 API 主类。

（自 music_api.py 拆出，行为与原实现一致。）
"""

import json
import time
from collections.abc import Callable
from random import randrange
from typing import Any

from core.http_client import UpstreamError
from core.http_client import client as _http

from .client import (
    HTTPClient,
    _as_dict,
    _ensure_dict,
)
from .constants import (
    APIConstants,
)
from .crypto import (
    CryptoUtils,
    WeapiCrypto,
)
from .errors import (
    APIException,
)


def _song_base_fields(item: dict[str, Any]) -> dict[str, Any]:
    """歌曲条目公共字段（搜索四路与歌单详情共用同一容错口径）。"""
    return {
        'id': item['id'],
        'name': item.get('name') or '未知歌曲',
        'artists': '/'.join(filter(None, (a.get('name') for a in item.get('ar') or []))) or '未知歌手',
        'album': (item.get('al') or {}).get('name') or '未知专辑',
        'picUrl': (item.get('al') or {}).get('picUrl') or ''
    }


def _map_song_item(item: dict[str, Any]) -> dict[str, Any]:
    """歌曲条目映射。"""
    return {
        'type': 'song',
        **_song_base_fields(item)
    }


def _playlist_track_ids(playlist: dict[str, Any]) -> list[str]:
    """trackIds → 字符串 ID 列表（缺 id 的条目跳过）。"""
    return [str(t['id']) for t in (playlist.get('trackIds') or []) if t.get('id')]


async def _collect_playlist_tracks(track_ids: list[str], headers: dict[str, str],
                                   cookies: dict[str, str]) -> tuple[list[dict[str, Any]], int]:
    """分批拉取歌单全部曲目明细，返回 (tracks, failed_batches)。

    明细批次的业务 code 也要校验，失败批次记入 failed_batches，不再静默缺歌。
    脏 ID 跳过整批（'①'.isdigit() 为真但 int() 抛 ValueError——原实现此处
    未捕获会整歌单 500，现为跳批）；明细缺 id 的歌跳过不拖整批。
    """
    tracks: list[dict[str, Any]] = []
    failed_batches = 0
    for i in range(0, len(track_ids), 100):
        batch_ids = track_ids[i:i + 100]
        try:
            song_data = {'c': json.dumps([{'id': int(sid), 'v': 0} for sid in batch_ids
                                          if str(sid).isdigit()])}
        except (TypeError, ValueError):
            continue

        song_resp = await _http.request('POST', APIConstants.SONG_DETAIL_V3, data=song_data,
                                        headers=headers, cookies=cookies, timeout=30)

        song_result = song_resp.json()
        if not isinstance(song_result, dict):
            # 上游 null/数组体按失败批次计，不让 AttributeError 失真
            failed_batches += 1
            continue
        if song_result.get('code') != 200:
            # 本批次业务失败：计为失败批次，交付方可见残缺状态
            failed_batches += 1
            continue
        for song in (song_result.get('songs') or []):
            if not song.get('id'):
                continue
            tracks.append(_song_base_fields(song))
    return tracks, failed_batches


def _map_album_item(item: dict[str, Any]) -> dict[str, Any]:
    """专辑条目映射（artists 为空**不**兜底成"未知歌手"——与歌曲路不同口径）。"""
    return {
        'type': 'album',
        'id': item['id'],
        'name': item.get('name') or '未知名称',
        'artists': '/'.join(filter(None, (a.get('name') for a in item.get('artists') or []))),
        'alias': '',
        'album': '',
        'picUrl': item.get('picUrl')
    }


def _map_artist_item(item: dict[str, Any]) -> dict[str, Any]:
    """歌手条目映射（alias 缺失/为空 → ''）。"""
    return {
        'type': 'artist',
        'id': item['id'],
        'name': item.get('name') or '未知名称',
        'artists': '',
        'alias': '/'.join(item.get('alias', [])) if item.get('alias') else '',
        'album': '',
        'picUrl': item.get('picUrl')
    }


def _map_playlist_item(item: dict[str, Any]) -> dict[str, Any]:
    """歌单条目映射。"""
    return {
        'type': 'playlist',
        'id': item['id'],
        'name': item.get('name') or '未知名称',
        'artists': '',
        'creator': (item.get('creator') or {}).get('nickname', ''),
        'trackCount': item.get('trackCount', 0),
        'album': '',
        'picUrl': item.get('coverImgUrl')
    }


# search_type → (上游结果键, 条目映射)；未知类型不在表内 → 空结果
_SEARCH_RESULT_KEYS: dict[int, str] = {
    1: 'songs', 10: 'albums', 100: 'artists', 1000: 'playlists',
}
_SEARCH_ITEM_MAPPERS: dict[int, Callable[[dict[str, Any]], dict[str, Any]]] = {
    1: _map_song_item, 10: _map_album_item, 100: _map_artist_item, 1000: _map_playlist_item,
}


def _collect_search_items(result_data: dict[str, Any], search_type: int) -> list[dict[str, Any]]:
    """按类型取上游结果键并逐条映射；残缺条目（无 id，下架/屏蔽）跳过，不拖死整页。"""
    key = _SEARCH_RESULT_KEYS.get(search_type)
    mapper = _SEARCH_ITEM_MAPPERS.get(search_type)
    if key is None or mapper is None:
        return []
    items: list[dict[str, Any]] = []
    for item in result_data.get(key, []):
        if not item.get('id'):
            continue
        items.append(mapper(item))
    return items


class NeteaseAPI:
    """网易云音乐API主类"""

    def __init__(self):
        self.http_client = HTTPClient()
        self.crypto_utils = CryptoUtils()

    async 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': 'mp4' if quality == 'dolby' else 'flac',
                'header': json.dumps(config),
            }

            if quality == 'sky':
                payload['immerseType'] = 'c51'

            params = self.crypto_utils.encrypt_params(APIConstants.SONG_URL_V1, payload)
            response_text = await self.http_client.post_request(APIConstants.SONG_URL_V1, params, cookies)

            result = json.loads(response_text)
            # 合法 JSON 但非对象（null/数组）时给明确错误，不让 AttributeError 失真
            if not isinstance(result, dict):
                raise APIException("获取歌曲URL失败: 响应格式异常")
            if result.get('code') != 200:
                raise APIException(f"获取歌曲URL失败: {result.get('message', '未知错误')}")

            return result
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析响应数据失败: {e}") from e

    async 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}])}
            # 补齐 UA/Referer——本方法是全站唯一裸 UA 请求，风控/460 单点偏高
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }
            response = await _http.request('POST', APIConstants.SONG_DETAIL_V3, data=data,
                                     headers=headers, timeout=30)

            result = response.json()
            # 合法 JSON 但非对象（null/数组）时给明确错误，不让 AttributeError 失真
            if not isinstance(result, dict):
                raise APIException("获取歌曲详情失败: 响应格式异常")
            if result.get('code') != 200:
                raise APIException(f"获取歌曲详情失败: {result.get('message', '未知错误')}")

            return result
        except UpstreamError as e:
            raise APIException(f"获取歌曲详情请求失败: {e}") from e
        except json.JSONDecodeError as e:
            raise APIException(f"解析歌曲详情响应失败: {e}") from e

    async 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': '1',
                'ytv': '0',
                'yrv': '0'
            }

            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }

            response = await _http.request('POST', APIConstants.LYRIC_API, data=data,
                                   headers=headers, cookies=cookies, timeout=30)

            result = response.json()
            _ensure_dict(result, 'get_lyric 接口响应异常（非标准 JSON 对象）')
            if result.get('code') != 200:
                raise APIException(f"获取歌词失败: {result.get('message', '未知错误')}")

            return result
        except UpstreamError as e:
            raise APIException(f"获取歌词请求失败: {e}") from e
        except json.JSONDecodeError as e:
            raise APIException(f"解析歌词响应失败: {e}") from e

    async def search_music(self, keywords: str, cookies: dict[str, str], limit: int = 10,
                     search_type: int = 1, offset: int = 0) -> list[dict[str, Any]]:
        """搜索音乐

        Args:
            keywords: 搜索关键词
            cookies: 用户cookies
            limit: 返回数量限制
            search_type: 搜索类型 (1-歌曲, 10-专辑, 100-歌手, 1000-歌单)
            offset: 分页偏移（透传上游，翻页生效）

        Returns:
            搜索结果列表

        Raises:
            APIException: API调用失败时抛出
        """
        try:
            data = {'s': keywords, 'type': search_type, 'limit': limit,
                    'offset': max(0, int(offset or 0))}
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }

            response = await _http.request('POST', APIConstants.SEARCH_API, data=data,
                                   headers=headers, cookies=cookies, timeout=30)

            result = response.json()
            # 非 dict 响应（null/数组）给明确错误
            if not isinstance(result, dict):
                raise APIException("搜索失败: 响应格式异常")
            if result.get('code') != 200:
                raise APIException(f"搜索失败: {result.get('message', '未知错误')}")

            result_data = result.get('result') or {}
            return _collect_search_items(result_data, search_type)
        except UpstreamError as e:
            raise APIException(f"搜索请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析搜索响应失败: {e}") from e

    async def get_playlist_detail(self, playlist_id: int, cookies: dict[str, str]) -> dict[str, Any]:
        """获取歌单详情

        Args:
            playlist_id: 歌单ID
            cookies: 用户cookies

        Returns:
            歌单详情信息（曲目数与 trackIds 不一致时附 incomplete/missing/failed_batches）

        Raises:
            APIException: API调用失败时抛出
        """
        try:
            data = {'id': playlist_id}
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }

            response = await _http.request('POST', APIConstants.PLAYLIST_DETAIL_API, data=data,
                                   headers=headers, cookies=cookies, timeout=30)

            result = response.json()
            # 非 dict 响应（null/数组）给明确错误
            if not isinstance(result, dict):
                raise APIException("获取歌单详情失败: 响应格式异常")
            if result.get('code') != 200:
                raise APIException(f"获取歌单详情失败: {result.get('message', '未知错误')}")

            # 上游对不存在/私密/无权限歌单返回 code:200 + playlist:null ——
            # 显式转为 APIException（四条路由统一转"歌单不存在"语义，不再 500）
            playlist = result.get('playlist') or {}
            if not playlist:
                raise APIException("歌单不存在或无权访问")
            info = {
                'id': playlist.get('id'),
                'name': playlist.get('name'),
                'coverImgUrl': playlist.get('coverImgUrl'),
                'creator': (playlist.get('creator') or {}).get('nickname', ''),
                'trackCount': playlist.get('trackCount'),
                'description': playlist.get('description', ''),
                'tracks': []
            }

            track_ids = _playlist_track_ids(playlist)
            tracks, failed_batches = await _collect_playlist_tracks(track_ids, headers, cookies)
            info['tracks'] = tracks
            # trackIds 数与实际拿到的不一致时显式标记"不完整"，不冒充完整结果
            if len(tracks) != len(track_ids):
                info['incomplete'] = True
                info['missing'] = len(track_ids) - len(tracks)
                info['failed_batches'] = failed_batches

            return info
        except UpstreamError as e:
            raise APIException(f"获取歌单详情请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析歌单详情响应失败: {e}") from e

    async 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 = await _http.request('GET', url, headers=headers, cookies=cookies, timeout=30)

            result = response.json()
            _ensure_dict(result, 'get_album_detail 接口响应异常（非标准 JSON 对象）')
            if result.get('code') != 200:
                raise APIException(f"获取专辑详情失败: {result.get('message', '未知错误')}")

            album = result.get('album') or {}
            info = {
                'id': album.get('id'),
                'name': album.get('name'),
                'coverImgUrl': self.get_pic_url(album.get('pic')),
                'artist': (album.get('artist') or {}).get('name', ''),
                'publishTime': album.get('publishTime'),
                'description': album.get('description', ''),
                'songs': []
            }

            for song in result.get('songs', []):
                if not song.get('id'):   # 残缺条目跳过
                    continue
                info['songs'].append({
                    'id': song['id'],
                    'name': song.get('name') or '未知歌曲',
                    'artists': '/'.join(filter(None, (a.get('name') for a in song.get('ar') or []))) or '未知歌手',
                    'album': (song.get('al') or {}).get('name') or '未知专辑',
                    'picUrl': self.get_pic_url((song.get('al') or {}).get('pic'))
                })

            return info
        except UpstreamError as e:
            raise APIException(f"获取专辑详情请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析专辑详情响应失败: {e}") from e

    async def get_toplist(self) -> list[dict[str, Any]]:
        """获取网易云官方榜单列表

        Returns:
            榜单列表信息

        Raises:
            APIException: API调用失败时抛出
        """
        try:
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }

            response = await _http.request('GET', APIConstants.TOPLIST_API, headers=headers, timeout=30)

            result = response.json()
            _ensure_dict(result, 'get_toplist 接口响应异常（非标准 JSON 对象）')
            if result.get('code') != 200:
                raise APIException(f"获取榜单失败: {result.get('message', '未知错误')}")

            toplists = []
            for item in (result.get('list') or []):
                toplists.append({
                    'id': item.get('id'),
                    'name': item.get('name'),
                    'coverImgUrl': item.get('coverImgUrl'),
                    'description': item.get('description', ''),
                    'updateFrequency': item.get('updateFrequency', ''),
                    'playCount': item.get('playCount', 0),
                    'trackCount': item.get('trackCount', 0),
                    'topTitle': item.get('ToplistType', '')
                })
            return toplists
        except UpstreamError as e:
            raise APIException(f"获取榜单请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析榜单响应失败: {e}") from e

    async def get_artist_hot_songs(self, artist_id: int) -> dict[str, Any]:
        """获取歌手信息和热门歌曲

        Args:
            artist_id: 歌手ID

        Returns:
            歌手信息和热门歌曲列表

        Raises:
            APIException: API调用失败时抛出
        """
        try:
            url = f'{APIConstants.ARTIST_API}{artist_id}'
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }

            response = await _http.request('GET', url, headers=headers, timeout=30)

            result = response.json()
            _ensure_dict(result, 'get_artist_hot_songs 接口响应异常（非标准 JSON 对象）')
            if result.get('code') != 200:
                raise APIException(f"获取歌手信息失败: {result.get('message', '未知错误')}")

            artist = result.get('artist') or {}
            songs = []
            for song in (result.get('hotSongs') or []):
                songs.append({
                    'id': song.get('id'),
                    'name': song.get('name'),
                    'artists': '/'.join(a.get('name', '') for a in (song.get('ar') or [])),
                    'album': ((song.get('al') or {}).get('name', '')),
                    'picUrl': (song.get('al') or {}).get('picUrl', '')
                })

            return {
                'artist': {
                    'id': artist.get('id'),
                    'name': artist.get('name'),
                    'picUrl': artist.get('picUrl'),
                    'alias': '/'.join(artist.get('alias', [])) if artist.get('alias') else '',
                    'briefDesc': artist.get('briefDesc', ''),
                    'musicSize': artist.get('musicSize', len(songs))
                },
                'songs': songs
            }
        except UpstreamError as e:
            raise APIException(f"获取歌手信息请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析歌手信息响应失败: {e}") from e

    async def get_song_comments(self, song_id: int, limit: int = 20, offset: int = 0) -> dict[str, Any]:
        """获取歌曲评论

        Args:
            song_id: 歌曲ID
            limit: 每页数量
            offset: 偏移量

        Returns:
            评论数据（热门评论 + 全部评论）

        Raises:
            APIException: API调用失败时抛出
        """
        try:
            url = f'{APIConstants.COMMENT_API}{song_id}?limit={limit}&offset={offset}'
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }

            response = await _http.request('GET', url, headers=headers, timeout=30)

            result = response.json()
            _ensure_dict(result, 'get_song_comments 接口响应异常（非标准 JSON 对象）')
            if result.get('code') != 200:
                raise APIException(f"获取评论失败: {result.get('message', '未知错误')}")

            def _comment_dict(c):
                c = c or {}
                user = c.get('user') or {}
                be_replied = c.get('beReplied') or []
                replied = (be_replied[0] or {}) if be_replied else {}
                replied_user = replied.get('user') or {}
                vip_rights = _as_dict(user.get('vipRights'))
                associator = _as_dict(vip_rights.get('associator'))
                location_info = _as_dict(user.get('locationInfo'))
                return {
                    'id': c.get('commentId'),
                    'content': c.get('content', ''),
                    'time': c.get('time', 0),
                    'likedCount': c.get('likedCount', 0),
                    'user': {
                        'nickname': user.get('nickname', ''),
                        'avatarUrl': user.get('avatarUrl', ''),
                        'vipType': user.get('vipType', 0),
                        'vipLevel': associator.get('level', 0),
                        'location': location_info.get('location', '')
                    },
                    'beReplied': [{
                        'nickname': replied_user.get('nickname', '') or '',
                        'content': replied.get('content') or ''
                    }]
                }

            return {
                'total': result.get('total', 0),
                'hasMore': result.get('hasMore', False),
                'hotComments': [_comment_dict(c) for c in (result.get('hotComments') or [])],
                'comments': [_comment_dict(c) for c in (result.get('comments') or [])],
            }
        except UpstreamError as e:
            raise APIException(f"获取评论请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析评论响应失败: {e}") from e

    async def get_account_info(self, cookies: dict[str, str]) -> dict[str, Any] | None:
        """获取当前账号信息（用于个人登录状态展示）

        Args:
            cookies: 用户cookies

        Returns:
            账号信息字典，获取失败返回 None
        """
        try:
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }

            response = await _http.request('GET', APIConstants.USER_ACCOUNT_API, headers=headers,
                                    cookies=cookies, timeout=30)

            result = response.json()
            _ensure_dict(result, 'get_account_info 接口响应异常（非标准 JSON 对象）')
            if result.get('code') != 200:
                return None

            profile = result.get('profile') or {}
            # 无 userId 说明 Cookie 未关联有效登录态（可能已过期）
            if not profile.get('userId'):
                return None

            return {
                'userId': profile.get('userId'),
                'nickname': profile.get('nickname', ''),
                'avatarUrl': profile.get('avatarUrl', ''),
                'signature': profile.get('signature', ''),
                'vipType': profile.get('vipType', 0),
                'vipLevel': profile.get('vipLevel', 0)
            }
        except Exception:
            return None

    async def get_user_playlists(self, user_id: int, cookies: dict[str, str]) -> list[dict[str, Any]]:
        """获取用户收藏/创建的歌单列表

        Args:
            user_id: 用户ID
            cookies: 用户cookies

        Returns:
            歌单列表（含「我喜欢的音乐」等个人歌单）

        Raises:
            APIException: API调用失败时抛出

        超过 100 个歌单时按上游 more 标志续页（上限 20 页 = 2000 个），
        不再只取第一页就静默截断。
        """
        try:
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }

            playlists = []
            offset = 0
            page_size = 100
            for _ in range(20):   # 有界翻页，防上游 more 异常导致无限循环
                params = {'uid': user_id, 'limit': page_size, 'offset': offset}
                response = await _http.request('GET', APIConstants.USER_PLAYLIST_API, params=params,
                                        headers=headers, cookies=cookies, timeout=30)

                result = response.json()
                _ensure_dict(result, 'get_user_playlists 接口响应异常（非标准 JSON 对象）')
                if result.get('code') != 200:
                    raise APIException(f"获取用户歌单失败: {result.get('message', '未知错误')}")

                page_items = result.get('playlist') or []
                seen_ids = {p.get('id') for p in playlists}
                for item in page_items:
                    if item.get('id') in seen_ids:
                        continue   # 上游列表变动导致重复时跳过
                    creator = item.get('creator') or {}
                    playlists.append({
                        'id': item.get('id'),
                        'name': item.get('name', ''),
                        'coverImgUrl': item.get('coverImgUrl'),
                        'trackCount': item.get('trackCount', 0),
                        'playCount': item.get('playCount', 0),
                        'creator': creator.get('nickname', ''),
                        'is_own': creator.get('userId') == user_id,
                        'specialType': item.get('specialType', 0),
                        'subscribed': item.get('subscribed', False),
                        'description': item.get('description', '')
                    })
                more = bool(result.get('more'))
                if not more or len(page_items) < page_size:
                    break
                offset += page_size
            return playlists
        except UpstreamError as e:
            raise APIException(f"获取用户歌单请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析用户歌单响应失败: {e}") from e

    async def detect_vip_by_probe(self, cookies: dict[str, str],
                            probe_song_id: int = 347230) -> dict[str, Any]:
        """通过歌曲URL接口探测账号可用的最高音质，推断会员类型

        直接询问网易云接口「这个账号能把歌解析到什么音质」，
        比 account/get 更可靠（旧 Cookie 可能查不到 profile）。

        Args:
            cookies: 用户cookies
            probe_song_id: 用于探测的歌曲ID（默认《海阔天空》，支持全音质）

        Returns:
            {'vip_type': int, 'highest_level': str, 'levels_ok': List[str]}
        """
        levels_ok = await self.probe_levels(cookies, probe_song_id)
        highest = levels_ok[0] if levels_ok else 'standard'

        if highest in ('jymaster', 'jyeffect', 'sky'):
            vip_type = 110  # 黑胶SVIP（可解析沉浸/母带级）
        elif highest in ('hires', 'lossless'):
            vip_type = 11   # 黑胶VIP
        else:
            vip_type = 0    # 普通用户
        return {'vip_type': vip_type, 'highest_level': highest, 'levels_ok': levels_ok}

    async def probe_levels(self, cookies: dict[str, str],
                     probe_song_id: int = 347230) -> list[str]:
        """实测账号对每个音质级别能否真正解析

        只有接口返回的 level 与请求一致且带 URL，才算该音质可解析。
        返回按音质从高到低排序的可解析列表。

        Args:
            cookies: 用户cookies
            probe_song_id: 探测歌曲ID

        Returns:
            可解析音质列表（从高到低）
        """
        ladder = ['jymaster', 'jyeffect', 'sky', 'hires', 'lossless', 'exhigh', 'standard', 'dolby']
        levels_ok = []
        # 探测总预算 60s——上游黑洞时 8 级串行最坏 ~16 分钟线程滞留，
        # 超预算即返回已测得的结果（调用方 5 分钟缓存内复用）
        _deadline = time.monotonic() + 60
        for level in ladder:
            if time.monotonic() > _deadline:
                break
            try:
                result = await self.get_song_url(probe_song_id, level, cookies)
                if result and result.get('data'):
                    d = result['data'][0] or {}
                    # 请求与返回一致且带直链，才说明该音质真正可用
                    if d.get('url') and d.get('level') == level:
                        levels_ok.append(level)
            except Exception:
                continue
        return levels_ok

    async def subscribe_playlist(self, playlist_id, cookies: dict[str, str],
                           action: int = 1) -> dict[str, Any]:
        """收藏/取消收藏歌单（官方接口）

        Args:
            playlist_id: 歌单ID
            cookies: 个人账号cookies
            action: 1=收藏, 2=取消收藏

        Returns:
            接口响应

        Raises:
            APIException: 操作失败时抛出
        """
        try:
            data = {'id': playlist_id, 't': action}
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }
            response = await _http.request('POST', APIConstants.PLAYLIST_SUBSCRIBE_API, data=data,
                                     headers=headers, cookies=cookies, timeout=30)
            result = response.json()
            _ensure_dict(result, 'subscribe_playlist 接口响应异常（非标准 JSON 对象）')
            if result.get('code') != 200:
                raise APIException(f"收藏歌单失败: {result.get('message', '未知错误')}")
            return result
        except UpstreamError as e:
            raise APIException(f"收藏歌单请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析收藏歌单响应失败: {e}") from e

    async def subscribe_album(self, album_id, cookies: dict[str, str],
                        action: int = 1) -> dict[str, Any]:
        """收藏/取消收藏专辑（官方接口）

        Args:
            album_id: 专辑ID
            cookies: 个人账号cookies
            action: 1=收藏, 2=取消收藏

        Returns:
            接口响应

        Raises:
            APIException: 操作失败时抛出
        """
        try:
            data = {'id': album_id, 't': action}
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER
            }
            response = await _http.request('POST', APIConstants.ALBUM_SUBSCRIBE_API, data=data,
                                     headers=headers, cookies=cookies, timeout=30)
            result = response.json()
            _ensure_dict(result, 'subscribe_album 接口响应异常（非标准 JSON 对象）')
            if result.get('code') != 200:
                raise APIException(f"收藏专辑失败: {result.get('message', '未知错误')}")
            return result
        except UpstreamError as e:
            raise APIException(f"收藏专辑请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析收藏专辑响应失败: {e}") from e

    async def post_comment(self, thread_id: str, content: str, action: int = 1,
                     comment_id: int | None = None,
                     cookies: dict[str, str] | None = None) -> dict[str, Any]:
        """发送/回复歌曲评论（官方 weapi 接口）

        Args:
            thread_id: 资源线程ID，如 R_SO_4_{songId}
            content: 评论内容
            action: 1=发送评论, 2=回复评论
            comment_id: 回复目标评论ID（action=2 时必填）
            cookies: 个人账号cookies

        Returns:
            接口响应

        Raises:
            APIException: 操作失败时抛出
        """
        try:
            payload: dict[str, Any] = {
                'threadId': thread_id,
                'content': content,
                't': action,
                'type': 6,
            }
            if comment_id:
                payload['commentId'] = comment_id
            # 官方 /weapi/comment 要求携带 csrf_token（取自 Cookie
            # 的 __csrf）——原实现缺失，与加密偏差叠加导致评论疑似从未成功
            csrf = (cookies or {}).get('__csrf') or ''
            if csrf:
                payload['csrf_token'] = csrf

            body = WeapiCrypto.encrypt(payload)
            headers = {
                'User-Agent': APIConstants.USER_AGENT,
                'Referer': APIConstants.REFERER,
            }
            response = await _http.request('POST', APIConstants.COMMENT_POST_API, data=body,
                                     headers=headers, cookies=cookies or {}, timeout=30)

            text = response.text.strip()
            if not text:
                # weapi 被网络层吞掉（数据中心 IP 风控），无法确认是否成功
                raise APIException("网易云未返回操作确认（可能被网络风控拦截），请稍后查看评论区确认")

            result = response.json()
            _ensure_dict(result, 'post_comment 接口响应异常（非标准 JSON 对象）')
            if result.get('code') != 200:
                raise APIException(f"评论失败: {result.get('message', '未知错误')}")
            return result
        except UpstreamError as e:
            raise APIException(f"评论请求失败: {e}") from e
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析评论响应失败: {e}") from 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: int | None, 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}'