OpenClaw龙虾在飞书中可以发语音消息吗？

可以，参考这个SOP

# Feishu Voice SOP — 飞书语音消息技能

## 技能简介

将文本转换为语音消息，通过飞书 API 发送为真正的语音播放条（而非文件附件）。

**核心流程：**
```
文本 → gTTS 生成 MP3 → ffmpeg 转 Opus/OGG → 飞书 API 上传并发送
```

---

## 目录结构

```
~/.openclaw/skills/feishu-voice/
├── SKILL.md               ← 说明文档（供 AI 阅读）
├── SOP.md                 ← 本操作手册
└── scripts/
    ├── feishu_voice.py    ← TTS → Opus 生成（文本转语音）
    └── feishu_voice_api.py ← 飞书 API 发送（上传+发消息）
```

---

## 前置依赖

| 依赖 | 检查命令 | 安装命令 |
|------|---------|---------|
| ffmpeg | `which ffmpeg` | `yum install -y ffmpeg` |
| gtts | `python3 -c "import gtts"` | `pip install gtts` |
| requests | `python3 -c "import requests"` | `pip install requests` |

---

## 环境变量（必须配置）

| 变量名 | 说明 | 示例 |
|--------|------|------|
| `OPENCLAW_FEISHU_ACCOUNT_ID` | 飞书机器人账号名（对应 `openclaw.json` 中 `channels.feishu.accounts` 下的 key） | `coding` |
| `FEISHU_USER_ID` | 接收人的飞书 open_id | `ou_a34112ddf5e81c296cea080a4f3a9f4d` |

**配置方法：** 在调用脚本前设置，或写入 `~/.bashrc` / `~/.zshrc`。

---

## 标准操作流程

### 步骤 1：生成语音文件

```bash
python3 ~/.openclaw/skills/feishu-voice/scripts/feishu_voice.py "要说的文本内容" /tmp/voice
```

**输出：** `/tmp/voice.ogg`（Opus/OGG 格式，16kHz 单声道）

### 步骤 2：发送语音消息

```bash
FEISHU_USER_ID=<open_id> OPENCLAW_FEISHU_ACCOUNT_ID=<账号名> \
  python3 ~/.openclaw/skills/feishu-voice/scripts/feishu_voice_api.py audio /tmp/voice.ogg
```

**成功输出：**
```
=== 飞书消息发送 (audio) ===
账号: cli_xxx
文件: /tmp/voice.ogg
用户: ou_xxx
[1/3] ✓ token 获取成功
[2/3] ✓ 文件上传成功 (key: file_v3_...)
[3/3] 发送语音消息 (时长: 38142ms)...
  ✓ 消息发送成功! message_id: om_xxx
```

---

## OpenClaw Agent 内使用

当 AI 需要回复语音消息时，执行以下步骤：

```bash
# 1. 生成语音
TEXT="要说的内容"
python3 ~/.openclaw/skills/feishu-voice/scripts/feishu_voice.py "$TEXT" /tmp/voice

# 2. 发送语音（自动使用当前 agent 的飞书账号）
FEISHU_USER_ID=<open_id> \
  python3 ~/.openclaw/skills/feishu-voice/scripts/feishu_voice_api.py audio /tmp/voice.ogg
```

---

## 常见问题

### Q1：发送后显示为📎附件而不是语音条

**原因：** 没有正确传 `duration` 参数，或用了错误的发送方式。

**解决：** 确认使用 `feishu_voice_api.py audio` 命令，且 `file_type=opus`、`duration` 在 form data 中传递。**不要**使用飞书插件的 `sendFileFeishu`，那个不传 duration。

### Q2：`open_id cross app` 错误

**原因：** `open_id` 所属的应用和发送用的机器人账号不是同一个应用。

**解决：** 确保 `OPENCLAW_FEISHU_ACCOUNT_ID` 设置为该 `open_id` 所属的飞书机器人账号名（对应 `channels.feishu.accounts` 下的 key）。可在 `openclaw.json` 中查看有哪些可用账号。

### Q3：收到 `400 Bad Request` 或 `Access denied`

**原因：** 可能是 `receive_id_type` 不正确、文件格式不支持、或权限不足。

