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

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

# 解析Al Jazeera新闻页面
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():
    # 每隔10分钟抓取一次新闻
    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的配置示例：

# 代理API配置
PROXY_API_URL = 'https://api.brightdata.com/proxy'
API_KEY = 'your_api_key'  # 请替换为实际API密钥

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'
}

# 亮数据代理API配置
PROXY_API_URL = 'https://api.brightdata.com/proxy'
API_KEY = 'your_api_key'  # 请替换为实际API密钥

# 爬虫请求函数
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

# 解析Al Jazeera新闻页面
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()  # 继续运行定时任务