懒人听书 MCP 工具实现

一、工具基本信息

1.1 工具名称

lanrentingshu - 懒人听书有声书播放工具

1.2 提供的 MCP Tools

Tool 名称	功能描述
lanrentingshu.search_album	搜索懒人听书平台的有声书专辑
lanrentingshu.get_album_detail	获取专辑详情（集数列表、主播、简介）
lanrentingshu.search_and_play	一站式搜索并播放（搜索+下载+播放），支持本地缓存
lanrentingshu.pause	暂停当前播放，保持播放位置
lanrentingshu.resume	从暂停位置恢复播放
lanrentingshu.stop	停止播放，重置位置
lanrentingshu.get_status	获取播放器当前状态

1.3 功能描述

• 搜索专辑：通过关键词搜索有声书，返回专辑名、主播、作者、推荐语
• 专辑详情：获取指定专辑的完整集数列表，最多展示前50集
• 搜索播放：一站式播放链路（搜索→取第一结果→查详情→找目标集→获取流URL→下载→播放），优先从本地缓存命中
• 播放控制：暂停/恢复/停止，其中暂停区分TTS打断和手动暂停两种来源
• 状态查询：查看当前播放的有声书名、集数、播放状态

---

二、整体架构

2.1 MCP Server 入口

入口文件：/home/user/xiaozhi/src/mcp/mcp_server.py

工具注册入口：McpServer.add_common_tools() 方法（第377-380行）

# 添加懒人听书工具
from src.mcp.tools.lanrentingshu import get_lanrentingshu_tools_manager

lanrentingshu_manager = get_lanrentingshu_tools_manager()
lanrentingshu_manager.init_tools(self.add_tool, PropertyList, Property, PropertyType)

2.2 文件结构

src/mcp/tools/lanrentingshu/
├── __init__.py           # 包导出（Manager + Player 单例）
├── manager.py           # 工具管理器（注册7个 MCP Tool）
├── lanrentingshu_player.py  # 核心播放器（API调用、下载、解码、播放）
└── lanrentingshu.md     # API文档（52api.cn接口说明）

2.3 架构分层

┌─ manager.py ───────────────────────────────┐
│ LanrentingshuToolsManager                   │
│   - 7 个 MCP Tool wrapper                   │
│   - 参数校验 (PropertyList)                  │
│   - 结果格式化为可读文本                      │
└───────────┬─────────────────────────────────┘
            │ 调用 _player
            ▼
┌─ lanrentingshu_player.py ──────────────────┐
│ LanrentingshuPlayer                         │
│   - _call_api()      → 52api.cn HTTP        │
│   - search_album()    → type=search         │
│   - get_album_detail() → type=detail        │
│   - get_stream_url()   → type=stream        │
│   - _download_file()   → requests.stream    │
│   - _start_playback()  → FFmpeg解码→AudioCodec │
│   - _playback_loop()   → asyncio.Queue循环   │
└─────────────────────────────────────────────┘

2.4 Server 初始化流程

McpServer.__init__()
  ↓
McpPlugin.setup()
  ↓
server.add_common_tools()
  ↓
get_lanrentingshu_tools_manager()     ← 全局单例
  ↓
manager.init_tools(add_tool, ...)
  ↓
player = get_lanrentingshu_player_instance()  ← 全局单例
  ↓
注册 7 个 Tool:
  self._register_search_album_tool()
  self._register_get_album_detail_tool()
  self._register_search_and_play_tool()
  self._register_pause_tool()
  self._register_resume_tool()
  self._register_stop_tool()
  self._register_get_status_tool()

---

三、每个 Tool 详细拆解

Tool 1: lanrentingshu.search_album

输入参数

参数名	类型	必填	默认值	含义
keyword	string	是	-	搜索关键词（如"西游记"、“三体”）
page	integer	否	1	页码

核心逻辑

1. 从 args 提取 keyword 和 page
  ↓
2. 构建 API 请求体: {type: "search", keyword, page}
  ↓
3. await self._player.search_album(keyword, page)
  ↓ (内部)
  3.1 _call_api() → HTTP POST → https://www.52api.cn/api/lanren
  ↓
