Python Django电商比价系统：从爬虫到智能Agent的全栈实践

这次我们来看一个基于Python的电商比价系统，它整合了Django框架、requests爬虫、MySQL数据库以及数据分析和可视化功能，甚至还引入了智能体（Agent）的概念。这个项目的核心价值在于，它不是一个简单的爬虫脚本，而是一个完整的、可部署的Web应用系统，能够自动化地从多个电商平台抓取商品价格信息，进行数据清洗、存储、分析和可视化展示，最终辅助用户做出更明智的购物决策。

对于开发者而言，这个项目的吸引力在于其技术栈的完整性和实用性。它覆盖了从后端开发（Django）、数据抓取（requests）、数据存储（MySQL）到前端展示和数据科学（数据分析与可视化）的全链路。更关键的是，它探讨了如何将“智能体”思想融入比价流程，比如自动监控价格波动、触发抓取任务或生成比价报告，这为系统赋予了更高的自动化水平和智能性。

本文将带你从零开始，理解这个系统的架构，并完成一套可运行的部署与测试流程。我们会重点关注几个核心问题：系统如何搭建？爬虫如何稳定运行并规避反爬？数据如何高效存储和分析？可视化图表如何生成？以及所谓的“Agent”在系统中扮演什么角色？无论你是想学习Django全栈开发，还是想构建自己的数据采集与分析项目，这篇文章都能提供直接的参考。

1. 核心能力速览

2. 适用场景与使用边界

这个电商比价系统主要适合以下几类用户：

1. Python全栈学习者：希望通过一个完整的项目实践Django、爬虫、数据库和数据分析。
2. 价格敏感型消费者：希望自己搭建一个工具，长期监控心仪商品的价格变化。
3. 电商数据分析爱好者：想研究不同平台的价格策略、折扣规律和市场趋势。
4. 中小型电商运营者：监控竞品价格，调整自身定价策略。

使用边界与注意事项：

• 合法合规爬取：务必遵守目标网站的robots.txt协议，控制请求频率，避免对目标网站服务器造成压力。本文涉及的爬虫技术仅用于学习与个人合法数据收集。
• 反爬虫策略：电商网站通常有复杂的反爬机制。系统需要集成User-Agent轮换、IP代理池、请求间隔、模拟登录等技术来维持稳定运行。
• 数据准确性：爬取的价格可能不包含实时优惠券、满减活动，分析结果仅供参考。
• 商业用途：若用于商业监控，需确保其符合相关法律法规和平台用户协议。
• Agent的局限性：系统中的“Agent”通常指代自动化程序或智能模块，而非强人工智能。它可能是一个基于规则的价格监控触发器，或一个简单的数据分析报告生成器。

3. 环境准备与前置条件

在开始部署之前，请确保你的开发环境满足以下基本要求。

基础软件环境：

• 操作系统：Windows 10/11, macOS, 或 Linux (如Ubuntu)。推荐使用Linux或macOS进行生产部署。
• Python：版本 3.8 或以上。这是运行Django和大多数数据科学库的基础。
• 包管理工具：pip（通常随Python安装）。
• 版本控制：Git（可选，但强烈推荐用于代码管理）。
• 代码编辑器：VS Code, PyCharm 等。

核心组件清单：

1. Django：Web框架。
2. requests：HTTP请求库，用于爬虫。
3. MySQL客户端与驱动：如mysqlclient或pymysql。
4. 数据分析库：pandas（数据处理），numpy（数值计算）。
5. 可视化库：matplotlib（基础绘图），plotly（交互式图表，更适合Web），或seaborn。
6. 任务队列（可选）：celery+redis/RabbitMQ（用于异步任务和定时爬取）。
7. HTTP请求处理辅助：lxml或beautifulsoup4（解析HTML），fake-useragent（生成随机User-Agent）。

4. 安装部署与启动方式

我们假设项目已经有一个基本的Django结构。以下是搭建环境的通用步骤。

4.1 创建虚拟环境与安装依赖

首先，隔离项目环境。

