以前找moe.one要了一个TG新帖推送插件,我上TG上的少没细心关注,这几天才发现这个推送机器人抽风很厉害,不能做到实时推送,中间会间隔一堆才推一次,看不懂代码逻辑,于是自己在网上找了RSS订阅BOT部署,试了两个用的人多的项目,都是docker部署,我想改改消息格式都不知道怎么下手,实在对docker不懂...默认的消息格式让我强迫症实在难受,于是让AI配合写了如下Python脚本,此文记录一下搭建步骤,避免以后有需要时又要去找AI掰扯。

打开服务器终端,安装Python和所需库

安装 pip
更新包列表:

apt update

安装 python3-pip:

apt install python3-pip

安装完成后,确认 pip 是否正确安装:

pip3 --version

正常会返回 pip 的版本信息。

安装所需库

pip3 install feedparser requests python-telegram-bot

下一步,我们需要去Telegram创建Bot,在TG里搜索@BotFather,进入对话框,输入/newbot以创建新机器人,然后回复一个机器人名字,再回复一个机器人的用户名,需要带bot,可以参考截图设置:

1728448409299.png

@BotFather会回复你一个机器人的HTTP API,保存它,后面会用到。

进入刚刚创建的机器人对话框,将他拉进一个你需要推送的群,或者直接给你自己推送也行,拉群记得给它管理员和消息权限。

TG搜索@get_id_bot,进入对话框点击右上角,将他拉到刚刚的群里,在群聊对话框输入:/my_id@get_id_bot发送,会得到一个-开头的群聊ID,记录它,包括-符号。

1728449020735.png

信息获取完成,接下来就是创建脚本了,回到服务器终端

  1. 创建目录
    假设创建一个名为 dalaorss 的目录:
mkdir dalaorss
cd dalaorss
  1. 创建 Python 文件
    在 dalaorss 目录中,创建一个 Python 文件,这里使用 nano 编辑器创建一个名为 rss_bot.py 的文件:
nano rss_bot.py
  1. 添加代码
    将下面代码复制进脚本中,注意修改为你的信息
import feedparser
import logging
import asyncio
import json
import os
from telegram import Bot
from telegram.error import TelegramError

# Telegram Bot Token 和目标聊天 ID
TELEGRAM_TOKEN = 'Bot Token'
CHAT_ID = '群ID'

# 存储已发送的帖子 ID 的文件
POSTS_FILE = 'sent_posts.json'

# 读取已发送的帖子 ID
def load_sent_posts():
    if os.path.exists(POSTS_FILE):
        with open(POSTS_FILE, 'r') as f:
            return json.load(f)
    return []

# 保存已发送的帖子 ID
def save_sent_posts(post_ids):
    with open(POSTS_FILE, 'w') as f:
        json.dump(post_ids, f)

# 从 RSS 源获取更新
def fetch_updates():
    feed_url = "订阅地址"
    try:
        return feedparser.parse(feed_url)
    except Exception as e:
        logging.error(f"获取 RSS 更新时出错: {e}")
        return None

# 转义 Markdown 特殊字符
def escape_markdown(text):
    special_chars = r"_*~`>#+-.!"
    for char in special_chars:
        text = text.replace(char, f"\{char}")
    return text

# 发送消息到 Telegram
async def send_message(bot, title, link):
    # 转义 Markdown 特殊字符
    escaped_title = escape_markdown(title)  # 转义 Markdown 特殊字符
    escaped_link = escape_markdown(link)    # 转义 Markdown 特殊字符

    # 使用 Markdown 格式,将标题包裹在反引号中以避免超链接,链接直接显示
    message = f"`{escaped_title}`\n{escaped_link}"
    try:
        await bot.send_message(chat_id=CHAT_ID, text=message, parse_mode='MarkdownV2')
        logging.info(f"消息发送成功: {escaped_title}")
    except TelegramError as e:
        logging.error(f"发送消息时出错: {e}")

# 主函数
async def check_for_updates(sent_post_ids):
    updates = fetch_updates()

    if updates is None:
        return  # 如果获取更新出错,则返回

    new_post_ids = []  # 用于存储新帖子 ID

    for entry in updates.entries:
        # 从 guid 中提取帖子 ID
        post_id = entry.guid.split('-')[-1].split('.')[0]  # 提取 ID

        # 检查是否为新帖子
        if post_id not in sent_post_ids:
            new_post_ids.append((post_id, entry.title, entry.link))  # 存储 ID, 标题和链接

    # 如果有新帖子,按 ID 升序排序并发送最新帖子
    if new_post_ids:
        new_post_ids.sort(key=lambda x: int(x[0]))  # 升序排序
        latest_post_id, title, link = new_post_ids[0]  # 获取最新的帖子
        async with Bot(token=TELEGRAM_TOKEN) as bot:
            await send_message(bot, title, link)

        # 更新已发送的帖子 ID
        sent_post_ids.append(latest_post_id)
        save_sent_posts(sent_post_ids)  # 保存到文件

# 主循环
async def main():
    logging.basicConfig(level=logging.INFO)

    # 加载已发送的帖子 ID
    sent_post_ids = load_sent_posts()

    while True:
        try:
            await check_for_updates(sent_post_ids)
        except Exception as e:
            logging.error(f"检查更新时出错: {e}")
        await asyncio.sleep(60)  # 每 60 秒检查一次

if __name__ == "__main__":
    asyncio.run(main())
  1. 保存并退出
    在 nano 编辑器中,按 Ctrl + O 保存文件(字母O不是数字0),然后按 Enter 确认,接着按 Ctrl + X 退出编辑器。
  2. 运行脚本
    nohup(no hang up)命令可以让脚本在后台运行,即使您关闭终端会话也能保持运行。
nohup python3 rss_bot.py &

现在就可以去更新rss看看推送状态了,这里的脚本内容以我自己论坛为例,如果你也是xiuno论坛那么照抄就行,如果是博客或者其它程序,可能还需要做一些修改。

推送效果:

1728449810062.png

个人比较喜欢这种格式,如果你喜欢别的模式比如消息内预览,可以使用docker版本的,网上搜一下就有。

脚本运行后,怎么查看或关闭:

检查当前运行的进程

ps aux | grep python

会得到正在运行的进程,查看到rss_bot那条,最前面有一个ID,命令行输入

kill ID

即可停止。