NeteaseUrl/podcast_api.py

# -*- coding: utf-8 -*-
"""网易云音乐播客（电台 / 单集声音）API 模块

提供网易云"播客"板块相关接口封装，包括：
- 播客搜索（电台 / 单集声音）
- 播客电台详情与单集列表（分页）
- 单集（program）详情
- 热门播客推荐
- 单集音频直链解析（音质自动降级）

说明：
- 播客单集的音频本质上是网易云曲库中的 song（program.mainSong），
  因此直链复用 music_api.url_v1（eapi song/enhance/player/url/v1）获取。
- 支持格式：mp3 / m4a / aac 等（以接口返回 type 为准）。
- 支持链接协议：music.163.com/djradio?id=（电台）、/program?id= 与 /sound?id=（单集）、
  以及 163cn.tv 短链。
"""

import json
import re
import time
from typing import Any

from core import sync_http as requests  # 同步门面（底层 aiohttp）
from music_api import APIConstants, APIException, url_v1


class PodcastConstants:
    """播客接口相关常量"""

    # 搜索（cloudsearch：type=1009 播客电台 / 2000 单集声音）
    SEARCH_API = 'https://music.163.com/api/cloudsearch/pc'
    # 电台（播客）详情
    DJRADIO_GET_API = 'https://music.163.com/api/djradio/get'
    # 电台下的单集列表（分页）
    DJ_PROGRAM_BYRADIO_API = 'https://music.163.com/api/dj/program/byradio'
    # 单集详情
    DJ_PROGRAM_DETAIL_API = 'https://music.163.com/api/dj/program/detail'
    # 热门播客（必须携带有效 cateId）
    DJRADIO_HOT_API = 'https://music.163.com/api/djradio/hot'
    # 播客分类列表
    DJRADIO_CATEGORY_API = 'https://music.163.com/api/djradio/category/get'

    # 播客音频常用音质降级阶梯（无损对播客基本无意义）
    QUALITY_LADDER = ['exhigh', 'standard']

    # 热门接口的兜底分类ID（生活 / 情感 / 脱口秀 / 有声书）
    DEFAULT_HOT_CATE_IDS = [6, 3, 8, 10001]


# 与 music_api 同口径的上游 JSON 响应体上限
_MAX_UPSTREAM_JSON_BYTES = 8 * 1024 * 1024


def _post_json(url: str, data: dict[str, Any], cookies: dict[str, str] | None = None) -> dict[str, Any]:
    """发送表单 POST 请求并返回 JSON"""
    headers = {
        'User-Agent': APIConstants.USER_AGENT,
        'Referer': APIConstants.REFERER,
    }
    try:
        response = requests.post(url, data=data, headers=headers,
                                 cookies=cookies or {}, timeout=30)
        response.raise_for_status()
        if len(response.content) > _MAX_UPSTREAM_JSON_BYTES:
            raise APIException('播客接口响应体异常（超过大小上限）')
        result = response.json()
        if not isinstance(result, dict):
            raise APIException('播客接口响应异常（非标准 JSON 对象）')
    except requests.RequestException as e:
        raise APIException(f"播客接口请求失败: {e}") from e
    except json.JSONDecodeError as e:
        raise APIException(f"解析播客接口响应失败: {e}") from e
    if result.get('code') not in (200, None):
        raise APIException(f"播客接口返回错误: {result.get('message') or result.get('code')}")
    return result


def _to_int(v: Any, default: int | None = None) -> int | None:
    """安全转 int：垃圾值/None 回退默认值（上游字段常为 null/字符串）。"""
    try:
        return int(v)
    except (TypeError, ValueError):
        return default


def _ts_to_date(ts: int | None) -> str:
    """毫秒时间戳 → 日期字符串"""
    try:
        if not ts:
            return ''
        return time.strftime('%Y-%m-%d', time.localtime(int(ts) / 1000))
    except (TypeError, ValueError, OSError, OverflowError):
        return ''


def _fmt_duration(ms: int | None) -> str:
    """毫秒 → mm:ss / hh:mm:ss"""
    try:
        sec = int((ms or 0) / 1000)
    except (TypeError, ValueError):
        return ''
    if sec <= 0:
        return ''
    h, rem = divmod(sec, 3600)
    m, s = divmod(rem, 60)
    return f"{h}:{m:02d}:{s:02d}" if h else f"{m:02d}:{s:02d}"