4. 格式化返回结果为可读文本
  ↓
5. 返回专辑列表字符串

关键代码（manager.py 第70-106行）

async def search_album_wrapper(args: Dict[str, Any]) -> str:
    keyword = args.get("keyword", "")
    page = args.get("page", 1)
    result = await self._player.search_album(keyword, page)

    if result.get("status") == "success":
        albums = result.get("albums", [])
        if not albums:
            return f"未找到与'{keyword}'相关的有声书"

        lines = [
            f"找到 {result['total_count']} 个有声书专辑"
            f"（第{result['current_page']}/{result['total_pages']}页）:"
        ]
        for i, album in enumerate(albums, 1):
            lines.append(
                f"{i}. {album['name']} - 主播: {album['announcer']}"
                f" - 作者: {album['author']}"
            )
            if album.get("shortRecReason"):
                lines.append(f"   推荐: {album['shortRecReason']}")
        return "\n".join(lines)
    else:
        return result.get("message", "搜索失败")

Player 内部实现（player.py 第177-239行）

async def search_album(self, keyword: str, page: int = 1) -> dict:
    payload = {
        "key": self.config["API_KEY"],
        "type": "search",
        "keyword": keyword,
        "id": "",
        "stream_id": "",
        "stream_section": "",
        "page": str(page) if page > 1 else "",
        "cookie": "",
    }

    data = await self._call_api(payload)

    # 解析响应
    result_data = data.get("data", {})
    album_list = result_data.get("lists", [])
    total_count = result_data.get("count", 0)
    total_pages = result_data.get("total_pages", 1)

    # 缓存搜索结果（供后续 detail/play 使用）
    self._last_search_results = album_list

    # 格式化每个专辑
    formatted_albums = []
    for album in album_list:
        formatted_albums.append({
            "id": album.get("id"),        # ← 后续 detail 用
            "name": album.get("name"),
            "announcer": album.get("announcer"),
            "author": album.get("author"),
            "desc": album.get("desc", ""),
            "recReason": album.get("recReason", ""),
            "cover": album.get("cover", ""),
        })

    return {
        "status": "success",
        "albums": formatted_albums,
        "total_count": total_count,
        "total_pages": total_pages,
    }

调用的外部 API

项目	内容
URL	https://www.52api.cn/api/lanren
方法	HTTP POST
type	search
关键参数	keyword（搜索词）、page（页码）

返回值结构示例

找到 358 个有声书专辑（第1/8页）:
1. 米小圈快乐西游记 - 主播: 米小圈 - 作者: 佚名
   推荐: 米小圈带你听名著取真经
2. 西游记|甄齐播讲 - 主播: 甄齐 - 作者: 卡尔博学
   推荐: 西游记无障碍收听版
3. 父与子新编5：爆笑西游记|亲子笑话|睡前故事 - 主播: 大有叔叔 - 作者: 佚名
   推荐: 唐僧老爸·西游魔改笑劈叉

---

Tool 2: lanrentingshu.get_album_detail

输入参数

参数名	类型	必填	默认值	含义
album_id	string	是	-	专辑ID（从搜索结果取 id 字段）
keyword	string	否	“”	专辑关键词（辅助匹配）

核心逻辑

1. 从 args 提取 album_id（必填）和 keyword（可选）
  ↓
2. 构建 API 请求体: {type: "detail", id: album_id, keyword}
  ↓
3. await self._player.get_album_detail(album_id, keyword)
  ↓ (内部)
  3.1 _call_api() → HTTP POST
  ↓
4. 解析结果：专辑信息 + 集数列表（前50集）
  ↓
5. 格式化为可读文本返回

关键代码（manager.py 第112-159行）

