一、项目背景
电商数据采集是爬虫最常见的应用场景之一。无论是价格监控、竞品分析还是市场研究，都需要从电商平台获取商品信息。
这篇文章以一个简化的电商数据采集项目为例，完整展示从需求分析到代码实现的整个过程。
项目目标
- 采集商品列表页数据
- 提取商品名称、价格、销量等信息
- 处理分页和反爬
- 数据保存到 JSON 文件
二、需求分析
2.1 数据来源
以某电商平台的商品列表页为例，URL 格式为：
https://api.example.com/items?keyword=手机&page=1&size=20
2.2 数据结构
{
"code": 0,
"data": {
"items": [
{
"id": "12345",
"title": "iPhone 15 Pro",
"price": 7999.00,
"sales": 10000,
"shop": "Apple 官方旗舰店"
}
],
"total": 1000
}
}
三、环境准备
# 安装依赖
pip install requests
pip install fake-useragent
四、代码实现
4.1 基础爬虫
import requests
import json
import time
import random
from fake_useragent import UserAgent
class EcommerceCrawler:
def __init__(self):
self.ua = UserAgent()
self.session = requests.Session()
self.base_url = 'https://api.example.com/items'
def get_headers(self):
return {
'User-Agent': self.ua.random,
'Accept': 'application/json',
'Referer': 'https://www.example.com'
}
def fetch_page(self, keyword, page=1, size=20):
"""获取一页数据"""
params = {
'keyword': keyword,
'page': page,
'size': size
}
try:
response = self.session.get(
self.base_url,
params=params,
headers=self.get_headers(),
timeout=10
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"请求失败: {e}")
return None
def parse_items(self, data):
"""解析商品数据"""
if not data or data.get('code') != 0:
return []
items = data['data']['items']
results = []
for item in items:
results.append({
'id': item.get('id'),
'title': item.get('title'),
'price': item.get('price'),
'sales': item.get('sales'),
'shop': item.get('shop')
})
return results
def crawl(self, keyword, max_pages=5):
"""采集数据"""
all_items = []
for page in range(1, max_pages + 1):
print(f"正在采集第 {page} 页...")
data = self.fetch_page(keyword, page)
if not data:
break
items = self.parse_items(data)
all_items.extend(items)
print(f"第 {page} 页采集完成，获取 {len(items)} 条数据")
# 随机延时
time.sleep(random.uniform(1, 3))
return all_items
def save(self, data, filename='items.json'):
"""保存数据"""
with open(filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"数据已保存到 {filename}")
# 使用
crawler = EcommerceCrawler()
items = crawler.crawl('手机', max_pages=3)
crawler.save(items)
4.2 添加代理
class EcommerceCrawler:
def __init__(self):
# ... 其他初始化 ...
self.proxies = [
'http://proxy1.example.com:8080',
'http://proxy2.example.com:8080',
]
def get_proxy(self):
return {'http': random.choice(self.proxies)}
def fetch_page(self, keyword, page=1, size=20):
# ... 添加代理 ...
response = self.session.get(
self.base_url,
params=params,
headers=self.get_headers(),
proxies=self.get_proxy(),
timeout=10
)
五、运行与测试
# 运行爬虫
python crawler.py
# 输出：
# 正在采集第 1 页...
# 第 1 页采集完成，获取 20 条数据
# 正在采集第 2 页...
# 第 2 页采集完成，获取 20 条数据
# 正在采集第 3 页...
# 第 3 页采集完成，获取 20 条数据
# 数据已保存到 items.json
六、优化与扩展
6.1 添加重试机制
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class EcommerceCrawler:
def __init__(self):
self.session = requests.Session()
# 配置重试
retries = Retry(
total=3,
backoff_factor=1,
status_forcelist=[500, 502, 503, 504]
)
self.session.mount('http://', HTTPAdapter(max_retries=retries))
self.session.mount('https://', HTTPAdapter(max_retries=retries))
6.2 多线程采集
from concurrent.futures import ThreadPoolExecutor
class EcommerceCrawler:
def crawl_multi_thread(self, keyword, max_pages=5, workers=3):
"""多线程采集"""
all_items = []
with ThreadPoolExecutor(max_workers=workers) as executor:
futures = [
executor.submit(self.fetch_page, keyword, page)
for page in range(1, max_pages + 1)
]
for future in futures:
data = future.result()
if data:
items = self.parse_items(data)
all_items.extend(items)
return all_items
七、总结
这个电商爬虫项目展示了爬虫开发的基本流程：
- 分析接口和数据结构
- 编写基础请求代码
- 添加反爬处理
- 优化性能和稳定性