使用Python爬虫实时监控行业新闻

Posted 2025-02-14 Updated 2025-02- 14

By power 已删除用户

19~24 min read

本文介绍了使用 Python 爬虫实时监控行业新闻的案例。包括环境准备、请求与解析网页数据、设置定时任务，还提及使用代理 IP 提升稳定性，最终实现了实时抓取 Al Jazeera 新闻网站数据并定时自动抓取，同时避免 IP 被封禁。

关联问题: 如何选代理IP 能爬其他网站吗怎样修改定时时长

在互联网时代，新闻的实时性和时效性变得尤为重要。很多行业、技术、商业等领域的新闻都可以为公司或者个人发展提供有价值的信息。如果你有一项需求是要实时监控某个行业的新闻，自动化抓取并定期输出这些新闻，Python爬虫可以帮你轻松实现这一目标。

本文将通过一个案例，带你一步一步实现一个简单的Python爬虫，用于实时监控新闻网站的数据。

背景

在某些行业中，获取最新的新闻信息至关重要。通过定期抓取新闻网站的头条新闻，我们可以为用户提供行业热点的动态变化。本文的目标是创建一个爬虫，定期访问一个新闻网站，获取新闻的标题和链接，并打印出来。

环境准备

在开始编写代码之前，我们需要安装几个Python的第三方库：

requests：用于发送HTTP请求。
beautifulsoup4：用于解析网页HTML内容。
schedule：用于设置定时任务，使爬虫能够自动运行。

可以通过以下命令安装这些库：

pip install requests beautifulsoup4 schedule

请求网页数据

在爬取新闻之前，我们首先要获取目标网页的HTML内容。通过requests库可以方便地发送GET请求，并返回页面内容。以下是请求网页的代码：

import requests


HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}


