本文档详细阐述了如何利用 Python 结合
edge-tts与飞书开放平台 API,实现从文本生成到语音消息发送的完整自动化流程。
1. 概述
本方案通过以下核心步骤实现语音消息推送:
-
语音合成:利用
edge-tts将文本转换为高质量 MP3 音频。 -
文件上传:调用飞书 API 将音频文件上传至云端获取
file_key。 -
消息发送:构造
audio类型消息并推送给指定用户。
核心流程图
graph TD
A[用户请求/文本输入] --> B[edge-tts 生成 MP3]
B --> C[获取飞书 Access Token]
C --> D[上传 MP3 文件\n(file_type=opus)]
D --> E[获取 file_key]
E --> F[发送 audio 类型消息]
F --> G[用户在飞书收到语音]
2. 前置准备
2.1 飞书应用配置
需在 完成以下配置:
创建应用:新建“自建应用”,并添加 机器人 能力。
权限配置:在“权限管理”中开通以下接口权限:
-
im:message(发送消息) -
im:message.p2p_msg(发送单聊消息) -
im:message.group_at_msg(发送群聊@消息)
事件订阅:
-
添加事件
im.message.receive_v1。 -
事件配置选择 WebSocket 长连接(若仅需发送消息,此步可选,但建议配置以便双向交互)。
获取凭证:发布应用后,记录 App ID 和 App Secret。
2.2 环境依赖
确保本地已安装 Python 及以下库:
1pip install edge-tts aiohttp
2.3 敏感信息管理
建议将敏感信息存储在本地文件中,避免硬编码在脚本里。
目录结构:~/.clawdbot/secrets/ (Windows: C:\Users\Administrator\.clawdbot\secrets\)
存储内容:
-
feishu_app_secret: 存放 App Secret。 -
(可选)
feishu_app_id: 存放 App ID。
2.4 获取用户 OpenID
飞书消息发送需指定接收者的 open_id。
-
方法:让用户给机器人发送任意消息。
-
查询:在飞书开放平台“事件订阅” -> “事件日志”中查看最近事件。
-
提取路径:
event.message.sender.sender_id.open_id(格式通常为ou_xxxxxxxxxx)。
3. 核心实现代码
以下脚本实现了完整的语音生成、上传及发送流程。
import asyncio
import aiohttp
import json
import os
import edge_tts
# ==================== 配置区域 ====================
# 替换为实际用户的 open_id
USER_OPEN_ID = "ou_xxxxxxxxxx"
# 替换为实际 App ID
APP_ID = "cli_a9xxxx28cf9dbce"
# 读取本地存储的 App Secret (根据实际路径调整)
SECRET_PATH = r"C:\Users\Administrator\.clawdbot\secrets\feishu_app_secret"
with open(SECRET_PATH, "r", encoding='utf-8') as f:
APP_SECRET = f.read().strip()
# 临时音频文件路径
AUDIO_FILE_PATH = r"./temp_voice.mp3"
# ================================================
async def get_access_token():
"""获取飞书访问令牌 (有效期 2 小时)"""
url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
payload = {"app_id": APP_ID, "app_secret": APP_SECRET}
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload) as resp:
result = await resp.json()
if result.get("code") == 0:
return result.get("tenant_access_token")
else:
print(f"获取 Token 失败: {result}")
return None
async def upload_file(token, file_path):
"""上传音频文件到飞书"""
url = "https://open.feishu.cn/open-apis/im/v1/files"
# 读取文件内容
with open(file_path, "rb") as f:
file_content = f.read()
# 构建 FormData
form = aiohttp.FormData()
form.add_field('file', file_content, filename=os.path.basename(file_path), content_type='audio/mpeg')
# ⚠️ 关键点:飞书语音消息上传必须指定 file_type 为 'opus',即使源文件是 mp3
form.add_field('file_type', 'opus')
headers = {"Authorization": f"Bearer {token}"}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, data=form) as resp:
result = await resp.json()
if result.get("code") == 0:
return result["data"]["file_key"]
else:
print(f"上传文件失败: {result}")
return None
async def send_audio_message(token, receive_id, file_key):
"""发送音频消息"""
url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=open_id"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
# 消息体构造
data = {
"receive_id": receive_id,
"msg_type": "audio",
"content": json.dumps({"file_key": file_key})
}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=data) as resp:
result = await resp.json()
if result.get("code") == 0:
return True
else:
print(f"发送消息失败: {result}")
return False
async def generate_and_send(text, voice_name="zh-CN-XiaoxiaoNeural"):
"""主流程:生成语音并发送"""
print(f"正在生成语音:{text} ...")
# 1. 使用 edge-tts 生成 MP3
communicate = edge_tts.Communicate(text, voice_name)
await communicate.save(AUDIO_FILE_PATH)
print("语音文件生成完毕。")
# 2. 获取 Access Token
token = await get_access_token()
if not token:
return False
# 3. 上传文件
file_key = await upload_file(token, AUDIO_FILE_PATH)
if not file_key:
return False
print(f"文件上传成功,File Key: {file_key}")
# 4. 发送消息
success = await send_audio_message(token, USER_OPEN_ID, file_key)
# 5. 清理临时文件
if os.path.exists(AUDIO_FILE_PATH):
os.remove(AUDIO_FILE_PATH)
if success:
print("✅ 语音消息发送成功!")
else:
print("❌ 发送失败。")
return success
if __name__ == "__main__":
# 测试文本
test_text = "你好!这是通过 Python 自动发送的飞书语音消息测试。"
# 运行异步主函数
asyncio.run(generate_and_send(test_text))
4. 进阶场景:语音 + 文字组合发送
在实际业务中,通常需要同时发送文字说明和语音消息。由于飞书 API 限制,需分两次调用发送接口。
实现逻辑:
-
执行上述
generate_and_send流程发送语音。 -
调用飞书
msg_type: "text"接口发送配套文字。
代码片段示例:
async def send_text_message(token, receive_id, text):
url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=open_id"
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
data = {
"receive_id": receive_id,
"msg_type": "text",
"content": json.dumps({"text": text})
}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=data) as resp:
return (await resp.json()).get("code") == 0
# 在主流程中调用
# ... (生成语音和获取 file_key 后)
await send_audio_message(token, USER_OPEN_ID, file_key) # 发语音
await send_text_message(token, USER_OPEN_ID, f"🎧 语音回复:{test_text}") # 发文字
5. 可用音色列表 (edge-tts)
edge-tts 支持多种高质量的中文神经语音,常用选项如下:
| 音色代码 | 名称 | 风格描述 | 推荐场景 |
|---|---|---|---|
zh-CN-XiaoxiaoNeural |
晓晓 | 女声,温暖、自然 | 默认推荐,通用场景 |
zh-CN-YunjianNeural |
云健 | 男声,磁性、稳重 | 新闻播报、正式通知 |
zh-CN-XiaoyiNeural |
小艺 | 女声,活泼、亲切 | 客服助手、轻松对话 |
zh-CN-YunxiNeural |
云希 | 男声,温柔、柔和 | 情感陪伴、故事讲述 |
6. 常见问题排查 (FAQ)
Q1: 上传文件失败,错误码 234001
原因:file_type 参数不正确。
解决:飞书语音消息上传接口强制要求 file_type 参数为 "opus",即使你上传的是 .mp3 文件。请检查代码中 form.add_field('file_type', 'opus') 是否已正确设置。
Q2: 发送消息失败,错误码 99991664 或 99991663
原因:应用缺乏相应的消息发送权限,或未发布应用。
解决:
-
检查飞书开放平台是否已开启
im:message和im:message.p2p_msg权限。 -
确认应用状态是否为 “已发布”(开发版仅限开发者本人接收消息)。
-
确认接收者是否已在企业中,且未拉黑机器人。
Q3: 如何获取用户的 open_id?
解决:
-
让目标用户在飞书中给机器人发送一条消息(任意内容)。
-
登录飞书开放平台 -> 点击应用 -> 事件订阅 -> 事件日志。
-
找到最近的
im.message.receive_v1事件,展开 JSON,查找event.message.sender.sender_id.open_id字段。
Q4: Token 有效期问题
说明:tenant_access_token 有效期为 2 小时。
策略:生产环境中建议增加缓存机制,Token 过期后再重新请求,避免每次发消息都请求接口触发频率限制。
7. 快速启动命令
# 1. 安装依赖
pip install edge-tts aiohttp
# 2. 确保已配置好 secrets 文件和代码中的 APP_ID/USER_OPEN_ID
# 3. 运行脚本
python send_voice.py