# 创建项目目录并进入 mkdir ecommerce_price_comparison cd ecommerce_price_comparison # 创建Python虚拟环境（以venv为例） python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate # 安装核心依赖 pip install django requests pandas matplotlib plotly mysqlclient # 如果mysqlclient安装失败，可以尝试pymysql # pip install pymysql # 并在Django的settings.py中稍作配置 # 安装可选依赖（用于爬虫和异步任务） pip install beautifulsoup4 lxml fake-useragent celery redis

4.2 初始化Django项目与数据库配置

# 创建Django项目 django-admin startproject price_comparison_project . # 创建核心应用，例如命名为 `core` 或 `comparison` python manage.py startapp core

接下来，配置MySQL数据库。首先确保你的MySQL服务已启动并创建了一个数据库。

# 登录MySQL（示例） mysql -u root -p # 创建数据库 CREATE DATABASE price_comparison CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

然后，修改price_comparison_project/settings.py文件中的数据库配置部分：

# settings.py DATABASES = { 'default': { 'ENGINE': 'django.db.backends.mysql', 'NAME': 'price_comparison', # 数据库名 'USER': 'your_username', # 你的MySQL用户名 'PASSWORD': 'your_password', # 你的MySQL密码 'HOST': 'localhost', # 数据库主机 'PORT': '3306', # 数据库端口 } }

如果使用pymysql，还需要在settings.py顶部或__init__.py中添加：

import pymysql pymysql.install_as_MySQLdb()

4.3 数据模型设计与迁移

在core/models.py中定义核心数据模型，例如商品、价格记录、电商平台等。

from django.db import models class Platform(models.Model): name = models.CharField(max_length=100) # 如：京东、淘宝、亚马逊 base_url = models.URLField() class Product(models.Model): name = models.CharField(max_length=255) platform = models.ForeignKey(Platform, on_delete=models.CASCADE) external_id = models.CharField(max_length=100, blank=True) # 商品在原平台的ID url = models.URLField() image_url = models.URLField(blank=True) created_at = models.DateTimeField(auto_now_add=True) class PriceHistory(models.Model): product = models.ForeignKey(Product, on_delete=models.CASCADE, related_name='price_history') price = models.DecimalField(max_digits=10, decimal_places=2) timestamp = models.DateTimeField(auto_now_add=True) # 记录抓取时间 # 可添加其他字段，如是否有优惠、库存状态等 class Meta: ordering = ['-timestamp'] # 默认按时间倒序排列

创建并应用数据库迁移：

python manage.py makemigrations core python manage.py migrate

4.4 启动开发服务器

完成基础配置后，可以启动Django开发服务器进行初步验证。

python manage.py runserver

在浏览器中访问http://127.0.0.1:8000，如果看到Django的欢迎页面，说明基础环境搭建成功。

5. 功能测试与效果验证

我们将分模块验证系统的核心功能。

5.1 爬虫模块测试

爬虫模块是系统的数据源头。我们编写一个简单的爬虫服务，放在core/services/scraper.py中。

# core/services/scraper.py import requests from bs4 import BeautifulSoup from fake_useragent import UserAgent import time from django.utils import timezone from core.models import Product, PriceHistory class EcommerceScraper: def __init__(self): self.ua = UserAgent() self.session = requests.Session() self.session.headers.update({ 'User-Agent': self.ua.random }) def fetch_product_price(self, product_url, platform_name): """ 模拟抓取商品价格（示例，实际需针对不同网站解析） """ try: # 1. 发送请求 resp = self.session.get(product_url, timeout=10) resp.raise_for_status() # 检查HTTP错误 # 2. 解析页面（这里需要根据实际网站HTML结构调整） soup = BeautifulSoup(resp.content, 'lxml') # 示例：假设价格在某个class为‘price’的标签里 price_tag = soup.find(class_='price') if price_tag: # 清理价格字符串，如“￥199.00” price_text = price_tag.get_text().strip().replace('￥', '').replace(',', '') price = float(price_text) else: price = None print(f"未在 {product_url} 中找到价格标签") # 3. 返回结果 return { 'success': True, 'price': price, 'timestamp': timezone.now(), 'raw_html_sample': resp.text[:500] # 记录部分HTML用于调试 } except requests.exceptions.RequestException as e: return { 'success': False, 'error': str(e), 'price': None, 'timestamp': timezone.now() } except Exception as e: return { 'success': False, 'error': f'解析错误: {str(e)}', 'price': None, 'timestamp': timezone.now() } # 测试函数 def test_scraper(): scraper = EcommerceScraper() # 用一个测试URL（此处为示例，请替换为真实且允许爬取的URL） test_url = "https://item.jd.com/100000000001.html" # 示例URL result = scraper.fetch_product_price(test_url, "京东") print(f"抓取结果: {result}") # 模拟保存到数据库 if result['success'] and result['price']: # 这里需要先有对应的Product对象 # product = Product.objects.get(url=test_url) # PriceHistory.objects.create(product=product, price=result['price']) print(f"模拟保存价格: {result['price']} 于 {result['timestamp']}") else: print(f"抓取失败: {result.get('error')}") # 礼貌性延迟，避免请求过快 time.sleep(2) if __name__ == '__main__': test_scraper()

