多平台商品数据聚合爬虫的架构设计与反爬对抗实战
一、业务需求
在一站式日淘全品类平台中，用户需要在一个搜索框里同时看到乐天、日本亚马逊、骏河屋、煤炉等多个平台的商品。这背后的技术挑战是：如何从多个异构平台高效、稳定地采集商品数据，并统一格式返回给用户。
我负责这个聚合爬虫系统的时候，面临的挑战包括：
·
每个平台的页面结构完全不同
·
·
反爬策略各异（IP限流、User-Agent检测、验证码、动态渲染）
·
·
数据量巨大（百万级商品需要保持更新）
·
·
时效性要求高（商品价格变化需要在分钟级内同步）
·
二、整体架构
我设计了一个分层、可插拔的爬虫架构：
text
┌─────────────────────────────────────────────────────────┐│ 调度层（Scheduler） ││ 定时触发 / 事件触发 / 手动触发 │└─────────────────────┬───────────────────────────────────┘ ▼┌─────────────────────────────────────────────────────────┐│ 任务队列（Redis） ││ 按平台分队列，支持优先级 │└─────────────────────┬───────────────────────────────────┘ ▼┌─────────────────────────────────────────────────────────┐│ Worker Pool（多进程） ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ││ │ Rakuten │ │ Amazon │ │ Surugaya │ ││ │ Worker │ │ Worker │ │ Worker │ ││ └──────────┘ └──────────┘ └──────────┘ │└─────────────────────┬───────────────────────────────────┘ ▼┌─────────────────────────────────────────────────────────┐│ 数据清洗与标准化（Pipeline） ││ 统一字段映射 / 价格转换 / 去重 │└─────────────────────┬───────────────────────────────────┘ ▼┌─────────────────────────────────────────────────────────┐│ 存储层（MySQL + ES） │└─────────────────────────────────────────────────────────┘
三、每个平台的适配器实现
每个平台都有一个独立的适配器，封装了该平台的请求构造、解析逻辑和反爬策略：
python
from abc import ABC, abstractmethodclass PlatformAdapter(ABC): @abstractmethod def search(self, keyword, page=1): """搜索商品""" pass @abstractmethod def get_detail(self, item_id): """获取商品详情""" pass @abstractmethod def get_price(self, item_id): """获取实时价格（用于雅虎代拍）""" passclass RakutenAdapter(PlatformAdapter): def __init__(self): self.base_url = "https://api.rms.rakuten.co.jp" self.session = self._create_session() def search(self, keyword, page=1): # 乐天有官方API，优先使用 url = f"{self.base_url}/es/1.0/search" params = { 'keyword': keyword, 'page': page, 'hits': 30 } response = self.session.get(url, params=params) return self._parse_search_response(response.json()) def _parse_search_response(self, data): items = [] for hit in data.get('hits', []): items.append({ 'platform': 'rakuten', 'item_id': hit['itemId'], 'title': hit['itemName'], 'price': hit['itemPrice'], 'image_url': hit.get('imageUrl'), 'url': hit.get('itemUrl') }) return itemsclass SurugayaAdapter(PlatformAdapter): def __init__(self): self.session = self._create_session_with_cookies() self.csrf_token = self._get_csrf_token() def search(self, keyword, page=1): # 骏河屋没有官方API，需要爬取HTML url = f"https://www.surugaya.jp/search.php" params = { 'keyword': keyword, 'page': page } response = self.session.get(url, params=params) return self._parse_html(response.text) def _parse_html(self, html): from bs4 import BeautifulSoup soup = BeautifulSoup(html, 'html.parser') items = [] for item_div in soup.select('.item_list'): items.append({ 'platform': 'surugaya', 'item_id': self._extract_id(item_div), 'title': self._extract_title(item_div), 'price': self._extract_price(item_div), 'condition': self._extract_condition(item_div), # 中古等级 'url': self._extract_url(item_div) }) return items
四、反爬策略的对抗
挑战一：IP限流
乐天和日亚对同一IP的请求频率有严格限制。我的解决方案是维护一个动态代理IP池：
python
import randomimport redisclass ProxyPool: def __init__(self): self.redis = redis.Redis(decode_responses=True) self.key = "proxy_pool" def get_proxy(self): # 从池中随机获取一个代理 proxies = self.redis.smembers(self.key) if not proxies: return None proxy = random.choice(list(proxies)) # 验证代理是否可用 if self._validate(proxy): return proxy else: self.redis.srem(self.key, proxy) return self.get_proxy() def _validate(self, proxy): try: response = requests.get( "http://httpbin.org/ip", proxies={'http': proxy, 'https': proxy}, timeout=5 ) return response.status_code == 200 except: return False
挑战二：动态渲染
骏河屋和煤炉的部分页面使用JavaScript渲染，传统爬虫拿不到完整内容。我用Playwright配合无头浏览器：
python
from playwright.sync_api import sync_playwrightdef fetch_with_playwright(url): with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() # 设置更真实的User-Agent page.set_extra_http_headers({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' }) page.goto(url, wait_until='networkidle') # 等待动态内容加载 page.wait_for_selector('.item-list', timeout=10000) content = page.content() browser.close() return content
五、数据标准化与去重
不同平台的数据格式差异很大，需要统一映射：
python
class DataNormalizer: def normalize(self, raw_item): """将各平台的原始数据转换为标准格式""" platform = raw_item.get('platform') # 价格统一转为日元 price = self._normalize_price(raw_item.get('price'), platform) # 商品状态统一 status = self._normalize_status(raw_item.get('status'), platform) return { 'platform': platform, 'item_id': raw_item['item_id'], 'title': raw_item['title'], 'price_jpy': price, 'price_cny': self._jpy_to_cny(price), 'status': status, 'url': raw_item.get('url'), 'image_url': raw_item.get('image_url'), 'crawled_at': datetime.now().isoformat() }
六、性能与稳定性
这套聚合爬虫系统目前：
·
日均采集商品数：50万+
·
·
搜索响应时间：平均1.2秒
·
·
数据更新延迟：< 5分钟
·
·
可用性：99.5%
·
对于需要日本代购数据聚合的场景，这套架构提供了稳定、高效的数据支撑。
- 点赞
- 收藏
- 关注作者
评论（0）