**排查顺序：**
1. 确认 `receive_id_type=open_id`
2. 确认音频是 Opus in OGG，16kHz，mono
3. 确认 bot 有 `im:message` 权限

### Q4：gTTS 生成慢或失败

**原因：** gTTS 需要访问 Google TTS 服务，网络问题。

**解决：** 检查网络，或考虑换用其他 TTS 引擎（如阿里云、腾讯云 TTS）。

---

## 飞书 API 关键参数

| 步骤 | API | file_type | msg_type | 特殊字段 |
|------|-----|-----------|----------|---------|
| 上传 | `POST /im/v1/files` | `opus` | — | `duration` 在 form data |
| 发消息 | `POST /im/v1/messages` | — | `audio` | `content = {file_key, duration}` |

---

## 配置文件 key 对照

| 配置文件（JSON） | 脚本变量 | 说明 |
|-----------------|---------|------|
| `channels.feishu.accounts.<name>.appId` | `acc["appId"]` | 应用 ID |
| `channels.feishu.accounts.<name>.appSecret` | `acc["appSecret"]` | 应用密钥 |
| `channels.feishu.appId`（顶层） | `feishu["appId"]` | 默认账号 |
| `channels.feishu.appSecret`（顶层） | `feishu["appSecret"]` | 默认账号密钥 |

> ⚠️ 配置文件中 key 是 **camelCase**（`appId`），传给 API 时要转成 **snake_case**（`app_id`）。

---

## 安全注意

- 脚本**不硬编码**任何凭证，所有凭证从 `~/.openclaw/openclaw.json` 读取
- `skills.entries.*.env` 只注入到 host 进程，不进入 sandbox
- 语音文件临时存放于 `/tmp`，不会持久化

--

#!/usr/bin/env python3
"""
飞书语音消息生成器
- 使用 gTTS 生成中文语音
- 用 ffmpeg 转换为飞书支持的 Opus 格式
"""

import sys
import tempfile
import os

try:
    from gtts import gTTS
except ImportError:
    print("ERROR: gTTS not installed. Run: pip install gtts")
    sys.exit(1)

def generate_voice(text: str, lang: str = "zh", output_path: str = None) -> str:
    """
    生成飞书语音消息音频文件（Opus/OGG格式）
    
    Args:
        text: 要转换的文本
        lang: 语言代码，默认中文(zh)
        output_path: 输出文件路径（不含扩展名），默认临时文件
    
    Returns:
        生成的 .ogg 文件路径
    """
    if not text or not text.strip():
        raise ValueError("文本不能为空")
    
    # 创建临时文件用于存放 mp3
    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as mp3_tmp:
        mp3_path = mp3_tmp.name
    
    try:
        # 生成语音
        tts = gTTS(text=text, lang=lang, slow=False)
        tts.save(mp3_path)
        print(f"[TTS] 语音生成完成: {mp3_path}")
        
        # 确定输出路径
        if output_path is None:
            ogg_path = mp3_path.replace(".mp3", ".ogg")
        else:
            ogg_path = output_path if output_path.endswith(".ogg") else f"{output_path}.ogg"
        
        # 用 ffmpeg 转换为 Opus 编码的 OGG
        # 飞书语音消息需要 Opus in OGG container
        cmd = [
            "ffmpeg",
            "-y",                      # 覆盖输出文件
            "-i", mp3_path,            # 输入 MP3
            "-c:a", "libopus",         # 使用 Opus 编码器
            "-b:a", "32k",             # 比特率 32k（语音够用）
            "-ar", "16000",            # 采样率 16k（飞书语音标准）
            "-ac", "1",                # 单声道
            "-application", "voip",    # 优化语音
            ogg_path                   # 输出
        ]
        
        import subprocess
        result = subprocess.run(cmd, capture_output=True, text=True)
        
        if result.returncode != 0:
            print(f"[FFmpeg ERROR]: {result.stderr}", file=sys.stderr)
            raise RuntimeError(f"ffmpeg 转换失败: {result.stderr}")
        
        print(f"[FFmpeg] 转换完成: {ogg_path}")
        return ogg_path
        
    finally:
        # 清理临时 MP3
        if os.path.exists(mp3_path):
            os.remove(mp3_path)

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python3 feishu_voice.py <文本> [输出路径]")
        print("示例: python3 feishu_voice.py '你好，这是测试' /tmp/voice")
        sys.exit(1)
    
    text = sys.argv[1]
    output = sys.argv[2] if len(sys.argv) > 2 else None
    
    try:
        path = generate_voice(text, output_path=output)
        print(f"SUCCESS: {path}")
    except Exception as e:
        print(f"FAILED: {e}")
        sys.exit(1)

