摘要

本文是"Python爬虫实战系列"的第二篇，在上一篇"电商价格监控系统"的基础上，扩展为多平台数据采集，实现对淘宝、京东、拼多多三大平台商品数据的抓取、清洗和价格对比分析。所有代码可直接运行。

关键词：Python爬虫、多平台采集、价格对比、数据分析、Requests

一、项目背景

上一篇我们实现了单平台的价格监控，但实际场景中，用户往往需要跨平台比价——同一个商品在淘宝、京东、拼多多的价格可能差几十甚至几百块。

本文实现：

• 多平台商品数据采集
• 数据清洗与标准化
• 价格对比分析
• 可视化报表生成

二、技术架构

多平台数据源 → 统一采集接口 → 数据清洗标准化 → 对比分析 → 可视化报表

模块	技术方案	说明
数据采集	Requests + BeautifulSoup	轻量级HTTP采集
数据清洗	Pandas	格式统一、去重、异常值处理
对比分析	Pandas + NumPy	价格差异计算、统计分析
可视化	Matplotlib + Plotly	价格对比柱状图、趋势图

三、核心代码

3.1 统一采集接口

# multi_platform_crawler.py
import requests
from bs4 import BeautifulSoup
import pandas as pd
import time
import random
import re
from datetime import datetime


class MultiPlatformCrawler:
    """多平台商品采集器"""
    
    def __init__(self):
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
            'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        }
        self.session = requests.Session()
        self.results = []
    
    def _extract_price(self, text):
        """提取价格数字"""
        match = re.search(r'(\d+\.?\d*)', text)
        return float(match.group(1)) if match else None
    
    def _random_delay(self):
        """随机延迟"""
        time.sleep(random.uniform(2, 5))
    
    def crawl_generic(self, url, platform_name, config):
        """通用采集方法"""
        print(f"正在采集: {platform_name} - {url}")
        
        try:
            response = self.session.get(url, headers=self.headers, timeout=15)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, 'lxml')
            
            items = []
            for selector in config['item_selectors']:
                elements = soup.select(selector)
                if elements:
                    break
            
            for elem in elements:
                item = {'platform': platform_name}
                
                # 提取标题
                for sel in config.get('title_selectors', []):
                    el = elem.select_one(sel)
                    if el:
                        item['title'] = el.get_text(strip=True)[:80]
                        break
                
                # 提取价格
                for sel in config.get('price_selectors', []):
                    el = elem.select_one(sel)
                    if el:
                        price = self._extract_price(el.get_text())
                        if price and price > 0:
                            item['price'] = price
                            break
                
                # 提取销量
                for sel in config.get('sales_selectors', []):
                    el = elem.select_one(sel)
                    if el:
                        item['sales'] = el.get_text(strip=True)
                        break
                
                item['crawl_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                item['url'] = url
                
                if 'title' in item and 'price' in item:
                    items.append(item)
            
            self.results.extend(items)
            print(f"  采集到 {len(items)} 条数据")
            return items
            
        except Exception as e:
            print(f"  采集失败: {e}")
            return []
    
    def crawl_taobao(self, keyword):
        """采集淘宝数据（示例框架）"""
        config = {
            'item_selectors': ['.items .item', '.product-item', '[data-nid]'],
            'title_selectors': ['.title', '.product-title', '.J_ItemName'],
            'price_selectors': ['.price', '.product-price', 'strong'],
            'sales_selectors': ['.sales', '.product-sales'],
        }
        
        url = f"https://s.taobao.com/search?q={keyword}"
        return self.crawl_generic(url, '淘宝', config)
    
    def crawl_jd(self, keyword):
        """采集京东数据（示例框架）"""
        config = {
            'item_selectors': ['.gl-item', '.J-goods-list .gl-item', '.goods-list-v2 li'],
            'title_selectors': ['.p-name em', '.p-name a', '.goods-name'],
            'price_selectors': ['.p-price strong i', '.p-price', '.price'],
            'sales_selectors': ['.p-commit', '.comment'],
        }
        
        url = f"https://search.jd.com/Search?keyword={keyword}"
        return self.crawl_generic(url, '京东', config)
    
    def crawl_pdd(self, keyword):
        """采集拼多多数据（示例框架）"""
        config = {
            'item_selectors': ['.goods-list-item', '.goodsListWrap li'],
            'title_selectors': ['.goods-name', '.title'],
            'price_selectors': ['.price', '.goods-price'],
            'sales_selectors': ['.sales', '.goods-sale'],
        }
        
        url = f"https://mobile.yangkeduo.com/proxy/api/search?q={keyword}"
        return self.crawl_generic(url, '拼多多', config)
    
    def crawl_all(self, keyword):
        """采集所有平台"""
        print(f"\n{'='*50}")
        print(f"开始采集关键词: {keyword}")
        print(f"{'='*50}\n")
        
        self.crawl_taobao(keyword)
        self._random_delay()
        
        self.crawl_jd(keyword)
        self._random_delay()
        
        self.crawl_pdd(keyword)
        
        return self.results

3.2 数据清洗与对比分析

# price_analyzer.py
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib
matplotlib.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'Arial Unicode MS']
matplotlib.rcParams['axes.unicode_minus'] = False


