使用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.textelse:print(f"Failed to fetch {url}. Status code: {response.status_code}")return Noneexcept 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.textelse:print(f"Failed to fetch {url}. Status code: {response.status_code}")return Noneexcept 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.textelse:print(f"Failed to fetch {url}. Status code: {response.status_code}")return Noneexcept 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()  # 继续运行定时任务

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

总结

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