--

#!/usr/bin/env python3
"""
飞书语音消息发送器 - 直接调用飞书 API
流程: 获取 token → 上传音频文件 → 发送语音消息

凭证从 OpenClaw 配置文件读取，不硬编码。
"""

import sys
import json
import os
import requests
import subprocess

# 配置文件路径
CONFIG_PATH = os.path.expanduser("~/.openclaw/openclaw.json")
BASE_URL = "https://open.feishu.cn/open-apis"

def load_feishu_config(account_id: str = None) -> dict:
    """
    从 OpenClaw 配置文件读取飞书凭证。

    优先级（由高到低）：
    1. 环境变量 OPENCLAW_FEISHU_ACCOUNT_ID 指定账号名
    2. 参数 account_id（调用方传入）
    3. channels.feishu.accounts.<account_id>（需 account_id 不为空）
    4. channels.feishu（顶层默认账号配置）
    """
    if not os.path.exists(CONFIG_PATH):
        raise FileNotFoundError(f"配置文件不存在: {CONFIG_PATH}")
    
    with open(CONFIG_PATH) as f:
        config = json.load(f)
    
    feishu = config.get("channels", {}).get("feishu", {})
    accounts = feishu.get("accounts", {})
    
    # 环境变量最高优先；其次用调用方传入的 account_id
    effective_account = os.environ.get("OPENCLAW_FEISHU_ACCOUNT_ID", account_id)
    
    # 如果有有效的账号名，优先从 accounts 里查找
    if effective_account and effective_account in accounts:
        acc = accounts[effective_account]
        if acc.get("appId") and acc.get("appSecret"):
            return {"app_id": acc["appId"], "app_secret": acc["appSecret"]}

    # 兜底：尝试顶层配置（默认账号）—— 注意 key 是 camelCase
    if feishu.get("appId") and feishu.get("appSecret"):
        return {"app_id": feishu["appId"], "app_secret": feishu["appSecret"]}
    
    raise ValueError(
        f"未在 ~/.openclaw/openclaw.json 中找到飞书凭证。"
        f"请确保存在 channels.feishu.accounts.<账号名> 或 channels.feishu 顶层配置，"
        f"并通过环境变量 OPENCLAW_FEISHU_ACCOUNT_ID 指定账号名。"
    )

def get_tenant_access_token(app_id: str, app_secret: str) -> str:
    """获取 tenant access token"""
    url = f"{BASE_URL}/auth/v3/tenant_access_token/internal"
    resp = requests.post(url, json={"app_id": app_id, "app_secret": app_secret}, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"获取 token 失败: {data}")
    return data["tenant_access_token"]

def upload_file(token: str, file_path: str) -> str:
    """
    上传音频文件到飞书，返回 file_key
    file_type: opus=语音消息, mp4=视频, stream=其他文件
    """
    url = f"{BASE_URL}/im/v1/files"
    headers = {"Authorization": f"Bearer {token}"}
    
    file_size = os.path.getsize(file_path)
    file_name = os.path.basename(file_path)
    duration_ms = get_duration_ms(file_path)
    
    with open(file_path, "rb") as f:
        files = {
            "file": (file_name, f, "audio/ogg"),
            "file_name": (None, file_name),
            "file_type": (None, "opus"),
        }
        data = {"duration": str(duration_ms)}
        resp = requests.post(url, headers=headers, files=files, data=data, timeout=30)
    
    resp.raise_for_status()
    result = resp.json()
    if result.get("code") != 0:
        raise RuntimeError(f"上传文件失败: {result}")
    return result["data"]["file_key"]