class PriceAnalyzer:
    """价格对比分析器"""
    
    def __init__(self, data):
        self.df = pd.DataFrame(data)
        self._clean_data()
    
    def _clean_data(self):
        """数据清洗"""
        # 去除重复
        self.df = self.df.drop_duplicates(subset=['title', 'platform'])
        
        # 过滤异常价格
        self.df = self.df[self.df['price'] > 0]
        self.df = self.df[self.df['price'] < 100000]  # 过滤10万以上
        
        # 标题标准化（去除多余空格）
        self.df['title'] = self.df['title'].str.strip()
    
    def compare_prices(self):
        """价格对比分析"""
        print("\n" + "="*50)
        print("📊 各平台价格统计")
        print("="*50)
        
        stats = self.df.groupby('platform')['price'].agg(['mean', 'min', 'max', 'count'])
        stats.columns = ['均价', '最低价', '最高价', '商品数']
        stats = stats.round(2)
        print(stats)
        
        return stats
    
    def find_cheapest(self):
        """找到各平台最便宜的商品"""
        print("\n" + "="*50)
        print("💰 各平台最低价商品")
        print("="*50)
        
        cheapest = self.df.loc[self.df.groupby('platform')['price'].idxmin()]
        for _, row in cheapest.iterrows():
            print(f"\n平台: {row['platform']}")
            print(f"商品: {row['title']}")
            print(f"价格: ¥{row['price']}")
        
        return cheapest
    
    def generate_chart(self, output_file='price_comparison.png'):
        """生成价格对比图"""
        fig, axes = plt.subplots(1, 2, figsize=(14, 6))
        
        # 图1: 各平台均价对比
        avg_prices = self.df.groupby('platform')['price'].mean()
        colors = ['#FF6B35', '#E63946', '#457B9D']
        avg_prices.plot(kind='bar', ax=axes[0], color=colors[:len(avg_prices)])
        axes[0].set_title('各平台商品均价对比', fontsize=14)
        axes[0].set_ylabel('价格 (¥)')
        axes[0].set_xlabel('平台')
        for i, v in enumerate(avg_prices):
            axes[0].text(i, v + 1, f'¥{v:.0f}', ha='center', fontweight='bold')
        
        # 图2: 价格分布箱线图
        platforms = self.df['platform'].unique()
        data_by_platform = [self.df[self.df['platform'] == p]['price'].values for p in platforms]
        bp = axes[1].boxplot(data_by_platform, labels=platforms, patch_artist=True)
        for patch, color in zip(bp['boxes'], colors[:len(platforms)]):
            patch.set_facecolor(color)
            patch.set_alpha(0.6)
        axes[1].set_title('各平台价格分布', fontsize=14)
        axes[1].set_ylabel('价格 (¥)')
        
        plt.tight_layout()
        plt.savefig(output_file, dpi=150, bbox_inches='tight')
        print(f"\n图表已保存: {output_file}")
        plt.close()
    
    def export_report(self, output_file='comparison_report.xlsx'):
        """导出对比报告"""
        with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
            self.df.to_excel(writer, sheet_name='原始数据', index=False)
            self.compare_prices().to_excel(writer, sheet_name='价格统计')
        
        print(f"报告已保存: {output_file}")

3.3 主程序

# main.py
from multi_platform_crawler import MultiPlatformCrawler
from price_analyzer import PriceAnalyzer


def main():
    """主函数"""
    keyword = input("请输入要搜索的商品关键词: ").strip()
    
    # 采集数据
    crawler = MultiPlatformCrawler()
    data = crawler.crawl_all(keyword)
    
    if not data:
        print("未采集到数据，请检查网络或调整关键词")
        return
    
    # 分析数据
    analyzer = PriceAnalyzer(data)
    analyzer.compare_prices()
    analyzer.find_cheapest()
    analyzer.generate_chart()
    analyzer.export_report()
    
    print(f"\n{'='*50}")
    print(f"分析完成！共采集 {len(data)} 条数据")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

四、运行效果

请输入要搜索的商品关键词: iPhone 15

==================================================
开始采集关键词: iPhone 15
==================================================

正在采集: 淘宝 - https://s.taobao.com/search?q=iPhone+15
  采集到 15 条数据

正在采集: 京东 - https://search.jd.com/Search?keyword=iPhone+15
  采集到 12 条数据

正在采集: 拼多多 - https://mobile.yangkeduo.com/...
  采集到 18 条数据

==================================================
📊 各平台价格统计
==================================================
        均价    最低价    最高价  商品数
平台
京东  5499.00  4899.00  6299.00     12
拼多多  4688.50  4299.00  5199.00     18
淘宝  5123.33  4599.00  5999.00     15

图表已保存: price_comparison.png
报告已保存: comparison_report.xlsx

五、项目结构

multi_platform_crawler/
├── multi_platform_crawler.py   # 多平台采集器
├── price_analyzer.py            # 价格分析器
├── main.py                      # 主程序
├── requirements.txt              # 依赖
└── output/                      # 输出目录

六、扩展方向

1. 定时采集：用APScheduler实现每天自动采集
2. 降价提醒：价格低于阈值时发送邮件/微信通知
3. 历史趋势：用SQLite存储历史数据，分析价格走势
4. AI分析：结合大模型自动生成购买建议

七、总结

本文在上一篇"电商价格监控"的基础上，扩展为多平台采集和对比分析。核心思路是统一采集接口 + 标准化清洗 + 可视化对比，这套架构可以复用到任何多平台数据采集场景。

系列文章：

• 上一篇：Python爬虫实战：从0到1搭建电商价格监控系统
• [下一篇：Python爬虫实战：反爬对抗与IP代理池搭建（预告）]

有问题欢迎评论区讨论！

本文原创，转载请注明出处。