求助：搭建基于天气触发的Python Telegram自动呼叫机器人（pytgcalls）

一、必备凭证说明 #

• API ID & API Hash：从Telegram官方平台获取，登录自己的Telegram账号后创建应用即可拿到（无需复杂配置）。
• 会话文件：pytgcalls依赖Telegram客户端会话，可通过Pyrogram第一次运行时自动生成（输入手机号+验证码即可）；如果用机器人，需要从@BotFather获取BOT_TOKEN。
• 目标聊天ID：用@userinfobot获取你自己的聊天ID，用于指定呼叫对象。

二、环境搭建 #

安装所需依赖包：

pip install pytgcalls pyrogram python-dotenv ffmpeg-python

注：ffmpeg用于处理音频格式转换，若系统已安装可跳过ffmpeg-python。

三、基础功能代码示例（!call命令触发） #

1. 创建.env文件存储敏感信息：

API_ID=你的API_ID
API_HASH=你的API_HASH
# 二选一：用机器人的话填BOT_TOKEN，用个人账号则注释掉这行
BOT_TOKEN=你的机器人TOKEN
TARGET_CHAT_ID=你的目标聊天ID
AUDIO_PATH=./coat_reminder.ogg  # 预录语音文件路径

2. 主程序代码：

from pyrogram import Client, filters
from pytgcalls import PyTgCalls
from pytgcalls.types import MediaStream
import os
from dotenv import load_dotenv

load_dotenv()

# 初始化Pyrogram客户端
app = Client(
    'telegram_call_session',
    api_id=int(os.getenv('API_ID')),
    api_hash=os.getenv('API_HASH'),
    bot_token=os.getenv('BOT_TOKEN', None)  # 机器人模式才需要，个人账号自动忽略
)

# 绑定通话客户端
call_client = PyTgCalls(app)

# 处理!call命令
@app.on_message(filters.command('call', prefixes='!'))
async def trigger_call(_, message):
    try:
        # 发起语音呼叫并播放预录音频
        await call_client.join_group_call(
            os.getenv('TARGET_CHAT_ID'),
            MediaStream(os.getenv('AUDIO_PATH'))
        )
        await message.reply('语音呼叫已发起，正在播放提醒！')
    except Exception as e:
        await message.reply(f"呼叫失败：{str(e)}")

if __name__ == '__main__':
    call_client.start()
    app.run()

四、关键注意事项 #

• 音频格式要求：必须是OGG OPUS格式，用ffmpeg转换命令：

  ffmpeg -i input.mp3 -c:a libopus -b:a 128k output.ogg
  

• 权限问题：机器人需先加入目标聊天，且拥有发言权限；个人账号直接使用即可。
• 会话维护：第一次运行个人账号模式时，会提示输入手机号和验证码，生成的会话文件会自动保存，后续运行无需重复验证。

五、扩展：天气触发自动呼叫 #

在基础功能跑通后，添加天气监测逻辑（以WeatherAPI为例）：

import requests
import time
from threading import Thread

def check_temperature():
    # 替换为你的天气API接口和参数
    api_url = "https://api.weatherapi.com/v1/current.json?key=你的天气API_KEY&q=你的城市"
    resp = requests.get(api_url)
    if resp.status_code == 200:
        temp = resp.json()['current']['temp_c']
        return temp < 24
    return False

def weather_monitor():
    while True:
        if check_temperature():
            # 异步发起呼叫
            app.loop.create_task(
                call_client.join_group_call(
                    os.getenv('TARGET_CHAT_ID'),
                    MediaStream(os.getenv('AUDIO_PATH'))
                )
            )
            # 避免重复触发，可设置冷却时间
            time.sleep(3600)
        time.sleep(1800)  # 每30分钟检查一次

# 在main函数中启动监测线程
if __name__ == '__main__':
    call_client.start()
    Thread(target=weather_monitor, daemon=True).start()
    app.run()

内容的提问来源于stack exchange，提问作者Akon