async def get_album_detail_wrapper(args: Dict[str, Any]) -> str:
    album_id = args.get("album_id", "")
    keyword = args.get("keyword", "")
    result = await self._player.get_album_detail(album_id, keyword)

    if result.get("status") == "success":
        album = result.get("album", {})
        episodes = result.get("episodes", [])

        lines = [
            f"专辑: {album['name']}",
            f"主播: {album.get('announcer', '未知')}",
            f"作者: {album.get('author', '未知')}",
            f"总播放: {album.get('total_play', 0)}",
            f"总集数: {result['total_episodes']} 集",
            f"简介: {album.get('desc', '')[:200]}",
            "",
            "集数列表（前50集）:",
        ]

        for ep in episodes[:50]:
            lines.append(
                f"  第{ep['section']}集: {ep['name']} ({ep.get('size', '')})"
            )

        if len(episodes) > 50:
            lines.append(f"  ...共{result['total_episodes']}集，仅显示前50集")

        return "\n".join(lines)
    else:
        return result.get("message", "获取详情失败")

Player 内部实现（player.py 第243-307行）

每个 episode 的结构：

{
    "stream_id": "xxx",       # ← 后续 get_stream_url 用
    "section": 1,             # 集数编号
    "name": "第1集 猴王出世",  # 集名
    "size": "12.5MB"          # 文件大小
}

调用的外部 API

项目	内容
URL	https://www.52api.cn/api/lanren
type	detail
关键参数	id（专辑ID）、keyword（可选）

---

Tool 3: lanrentingshu.search_and_play ★核心

输入参数

参数名	类型	必填	默认值	含义
keyword	string	是	-	搜索关键词（如"西游记"）
section	integer	否	1	要播放的集数

核心逻辑

1. 从 args 提取 keyword 和 section
  ↓
2. await self._player.search_and_play(keyword, section)
  ↓ (内部)
  ┌─ 第一步：查本地缓存 ───────────────────┐
  │ _find_local_episode(keyword, section)   │
  │   精确匹配: {keyword}_{section:04d}.m4a │
  │   模糊匹配: 文件名含关键词+集数          │
  │   ↓ 命中?                               │
  │   Yes → _start_playback() → 直接返回    │
  │   No  → 进入后台模式                     │
  └─────────────────────────────────────────┘
  ↓
  ┌─ 第二步：后台异步下载播放 ───────────────┐
  │ asyncio.create_task(                     │
  │   _background_search_and_play())         │
  │ 立即返回 "正在下载中" 给 AI               │
  └─────────────────────────────────────────┘
  ↓ (后台任务，不阻塞 AI)
  ┌─ 后台任务详细流程 ──────────────────────┐
  │ ① search_album(keyword)                 │
  │    → 取第一个结果的 id                   │
  │ ② get_album_detail(album_id)            │
  │    → 找到 section 对应的 stream_id      │
  │ ③ get_stream_url(album_id, stream_id)   │
  │    → 获取音频流 URL                      │
  │ ④ _download_file(url)                   │
  │    → 流式下载到本地缓存                  │
  │ ⑤ _start_playback(file_path)            │
  │    → FFmpeg解码 → AudioCodec输出        │
  └─────────────────────────────────────────┘

关键代码：前台入口（player.py 第411-457行）

async def search_and_play(self, keyword: str, stream_section: int = 1) -> dict:
    # 保存请求参数（后台任务可能用到）
    self._pending_keyword = keyword
    self._pending_section = stream_section

    # ★ 第一步：优先查本地缓存（快速路径）
    local_file = await self._find_local_episode(keyword, stream_section)
    if local_file:
        success = await self._play_local_file(
            local_file, keyword, f"{keyword}_第{stream_section}集"
        )
        if success:
            return {
                "status": "success",
                "message": f"正在播放(本地): {keyword} - 第{stream_section}集",
                "album": keyword,
                "episode": f"第{stream_section}集",
                "section": stream_section,
            }

    # ★ 第二步：启动后台搜索下载，立即返回
    asyncio.create_task(
        self._background_search_and_play(keyword, stream_section)
    )
    return {
        "status": "success",
        "message": (
            f"正在搜索下载: {keyword} 第{stream_section}集，"
            "请告知用户正在下载中稍等一下"
        ),
        "album": keyword,
        "background": True,  # ← 标记后台任务
    }

关键代码：后台任务（player.py 第459-525行）

