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

产品销量数据爬虫通用模板

ds 2025/5/17 13:56:01

最近遇到各行各业的需要爬取销售数据，每次写一个教程相对麻烦，所以思前考后我还是觉得写一个通用模板更适合。所以模板需要足够的灵活性，让用户能够自定义选择器。比如，产品标题、价格、销量的CSS选择器可能因网站而异，所以可能需要能够传入这些参数。此外，分页处理也很重要，因为销量数据通常分布在多个页面。分页逻辑可能有不同形式，比如URL参数递增或者JavaScript加载，这里可能需要用户指定分页的URL模式或最大页数。

在这里插入图片描述

数据存储部分，可能需要将结果保存为CSV或JSON，方便后续分析。所以模板里应该包含保存数据的函数，比如保存到CSV文件，并处理可能的编码问题。

异常处理也是关键，网络请求可能会失败，页面结构可能变化，所以需要重试机制和日志记录。添加try-except块，记录错误信息，并在失败时重试几次请求。

另外，在实际操作中可能需要代理支持，以防止IP被封。所以在模板中应该允许用户传入代理配置，并在发送请求时使用。

以下就是我通宵几天写的一个通用的产品销量数据爬虫模板，使用 Python 编写，基于 requests 和 BeautifulSoup 库。使用时需要根据目标网站的具体结构进行调整：

import requests
from bs4 import BeautifulSoup
import csv
import time
import logging
from typing import Dict, Listclass ProductSpider:def __init__(self,base_url: str,headers: Dict[str, str] = None,proxies: Dict[str, str] = None,max_retries: int = 3,delay: float = 1.0):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.proxies = proxiesself.max_retries = max_retriesself.delay = delayself.session = requests.Session()# 配置日志logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s')def fetch_page(self, url: str) -> BeautifulSoup:"""获取页面并返回解析后的BeautifulSoup对象"""for _ in range(self.max_retries):try:response = self.session.get(url,headers=self.headers,proxies=self.proxies,timeout=10)response.raise_for_status()time.sleep(self.delay)  # 遵守爬取延迟return BeautifulSoup(response.text, 'lxml')except Exception as e:logging.error(f"请求失败: {str(e)}，剩余重试次数：{self.max_retries - _ - 1}")time.sleep(2)return Nonedef parse_product(self, soup: BeautifulSoup, selectors: Dict[str, str]) -> Dict[str, str]:"""解析产品信息"""product = {}try:product['title'] = soup.select_one(selectors['title']).text.strip()product['price'] = soup.select_one(selectors['price']).text.strip()product['sales'] = soup.select_one(selectors['sales']).text.strip()product['rating'] = soup.select_one(selectors['rating']).text.strip()# 添加更多需要提取的字段...except AttributeError as e:logging.warning(f"解析失败: {str(e)}")return Nonereturn productdef get_page_products(self, page_url: str, selectors: Dict[str, str]) -> List[Dict]:"""获取单页产品数据"""soup = self.fetch_page(page_url)if not soup:return []products = []product_items = soup.select(selectors['product_list'])  # 产品列表选择器for item in product_items:product = self.parse_product(item, selectors)if product:products.append(product)return productsdef save_to_csv(self, data: List[Dict], filename: str):"""保存数据到CSV文件"""if not data:returnkeys = data[0].keys()with open(filename, 'w', newline='', encoding='utf-8-sig') as f:writer = csv.DictWriter(f, fieldnames=keys)writer.writeheader()writer.writerows(data)def run(self,output_file: str = 'products.csv',page_param: str = 'page',max_pages: int = 10,selectors: Dict[str, str] = None):"""执行爬虫"""all_products = []current_page = 1while current_page <= max_pages:# 构造分页URL（根据目标网站实际情况调整）page_url = f"{self.base_url}?{page_param}={current_page}"logging.info(f"正在爬取第 {current_page} 页: {page_url}")products = self.get_page_products(page_url, selectors)if not products:breakall_products.extend(products)current_page += 1self.save_to_csv(all_products, output_file)logging.info(f"爬取完成，共获取 {len(all_products)} 条数据，已保存到 {output_file}")if __name__ == "__main__":# 使用示例（需要根据目标网站修改选择器）spider = ProductSpider(base_url="https://example.com/products",headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36...'},delay=1.5)# CSS选择器配置（需要根据目标网站实际结构修改）SELECTORS = {'product_list': '.product-item',       # 产品列表容器'title': '.product-title',             # 产品标题'price': '.price-value',               # 产品价格'sales': '.sales-count',               # 销量信息'rating': '.rating-score'              # 评分信息}spider.run(output_file='products_data.csv',page_param='page',max_pages=5,selectors=SELECTORS)