如何让基于BeautifulSoup的爬虫仅推送新增新闻至Telegram Bot?
嘿,这个问题其实挺常见的,根本不需要复杂的排序算法就能搞定~核心思路就是记录已经推送过的新闻,下次爬取的时候只处理那些没记录过的内容就行。下面给你具体的实现方案和修改后的代码:
核心解决思路
你需要一个持久化的方式来保存已经推送过的新闻标识(比如新闻的唯一链接,比标题更靠谱),每次爬取新内容时,对比这个记录列表,只推送不在列表里的新闻,之后再把新推送的内容加入列表。
具体实现步骤
- 抓取新闻的唯一标识:优先用新闻的链接(每个新闻的链接都是唯一的,不会重复),比单纯用标题判断更准确。
- 持久化存储已推送记录:用一个本地文本文件(比如
sent_news.txt)来保存已推送的链接,每次运行程序前读取这个文件,运行结束后更新。 - 筛选新增内容:爬取到新闻后,检查链接是否在已推送列表中,不在的话就推送,并把链接加入列表。
修改后的完整代码
import requests import bs4 # 存储已推送新闻链接的本地文件 SENT_NEWS_FILE = "sent_news.txt" def load_sent_news(): """读取已推送过的新闻链接集合""" try: with open(SENT_NEWS_FILE, "r", encoding="utf-8") as f: return set(f.read().splitlines()) except FileNotFoundError: # 如果文件不存在,返回空集合 return set() def save_sent_news(sent_news): """把已推送的新闻链接保存到文件""" with open(SENT_NEWS_FILE, "w", encoding="utf-8") as f: f.write("\n".join(sent_news)) def get_telegram_chat_id(bot_token): """获取Telegram的Chat ID(只需调用一次,不用每次循环都请求)""" response = requests.get(f'https://api.telegram.org/bot{bot_token}/getUpdates') updates = response.json().get('result', []) if not updates: raise ValueError("请先给你的Telegram Bot发送一条消息,这样才能获取到Chat ID哦") return updates[0]['message']['from']['id'] def main(): # 替换成你的Bot Token BOT_TOKEN = "<你的Bot Token>" # 加载已推送的新闻列表 sent_news = load_sent_news() # 获取Chat ID chat_id = get_telegram_chat_id(BOT_TOKEN) # 爬取目标网站 fonte = requests.get('https://www.noticiasaominuto.com/') soup = bs4.BeautifulSoup(fonte.text, 'lxml') # 遍历所有新闻条目(这里假设新闻的外层是a标签,class为article-thumb,你可以根据实际页面调整) for article_link in soup.find_all('a', class_='article-thumb'): news_url = article_link['href'] # 补全相对链接为完整URL if not news_url.startswith('http'): news_url = f"https://www.noticiasaominuto.com{news_url}" # 检查是否已经推送过这条新闻 if news_url not in sent_news: # 获取新闻标题 news_title = article_link.find('p', class_='article-thumb-text').text.strip() print(f"发现新新闻,准备推送:{news_title}") # 发送到Telegram send_response = requests.post( f'https://api.telegram.org/bot{BOT_TOKEN}/sendMessage', data={ 'chat_id': chat_id, 'text': f"📰 {news_title}\n{news_url}" } ) # 检查发送是否成功 if send_response.status_code == 200: print("推送成功!") sent_news.add(news_url) else: print(f"推送失败,错误信息:{send_response.json()}") # 保存更新后的已推送列表 save_sent_news(sent_news) if __name__ == "__main__": main()
关键细节说明
- 为什么用链接做标识:新闻标题可能存在重复的情况,但每个新闻的链接都是唯一的,用它来判断是否是新增内容更准确。
- 用集合存储已推送链接:集合的查找操作效率是O(1),比列表的O(n)快很多,尤其是当已推送的新闻越来越多时,优势更明显。
- 优化Chat ID获取:原来的代码每次循环都调用
getUpdates,会浪费请求次数,现在改成只调用一次,更高效。 - 持久化存储:用文本文件是最简单的方案,如果你的需求更复杂(比如多用户、跨设备运行),可以换成SQLite数据库,逻辑是一样的。
另外补充一句:完全不需要排序算法哦,我们只需要对比新闻是否已经推送过,不需要对新闻进行排序。如果目标网站的新闻是按时间倒序排列的(最新的在最前面),你还可以在遇到第一条已推送的新闻后就停止循环,进一步提升效率。
内容的提问来源于stack exchange,提问作者Tiago Gonçalves
相关产品推荐
相关产品推荐

