本教程旨在指导您如何使用Python构建一个Telegram机器人，通过集成python-telegram-bot和pytgcalls库，实现基于聊天命令或外部事件触发的自动化语音通知功能。我们将重点讲解pytgcalls的配置、用户会话管理以及如何在Telegram群组语音聊天中播放预录消息，帮助您从零开始搭建一个能够进行语音交互的智能通知系统。
引言：自动化Telegram语音通知的挑战与机遇
在日常工作和生活中，我们常常需要及时获取关键信息并采取行动。传统的文本通知虽然便捷，但在某些紧急或需要即时关注的场景下，语音通知能提供更强的提醒效果。设想一个场景：当特定天气条件（如气温骤降）发生时，您的Telegram机器人能自动发起一个语音通话，播放“请注意保暖，出门记得穿外套”的提醒。这种自动化语音通知不仅能提升信息传达的效率，还能为用户带来更个性化、更直观的体验。
实现这一目标，我们需要结合Telegram Bot API进行消息交互，并利用Telegram的用户客户端功能来处理语音通话。python-telegram-bot库为前者提供了强大的支持，而pytgcalls库则允许我们以编程方式控制Telegram群组语音聊天，从而实现播放音频的需求。本教程将引导您完成从环境搭建到代码实现的整个过程。
核心库介绍与环境准备
要构建我们的Telegram语音通知机器人，主要依赖以下Python库：
立即学习“Python免费学习笔记（深入）”；
- python-telegram-bot: 用于与Telegram Bot API交互，处理消息、命令等。
- pytgcalls: 用于控制Telegram群组语音聊天，包括加入、播放音频和离开。
- pyrogram (或 telethon): pytgcalls的底层依赖，用于创建和管理Telegram用户会话。pytgcalls需要一个用户会话来模拟一个真实用户加入语音聊天。
安装所有必要的库：
pip install python-telegram-bot pytgcalls pyrogram ffmpeg-python
FFmpeg要求：pytgcalls依赖FFmpeg来处理音频流。请确保您的系统上安装了FFmpeg。您可以从FFmpeg官网下载并安装，或使用系统包管理器（如sudo apt install ffmpeg在Debian/Ubuntu上，brew install ffmpeg在macOS上）。
pytgcalls认证与会话管理
pytgcalls需要一个Telegram用户账户的授权才能操作语音聊天。这与机器人账户（通过BotFather创建）不同。我们需要获取以下凭证：
- api_id 和 api_hash: 这是Telegram API的应用程序凭证。访问my.telegram.org登录您的Telegram账户，然后创建一个新的应用程序即可获得。
- session_string: 这是您的用户账户的授权会话字符串。pytgcalls使用它来模拟您的账户加入语音聊天。由于session_string包含了您的账户授权信息，请务必妥善保管，切勿泄露。
生成 session_string：
我们将使用pyrogram库来生成session_string。创建一个名为 generate_session.py 的文件：
from pyrogram import Client
API_ID = YOUR_API_ID # 替换为您的api_id
API_HASH = "YOUR_API_HASH" # 替换为您的api_hash
async def generate_session():
# 'my_account' 是会话文件的名称，可以随意命名
async with Client("my_account", api_id=API_ID, api_hash=API_HASH) as app:
print("请在控制台输入您的手机号（带国家代码，如 +8613800138000）和验证码。")
session_string = await app.export_session_string()
print(f"\n您的 session_string 是：\n{session_string}\n")
print("请将此 session_string 复制到您的配置文件中。")
if __name__ == "__main__":
import asyncio
asyncio.run(generate_session())运行此脚本：python generate_session.py。按照提示输入您的手机号和Telegram发送的验证码。成功后，脚本将输出您的session_string。
构建Telegram机器人基础框架
首先，您需要通过与BotFather对话来创建一个新的Telegram机器人，并获取其BOT_TOKEN。
创建一个 config.py 文件来存储所有敏感信息：
# config.py BOT_TOKEN = "YOUR_BOT_TOKEN" # 从BotFather获取 API_ID = YOUR_API_ID # 从my.telegram.org获取 API_HASH = "YOUR_API_HASH" # 从my.telegram.org获取 SESSION_STRING = "YOUR_SESSION_STRING" # 通过generate_session.py生成 # 目标群组ID，机器人和您的用户账户都必须是这个群组的成员 # 且您的用户账户（通过pytgcalls控制的）需要有管理语音聊天的权限 TARGET_CHAT_ID = -1001234567890 # 替换为您的目标群组ID，通常以-100开头
注意：
- TARGET_CHAT_ID 必须是一个群组的ID。pytgcalls主要用于管理群组语音聊天，不直接支持向个人用户发起一对一的语音通话。
- 您的用户账户（由pytgcalls控制的那个）必须是 TARGET_CHAT_ID 群组的成员，并且拥有发起和管理语音聊天的权限。
- 机器人账户也必须是 TARGET_CHAT_ID 群组的成员。
实现语音通话功能：pytgcalls实战
现在，我们将把机器人逻辑和pytgcalls功能结合起来。
创建一个 main.py 文件：
# main.py
import logging
import asyncio
from pyrogram import Client
from pytgcalls import PyTgCalls, idle
from pytgcalls.types import AudioPiped
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
from config import BOT_TOKEN, API_ID, API_HASH, SESSION_STRING, TARGET_CHAT_ID
# 配置日志
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logging.getLogger("httpx").setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
# 初始化 Pyrogram 客户端 (用于 pytgcalls)
pyrogram_client = Client(
name="my_account",
api_id=API_ID,
api_hash=API_HASH,
session_string=SESSION_STRING,
in_memory=True # 如果不想生成session文件，可以设置为True
)
# 初始化 PyTgCalls 客户端
pytgcalls_client = PyTgCalls(pyrogram_client)
# 定义一个异步函数来处理 !call 命令
async def call_me(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""处理 !call 命令，加入群组语音聊天并播放音频。"""
user_id = update.effective_user.id
logger.info(f"用户 {user_id} 发送了 !call 命令。")
# 检查是否已在语音聊天中
if pytgcalls_client.is_connected:
await update.message.reply_text("机器人已在语音聊天中，请稍后再试。")
return
try:
await update.message.reply_text(f"正在尝试加入群组 {TARGET_CHAT_ID} 的语音聊天并播放提醒...")
# 加入语音聊天并播放音频
# 这里的 'audio.mp3' 是您预录制的音频文件路径
# 确保音频文件存在，并且是 pytgcalls 支持的格式 (如 MP3, WAV)。
# 建议使用 FFmpeg 将音频转换为 48kHz, mono, 16bit PCM 或 Opus 编码。
# 例如：ffmpeg -i input.wav -acodec libopus -b:a 32k -ar 48000 -ac 1 output.ogg
await pytgcalls_client.join_group_call(
TARGET_CHAT_ID,
AudioPiped("audio.mp3") # 替换为您的音频文件路径
)
logger.info(f"成功加入群组 {TARGET_CHAT_ID} 的语音聊天并开始播放。")
await update.message.reply_text("已成功加入语音聊天并播放提醒。请加入群组语音聊天收听。")
# 播放一段时间后自动离开
# 这里只是一个示例，您可以根据需要调整播放时长或手动控制
await asyncio.sleep(15) # 播放15秒
await pytgcalls_client.leave_group_call(TARGET_CHAT_ID)
logger.info(f"已离开群组 {TARGET_CHAT_ID} 的语音聊天。")
await update.message.reply_text("已离开语音聊天。")
except Exception as e:
logger.error(f"处理 !call 命令时发生错误: {e}", exc_info=True)
await update.message.reply_text(f"抱歉，在尝试语音通话时发生错误: {e}")
# 定义启动命令处理器
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""发送欢迎消息。"""
await update.message.reply_text("你好！我是一个语音通知机器人。发送 `!call` 我会尝试在指定群组语音聊天中播放消息。")
async def main() -> None:
"""主函数，设置并启动机器人。"""
# 构建 Telegram 机器人应用
application = Application.builder().token(BOT_TOKEN).build()
# 添加命令处理器
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler("call", call_me))
# 启动 pytgcalls 客户端
# pytgcalls_client.start() 是一个同步操作，需要在一个单独的线程或协程中运行
# 这里我们使用 asyncio.gather 来同时运行机器人和 pytgcalls 客户端
# 启动 pytgcalls 客户端
await pytgcalls_client.start()
logger.info("PyTgCalls 客户端已启动。")
# 启动 Telegram 机器人
await application.run_polling(allowed_updates=Update.ALL_TYPES)
logger.info("Telegram 机器人已启动并开始轮询。")
# 保持 pytgcalls 客户端运行，直到程序终止
await idle()
logger.info("PyTgCalls 客户端已停止。")
if __name__ == "__main__":
# 确保 audio.mp3 文件存在于当前目录或指定正确路径
# 您可以使用任何音频编辑软件录制并保存为 audio.mp3
# 示例：一个简单的“你好，这是一个测试通知”的MP3文件
# 注意：pytgcalls 对音频格式有特定要求，推荐使用 48kHz, mono, 16bit PCM 或 Opus 编码。
# 如果遇到播放问题，请尝试转换音频文件。
# 例如，使用 ffmpeg 转换：
# ffmpeg -i your_original_audio.wav -acodec libopus -b:a 32k -ar 48000 -ac 1 audio.mp3
# 运行主函数
asyncio.run(main())准备音频文件： 创建一个名为 audio.mp3 的音频文件，并放置在与 main.py 相同的目录下。这个文件将是机器人播放的预录消息。为了最佳兼容性，请确保音频文件符合以下建议：
- 格式： MP3 或 OGG (Opus编码)。
- 采样率： 48 kHz。
- 声道： 单声道 (mono)。
- 比特率： 推荐 32kbps 或更高。
您可以使用FFmpeg进行转换： ffmpeg -i your_original_audio.wav -acodec libopus -b:a 32k -ar 48000 -ac 1 audio.mp3
运行与测试
- 填写 config.py: 确保 BOT_TOKEN, API_ID, API_HASH, SESSION_STRING 和 TARGET_CHAT_ID 都已正确填写。
- 创建 audio.mp3: 放置一个预录制的音频文件。
-
运行 main.py:
python main.py
-
在Telegram中测试:
- 确保您的机器人和通过pytgcalls控制的用户账户都在 TARGET_CHAT_ID 指定的群组中。
- 在与机器人的私人聊天或 TARGET_CHAT_ID 群组中发送 /start 命令，机器人会回复欢迎消息。
- 发送 !call 命令。机器人将尝试加入群组语音聊天并播放 audio.mp3。
- 您需要手动加入该群组的语音聊天才能听到机器人播放的消息。
注意事项与进阶考量
pytgcalls与群组语音聊天： 再次强调，pytgcalls用于操作群组语音聊天，而非直接向个人用户发起一对一通话。当机器人“打电话”时，它实际上是加入了一个群组的语音聊天，并在其中播放音频。用户需要主动加入该群组的语音聊天才能听到。
-
权限管理：
- 通过pytgcalls控制的用户账户必须是目标群组的成员，并且拥有发起和管理语音聊天的权限。
- 机器人账户也必须是目标群组的成员。
错误处理： 实际应用中需要更完善的错误处理机制，例如网络中断、API限制、音频文件不存在或格式不正确等情况。
音频流控制： pytgcalls提供了更高级的音频控制功能，例如循环播放、切换音频流、调整音量等。您可以查阅其官方文档以获取更多信息。
-
外部触发器集成：
天气条件： 要实现基于天气条件的触发，您需要集成一个天气API（例如OpenWeatherMap）。定期调用API获取天气数据，当满足特定条件（如温度低于24度）时，调用 call_me 函数中的核心逻辑（加入语音聊天并播放音频）。这通常需要一个独立的异步任务或定时器。
-
示例 (伪代码):
async def check_weather_and_call(): while True: # 调用天气API获取当前温度 current_temp = await get_weather_from_api() # 假设这是异步函数 if current_temp < 24: logger.info("温度低于24度，触发语音通知！") # 直接调用 pytgcalls 逻辑，或通过消息队列触发 bot 命令 # 这里简化为直接调用 await pytgcalls_client.join_group_call(TARGET_CHAT_ID, AudioPiped("coat_reminder.mp3")) await asyncio.sleep(15) await pytgcalls_client.leave_group_call(TARGET_CHAT_ID) await asyncio.sleep(3600) # 避免频繁通知，一小时检查一次 else: logger.info(f"当前温度 {current_temp}，无需通知。") await asyncio.sleep(600) # 每10分钟检查一次 # 在 main() 函数中，与 application.run_polling 和 pytgcalls_client.start() 并行运行 # asyncio.create_task(check_weather_and_call())
部署： 对于生产环境，您需要将机器人部署到云服务器（如
