MindOS_CLI/mindcli/cli.py

"""
Mind CLI — 拾音采集端（内容枢纽时代 v0.3.0 瘦身版）。

定位（SPEC_mindos_content_hub 决策）：MindOS 不再自建"本地执行体"，
MindCLI 降级为**最小系统拾音工具**：采集系统音频/麦克风 → 实时流式
推送到 Cloud ASR（dashscope_realtime 管线），转写由云端负责。

保留（2026-08 决策）：
  - record 命令族（实时流式，不落盘）
  - _vendor/ 快照机制（scripts/vendor_hermes.sh）
砍掉（宿主层由 WorkBuddy 等 MCP 消费端承接）：
  - chat/ask 命令集、tunnel、tool_proxy、capability
  - 守护进程（8660 health server）、launchd 服务（service.py）
  - pipx 自更新（update）
"""

import asyncio
import json
import os
import signal
import sys
import time
from pathlib import Path

import click

# 确保 _vendor/ 已注入 sys.path
import mindcli  # noqa: F401 — 触发 __init__.py 的 sys.path 注入


# ── 常量 ──────────────────────────────────────────────────────
_RECORD_WS_BASE = os.environ.get(
    "MINDOS_RECORD_WS_URL",
    "wss://agent.brainwork.club/mindos-next/ws/record",
)
_STATE_DIR = Path.home() / ".mindcli"
_STATE_FILE = _STATE_DIR / "record.json"


def _get_jwt() -> str | None:
    """获取 MindPass JWT（环境变量 MINDOS_JWT）。"""
    return os.environ.get("MINDOS_JWT")


def _load_state() -> dict | None:
    try:
        with open(_STATE_FILE) as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return None


def _save_state(state: dict) -> None:
    _STATE_DIR.mkdir(parents=True, exist_ok=True)
    with open(_STATE_FILE, "w") as f:
        json.dump(state, f, ensure_ascii=False)


def _clear_state() -> None:
    try:
        _STATE_FILE.unlink()
    except FileNotFoundError:
        pass


def _pid_alive(pid: int) -> bool:
    try:
        os.kill(pid, 0)
        return True
    except ProcessLookupError:
        return False
    except PermissionError:
        return True


@click.group()
def main():
    """MindOS 拾音采集端 — 实时流式推送 Cloud ASR。"""


@main.group()
def record():
    """管理系统录音（实时流式 → Cloud ASR，不落盘）。"""


@record.command(name="start")
@click.option("--token", default="", help="MindPass JWT（缺省读 MINDOS_JWT）")
@click.option("--chat-id", default="", help="对话 ID")
@click.option("--source", type=click.Choice(["system", "mic"]), default="system",
              help="音频源：system=系统音频（默认），mic=麦克风")
def record_start(token, chat_id, source):
    """开始录音（前台运行，Ctrl+C 停止）。

    实时流式：采集 PCM16 帧 → WS 直推 Cloud ASR，不落盘。
    """
    jwt = token or _get_jwt()
    if not jwt:
        click.echo("❌ 未找到 JWT：请传 --token 或设置环境变量 MINDOS_JWT")
        sys.exit(1)

    meeting_id = f"cli_rec_{int(time.time() * 1000)}"
    ws_url = (
        f"{_RECORD_WS_BASE}?token={jwt}&chatId={chat_id}"
        f"&meetingId={meeting_id}&source={source}"
    )

    # 保存状态（pid 供 record stop/status 跨进程识别）
    _save_state({
        "pid": os.getpid(),
        "meetingId": meeting_id,
        "chatId": chat_id,
        "source": source,
        "startedAt": time.time(),
    })

    def _on_text(msg_type: str, text: str) -> None:
        click.echo(f"[{msg_type}] {text}")

    try:
        from mindcli.pipelines.audio_capture import capture as capture_audio
        asyncio.run(_run_capture(
            capture_audio, ws_url, chat_id, meeting_id, source, _on_text,
        ))
    except KeyboardInterrupt:
        pass
    finally:
        _clear_state()


async def _run_capture(capture_audio, ws_url, chat_id, meeting_id, source, on_text):
    """前台录音循环：启动采集 → 等 SIGINT/SIGTERM → 优雅停止。"""
    handle = await capture_audio(
        ws_url=ws_url, chat_id=chat_id, meeting_id=meeting_id,
        source=source, on_text=on_text,
    )
    click.echo(f"🎙️ 录音已开始 source={source} meetingId={meeting_id}（Ctrl+C 停止）")

    stop_event = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, stop_event.set)
    loop.add_signal_handler(signal.SIGTERM, stop_event.set)

    try:
        await stop_event.wait()
    finally:
        await handle.stop()
        click.echo(f"⏹️ 录音已停止 meetingId={meeting_id}")


@record.command(name="stop")
def record_stop():
    """停止录音（向 record start 进程发 SIGTERM）。"""
    state = _load_state()
    if not state or not _pid_alive(state.get("pid", -1)):
        _clear_state()
        click.echo("⏹️ 未在录音")
        return
    os.kill(state["pid"], signal.SIGTERM)
    click.echo(f"⏹️ 已发送停止信号 pid={state['pid']}")


@record.command(name="status")
def record_status():
    """查看录音状态。"""
    state = _load_state()
    if not state or not _pid_alive(state.get("pid", -1)):
        click.echo("⏹️ 未在录音")
        return
    duration = round(time.time() - state.get("startedAt", time.time()), 1)
    click.echo(f"🎙️ 录音中 duration={duration}s chatId={state.get('chatId', '')} "
               f"source={state.get('source')} meetingId={state.get('meetingId')}")


if __name__ == "__main__":
    main()