def _fmt_count(n: int | None) -> str:
    """数量 → 万单位简写"""
    try:
        n = int(n or 0)
    except (TypeError, ValueError):
        return '0'
    if n >= 100000000:
        v = n / 100000000
        return f"{v:.1f}亿".replace('.0', '')
    if n >= 10000:
        v = n / 10000
        return f"{v:.1f}万".replace('.0', '')
    return str(n)


# ============ 数据规范化 ============

def _norm_radio(raw: dict[str, Any]) -> dict[str, Any]:
    """规范化电台（播客）信息"""
    dj = raw.get('dj') or {}
    return {
        'id': raw.get('id'),
        'name': raw.get('name', ''),
        'pic_url': raw.get('picUrl', ''),
        'desc': str(raw.get('desc') or raw.get('description') or '').strip(),
        'category': raw.get('category', ''),
        'sub_count': raw.get('subCount', 0),
        'sub_count_text': _fmt_count(raw.get('subCount')),
        'program_count': raw.get('programCount', 0),
        'play_count': raw.get('playCount', 0),
        'play_count_text': _fmt_count(raw.get('playCount')),
        'dj_name': dj.get('nickname', ''),
        'dj_avatar': dj.get('avatarUrl', ''),
        'create_date': _ts_to_date(raw.get('createTime')),
        'second_category': raw.get('secondCategory', ''),
    }


def _norm_program(raw: dict[str, Any]) -> dict[str, Any]:
    """规范化单集（program）信息"""
    radio = raw.get('radio') or {}
    dj = raw.get('dj') or radio.get('dj') or {}
    main_song = raw.get('mainSong') or raw.get('mainMusic') or {}
    return {
        'id': raw.get('id'),
        'name': raw.get('name', ''),
        'cover_url': raw.get('coverUrl') or raw.get('cover') or radio.get('picUrl', ''),
        'description': str(raw.get('description') or raw.get('desc') or '').strip(),
        'duration': raw.get('duration', 0),
        'duration_text': _fmt_duration(raw.get('duration') or main_song.get('duration')),
        'create_date': _ts_to_date(raw.get('createTime')),
        'serial_num': raw.get('serialNum', 0),
        'listener_count': raw.get('listenerCount', 0),
        'listener_count_text': _fmt_count(raw.get('listenerCount')),
        'liked_count': raw.get('likedCount', 0),
        'comment_count': raw.get('commentCount', 0),
        'main_song_id': main_song.get('id'),
        'main_song_name': main_song.get('name', ''),
        'radio_id': radio.get('id'),
        'radio_name': radio.get('name', ''),
        'dj_name': dj.get('nickname', ''),
        'category': raw.get('category') or radio.get('category', ''),
    }


# ============ 链接 / ID 解析 ============

_SHORTLINK_CACHE: dict[str, str] = {}