async def _background_search_and_play(
    self, keyword: str, stream_section: int
) -> None:
    try:
        # ① 搜索专辑 → 取第一个结果的 id
        search_result = await self.search_album(keyword)
        first_album = search_result["albums"][0]
        album_id = str(first_album["id"])
        album_name = first_album["name"]

        # ② 获取专辑详情 → 找目标集数
        detail_result = await self.get_album_detail(album_id, keyword)
        episodes = detail_result["episodes"]
        target_episode = None
        for ep in episodes:
            if ep.get("section") == stream_section:
                target_episode = ep
                break
        if not target_episode and episodes:
            target_episode = episodes[0]  # fallback 第一集

        # ③ 获取音频流 URL
        stream_url = await self.get_stream_url(
            album_id,
            str(target_episode["stream_id"]),
            str(stream_section),
        )

        # ④ 设置当前播放信息
        self.current_album_name = album_name
        self.current_stream_name = target_episode["name"]
        self.current_stream_id = str(target_episode["stream_id"])
        self.current_stream_section = stream_section

        # 缓存文件名（用于后续本地查找）
        cache_filename = self._sanitize_filename(
            f"{album_name}_{stream_section:04d}"
        )
        self._episode_cache[self.current_stream_id] = cache_filename

        # ⑤ 下载 + 播放
        success = await self._play_url(stream_url)

    except asyncio.CancelledError:
        logger.info("[懒人听书] 后台搜索播放任务被取消")
    except Exception as e:
        logger.error(f"[懒人听书] 后台搜索播放异常: {e}")

播放链路内部细节

_play_url → _download_file → _start_playback → _playback_loop

下载文件（player.py 第551-596行）：

async def _download_file(self, url: str) -> Optional[Path]:
    # 构建缓存文件名: {专辑名}_{集数:04d}.m4a
    filename = self._sanitize_filename(
        f"{self.current_album_name}_{self.current_stream_section:04d}"
    )
    cache_path = self.cache_dir / f"{filename}.m4a"

    # 检查缓存
    if cache_path.exists():
        return cache_path

    # 流式下载到临时文件，再 rename 到正式目录
    temp_path = self.temp_cache_dir / f"temp_{int(time.time())}_{filename}.m4a"

    response = requests.get(url, headers=..., stream=True, timeout=60)
    with open(temp_path, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)

    shutil.move(str(temp_path), str(cache_path))  # 原子操作
    return cache_path

启动播放（player.py 第598-647行）：

async def _start_playback(self, file_path: Path, start_position: float = 0.0) -> bool:
    # 1. TTS 检查 → 如果机器人在说话，延迟启动
    if self.app and self.app.is_speaking():
        self._deferred_start_path = file_path
        self._deferred_start_position = start_position
        self.is_playing = True
        self.paused = True  # 先暂停，TTS结束后自动恢复
        return True

    # 2. 创建音频队列（内存缓冲）
    self._audio_queue = asyncio.Queue(maxsize=5000)

    # 3. 启动 FFmpeg 解码器
    self.decoder = MusicDecoder(
        sample_rate=AudioConfig.OUTPUT_SAMPLE_RATE,  # 24000
        channels=AudioConfig.CHANNELS,               # 1 (单声道)
    )
    await self.decoder.start_decode(file_path, self._audio_queue, start_position)

    # 4. 启动播放循环
    self._playback_task = asyncio.create_task(self._playback_loop())

    self.is_playing = True
    self.paused = False
    self.start_play_time = time.time() - start_position

播放循环（player.py 第649-674行）：

async def _playback_loop(self):
    while self.is_playing:
        if self.paused:
            await asyncio.sleep(0.1)  # 暂停时轮询等待
            continue

        try:
            # 从队列取 PCM 帧，超时 5 秒
            audio_data = await asyncio.wait_for(
                self._audio_queue.get(), timeout=5.0
            )
        except asyncio.TimeoutError:
            continue

        if audio_data is None:  # EOF → 播放结束
            await self._handle_playback_finished()
            break

        await self._write_to_audio_codec(audio_data)

写入 AudioCodec（player.py 第676-688行）：

