当前位置：首页 > news >正文

Python实例题：基于 Python 的简单爬虫与数据可视化

news 2025/7/2 11:09:20

Python实例题

题目

基于 Python 的简单爬虫与数据可视化

要求：

使用 Python 构建一个简单的爬虫程序，支持以下功能：
- 从指定网站抓取数据（如新闻、商品信息等）
- 解析 HTML 内容并提取所需信息
- 将数据存储到文件或数据库
- 对抓取的数据进行统计和可视化分析
使用 requests 和 BeautifulSoup 进行网页爬取和解析。
使用 pandas 和 matplotlib 进行数据处理和可视化。
添加命令行界面，支持用户输入爬取参数。

解题思路：

使用 requests 发送 HTTP 请求获取网页内容。
通过 BeautifulSoup 解析 HTML 内容并提取数据。
使用 pandas 处理和分析数据。
利用 matplotlib 和 seaborn 绘制图表。
实现基本的反爬策略（如随机 User-Agent、请求间隔）。

代码实现：

import requests
from bs4 import BeautifulSoup
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import time
import random
import argparse
import json
import os
from datetime import datetime
import re# 设置中文字体
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]
plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示问题class WebScraper:def __init__(self, base_url, headers=None, delay_min=1, delay_max=3):self.base_url = base_urlself.headers = headers or {'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'}self.delay_min = delay_minself.delay_max = delay_maxself.data = []def fetch_page(self, url):"""发送HTTP请求获取页面内容"""try:# 添加随机延迟，避免频繁请求delay = random.uniform(self.delay_min, self.delay_max)time.sleep(delay)response = requests.get(url, headers=self.headers)response.raise_for_status()  # 检查请求是否成功return response.textexcept requests.exceptions.RequestException as e:print(f"请求错误: {e}")return Nonedef parse_page(self, html_content, parser='html.parser'):"""解析HTML内容"""if not html_content:return Nonereturn BeautifulSoup(html_content, parser)def extract_data(self, soup):"""从页面中提取数据（需要在子类中实现）"""raise NotImplementedError("子类必须实现extract_data方法")def scrape(self, pages=1, url_pattern=None):"""执行爬取任务"""self.data = []for page in range(1, pages + 1):print(f"正在爬取第 {page}/{pages} 页...")if url_pattern:url = url_pattern.format(page=page)else:url = self.base_url if page == 1 else f"{self.base_url}?page={page}"html_content = self.fetch_page(url)if not html_content:continuesoup = self.parse_page(html_content)if not soup:continuepage_data = self.extract_data(soup)if page_data:self.data.extend(page_data)print(f"第 {page} 页爬取完成，获取 {len(page_data)} 条数据")print(f"爬取完成，共获取 {len(self.data)} 条数据")return self.datadef save_to_csv(self, filename=None):"""将数据保存为CSV文件"""if not self.data:print("没有数据可保存")returnfilename = filename or f"data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"df = pd.DataFrame(self.data)df.to_csv(filename, index=False, encoding='utf-8-sig')print(f"数据已保存到 {filename}")def save_to_json(self, filename=None):"""将数据保存为JSON文件"""if not self.data:print("没有数据可保存")returnfilename = filename or f"data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"with open(filename, 'w', encoding='utf-8') as f:json.dump(self.data, f, ensure_ascii=False, indent=2)print(f"数据已保存到 {filename}")class NewsScraper(WebScraper):def __init__(self, base_url, headers=None):super().__init__(base_url, headers)def extract_data(self, soup):"""从新闻页面提取数据"""articles = []# 查找所有新闻条目news_items = soup.select('div.news-item')  # 根据实际页面结构调整选择器for item in news_items:try:title_elem = item.select_one('h2.title a')title = title_elem.text.strip() if title_elem else "未找到标题"link = title_elem['href'] if title_elem and 'href' in title_elem.attrs else "#"if link.startswith('/'):link = self.base_url + linkdate_elem = item.select_one('span.date')date = date_elem.text.strip() if date_elem else "未知日期"summary_elem = item.select_one('p.summary')summary = summary_elem.text.strip() if summary_elem else "无摘要"articles.append({'title': title,'link': link,'date': date,'summary': summary})except Exception as e:print(f"提取新闻条目时出错: {e}")continuereturn articlesdef analyze(self):"""分析新闻数据"""if not self.data:print("没有数据可分析")returndf = pd.DataFrame(self.data)# 分析发布日期try:# 尝试解析日期df['date'] = pd.to_datetime(df['date'])df['day_of_week'] = df['date'].dt.day_name()# 按星期几统计新闻数量day_counts = df['day_of_week'].value_counts()plt.figure(figsize=(10, 6))sns.barplot(x=day_counts.index, y=day_counts.values)plt.title('新闻发布的星期分布')plt.xlabel('星期')plt.ylabel('新闻数量')plt.xticks(rotation=45)plt.tight_layout()plt.savefig('news_by_day.png')plt.close()print("新闻发布星期分布分析完成，图表已保存为 news_by_day.png")except Exception as e:print(f"日期分析出错: {e}")# 分析标题长度df['title_length'] = df['title'].str.len()plt.figure(figsize=(10, 6))sns.histplot(df['title_length'], bins=20, kde=True)plt.title('新闻标题长度分布')plt.xlabel('标题长度（字符）')plt.ylabel('频率')plt.tight_layout()plt.savefig('title_length_distribution.png')plt.close()print("新闻标题长度分析完成，图表已保存为 title_length_distribution.png")class ProductScraper(WebScraper):def __init__(self, base_url, headers=None):super().__init__(base_url, headers)def extract_data(self, soup):"""从商品页面提取数据"""products = []# 查找所有商品条目product_items = soup.select('div.product-item')  # 根据实际页面结构调整选择器for item in product_items:try:name_elem = item.select_one('h3.product-name')name = name_elem.text.strip() if name_elem else "未找到商品名称"price_elem = item.select_one('span.price')price_text = price_elem.text.strip() if price_elem else "0"# 提取价格中的数字部分price = re.search(r'\d+\.?\d*', price_text)price = float(price.group(0)) if price else 0rating_elem = item.select_one('div.rating')rating = rating_elem['data-rating'] if rating_elem and 'data-rating' in rating_elem.attrs else "0"rating = float(rating)review_count_elem = item.select_one('span.review-count')review_count_text = review_count_elem.text.strip() if review_count_elem else "0"# 提取评论数中的数字部分review_count = re.search(r'\d+', review_count_text)review_count = int(review_count.group(0)) if review_count else 0products.append({'name': name,'price': price,'rating': rating,'review_count': review_count})except Exception as e:print(f"提取商品条目时出错: {e}")continuereturn productsdef analyze(self):"""分析商品数据"""if not self.data:print("没有数据可分析")returndf = pd.DataFrame(self.data)# 价格分布plt.figure(figsize=(10, 6))sns.histplot(df['price'], bins=20, kde=True)plt.title('商品价格分布')plt.xlabel('价格')plt.ylabel('频率')plt.tight_layout()plt.savefig('price_distribution.png')plt.close()print("商品价格分布分析完成，图表已保存为 price_distribution.png")# 价格与评分关系plt.figure(figsize=(10, 6))sns.scatterplot(x='price', y='rating', data=df, alpha=0.6)plt.title('商品价格与评分关系')plt.xlabel('价格')plt.ylabel('评分')plt.tight_layout()plt.savefig('price_rating_relationship.png')plt.close()print("商品价格与评分关系分析完成，图表已保存为 price_rating_relationship.png")# 相关性分析corr_matrix = df[['price', 'rating', 'review_count']].corr()plt.figure(figsize=(10, 8))sns.heatmap(corr_matrix, annot=True, cmap='coolwarm', fmt='.2f', linewidths=0.5)plt.title('商品属性相关性分析')plt.tight_layout()plt.savefig('correlation_analysis.png')plt.close()print("商品属性相关性分析完成，图表已保存为 correlation_analysis.png")def main():parser = argparse.ArgumentParser(description='简单网页爬虫与数据分析工具')parser.add_argument('--type', type=str, choices=['news', 'product'], default='news', help='爬取类型：news（新闻）或product（商品）')parser.add_argument('--url', type=str, required=True, help='目标网站URL')parser.add_argument('--pages', type=int, default=1, help='要爬取的页数')parser.add_argument('--output', type=str, choices=['csv', 'json'], default='csv', help='输出格式：csv或json')parser.add_argument('--delay_min', type=float, default=1, help='最小请求延迟（秒）')parser.add_argument('--delay_max', type=float, default=3, help='最大请求延迟（秒）')parser.add_argument('--analyze', action='store_true', help='是否进行数据分析')args = parser.parse_args()# 创建爬虫实例if args.type == 'news':scraper = NewsScraper(args.url, delay_min=args.delay_min, delay_max=args.delay_max)else:scraper = ProductScraper(args.url, delay_min=args.delay_min, delay_max=args.delay_max)# 执行爬取scraper.scrape(pages=args.pages)# 保存数据if args.output == 'csv':scraper.save_to_csv()else:scraper.save_to_json()# 数据分析if args.analyze:scraper.analyze()if __name__ == "__main__":main()

查看全文

http://www.dtcms.com/a/264115.html