def _resolve_short_link(url: str) -> str:
    """解析 163cn.tv 短链，返回跳转后的真实链接（仅接受整串短链，子串不触发请求）"""
    if not re.match(r'^https?://(www\.)?163cn\.tv/\S+$', str(url or '').strip(), re.IGNORECASE):
        return url
    # 带浏览器 UA（短链 CDN 对默认 UA 区别对待）+ 模块级缓存
    #（此前仅音乐路径有缓存，播客同一分享链接每次解析都发真实 GET）
    global _SHORTLINK_CACHE
    cached = _SHORTLINK_CACHE.get(url)
    if cached is not None:
        return cached
    resp = None
    try:
        resp = requests.get(url, allow_redirects=False, timeout=10,
                            headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'})
        real = resp.headers.get('Location', url)
    except requests.RequestException:
        return url
    finally:
        if resp is not None:
            resp.close()
    # 短链展开口径——展开失败（200 无 Location/5xx，
    # real==url）不写缓存，一次 CDN 抖动不再让该链接整个缓存存续期必然失败
    if real == url:
        return real
    if len(_SHORTLINK_CACHE) > 500:
        _SHORTLINK_CACHE.clear()
    _SHORTLINK_CACHE[url] = real
    return real


def extract_podcast_ref(raw: str) -> dict[str, str]:
    """从用户输入中提取播客引用（正规 URL 解析，替代子串匹配与非数字剔除）

    Returns:
        {'kind': 'radio'|'program'|'unknown', 'id': '...'}
        - radio:   djradio?id= / radio?id= 链接
        - program: program?id= / sound?id= 链接
        - unknown: 纯数字 ID（需依次尝试单集/电台）
    """
    from urllib.parse import parse_qs, urlsplit
    text = str(raw or '').strip()
    # 短链先还原（仅当整串就是一个短链 URL）
    if '163cn.tv' in text.lower():
        text = _resolve_short_link(text)

    # 有 URL 结构：按路径判断类型，query 正规解析 id（参数顺序无关）
    if '://' in text or text.startswith('//'):
        try:
            parts = urlsplit(text if '://' in text else 'https:' + text)
            q = parse_qs(parts.query)
            # 兼容老式 #/program?id= 片段路由
            if not q.get('id'):
                m = re.search(r'[?&#]id=(\d+)', text)
                if m:
                    q = {'id': [m.group(1)]}
            pid = (q.get('id') or [''])[0]
            if pid.isdigit():
                path_l = parts.path.lower()
                if 'djradio' in path_l or 'radio' in path_l:
                    return {'kind': 'radio', 'id': pid}
                if 'program' in path_l or 'sound' in path_l or 'song' in path_l:
                    return {'kind': 'program', 'id': pid}
                # 路径无类型线索（如 y2027.d.popchan.cn 之类播客域名）：按 unknown 走探测
                return {'kind': 'unknown', 'id': pid}
        except ValueError:
            pass
        raise APIException('无法识别的播客链接，请粘贴电台/单集链接或直接输入纯数字 ID')

    # 纯数字 ID
    if text.isdigit():
        return {'kind': 'unknown', 'id': text}
    raise APIException('无法识别的播客链接或ID，支持 电台链接(djradio?id=)、单集链接(program?id=) 或纯数字ID')


# ============ 核心 API ============

def search_podcast(keywords: str, cookies: dict[str, str] | None = None,
                   limit: int = 20, offset: int = 0,
                   search_type: str = 'all') -> dict[str, Any]:
    """搜索播客

    Args:
        keywords: 关键词
        cookies: 请求 Cookie（可选）
        limit: 每类返回数量（1-50）
        offset: 偏移量
        search_type: all=电台+单集 / radio=仅电台 / voice=仅单集

    Returns:
        {'radios': [...], 'voices': [...]}
    """
    keywords = str(keywords or '').strip()
    if not keywords:
        raise APIException('搜索关键词不能为空')
    limit = max(1, min(int(limit or 20), 50))

    result: dict[str, Any] = {'radios': [], 'voices': []}

    def _do_search(stype: int) -> dict[str, Any]:
        return _post_json(PodcastConstants.SEARCH_API,
                          {'s': keywords, 'type': stype, 'limit': limit, 'offset': offset},
                          cookies)

    if search_type in ('all', 'radio'):
        res = _do_search(1009)
        radios = (res.get('result') or {}).get('djRadios') or []
        result['radios'] = [_norm_radio(r) for r in radios]

    if search_type in ('all', 'voice'):
        res = _do_search(2000)
        rdata = res.get('result') or {}
        voices = rdata.get('songs') or rdata.get('voices') or rdata.get('programs') or []
        result['voices'] = [_norm_program(v) for v in voices]

    return result


def get_radio_detail(radio_id: Any, cookies: dict[str, str] | None = None) -> dict[str, Any]:
    """获取播客电台（djradio）详情"""
    try:
        rid = int(str(radio_id).strip())
    except (TypeError, ValueError):
        raise APIException('无效的电台ID') from None
    result = _post_json(PodcastConstants.DJRADIO_GET_API, {'id': rid}, cookies)
    # 正常返回键为 djRadio；不存在时 code=404（_post_json 已抛错）或 data 为空
    data = result.get('djRadio') or result.get('data') or {}
    if not data:
        raise APIException('未找到该播客电台（可能已下架或ID错误）')
    return _norm_radio(data)


def get_radio_programs(radio_id: Any, cookies: dict[str, str] | None = None,
                       limit: int = 30, offset: int = 0, asc: bool = True) -> dict[str, Any]:
    """获取电台下的单集列表（分页）

    Args:
        radio_id: 电台ID
        limit: 每页数量（1-100）
        offset: 偏移量
        asc: True=从第1集开始（正序），False=最新在前

    Returns:
        {'programs': [...], 'count': 总数, 'has_more': bool}
    """
    try:
        rid = int(str(radio_id).strip())
    except (TypeError, ValueError):
        raise APIException('无效的电台ID') from None
    limit = max(1, min(int(limit or 30), 100))
    result = _post_json(PodcastConstants.DJ_PROGRAM_BYRADIO_API,
                        {'radioId': rid, 'limit': limit, 'offset': max(0, int(offset or 0)),
                         'asc': 'true' if asc else 'false'},
                        cookies)
    programs = [_norm_program(p) for p in (result.get('programs') or [])]
    count = _to_int(result.get('count'))
    offset_val = max(0, int(offset or 0))
    return {
        'programs': programs,
        'count': count,
        'has_more': (offset_val + len(programs)) < count if count else len(programs) >= limit,
    }


def get_program_detail(program_id: Any, cookies: dict[str, str] | None = None) -> dict[str, Any]:
    """获取单集（program）详情"""
    try:
        pid = int(str(program_id).strip())
    except (TypeError, ValueError):
        raise APIException('无效的单集ID') from None
    result = _post_json(PodcastConstants.DJ_PROGRAM_DETAIL_API, {'id': pid}, cookies)
    program = result.get('program') or {}
    if not program:
        raise APIException('未找到该播客单集（可能已下架或ID错误）')
    return _norm_program(program)


def get_categories() -> list[dict[str, Any]]:
    """获取播客分类列表（GET 接口）

    Returns:
        [{'id': 6, 'name': '生活'}, ...]
    """
    headers = {
        'User-Agent': APIConstants.USER_AGENT,
        'Referer': APIConstants.REFERER,
    }
    try:
        response = requests.get(PodcastConstants.DJRADIO_CATEGORY_API,
                                headers=headers, timeout=30)
        response.raise_for_status()
        if len(response.content) > _MAX_UPSTREAM_JSON_BYTES:
            raise APIException('播客接口响应体异常（超过大小上限）')
        result = response.json()
    except requests.RequestException as e:
        raise APIException(f"获取播客分类失败: {e}") from e
    except json.JSONDecodeError as e:
        raise APIException(f"解析播客分类响应失败: {e}") from e
    if not isinstance(result, dict):
        # 非 dict 响应（null/数组）显式报错，不再 AttributeError
        raise APIException('解析播客分类响应失败（非标准 JSON 对象）')
    cats = []
    for c in (result.get('categories') or []):
        if not isinstance(c, dict):
            continue
        try:
            cid = int(c.get('id') or 0)
        except (TypeError, ValueError):
            continue   # 垃圾 id 跳过该条，不再拖死整页分类
        name = c.get('name')
        if cid and name:
            cats.append({'id': cid, 'name': str(name)})
    if not cats:
        raise APIException('播客分类列表为空')
    return cats


def get_hot_radios(cookies: dict[str, str] | None = None,
                   limit: int = 30, offset: int = 0,
                   cate_id: int | None = None) -> list[dict[str, Any]]:
    """获取分类下热门播客

    Args:
        cate_id: 播客分类ID；用户明确指定分类时只查该分类
                 （即便为空也不悄悄混入其他分类的内容），未指定时按默认分类依次尝试

    Returns:
        电台列表（已规范化）
    """
    limit = max(1, min(int(limit or 30), 100))
    cate_ids: list[int] = []
    user_cate = False
    if cate_id:
        try:
            cate_ids.append(int(cate_id))
            user_cate = True
        except (TypeError, ValueError):
            pass
    if not user_cate:
        for cid in PodcastConstants.DEFAULT_HOT_CATE_IDS:
            if cid not in cate_ids:
                cate_ids.append(cid)

    for cid in cate_ids:
        result = _post_json(PodcastConstants.DJRADIO_HOT_API,
                            {'limit': limit, 'offset': max(0, int(offset or 0)), 'cateId': cid},
                            cookies)
        radios = [_norm_radio(r) for r in (result.get('djRadios') or [])]
        if radios:
            return radios
        if user_cate:
            break   # 指定分类为空就返回空，不注入默认分类内容
    return []


def resolve_episode_url(song_id: Any, cookies: dict[str, str] | None = None,
                        level: str = 'auto') -> dict[str, Any] | None:
    """解析单集音频直链（带音质降级）

    Args:
        song_id: 单集 mainSong 的歌曲ID
        cookies: 全局账号 Cookie
        level: 期望音质（auto=按阶梯自动尝试）

    Returns:
        {'url','level','type','size','br','md5'} 或 None（全部失败）
    """
    # 降级阶梯只向下不向上——原实现指定 standard 会回退到更高的 exhigh
    _FALLDOWN_LADDER = ['hires', 'lossless', 'exhigh', 'standard']
    order: list[str] = []
    level = str(level or 'auto').strip()
    if level and level != 'auto':
        order.append(level)
        # 追加"低于"请求音质的等级（hires > lossless > exhigh > standard）
        if level in _FALLDOWN_LADDER:
            idx = _FALLDOWN_LADDER.index(level)
            for lv in _FALLDOWN_LADDER[idx + 1:]:
                order.append(lv)
        else:
            # 未知等级：退到通用阶梯
            order.extend(PodcastConstants.QUALITY_LADDER)
    else:
        order.extend(PodcastConstants.QUALITY_LADDER)
    # 去重保序
    dedup: list[str] = []
    for lv in order:
        if lv not in dedup:
            dedup.append(lv)
    order = dedup

    for lv in order:
        try:
            # song_id 可能是 None/垃圾值（防御：int 失败按无音频处理，不再 TypeError 500）
            sid = int(str(song_id).strip())
        except (TypeError, ValueError):
            return None
        try:
            res = url_v1(sid, lv, cookies or {})
        except (APIException, ValueError):
            continue
        data = (res or {}).get('data') or []
        if data and data[0] and data[0].get('url'):
            d = data[0]
            return {
                'url': d.get('url'),
                'level': d.get('level') or lv,
                'type': d.get('type') or 'mp3',
                'size': _to_int(d.get('size')),
                'br': d.get('br'),
                'md5': d.get('md5', ''),
                'requested_level': lv,
            }
    return None


# ============ 组合能力 ============

def parse_podcast_ref(raw: str, cookies: dict[str, str] | None = None,
                      page_size: int = 30) -> dict[str, Any]:
    """智能解析用户输入：电台链接 → 电台详情+单集列表；单集链接 → 单集详情

    Returns:
        {'kind': 'radio', 'radio': {...}, 'programs': [...], 'count': n, 'has_more': bool}
        或 {'kind': 'program', 'program': {...}}
    """
    ref = extract_podcast_ref(raw)

    if ref['kind'] == 'radio':
        radio = get_radio_detail(ref['id'], cookies)
        page = get_radio_programs(ref['id'], cookies, limit=page_size, offset=0, asc=True)
        return {'kind': 'radio', 'radio': radio, **page}

    if ref['kind'] == 'program':
        return {'kind': 'program', 'program': get_program_detail(ref['id'], cookies)}

    # 纯数字：先试单集，再试电台
    try:
        return {'kind': 'program', 'program': get_program_detail(ref['id'], cookies)}
    except APIException:
        pass
    radio = get_radio_detail(ref['id'], cookies)
    page = get_radio_programs(ref['id'], cookies, limit=page_size, offset=0, asc=True)
    return {'kind': 'radio', 'radio': radio, **page}


def list_radio_all_program_ids(radio_id: Any, cookies: dict[str, str] | None = None,
                               max_count: int = 200) -> list[dict[str, Any]]:
    """拉取电台全部单集（用于整台打包，正序），返回 [{'id','name','main_song_id','serial_num','cover_url'}]

    最多拉取 max_count 集，防止超大电台拖垮磁盘/任务。
    按节目 ID 去重；一页没有任何新增（短页/重复页/列表变动）即有界停止，
    不再固定 offset+=100 盲目翻页。
    """
    out: list[dict[str, Any]] = []
    seen_ids = set()
    offset = 0
    page_size = 100
    max_pages = 30   # 有界保护：即使上游 has_more 恒真也最多翻 30 页
    for _ in range(max_pages):
        page = get_radio_programs(radio_id, cookies, limit=page_size, offset=offset, asc=True)
        new_count = 0
        for p in page['programs']:
            if p['id'] in seen_ids:
                continue
            seen_ids.add(p['id'])
            out.append({
                'id': p['id'],
                'name': p['name'],
                'main_song_id': p['main_song_id'],
                'serial_num': p['serial_num'],
                'cover_url': p['cover_url'],
            })
            new_count += 1
            if len(out) >= max_count:
                break
        if len(out) >= max_count:
            break
        # 本页无新增（重复页/短页）或上游明确说没有更多 → 停止
        if new_count == 0 or not page['has_more'] or not page['programs']:
            break
        offset += page_size
    return out