You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何让基于BeautifulSoup的爬虫仅推送新增新闻至Telegram Bot?

嘿,这个问题其实挺常见的,根本不需要复杂的排序算法就能搞定~核心思路就是记录已经推送过的新闻,下次爬取的时候只处理那些没记录过的内容就行。下面给你具体的实现方案和修改后的代码:

核心解决思路

你需要一个持久化的方式来保存已经推送过的新闻标识(比如新闻的唯一链接,比标题更靠谱),每次爬取新内容时,对比这个记录列表,只推送不在列表里的新闻,之后再把新推送的内容加入列表。

具体实现步骤

  1. 抓取新闻的唯一标识:优先用新闻的链接(每个新闻的链接都是唯一的,不会重复),比单纯用标题判断更准确。
  2. 持久化存储已推送记录:用一个本地文本文件(比如sent_news.txt)来保存已推送的链接,每次运行程序前读取这个文件,运行结束后更新。
  3. 筛选新增内容:爬取到新闻后,检查链接是否在已推送列表中,不在的话就推送,并把链接加入列表。

修改后的完整代码

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.05.28 06:19:27