如何用Python实现Telegram机器人自动推送新闻至所有加入群组？

看起来你已经迈出了第一步，接下来咱们把这个机器人的功能补全，搞定批量推送和自动化运行这两个核心需求。我分几个部分给你讲清楚：

1. 维护所有已加入群组的Chat ID列表 #

你现在用bot.get_updates()[-1].message.chat.id只能拿到最新交互的Chat ID，要推送到所有群组，得主动记录机器人加入的每个群组。最简单的方式是用JSON文件存Chat ID列表，同时监听机器人被添加到群组的事件：

示例代码（基于pyTelegramBotAPI） #

import telebot
import json
from telebot.types import ChatMemberUpdated

# 初始化机器人
BOT_TOKEN = "你的机器人Token"
bot = telebot.TeleBot(BOT_TOKEN)

# 存储Chat ID的文件路径
CHAT_IDS_FILE = "chat_ids.json"

# 加载已存储的Chat ID
def load_chat_ids():
    try:
        with open(CHAT_IDS_FILE, "r") as f:
            return set(json.load(f))  # 用集合避免重复
    except FileNotFoundError:
        return set()

# 保存Chat ID到文件
def save_chat_ids(chat_ids):
    with open(CHAT_IDS_FILE, "w") as f:
        json.dump(list(chat_ids), f)

# 监听机器人被添加到群组的事件
@bot.chat_member_handler(func=lambda update: update.new_chat_member.user.id == bot.get_me().id)
def on_bot_added(update: ChatMemberUpdated):
    chat_id = update.chat.id
    chat_ids = load_chat_ids()
    if chat_id not in chat_ids:
        chat_ids.add(chat_id)
        save_chat_ids(chat_ids)
        print(f"已记录新群组：{chat_id}")

2. 实现新闻更新检测 #

要自动识别最新新闻，推荐用RSS订阅（如果目标网站提供的话），或者爬取页面的更新时间/新闻ID。这里以RSS为例，用feedparser库解析：

先安装依赖 #

pip install feedparser

新闻检测代码 #

import feedparser

# 目标新闻的RSS地址
NEWS_RSS_URL = "https://example.com/news/rss"
# 存储最后推送的新闻ID的文件
LAST_NEWS_ID_FILE = "last_news_id.txt"

def get_last_news_id():
    try:
        with open(LAST_NEWS_ID_FILE, "r") as f:
            return f.read().strip()
    except FileNotFoundError:
        return None

def save_last_news_id(news_id):
    with open(LAST_NEWS_ID_FILE, "w") as f:
        f.write(news_id)

def check_new_news():
    feed = feedparser.parse(NEWS_RSS_URL)
    if not feed.entries:
        return None
    
    latest_news = feed.entries[0]
    last_id = get_last_news_id()
    
    # 用新闻的唯一ID判断是否是新内容
    news_id = latest_news.id if "id" in latest_news else latest_news.link
    if news_id != last_id:
        save_last_news_id(news_id)
        # 组装推送内容
        message = f"📢 最新新闻：\n{latest_news.title}\n{latest_news.link}"
        return message
    return None

3. 定时推送与自动化运行 #

接下来要让机器人定时检查新闻并推送到所有群组，同时保持脚本持续运行。推荐用schedule库做定时任务：

安装依赖 #

pip install schedule

整合推送与定时任务 #

import schedule
import time
from functools import partial

def push_to_all_groups(message):
    if not message:
        return
    chat_ids = load_chat_ids()
    for chat_id in chat_ids:
        try:
            bot.send_message(chat_id, message)
            print(f"已推送至群组：{chat_id}")
        except Exception as e:
            print(f"推送至群组{chat_id}失败：{str(e)}")
            # 移除无效的Chat ID（比如机器人被踢出群组）
            chat_ids.remove(chat_id)
            save_chat_ids(chat_ids)

def check_and_push_news():
    new_news = check_new_news()
    push_to_all_groups(new_news)

# 设置定时任务，比如每10分钟检查一次
schedule.every(10).minutes.do(check_and_push_news)

# 启动机器人的监听和定时任务
if __name__ == "__main__":
    # 先启动机器人的轮询（非阻塞模式）
    import threading
    threading.Thread(target=bot.infinity_polling, daemon=True).start()
    
    print("机器人已启动，开始监听新闻更新...")
    while True:
        schedule.run_pending()
        time.sleep(60)

关键注意事项 #

• 机器人权限：确保你的机器人在群组中有发送消息的权限，开启机器人的「群组权限」（在BotFather中设置）。
• 后台运行：如果要让机器人一直运行，推荐用nohup命令：

  nohup python your_bot.py &
  

  或者用systemd配置成系统服务，这样重启后会自动启动。
• 异常处理：代码中已经加入了基本的异常捕获，你可以根据需要扩展（比如网络超时重试）。

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