def fetch_news(url):
    try:
        print(f"Attempting to fetch: {url}")  
        response = requests.get(url, headers=HEADERS, timeout=10)
        print(f"Status code: {response.status_code}")  
        if response.status_code == 200:
            return response.text
        else:
            print(f"Failed to fetch {url}. Status code: {response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"Error fetching {url}: {e}")
        return None

HEADERS用于模拟浏览器访问，避免被网站屏蔽。
fetch_news函数发送GET请求并返回网页内容。如果请求成功，则返回HTML内容。

解析网页数据

一旦我们获取了网页的HTML内容，就需要解析这些内容，提取出我们关心的数据（例如新闻标题和链接）。这里我们使用beautifulsoup4来解析HTML并提取新闻数据。

from bs4 import BeautifulSoup


def parse_aljazeera_page(page_content):
    soup = BeautifulSoup(page_content, 'html.parser')
    news_items = []

    articles = soup.find_all('a', class_='u-clickable-card__link')
    print(f"Found {len(articles)} articles on Al Jazeera")

    for article in articles:
        title_tag = article.find('h3')
        if title_tag:
            title = title_tag.text.strip()
            link = article['href']
            if link.startswith('http'):
                news_items.append({
                    'title': title,
                    'link': link
                })
            else:

                full_link = f"https://www.aljazeera.com{link}"
                news_items.append({
                    'title': title,
                    'link': full_link
                })

    return news_items

BeautifulSoup用于解析HTML内容。
parse_aljazeera_page函数从页面中找到所有新闻条目，并提取每个新闻的标题和链接。

定时任务

爬虫的核心功能是定期抓取新闻信息。为了实现这一点，我们可以使用schedule库来设置定时任务，定时运行爬虫。

import schedule
import time


def run_scheduler():

    schedule.every(10).minutes.do(monitor_news)

    while True:
        print("Scheduler is running...")  
        schedule.run_pending()
        time.sleep(1)

我们使用schedule.every(10).minutes.do(monitor_news)设置每10分钟执行一次monitor_news函数，获取并输出新闻。

综合代码

将之前的部分代码整合在一起，并加入一个监控新闻的函数：

def monitor_news():
    url = 'https://www.aljazeera.com/'
    page_content = fetch_news(url)

    if page_content:
        news_items = parse_aljazeera_page(page_content)
        if news_items:
            print(f"News from {url}:")
            for news in news_items:
                print(f"Title: {news['title']}")
                print(f"Link: {news['link']}")
                print("-" * 50)
        else:
            print(f"No news items found at {url}.")
    else:
        print(f"Failed to fetch {url}.")

if __name__ == '__main__':
    monitor_news()  
    run_scheduler()

使用代理IP提升稳定性

爬虫在运行时，可能会遇到反爬机制导致IP被封禁的情况。为了规避这一问题，我们可以通过配置代理IP来提高爬虫的稳定性。下面是如何使用亮数据代理API的配置示例：

PROXY_API_URL = 'https://api.brightdata.com/proxy'
API_KEY = 'your_api_key'

PROXY_API_URL：亮数据的代理API接口地址。
API_KEY：你的API密钥，用于认证API请求。

通过修改爬虫的请求函数，将代理配置加到请求中，可以让爬虫通过多个IP地址进行请求，从而降低被封禁的风险：

def fetch_news_with_proxy(url):
    try:
        print(f"Attempting to fetch with proxy: {url}")  
        response = requests.get(
            url,
            headers=HEADERS,
            proxies={"http": PROXY_API_URL, "https": PROXY_API_URL},
            timeout=10
        )
        print(f"Status code: {response.status_code}")  
        if response.status_code == 200:
            return response.text
        else:
            print(f"Failed to fetch {url}. Status code: {response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"Error fetching {url}: {e}")
        return None

运行截图与完整代码

运行截图：

完整代码如下

import requests
from bs4 import BeautifulSoup
import schedule
import time


HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}


PROXY_API_URL = 'https://api.brightdata.com/proxy'
API_KEY = 'your_api_key'  


def fetch_news(url):
    try:
        print(f"Attempting to fetch: {url}")  
        response = requests.get(url, headers=HEADERS, timeout=10)
        print(f"Status code: {response.status_code}")  
        if response.status_code == 200:
            return response.text
        else:
            print(f"Failed to fetch {url}. Status code: {response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"Error fetching {url}: {e}")
        return None


def parse_aljazeera_page(page_content):
    soup = BeautifulSoup(page_content, 'html.parser')
    news_items = []

    articles = soup.find_all('a', class_='u-clickable-card__link')
    print(f"Found {len(articles)} articles on Al Jazeera")

    for article in articles:
        title_tag = article.find('h3')
        if title_tag:
            title = title_tag.text.strip()
            link = article['href']
            if link.startswith('http'):
                news_items.append({
                    'title': title,
                    'link': link
                })
            else:

                full_link = f"https://www.aljazeera.com{link}"
                news_items.append({
                    'title': title,
                    'link': full_link
                })

    return news_items


def run_scheduler():
    schedule.every(10).minutes.do(monitor_news)

    while True:
        print("Scheduler is running...")  
        schedule.run_pending()
        time.sleep(1)


def monitor_news():
    url = 'https://www.aljazeera.com/'
    page_content = fetch_news(url)

    if page_content:
        news_items = parse_aljazeera_page(page_content)
        if news_items:
            print(f"News from {url}:")
            for news in news_items:
                print(f"Title: {news['title']}")
                print(f"Link: {news['link']}")
                print("-" * 50)
        else:
            print(f"No news items found at {url}.")
    else:
        print(f"Failed to fetch {url}.")


if __name__ == '__main__':
    monitor_news()  
    run_scheduler()

通过这一方式，爬虫不仅能抓取并显示新闻内容，还能避开反爬机制，提升抓取稳定性。

总结

通过上述步骤，我们实现了一个简单的Python爬虫，用于实时抓取Al Jazeera新闻网站的数据，并通过定时任务每隔一定时间自动抓取一次。在爬虫运行过程中，可能会遇到反爬机制导致IP被封禁的情况。为了避免这个问题，我们可以通过配置代理IP来提高爬虫的稳定性。

知识库

License: CC BY 4.0