使用Python和pytgcall创建Telegram机器人实现自动化语音通知

pip install python-telegram-bot pytgcalls pyrogram ffmpeg-python

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())

# 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开头

# 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())