audio_codebuddy/mcp-for-xiaozhi/push_book.py

"""
push_book.py — 书籍生成完成后，主动推送到小智音箱

使用小智服务端的新 /xiaozhi/push 端点。
当 book 生成完毕后，调用此脚本推送给音箱自动朗读。

用法：
    python push_book.py <book_id> [--device-id DEVICE_ID] [--server XIAOZHI_SERVER]

环境变量：
    XIAOZHI_SERVER: 小智服务端地址（默认 http://127.0.0.1:8003）
    DEVICE_ID: ESP32 设备ID
    AUDIOBOOK_API_BASE: 后端 API 地址
"""
from __future__ import annotations

import json
import os
import sys
import time
import urllib.request

AUDIOBOOK_API = os.environ.get("AUDIOBOOK_API_BASE", "http://127.0.0.1:38381")
XIAOZHI_SERVER = os.environ.get("XIAOZHI_SERVER", "http://127.0.0.1:8003")
DEVICE_ID = os.environ.get("DEVICE_ID", "")


def http_get(path: str) -> dict:
    with urllib.request.urlopen(f"{AUDIOBOOK_API}{path}", timeout=10) as r:
        return json.loads(r.read().decode("utf-8"))


def push_to_device(device_id: str, text: str) -> bool:
    """调小智服务端 POST /xiaozhi/push"""
    url = f"{XIAOZHI_SERVER}/xiaozhi/push"
    payload = json.dumps({
        "device_id": device_id,
        "text": text,
        "type": "chat",
    }).encode("utf-8")
    req = urllib.request.Request(url, data=payload, method="POST",
                                 headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            resp = json.loads(r.read().decode("utf-8"))
            return resp.get("code") == 0
    except Exception as e:
        print(f"推送失败: {e}", file=sys.stderr)
        return False


def collect_book_text(book_id: int) -> str | None:
    """收集书籍所有章节的文本"""
    data = http_get(f"/api/book-generator/books/{book_id}")
    info = data.get("data") or {}
    if not info:
        print(f"书籍 {book_id} 不存在", file=sys.stderr)
        return None
    chapters = info.get("chapters") or []
    if not chapters:
        print(f"书籍 {book_id} 尚无章节", file=sys.stderr)
        return None

    title = info.get("title") or "未命名"
    lines = [f"《{title}》已生成，共{len(chapters)}章。现在开始朗读：\n"]
    for i, ch in enumerate(chapters):
        ch_data = http_get(f"/api/book-generator/books/chapters/{ch['id']}")
        ch_info = ch_data.get("data") or {}
        ch_title = ch_info.get("title") or f"第{i+1}章"
        ch_content = ch_info.get("content") or ch_info.get("text") or ""
        # 每章取前 500 字推送给音箱
        snippet = ch_content[:500] if ch_content else "(暂无内容)"
        lines.append(f"{ch_title}\n\n{snippet}\n")
        if len(chapters) > 1 and i < len(chapters) - 1:
            lines.append("[请继续朗读下一章]\n")
        else:
            lines.append("[全书朗读完毕]")

    return "\n".join(lines)


def wait_and_push(book_id: int, device_id: str, timeout: int = 600) -> bool:
    """轮询等生成完成，然后推送"""
    deadline = time.time() + timeout
    print(f"等待书籍 {book_id} 生成完成（最多 {timeout} 秒）...", file=sys.stderr)
    while time.time() < deadline:
        data = http_get(f"/api/book-generator/books/{book_id}")
        info = data.get("data") or {}
        progress = info.get("progress", 0)
        chapters = info.get("chapters") or []
        if chapters and progress >= 30:
            # 验证真有内容
            fd = http_get(f"/api/book-generator/books/chapters/{chapters[0]['id']}")
            fc = ((fd.get("data") or {}).get("content") or (fd.get("data") or {}).get("text") or "")
            if fc and len(fc) > 50:
                text = collect_book_text(book_id)
                if text:
                    print(f"推送中 ({len(text)} 字)...", file=sys.stderr)
                    return push_to_device(device_id, text)
        time.sleep(10)
    print("超时", file=sys.stderr)
    return False


def main() -> int:
    if len(sys.argv) < 2:
        print("Usage: python push_book.py <book_id> [--device-id ID] [--server URL]")
        return 2

    book_id = int(sys.argv[1])
    device_id = DEVICE_ID
    server = XIAOZHI_SERVER

    # 解析命令行参数
    args = sys.argv[2:]
    i = 0
    while i < len(args):
        if args[i] == "--device-id" and i + 1 < len(args):
            device_id = args[i + 1]
            i += 2
        elif args[i] == "--server" and i + 1 < len(args):
            server = args[i + 1]
            os.environ["XIAOZHI_SERVER"] = server
            i += 2
        else:
            i += 1

    if not device_id:
        print("错误: 需要 device_id。用 --device-id 指定或在 .env 设置 DEVICE_ID", file=sys.stderr)
        print("提示: device_id 是小智音箱的唯一标识，在 app 或控制台可查看", file=sys.stderr)
        return 2

    print(f"XIAOZHI_SERVER={server}", file=sys.stderr)
    print(f"DEVICE_ID={device_id}", file=sys.stderr)

    ok = wait_and_push(book_id, device_id)
    if ok:
        print("推送成功！音箱应该开始朗读了")
        return 0
    else:
        print("推送失败")
        return 1


if __name__ == "__main__":
    sys.exit(main())