def get_duration_ms(file_path: str) -> int:
    """用 ffprobe 获取音频时长（毫秒）"""
    try:
        result = subprocess.run(
            ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", file_path],
            capture_output=True, text=True, check=True
        )
        data = json.loads(result.stdout)
        duration = float(data["format"]["duration"])
        return int(duration * 1000)
    except Exception as e:
        print(f"[WARN] ffprobe failed: {e}, using default 1000ms")
        return 1000

def send_audio_message(token: str, receive_id: str, file_key: str, duration_ms: int):
    """发送语音消息"""
    url = f"{BASE_URL}/im/v1/messages"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    content = json.dumps({
        "file_key": file_key,
        "duration": duration_ms
    })
    
    params = {"receive_id_type": "open_id"}
    payload = {
        "receive_id": receive_id,
        "msg_type": "audio",
        "content": content
    }
    
    resp = requests.post(url, headers=headers, params=params, json=payload, timeout=15)
    resp.raise_for_status()
    result = resp.json()
    if result.get("code") != 0:
        raise RuntimeError(f"发送消息失败: {result}")
    return result["data"]["message_id"]

def send_file_attachment(token: str, receive_id: str, file_key: str, file_name: str):
    """发送文件附件"""
    url = f"{BASE_URL}/im/v1/messages"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    content = json.dumps({
        "file_key": file_key,
        "file_name": file_name
    })
    
    params = {"receive_id_type": "open_id"}
    payload = {
        "receive_id": receive_id,
        "msg_type": "file",
        "content": content
    }
    
    resp = requests.post(url, headers=headers, params=params, json=payload, timeout=15)
    resp.raise_for_status()
    result = resp.json()
    if result.get("code") != 0:
        raise RuntimeError(f"发送文件失败: {result}")
    return result["data"]["message_id"]

def main():
    if len(sys.argv) < 2:
        print("用法:")
        print("  发送语音消息: python3 feishu_voice_api.py audio <音频文件.ogg> [用户open_id]")
        print("  发送文件附件: python3 feishu_voice_api.py file <文件路径> [用户open_id]")
        print("")
        print("凭证从 ~/.openclaw/openclaw.json 自动读取")
        print("可通过环境变量 OPENCLAW_FEISHU_ACCOUNT_ID 指定飞书账号")
        print("目标用户 open_id：优先从 argv > 环境变量 FEISHU_USER_ID > 配置文件")
        sys.exit(1)
    
    msg_type = sys.argv[1]
    file_path = sys.argv[2]
    user_id = sys.argv[3] if len(sys.argv) > 3 else os.environ.get("FEISHU_USER_ID")
    
    if not os.path.exists(file_path):
        print(f"错误: 文件不存在 {file_path}")
        sys.exit(1)
    
    if user_id is None:
        print("错误: 必须指定用户 open_id（作为第3个参数或设置 FEISHU_USER_ID 环境变量）")
        sys.exit(1)
    
    # 从配置文件读取凭证
    creds = load_feishu_config()
    print(f"=== 飞书消息发送 ({msg_type}) ===")
    print(f"账号: {creds['app_id']}")
    print(f"文件: {file_path}")
    print(f"用户: {user_id}")
    
    # 获取 token
    token = get_tenant_access_token(creds["app_id"], creds["app_secret"])
    print("\n[1/3] ✓ token 获取成功")
    
    # 上传文件
    file_key = upload_file(token, file_path)
    print(f"[2/3] ✓ 文件上传成功 (key: {file_key[:20]}...)")
    
    # 发送消息
    if msg_type == "audio":
        duration_ms = get_duration_ms(file_path)
        print(f"[3/3] 发送语音消息 (时长: {duration_ms}ms)...")
        msg_id = send_audio_message(token, user_id, file_key, duration_ms)
        print(f"  ✓ 消息发送成功! message_id: {msg_id}")
    elif msg_type == "file":
        file_name = os.path.basename(file_path)
        print(f"[3/3] 发送文件附件 ({file_name})...")
        msg_id = send_file_attachment(token, user_id, file_key, file_name)
        print(f"  ✓ 文件发送成功! message_id: {msg_id}")
    else:
        print(f"错误: 未知的消息类型 {msg_type}")
        sys.exit(1)

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"\n✗ 错误: {e}")
        sys.exit(1)