测试步骤：

1. 在项目根目录下运行python -m core.services.scraper。
2. 观察控制台输出，检查是否能成功获取HTTP响应并解析出价格。
3. 根据目标网站的实际HTML结构，调整BeautifulSoup的查找逻辑。

关键验证点：

• HTTP请求是否成功（状态码200）。
• HTML解析逻辑是否能准确定位价格元素。
• 是否处理了网络异常和解析异常。
• User-Agent是否随机切换（初步反爬）。

5.2 数据存储与查询测试

验证Django ORM能否正确操作数据库。 在Django Shell中进行测试：

python manage.py shell

# 在Django Shell中执行 from core.models import Platform, Product, PriceHistory from django.utils import timezone # 1. 创建平台 platform_jd, created = Platform.objects.get_or_create(name='京东', base_url='https://www.jd.com') print(f"平台: {platform_jd.name}, 创建状态: {created}") # 2. 创建商品 product, created = Product.objects.get_or_create( name='测试商品-手机', platform=platform_jd, defaults={'url': 'https://item.jd.com/123456.html', 'external_id': '123456'} ) print(f"商品: {product.name}, 创建状态: {created}") # 3. 插入价格历史 price_record = PriceHistory.objects.create(product=product, price=2999.99) print(f"插入价格记录: {price_record.price} at {price_record.timestamp}") # 4. 查询某个商品的最新价格 latest_price = product.price_history.first() # 因为Meta中设置了ordering为‘-timestamp’ if latest_price: print(f"商品【{product.name}】最新价格: {latest_price.price}") else: print("暂无价格记录") # 5. 查询价格历史趋势（最近10条） history = product.price_history.all()[:10] for h in history: print(f"{h.timestamp}: ￥{h.price}")

预期结果：能够成功创建和查询数据库记录，无报错。

5.3 数据分析与可视化测试

使用Pandas分析价格历史，并用Matplotlib或Plotly生成图表。创建一个视图函数或管理命令来测试。