async def _write_to_audio_codec(self, pcm_data: np.ndarray):
    if not self._refresh_audio_codec():
        return
    if pcm_data.ndim > 1:
        pcm_data = pcm_data.mean(axis=1).astype(np.int16)  # 多声道→单声道
    await self.audio_codec.write_pcm_direct(pcm_data)

调用的外部服务

服务	用途	格式
52api.cn	搜索/详情/流地址	HTTP POST JSON
FFmpeg	M4A → PCM 解码	本地进程
AudioCodec	PCM → 机器人喇叭	Python 对象

---

Tool 4: lanrentingshu.pause

输入参数

无

核心逻辑

1. 检查 is_playing → 没在播直接返回
2. 如果已暂停且是手动暂停 → 提示已暂停
3. 设置 paused = True, 记录暂停来源 = "manual"
4. 计算当前位置 current_position = now - start_play_time
5. 停止 FFmpeg 解码器
6. 清空音频队列

暂停来源区分（player.py 第734-765行）：

来源	值	触发方式	行为差异
TTS打断	"tts"	AI说话时自动触发	TTS结束后自动恢复
手动暂停	"manual"	用户说"暂停"	必须用户说"继续"才恢复

async def pause(self, source: str = "manual") -> dict:
    if not self.is_playing:
        return {"status": "info", "message": "没有正在播放的音频"}

    # 已暂停状态下的特殊处理
    if self.paused:
        if self._pause_source == "tts" and source == "manual":
            self._pause_source = source  # 升级：TTS打断→手动暂停
        return {"status": "info", "message": "已经处于暂停状态"}

    self.paused = True
    self._pause_source = source  # ← 记录谁暂停的

    # 保存播放位置
    if self.start_play_time > 0:
        self.current_position = time.time() - self.start_play_time

    # 停解码器 + 清队列
    if self.decoder:
        await self.decoder.stop()
        self.decoder = None
    # ... 清空 _audio_queue ...

---

Tool 5: lanrentingshu.resume

输入参数

无

核心逻辑

1. 检查 is_playing → 没在播直接返回
2. 检查 paused → 没暂停直接返回
3. 检查文件仍存在 → 不存在则报错
4. 重建音频队列
5. 重建 FFmpeg 解码器，从 current_position 开始解码
6. 重建播放循环任务
7. 恢复 paused = False

async def resume(self) -> dict:
    # 重建解码器，从暂停位置开始
    self.decoder = MusicDecoder(
        sample_rate=AudioConfig.OUTPUT_SAMPLE_RATE,
        channels=AudioConfig.CHANNELS,
    )
    success = await self.decoder.start_decode(
        self._current_file_path,    # 同一个文件
        self._audio_queue,
        self.current_position,      # ← 从上次暂停位置继续
    )

    self.paused = False
    self._pause_source = None
    self.start_play_time = time.time() - self.current_position

---

Tool 6: lanrentingshu.stop

输入参数

无

核心逻辑

1. 检查 is_playing → 没在播直接返回
2. 停止解码器
3. 取消播放任务 (asyncio.Task.cancel)
4. 清空音频队列
5. 重置所有播放状态 (position=0, paused=False)

---

Tool 7: lanrentingshu.get_status

输入参数

无

核心逻辑

直接读取播放器状态字段，三种状态互斥：

状态	条件
未播放	!is_playing
播放中	is_playing && !paused
已暂停	paused && _pause_source == "manual"

async def get_status(self) -> dict:
    if not self.is_playing:
        playing_state = "未播放"
    elif self.paused and self._pause_source == "manual":
        playing_state = "已暂停"
    elif self.is_playing:
        playing_state = "播放中"

    return {
        "album": self.current_album_name,
        "episode": self.current_stream_name,
        "is_playing": self.is_playing,
        "is_paused": self.paused,
    }

---

四、关键代码片段解读

4.1 API 统一调用层

代码位置：lanrentingshu_player.py 第155-173行

