登录
TG自建Python版的RSS订阅Bot
以前找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，可以参考截图设置：
@BotFather会回复你一个机器人的HTTP API，保存它，后面会用到。
进入刚刚创建的机器人对话框，将他拉进一个你需要推送的群，或者直接给你自己推送也行，拉群记得给它管理员和消息权限。
TG搜索@get_id_bot，进入对话框点击右上角，将他拉到刚刚的群里，在群聊对话框输入：/my_id@get_id_bot发送，会得到一个-开头的群聊ID，记录它，包括-符号。
信息获取完成，接下来就是创建脚本了，回到服务器终端
- 创建目录
假设创建一个名为 dalaorss 的目录：
mkdir dalaorss
cd dalaorss- 创建 Python 文件
在 dalaorss 目录中，创建一个 Python 文件，这里使用 nano 编辑器创建一个名为 rss_bot.py 的文件：
nano rss_bot.py- 添加代码
将下面代码复制进脚本中，注意修改为你的信息
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())
- 保存并退出
在 nano 编辑器中，按 Ctrl + O 保存文件（字母O不是数字0），然后按 Enter 确认，接着按 Ctrl + X 退出编辑器。 - 运行脚本
nohup（no hang up）命令可以让脚本在后台运行，即使您关闭终端会话也能保持运行。
nohup python3 rss_bot.py &现在就可以去更新rss看看推送状态了，这里的脚本内容以我自己论坛为例，如果你也是xiuno论坛那么照抄就行，如果是博客或者其它程序，可能还需要做一些修改。
推送效果：
个人比较喜欢这种格式，如果你喜欢别的模式比如消息内预览，可以使用docker版本的，网上搜一下就有。
脚本运行后，怎么查看或关闭：
检查当前运行的进程
ps aux | grep python会得到正在运行的进程，查看到rss_bot那条，最前面有一个ID，命令行输入
kill ID即可停止。