# core/services/analyzer.py import pandas as pd import matplotlib.pyplot as plt from io import BytesIO import base64 from django.db.models import Avg, Max, Min from core.models import Product, PriceHistory from django.utils import timezone from datetime import timedelta def generate_price_trend_chart(product_id, days=30): """ 生成指定商品最近N天的价格趋势图（Base64编码图片） """ end_date = timezone.now() start_date = end_date - timedelta(days=days) # 从数据库查询数据 records = PriceHistory.objects.filter( product_id=product_id, timestamp__gte=start_date, timestamp__lte=end_date ).order_by('timestamp').values('timestamp', 'price') if not records: return None # 转换为Pandas DataFrame df = pd.DataFrame(list(records)) df['timestamp'] = pd.to_datetime(df['timestamp']) df.set_index('timestamp', inplace=True) # 计算统计量 avg_price = df['price'].mean() min_price = df['price'].min() max_price = df['price'].max() # 绘制图表 plt.figure(figsize=(10, 6)) plt.plot(df.index, df['price'], marker='o', linestyle='-', linewidth=2, markersize=4) plt.axhline(y=avg_price, color='r', linestyle='--', alpha=0.7, label=f'平均价: ￥{avg_price:.2f}') plt.fill_between(df.index, min_price, max_price, alpha=0.1, color='gray', label=f'价格区间: ￥{min_price:.2f}-{max_price:.2f}') plt.title(f'商品价格趋势图 (最近{days}天)') plt.xlabel('日期') plt.ylabel('价格 (元)') plt.grid(True, alpha=0.3) plt.legend() plt.xticks(rotation=45) plt.tight_layout() # 将图表保存为Base64字符串，便于在HTML中显示 buffer = BytesIO() plt.savefig(buffer, format='png') buffer.seek(0) image_png = buffer.getvalue() buffer.close() graphic = base64.b64encode(image_png).decode('utf-8') plt.close() # 关闭图表，释放内存 return { 'chart_image': f"data:image/png;base64,{graphic}", 'stats': { 'avg': avg_price, 'min': min_price, 'max': max_price, 'data_points': len(df) } } # 测试函数 def test_analysis(): # 假设存在一个商品ID为1 result = generate_price_trend_chart(product_id=1, days=7) if result: print(f"分析完成。数据点数量: {result['stats']['data_points']}") print(f"平均价: {result['stats']['avg']:.2f}, 最低价: {result['stats']['min']:.2f}, 最高价: {result['stats']['max']:.2f}") # 在实际Web视图中，可以将 result['chart_image'] 直接传给模板的img标签src # <img src="{{ chart_image }}" alt="价格趋势图"> print("图表已生成为Base64字符串（长度省略）") else: print("该商品在指定时间内无价格数据。") if __name__ == '__main__': # 先确保数据库中有商品ID=1的价格历史数据 test_analysis()

测试步骤：

1. 确保数据库中有足够的价格历史数据。
2. 运行python -m core.services.analyzer。
3. 检查控制台输出的统计信息是否正确。
4. （可选）将Base64字符串解码保存为图片文件，查看图表是否正常生成。

5.4 Agent（智能体）功能测试

在此上下文中，“Agent”可以是一个自动化的监控与任务调度模块。一个简单的实现是使用Celery进行定时爬取。 首先，配置Celery。在项目根目录创建celery.py。

# price_comparison_project/celery.py import os from celery import Celery os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'price_comparison_project.settings') app = Celery('price_comparison_project') app.config_from_object('django.conf:settings', namespace='CELERY') app.autodiscover_tasks() @app.task(bind=True, ignore_result=True) def debug_task(self): print(f'Request: {self.request!r}')

在settings.py中添加Celery配置：

# settings.py CELERY_BROKER_URL = 'redis://localhost:6379/0' # 使用Redis作为消息代理 CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' CELERY_TIMEZONE = 'Asia/Shanghai'

然后，创建一个定时爬取任务。

# core/tasks.py from celery import shared_task from core.services.scraper import EcommerceScraper from core.models import Product import time @shared_task def scrape_all_products(): """ 定时任务：抓取所有已监控商品的价格 """ scraper = EcommerceScraper() products = Product.objects.all() results = [] for product in products: print(f"正在抓取商品: {product.name} ({product.platform.name})") result = scraper.fetch_product_price(product.url, product.platform.name) if result['success'] and result['price'] is not None: # 保存价格记录 from core.models import PriceHistory PriceHistory.objects.create(product=product, price=result['price']) results.append({ 'product': product.name, 'price': result['price'], 'status': 'success' }) else: results.append({ 'product': product.name, 'error': result.get('error'), 'status': 'failed' }) time.sleep(3) # 每个请求间隔3秒，避免被封 return results

配置Celery Beat定时调度（在settings.py或单独的celerybeat_schedule中）：

# settings.py from celery.schedules import crontab CELERY_BEAT_SCHEDULE = { 'scrape-every-hour': { 'task': 'core.tasks.scrape_all_products', 'schedule': crontab(minute=0, hour='*/1'), # 每小时执行一次 # 'schedule': 3600.0, # 或每3600秒执行一次 }, }

测试步骤：