async def _call_api(self, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """所有 API 调用的统一入口."""
    response = await asyncio.to_thread(
        requests.post,                          # 同步请求包在线程中执行
        self.config["API_BASE_URL"],           # https://www.52api.cn/api/lanren
        json=payload,                            # 请求体
        headers=self.config["HEADERS"],          # User-Agent, Referer 等
        timeout=15,
    )
    response.raise_for_status()
    data = response.json()
    if data.get("code") != 200:
        logger.error(f"API返回错误: code={data.get('code')}")
        return None
    return data

为什么用 asyncio.to_thread：requests 是同步库，直接调用会阻塞整个事件循环。to_thread 把它放到线程池执行，不阻塞协程。

4.2 后台任务模式（避免 AI 超时）

核心问题：MCP 工具调用有超时限制（通常几秒），但搜索+下载可能需要几十秒。

代码位置：lanrentingshu_player.py 第444-457行

# 本地缓存未命中时：启动后台任务，立即返回
asyncio.create_task(
    self._background_search_and_play(keyword, stream_section)
)
return {
    "status": "success",
    "message": f"正在搜索下载: {keyword} 第{stream_section}集，"
               "请告知用户正在下载中稍等一下",
    "background": True,  # ← 标记为后台任务
}

时序图：

AI 调用 search_and_play("西游记", 1)
  │
  ▼ (1ms)
查本地缓存 → 未命中
  │
  ▼ (1ms)
asyncio.create_task(后台任务)     ← 不等待，立即返回
  │
  ▼ (1ms)
返回 "正在下载中..." → AI 继续对话
                                        │
                                        ▼ (后台，不阻塞)
                                  ① search_album("西游记")      (~2s)
                                  ② get_album_detail(album_id)  (~1s)
                                  ③ get_stream_url(...)         (~1s)
                                  ④ _download_file(url)         (~10-30s)
                                  ⑤ _start_playback(file)       (~100ms)
                                  ⑥ _playback_loop() 开始        (持续)

对比同步模式的问题：

AI 调用 search_and_play("西游记", 1)
  ↓
  ... 等待 30 秒 ...
  ↓
超时！❌ AI 收不到回复

4.3 本地缓存策略

代码位置：lanrentingshu_player.py 第357-394行

缓存目录：~/.cache/xiaozhi/lanrentingshu/

文件命名规则：{专辑名}_{集数:04d}.m4a（如 西游记_0001.m4a）

查找策略（两级）：

async def _find_local_episode(self, album_name: str, stream_section: int):
    album_key = self._sanitize_filename(album_name).lower()
    section_str = f"_{stream_section:04d}"

    # 策略1：精确匹配
    cache_filename = f"{album_key}{section_str}"
    for ext in [".m4a", ".mp3"]:
        exact_path = self.cache_dir / f"{cache_filename}{ext}"
        if exact_path.exists():
            return exact_path  # ← 精确命中！

    # 策略2：模糊匹配（文件名含专辑名+集数）
    for file_path in self.cache_dir.glob("*.m4a"):
        if album_key in file_path.stem.lower() and section_str in file_path.stem:
            return file_path  # ← 模糊命中！

    return None  # 缓存未命中

缓存生命周期：

下载完成→ m4a 文件写入 cache_dir → 永久保留（除非手动清理）
临时文件→ temp_cache_dir → 程序退出时清理

4.4 TTS 感知机制

代码位置：lanrentingshu_player.py 第600-610行

# TTS 正在说话 → 延迟启动播放
if self.app and self.app.is_speaking():
    self._deferred_start_path = file_path
    self._deferred_start_position = start_position
    self.is_playing = True
    self.paused = True  # 先标记暂停
    return True

恢复机制：外部（TTS 完成后）调用 resume() 时，从 _deferred_start_path 开始播放。

TTS 说话中→ 有声书延迟启动（paused=True）
   │
   ▼ (TTS 结束)
  resume() → 从 _deferred_start_path 开始播放

4.5 文件名清理

代码位置：lanrentingshu_player.py 第144-151行

def _sanitize_filename(self, filename: str) -> str:
    # 移除 Windows/Linux 不允许的字符
    filename = re.sub(r'[<>:"/\\|?*\x00-\x1f]', '_', filename)
    filename = filename.strip('. ')
    if len(filename) > 100:
        filename = filename[:100]  # 截断过长文件名
    return filename or "unknown"

4.6 单例模式

代码位置：lanrentingshu_player.py 第874-883行

_lanrentingshu_player_instance = None

def get_lanrentingshu_player_instance() -> LanrentingshuPlayer:
    global _lanrentingshu_player_instance
    if _lanrentingshu_player_instance is None:
        _lanrentingshu_player_instance = LanrentingshuPlayer()
    return _lanrentingshu_player_instance

Manager 同理（manager.py 第267-276行）：

_lanrentingshu_tools_manager = None

def get_lanrentingshu_tools_manager() -> LanrentingshuToolsManager:
    global _lanrentingshu_tools_manager
    if _lanrentingshu_tools_manager is None:
        _lanrentingshu_tools_manager = LanrentingshuToolsManager()
    return _lanrentingshu_tools_manager

为什么用单例：

• Player 保持播放状态（当前曲目、位置、暂停状态等）
• Manager 只是注册工具，单例避免重复注册

---

五、52api.cn 懒人听书 API 接口

5.1 通用请求格式

URL：https://www.52api.cn/api/lanren
方法：HTTP POST
认证：API Key 放在请求体中

{
  "key": "mmMltSARhyRBnd5LYy16TLPcqS",
  "type": "search|detail|stream",
  "keyword": "",
  "id": "",
  "stream_id": "",
  "stream_section": "",
  "page": "",
  "cookie": ""
}

5.2 三个 type 对比

type	作用	核心输入	核心输出
search	搜索专辑	keyword, page	lists[{id, name, announcer, author, cover, …}]
detail	专辑详情+集数	id（专辑ID）	lists[{stream_id, section, stream_name, size, …}]
stream	获取流URL	id, stream_id, stream_section	stream_url（音频直链）

5.3 通用响应格式

{
  "code": 200,        // 200=成功
  "msg": "success",
  "data": { ... },    // 根据 type 不同结构不同
  "exec_time": 1.267, // 服务端耗时
  "ip": "xxx",
  "copyRight": "本接口由我爱API(www.52api.cn)独家开发..."
}

---

六、与 Bilibili 工具的对比

维度	Bilibili	Lanrentingshu
数据来源	bilibili_api + yt-dlp	52api.cn HTTP API
内容类型	视频（需提取音频）	纯音频（M4A直链）
音频输出	PCM → ROS Topic /aima/hal/audio/playback	PCM → AudioCodec.write_pcm_direct()
视频播放	HTTP API 调机器人播放画面	无（纯音频）
缓存格式	.mp4 + .pcm 双文件	.m4a 单文件
解码	FFmpeg 视频→PCM	FFmpeg M4A→PCM
搜索API	bilibili_api Python库	52api.cn HTTP POST
播放控制	无暂停/恢复	有暂停/恢复（区分TTS/手动）
TTS感知	无	有（TTS说话时延迟启动）
Tool数量	3 个	7 个
单例	Player 单例	Player + Manager 双单例
后台任务	_background_download_and_play	_background_search_and_play

---

七、架构图

┌───────────────────────────────────────────────────────────────────────┐
│                           AI / 用户                                   │
└───────────────────────────────┬───────────────────────────────────────┘
                                │ MCP 协议 (tools/list, tools/call)
                                ▼
┌───────────────────────────────────────────────────────────────────────┐
│                      xiaozhi (McpServer)                              │
│  add_common_tools()                                                   │
│    → get_lanrentingshu_tools_manager()                                │
│    → manager.init_tools(add_tool, PropertyList, ...)                  │
└───────────────────────────────┬───────────────────────────────────────┘
                                │
                                ▼
┌───────────────────────────────────────────────────────────────────────┐
│                   LanrentingshuToolsManager                            │
│                                                                       │
│  ┌─ search_album_wrapper(args)        lanrentingshu.search_album    ─┐│
│  │   await player.search_album(keyword, page)                        ││
│  │   返回 "找到 358 个有声书专辑..."                                   ││
│  └────────────────────────────────────────────────────────────────────┘│
│                                                                       │
│  ┌─ get_album_detail_wrapper(args)    lanrentingshu.get_album_detail ─┐│
│  │   await player.get_album_detail(album_id)                          ││
│  │   返回 "专辑: 西游记, 共 100 集..."                                 ││
│  └────────────────────────────────────────────────────────────────────┘│
│                                                                       │
│  ┌─ search_and_play_wrapper(args)     lanrentingshu.search_and_play  ─┐│
│  │   await player.search_and_play(keyword, section)                   ││
│  │   返回 "正在播放: 西游记 - 第1集" 或 "正在下载中..."                ││
│  └────────────────────────────────────────────────────────────────────┘│
│                                                                       │
│  ┌─ pause_wrapper / resume_wrapper / stop_wrapper / get_status_wrapper ││
└───────────────────────────────┬───────────────────────────────────────┘
                                │ _player 引用
                                ▼
┌───────────────────────────────────────────────────────────────────────┐
│                   LanrentingshuPlayer (单例)                           │
│                                                                       │
│  API 层（52api.cn）                                                   │
│  ┌─────────────────────────────────────────────────────────────────┐  │
│  │ _call_api(payload) → HTTP POST → https://www.52api.cn/api/lanren │  │
│  │   search_album()    type=search   搜索专辑                       │  │
│  │   get_album_detail() type=detail   专辑详情+集数列表              │  │
│  │   get_stream_url()   type=stream   音频流直链URL                  │  │
│  └─────────────────────────────────────────────────────────────────┘  │
│                    │                                                   │
│                    ▼ (stream_url)                                      │
│  下载层                                                                │
│  ┌─────────────────────────────────────────────────────────────────┐  │
│  │ _download_file(url)                                              │  │
│  │   requests.get(stream=True) → 本地 .m4a                          │  │
│  │   缓存到 ~/.cache/xiaozhi/lanrentingshu/{专辑}_{集数:04d}.m4a    │  │
│  └─────────────────────────────────────────────────────────────────┘  │
│                    │                                                   │
│                    ▼ (.m4a 文件)                                       │
│  解码层                                                                │
│  ┌─────────────────────────────────────────────────────────────────┐  │
│  │ MusicDecoder (FFmpeg) .m4a → PCM int16 numpy array               │  │
│  │   sample_rate: 24000, channels: 1                                │  │
│  └─────────────────────────────────────────────────────────────────┘  │
│                    │                                                   │
│                    ▼ (asyncio.Queue<PCM frame>)                        │
│  播放层                                                                │
│  ┌─────────────────────────────────────────────────────────────────┐  │
│  │ _playback_loop()                                                 │  │
│  │   while is_playing:                                              │  │
│  │     pcm = await audio_queue.get()                                │  │
│  │     await audio_codec.write_pcm_direct(pcm)  → 机器人喇叭         │  │
│  └─────────────────────────────────────────────────────────────────┘  │
│                                                                       │
│  缓存目录: ~/.cache/xiaozhi/lanrentingshu/                             │
│  临时目录: ~/.cache/xiaozhi/lanrentingshu/temp/                        │
└───────────────────────────────────────────────────────────────────────┘

---

八、可复用的模式总结

8.1 后台任务模式（避免超时）

# 快速路径：有缓存直接返回
if cache_hit:
    return immediate_result

# 慢速路径：启动后台任务，立即返回
asyncio.create_task(self._background_work(...))
return {"status": "success", "message": "正在处理中...", "background": True}

8.2 统一 API 层

async def _call_api(self, payload):
    response = await asyncio.to_thread(requests.post, ...)
    return response.json()

• 所有 HTTP 调用走同一个入口
• asyncio.to_thread 包装同步 requests

8.3 播放器状态模型

状态: 未播放 → 播放中 ⇄ 已暂停 → 停止

暂停区分来源: "tts" (自动恢复) vs "manual" (用户手动恢复)

8.4 两级缓存查找

1. 精确匹配: {sanitized_keyword}_{section:04d}.m4a
2. 模糊匹配: 文件名含 keyword + section_str

8.5 Manager + Player 双单例

Manager (工具注册) ──→ Player (业务逻辑)
  单例                   单例