1. 确保Redis服务已启动。
2. 启动Celery Worker：celery -A price_comparison_project worker --loglevel=info
3. 启动Celery Beat：celery -A price_comparison_project beat --loglevel=info
4. 观察Worker和Beat的日志，看定时任务是否被正确触发和执行。
5. 检查数据库，确认新的价格记录是否被添加。

6. 接口API与批量任务

系统可以提供RESTful API，方便其他系统集成或前端调用。

6.1 Django REST Framework API示例

首先安装Django REST Framework:pip install djangorestframework创建一个API视图，用于获取商品列表或触发抓取。

# core/api/views.py from rest_framework.decorators import api_view from rest_framework.response import Response from rest_framework import status from core.models import Product, PriceHistory from core.serializers import ProductSerializer, PriceHistorySerializer from core.tasks import scrape_all_products @api_view(['GET']) def product_list(request): """获取所有被监控的商品列表""" products = Product.objects.all() serializer = ProductSerializer(products, many=True) return Response(serializer.data) @api_view(['POST']) def trigger_scraping(request): """手动触发一次批量抓取任务（异步）""" task = scrape_all_products.delay() # 异步执行 return Response({ 'message': '批量抓取任务已触发', 'task_id': task.id }, status=status.HTTP_202_ACCEPTED) @api_view(['GET']) def price_history(request, product_id): """获取某个商品的价格历史""" history = PriceHistory.objects.filter(product_id=product_id).order_by('-timestamp')[:50] # 最近50条 serializer = PriceHistorySerializer(history, many=True) return Response(serializer.data)

配置URL路由：

# core/urls.py from django.urls import path from .api import views urlpatterns = [ path('api/products/', views.product_list, name='product_list'), path('api/trigger-scrape/', views.trigger_scraping, name='trigger_scraping'), path('api/products/<int:product_id>/prices/', views.price_history, name='price_history'), ]

API调用测试（使用curl或Python requests）：

# 获取商品列表 curl -X GET http://127.0.0.1:8000/api/products/ # 触发批量抓取 curl -X POST http://127.0.0.1:8000/api/trigger-scrape/ # 获取商品ID为1的价格历史 curl -X GET http://127.0.0.1:8000/api/products/1/prices/

6.2 批量任务管理

除了Celery定时任务，系统还应支持手动批量导入商品、导出数据等。

• 批量导入：通过Admin后台或上传CSV文件，批量添加监控商品。
• 批量导出：将价格历史导出为CSV或Excel文件，供进一步分析。
• 失败重试：在Celery任务中，对抓取失败的请求可以实现重试机制。

  @shared_task(bind=True, max_retries=3) def scrape_single_product(self, product_id): try: product = Product.objects.get(id=product_id) scraper = EcommerceScraper() result = scraper.fetch_product_price(product.url, product.platform.name) if result['success']: PriceHistory.objects.create(product=product, price=result['price']) return f"Success: {product.name}" else: raise ValueError(result.get('error')) except Exception as exc: # 重试逻辑 raise self.retry(exc=exc, countdown=60) # 60秒后重试

7. 资源占用与性能观察

作为一个Web应用加数据抓取系统，其资源消耗主要集中在数据库、网络I/O和周期性爬虫任务上。

• 数据库性能：

  • 观察点：随着价格历史数据增多，PriceHistory表会急剧膨胀。查询最近价格或生成趋势图可能变慢。
  • 优化建议：
    1. 为PriceHistory表的product_id和timestamp字段建立复合索引。

    CREATE INDEX idx_price_history_product_timestamp ON core_pricehistory (product_id, timestamp DESC);

    2. 考虑对历史数据进行分表或归档（例如，按月分表）。
    3. 在生成图表时，可以在数据库中先进行聚合（使用Django的Avg、Min、Max），减少传输到Python层的数据量。

• 网络与爬虫性能：

  • 观察点：爬虫请求频率过高会导致IP被封。同步抓取大量商品会导致总耗时很长。
  • 优化建议：
    1. 请求间隔：在爬虫代码中强制加入随机延迟（如time.sleep(random.uniform(2, 5))）。
    2. 使用代理IP池：对于大规模抓取，需要集成代理服务。
    3. 异步抓取：使用aiohttp或Scrapy框架替代requests进行异步并发抓取，可以大幅提升效率，但复杂度增加。
    4. 错误处理与重试：完善异常处理，对网络错误实现指数退避重试。

• Web服务器性能：

  • 观察点：当可视化图表生成频繁或数据量大的API被频繁调用时，Django开发服务器可能成为瓶颈。
  • 优化建议：
    1. 生产环境部署：使用Gunicorn/Uvicorn + Nginx。
    2. 缓存：对价格趋势图、商品列表等变化不频繁的数据使用Django缓存框架（如Redis）。
    3. 数据库连接池：考虑使用django-db-connections等工具。

8. 常见问题与排查方法

9. 最佳实践与使用建议

1. 从简单开始，逐步迭代：先实现一个平台、一种商品类型的抓取和展示，跑通全流程。再逐步增加平台、反爬策略、数据分析维度。
2. 重视数据模型设计：良好的数据库设计是系统稳定的基石。仔细规划Product、PriceHistory、Platform等模型的关系和字段。
3. 爬虫伦理与稳健性：
   • 遵守robots.txt：在抓取前检查目标网站的规则。
   • 设置合理的延迟：这是最基本的礼貌，也能让你的爬虫活得更久。
   • 处理异常：网络请求必须包含超时和重试逻辑。
   • 数据清洗：抓取到的原始文本（价格、标题）需要仔细清洗和格式化后再存入数据库。
4. 异步与队列化：对于爬虫这种I/O密集型任务，一定要使用Celery等工具将其异步化、队列化，避免阻塞Web请求。
5. 监控与日志：为爬虫任务和关键业务逻辑添加详细的日志记录。监控任务执行成功率、失败原因，便于及时发现问题。
6. 前端展示优化：使用Plotly.js或ECharts等前端图表库，将图表渲染压力转移到客户端，减轻服务器负担，并获得更好的交互体验。
7. 安全考虑：
   • 生产环境：务必关闭Django的Debug模式，设置安全的SECRET_KEY，配置好ALLOWED_HOSTS。
   • 数据库：不要使用弱密码，避免远程连接。
   • API：如果对外开放，需要添加认证和限流。
8. “Agent”的深化：可以将简单的定时任务升级为更智能的Agent，例如：
   • 价格预警：当价格低于设定阈值时，自动发送邮件或短信通知。
   • 自动抓取策略调整：根据商品热度或价格波动频率，动态调整抓取周期。
   • 竞品分析报告：定期自动生成PDF或邮件报告，对比多个平台的价格、折扣力度。

10. 总结与下一步

这个Python电商比价系统项目是一个绝佳的全栈开发学习案例，它串联了Web开发、数据抓取、存储、分析和可视化等多个关键技术环节。最值得尝试的点在于，你能亲手构建一个从数据采集到价值呈现的完整闭环系统，并且可以根据自己的需求无限扩展。

部署时，建议你按以下顺序验证：

1. 基础环境：确保Python、MySQL、Redis能正常联动。
2. 核心爬虫：针对一个简单的、反爬不严的测试网站，跑通单次抓取和解析。
3. 数据管道：验证数据能否正确存入MySQL并通过Django Admin查看。
4. 定时任务：配置Celery，实现自动化的定时抓取。
5. 可视化展示：在Django模板中成功渲染出一张价格趋势图。
6. API接口：通过curl或Postman测试API是否能正常返回数据。

最容易踩的坑集中在爬虫和部署环节：网站结构变化导致解析失败、请求过快被屏蔽、数据库连接配置错误、Celery任务无法触发。按照第8部分的排查方法，大部分问题都能解决。

完成基础版本后，你可以继续探索的方向包括：集成更多电商平台（需各自适配解析规则）、引入机器学习模型预测价格走势、构建更美观的前端界面、开发移动端小程序、或者将系统容器化（Docker）以便于部署。这个项目就像一棵技能树，你可以沿着任何一枝深入下去，都能获得宝贵的实战经验。