在电商购物时你是否遇到过同一件商品在不同平台、不同账号甚至不同时间点价格差异巨大的情况这种基于用户画像、行为数据甚至设备信息的差异化定价策略就是典型的“价格歧视”。作为消费者我们往往处于信息不对称的劣势方很难实时掌握全网价格动态。本文将以实际项目为例介绍如何利用 AI 技术构建一个智能购物助手自动监控多个电商平台的商品价格通过数据分析和智能提醒帮助用户做出更优的购买决策。这个项目最初是为 B 站 AI 创造公开赛设计的实战案例涉及网络爬虫、数据存储、价格分析和消息推送等完整技术链路。无论你是对 AI 应用开发感兴趣的初学者还是希望将 AI 技术落地到实际场景的开发者都可以通过本文了解从需求分析到技术实现的完整过程。我们将使用 Python 作为主要开发语言结合常见的开源工具和框架构建一个可运行、可扩展的智能购物助手原型。1. 理解价格监控的技术挑战与解决方案1.1 电商价格歧视的常见形式价格歧视并非简单的随机定价而是有策略的差异化定价。常见形式包括平台差异同一商品在淘宝、京东、拼多多等平台定价不同用户画像新老用户、会员等级、消费习惯不同导致价格差异时间策略节假日、促销活动期间价格波动频繁设备识别iOS 用户与 Android 用户可能看到不同价格地理位置不同地区的库存和定价策略有所区别1.2 智能购物助手的核心功能设计一个有效的价格监控系统需要具备以下能力多平台数据采集支持主流电商平台的商品信息获取定时监控机制按设定频率自动检查价格变化价格趋势分析识别历史价格波动规律智能提醒策略在合适时机推送降价通知数据持久化存储历史价格数据用于分析对比1.3 技术架构选型考虑基于功能需求和技术可行性我们选择以下技术栈爬虫框架Requests BeautifulSoup 用于静态页面Selenium 用于动态内容数据存储SQLite 用于轻量级存储MySQL/PostgreSQL 用于生产环境任务调度APScheduler 实现定时监控任务消息推送邮件、Server 酱、钉钉机器人等多渠道通知部署运行Docker 容器化部署Linux 服务器持续运行2. 环境准备与项目结构搭建2.1 开发环境要求确保你的开发环境满足以下要求环境组件版本要求说明Python3.8核心编程语言Chrome浏览器最新版用于动态页面渲染ChromeDriver匹配Chrome版本Selenium 浏览器驱动数据库SQLite3默认轻量级数据存储2.2 创建项目目录结构按照以下结构组织项目文件smart-shopping-assistant/ ├── config/ │ ├── __init__.py │ └── settings.py # 配置文件 ├── spiders/ │ ├── __init__.py │ ├── base_spider.py # 爬虫基类 │ ├── jd_spider.py # 京东爬虫 │ └── taobao_spider.py # 淘宝爬虫 ├── models/ │ ├── __init__.py │ └── product.py # 数据模型 ├── services/ │ ├── __init__.py │ ├── price_analyzer.py # 价格分析 │ └── notifier.py # 消息通知 ├── utils/ │ ├── __init__.py │ ├── database.py # 数据库操作 │ └── logger.py # 日志配置 ├── requirements.txt # 依赖列表 └── main.py # 主程序入口2.3 安装必要的Python依赖创建requirements.txt文件requests2.31.0 beautifulsoup44.12.2 selenium4.15.0 apscheduler3.10.4 sqlalchemy2.0.23 pandas2.1.3 lxml4.9.3 fake-useragent1.4.0 schedule1.2.0 python-dotenv1.0.0使用 pip 安装依赖pip install -r requirements.txt2.4 配置数据库连接在config/settings.py中配置数据库import os from dotenv import load_dotenv load_dotenv() class Config: # 数据库配置 DATABASE_URL os.getenv(DATABASE_URL, sqlite:///products.db) # 爬虫配置 REQUEST_TIMEOUT 30 RETRY_TIMES 3 # 价格监控配置 CHECK_INTERVAL 3600 # 1小时检查一次 PRICE_DROP_THRESHOLD 0.1 # 降价10%触发通知 # 通知配置 EMAIL_HOST os.getenv(EMAIL_HOST) EMAIL_PORT os.getenv(EMAIL_PORT, 587) EMAIL_USER os.getenv(EMAIL_USER) EMAIL_PASSWORD os.getenv(EMAIL_PASSWORD)3. 实现核心数据模型与爬虫功能3.1 设计商品数据模型在models/product.py中定义数据模型from sqlalchemy import Column, Integer, String, Float, DateTime, Text from sqlalchemy.ext.declarative import declarative_base from datetime import datetime Base declarative_base() class Product(Base): __tablename__ products id Column(Integer, primary_keyTrue) platform Column(String(50), nullableFalse) # 平台名称jd, taobao等 product_id Column(String(100), nullableFalse) # 平台商品ID title Column(Text, nullableFalse) # 商品标题 current_price Column(Float, nullableFalse) # 当前价格 original_price Column(Float) # 原价 url Column(Text, nullableFalse) # 商品链接 image_url Column(Text) # 商品图片 created_time Column(DateTime, defaultdatetime.now) updated_time Column(DateTime, defaultdatetime.now, onupdatedatetime.now) class PriceHistory(Base): __tablename__ price_history id Column(Integer, primary_keyTrue) product_id Column(Integer, nullableFalse) # 关联products.id price Column(Float, nullableFalse) # 历史价格 check_time Column(DateTime, defaultdatetime.now) # 检查时间3.2 实现爬虫基类创建通用的爬虫基类spiders/base_spider.pyimport requests from bs4 import BeautifulSoup from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from fake_useragent import UserAgent import time import logging logger logging.getLogger(__name__) class BaseSpider: def __init__(self, use_seleniumFalse): self.use_selenium use_selenium self.ua UserAgent() self.session requests.Session() self.setup_session() if use_selenium: self.driver self.setup_selenium() def setup_session(self): 配置请求会话 self.session.headers.update({ User-Agent: self.ua.random, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, }) def setup_selenium(self): 配置Selenium浏览器驱动 chrome_options Options() chrome_options.add_argument(--headless) chrome_options.add_argument(--no-sandbox) chrome_options.add_argument(--disable-dev-shm-usage) chrome_options.add_argument(f--user-agent{self.ua.random}) driver webdriver.Chrome(optionschrome_options) driver.implicitly_wait(10) return driver def get_page(self, url, wait_forNone): 获取页面内容 if self.use_selenium: return self.get_page_selenium(url, wait_for) else: return self.get_page_requests(url) def get_page_requests(self, url): 使用requests获取页面 try: response self.session.get(url, timeout30) response.raise_for_status() return response.text except requests.RequestException as e: logger.error(f请求失败: {url}, 错误: {e}) return None def get_page_selenium(self, url, wait_forNone): 使用Selenium获取页面 try: self.driver.get(url) if wait_for: WebDriverWait(self.driver, 10).until( EC.presence_of_element_located((By.CSS_SELECTOR, wait_for)) ) return self.driver.page_source except Exception as e: logger.error(fSelenium请求失败: {url}, 错误: {e}) return None def parse_price(self, price_text): 解析价格文本为浮点数 if not price_text: return 0.0 # 移除货币符号和空格保留数字和小数点 import re price_str re.sub(r[^\d.], , price_text) try: return float(price_str) except ValueError: logger.warning(f价格解析失败: {price_text}) return 0.0 def close(self): 清理资源 if hasattr(self, driver): self.driver.quit()3.3 实现京东商品爬虫创建spiders/jd_spider.pyfrom .base_spider import BaseSpider from bs4 import BeautifulSoup import logging logger logging.getLogger(__name__) class JDSpider(BaseSpider): def __init__(self): super().__init__(use_seleniumTrue) # 京东需要JS渲染 def get_product_info(self, url): 获取京东商品信息 html self.get_page(url, wait_for.sku-name) if not html: return None soup BeautifulSoup(html, lxml) try: # 提取商品标题 title_elem soup.select_one(.sku-name) title title_elem.get_text(stripTrue) if title_elem else 未知商品 # 提取当前价格 price_elem soup.select_one(.price .price-num) current_price self.parse_price(price_elem.get_text() if price_elem else 0) # 提取原价 original_price_elem soup.select_one(.price .price-jp) original_price self.parse_price( original_price_elem.get_text() if original_price_elem else str(current_price) ) # 提取商品图片 image_elem soup.select_one(#spec-img) image_url image_elem.get(data-origin) if image_elem else product_info { platform: jd, product_id: self.extract_product_id(url), title: title, current_price: current_price, original_price: original_price, url: url, image_url: image_url } logger.info(f京东商品信息获取成功: {title}, 价格: {current_price}) return product_info except Exception as e: logger.error(f解析京东商品信息失败: {e}) return None def extract_product_id(self, url): 从URL中提取商品ID import re match re.search(r/(\d)\.html, url) return match.group(1) if match else unknown3.4 实现数据库操作工具在utils/database.py中创建数据库管理类from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from models.product import Base, Product, PriceHistory import logging logger logging.getLogger(__name__) class DatabaseManager: def __init__(self, database_url): self.engine create_engine(database_url) self.Session sessionmaker(bindself.engine) self.create_tables() def create_tables(self): 创建数据表 Base.metadata.create_all(self.engine) logger.info(数据库表创建完成) def add_product(self, product_data): 添加或更新商品信息 session self.Session() try: # 检查商品是否已存在 product session.query(Product).filter_by( platformproduct_data[platform], product_idproduct_data[product_id] ).first() if product: # 更新现有商品 price_changed product.current_price ! product_data[current_price] product.current_price product_data[current_price] product.original_price product_data[original_price] product.title product_data[title] product.image_url product_data[image_url] product.updated_time product_data.get(updated_time) else: # 添加新商品 product Product(**product_data) session.add(product) price_changed True session.commit() # 如果价格变化记录历史价格 if price_changed: self.record_price_history(product.id, product_data[current_price]) logger.info(f商品信息保存成功: {product_data[title]}) return product, price_changed except Exception as e: session.rollback() logger.error(f保存商品信息失败: {e}) return None, False finally: session.close() def record_price_history(self, product_id, price): 记录价格历史 session self.Session() try: price_history PriceHistory(product_idproduct_id, priceprice) session.add(price_history) session.commit() except Exception as e: session.rollback() logger.error(f记录价格历史失败: {e}) finally: session.close() def get_price_history(self, product_id, days30): 获取商品价格历史 session self.Session() try: from datetime import datetime, timedelta start_date datetime.now() - timedelta(daysdays) history session.query(PriceHistory).filter( PriceHistory.product_id product_id, PriceHistory.check_time start_date ).order_by(PriceHistory.check_time.asc()).all() return history except Exception as e: logger.error(f获取价格历史失败: {e}) return [] finally: session.close()4. 构建价格分析与通知服务4.1 实现价格趋势分析创建services/price_analyzer.pyimport logging from datetime import datetime, timedelta from sqlalchemy import func from utils.database import DatabaseManager logger logging.getLogger(__name__) class PriceAnalyzer: def __init__(self, db_manager): self.db db_manager def analyze_price_trend(self, product_id, days30): 分析价格趋势 history self.db.get_price_history(product_id, days) if len(history) 2: return { trend: stable, change_percent: 0, min_price: history[0].price if history else 0, max_price: history[0].price if history else 0, suggestion: 数据不足无法分析趋势 } prices [h.price for h in history] current_price prices[-1] min_price min(prices) max_price max(prices) # 计算价格变化百分比 first_price prices[0] change_percent ((current_price - first_price) / first_price) * 100 # 判断趋势 if change_percent -5: trend down suggestion 价格呈下降趋势可以考虑购买 elif change_percent 5: trend up suggestion 价格呈上升趋势建议观望 else: trend stable suggestion 价格相对稳定可根据需要购买 # 检查是否接近历史最低价 if current_price min_price * 1.05: # 在当前最低价的5%范围内 suggestion 接近历史最低价建议立即购买 return { trend: trend, change_percent: round(change_percent, 2), min_price: min_price, max_price: max_price, current_price: current_price, suggestion: suggestion } def should_notify(self, product, previous_price, current_price, threshold0.1): 判断是否应该发送通知 if previous_price 0 or current_price 0: return False price_drop (previous_price - current_price) / previous_price # 价格下降超过阈值 if price_drop threshold: return True, f价格下降 {price_drop*100:.1f}% # 达到历史最低价 trend self.analyze_price_trend(product.id) if current_price trend[min_price] * 1.01: # 在最低价1%范围内 return True, 达到历史最低价 return False, 4.2 实现多渠道消息通知创建services/notifier.pyimport smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import requests import logging from config.settings import Config logger logging.getLogger(__name__) class Notifier: def __init__(self): self.config Config() def send_email(self, to_email, subject, content): 发送邮件通知 if not all([self.config.EMAIL_HOST, self.config.EMAIL_USER, self.config.EMAIL_PASSWORD]): logger.warning(邮件配置不完整跳过发送) return False try: msg MIMEMultipart() msg[From] self.config.EMAIL_USER msg[To] to_email msg[Subject] subject msg.attach(MIMEText(content, html)) server smtplib.SMTP(self.config.EMAIL_HOST, self.config.EMAIL_PORT) server.starttls() server.login(self.config.EMAIL_USER, self.config.EMAIL_PASSWORD) server.send_message(msg) server.quit() logger.info(f邮件发送成功: {to_email}) return True except Exception as e: logger.error(f邮件发送失败: {e}) return False def send_webhook(self, webhook_url, message): 发送Webhook通知支持钉钉、企业微信等 try: payload { msgtype: text, text: { content: message } } response requests.post(webhook_url, jsonpayload, timeout10) if response.status_code 200: logger.info(Webhook通知发送成功) return True else: logger.error(fWebhook通知发送失败: {response.status_code}) return False except Exception as e: logger.error(fWebhook通知发送异常: {e}) return False def format_price_alert_message(self, product, price_change_info, trend_analysis): 格式化价格提醒消息 message f 价格提醒{product.title} 当前价格¥{product.current_price} 价格变化{price_change_info} 趋势分析 - 近期趋势{trend_analysis[trend]} - 变化幅度{trend_analysis[change_percent]}% - 历史最低¥{trend_analysis[min_price]} - 历史最高¥{trend_analysis[max_price]} 建议{trend_analysis[suggestion]} 商品链接{product.url} return message5. 集成调度与主程序实现5.1 配置任务调度器创建任务调度管理类from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger import logging from spiders.jd_spider import JDSpider from utils.database import DatabaseManager from services.price_analyzer import PriceAnalyzer from services.notifier import Notifier logger logging.getLogger(__name__) class TaskScheduler: def __init__(self, db_manager): self.scheduler BackgroundScheduler() self.db db_manager self.price_analyzer PriceAnalyzer(db_manager) self.notifier Notifier() self.monitored_products [] # 监控的商品列表 def add_product_to_monitor(self, product_url, platformjd): 添加要监控的商品 product_info { url: product_url, platform: platform } self.monitored_products.append(product_info) logger.info(f添加监控商品: {product_url}) def check_product_price(self): 检查商品价格任务 for product_info in self.monitored_products: try: self._check_single_product(product_info) except Exception as e: logger.error(f检查商品价格失败: {e}) def _check_single_product(self, product_info): 检查单个商品价格 spider None try: # 根据平台选择爬虫 if product_info[platform] jd: spider JDSpider() else: logger.warning(f不支持的平台: {product_info[platform]}) return # 获取商品信息 product_data spider.get_product_info(product_info[url]) if not product_data: return # 保存到数据库 product, price_changed self.db.add_product(product_data) if price_changed and product: # 分析价格趋势 trend_analysis self.price_analyzer.analyze_price_trend(product.id) # 判断是否需要通知 should_notify, change_info self.price_analyzer.should_notify( product, product.current_price, product_data[current_price] ) if should_notify: # 发送通知 message self.notifier.format_price_alert_message( product, change_info, trend_analysis ) # 这里可以添加具体的通知逻辑 logger.info(f需要发送通知: {message}) finally: if spider: spider.close() def start_monitoring(self, interval_hours1): 启动监控任务 trigger IntervalTrigger(hoursinterval_hours) self.scheduler.add_job( self.check_product_price, triggertrigger, idprice_monitoring ) self.scheduler.start() logger.info(f价格监控任务已启动间隔: {interval_hours}小时) def stop_monitoring(self): 停止监控任务 self.scheduler.shutdown() logger.info(价格监控任务已停止)5.2 实现主程序入口创建main.py作为程序入口import logging import sys from config.settings import Config from utils.database import DatabaseManager from utils.logger import setup_logging from task_scheduler import TaskScheduler def main(): # 设置日志 setup_logging() logger logging.getLogger(__name__) try: # 初始化配置 config Config() logger.info(智能购物助手启动中...) # 初始化数据库 db_manager DatabaseManager(config.DATABASE_URL) logger.info(数据库初始化完成) # 初始化任务调度器 scheduler TaskScheduler(db_manager) # 添加示例监控商品实际使用时可以从配置文件或数据库读取 sample_products [ { url: https://item.jd.com/100000000001.html, # 示例商品 platform: jd } ] for product in sample_products: scheduler.add_product_to_monitor(product[url], product[platform]) # 启动监控任务 scheduler.start_monitoring(interval_hours1) logger.info(智能购物助手启动成功开始监控价格...) # 保持程序运行 try: while True: import time time.sleep(60) except KeyboardInterrupt: logger.info(接收到中断信号正在停止程序...) scheduler.stop_monitoring() logger.info(程序已安全退出) except Exception as e: logger.error(f程序启动失败: {e}) sys.exit(1) if __name__ __main__: main()5.3 配置日志系统在utils/logger.py中配置日志import logging import os from datetime import datetime def setup_logging(): 配置日志系统 # 创建日志目录 log_dir logs if not os.path.exists(log_dir): os.makedirs(log_dir) # 日志文件名包含日期 log_file os.path.join(log_dir, fshopping_assistant_{datetime.now().strftime(%Y%m%d)}.log) # 配置日志格式 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(log_file, encodingutf-8), logging.StreamHandler(sys.stdout) ] )6. 运行验证与常见问题排查6.1 首次运行测试创建测试脚本test_run.py#!/usr/bin/env python3 智能购物助手测试脚本 import logging from config.settings import Config from utils.database import DatabaseManager from spiders.jd_spider import JDSpider def test_database(): 测试数据库连接 print(测试数据库连接...) try: config Config() db DatabaseManager(config.DATABASE_URL) print(✅ 数据库连接成功) return True except Exception as e: print(f❌ 数据库连接失败: {e}) return False def test_jd_spider(): 测试京东爬虫 print(测试京东爬虫...) spider JDSpider() try: # 使用一个公开的测试商品 test_url https://item.jd.com/100000000001.html # 需要替换为真实商品URL product_info spider.get_product_info(test_url) if product_info: print(✅ 京东爬虫测试成功) print(f商品标题: {product_info[title]}) print(f商品价格: {product_info[current_price]}) return True else: print(❌ 京东爬虫测试失败) return False except Exception as e: print(f❌ 京东爬虫测试异常: {e}) return False finally: spider.close() if __name__ __main__: print(开始智能购物助手组件测试...) # 运行测试 db_success test_database() spider_success test_jd_spider() if db_success and spider_success: print(\n 所有组件测试通过可以运行主程序。) else: print(\n⚠️ 部分组件测试失败请检查配置。)6.2 常见问题与解决方案问题1ChromeDriver 版本不匹配现象Selenium 报错This version of ChromeDriver only supports Chrome version XXX解决# 查看Chrome版本 google-chrome --version # 下载对应版本的ChromeDriver # 访问 https://chromedriver.chromium.org/downloads # 或使用版本管理工具 pip install webdriver-manager问题2反爬虫检测现象请求被拒绝返回403错误或验证码页面解决增加请求头伪装使用代理IP轮换添加随机延迟使用更真实的User-Agent# 在BaseSpider中增强反爬措施 def enhance_anti_anti_crawler(self): self.session.headers.update({ Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.8,zh-TW;q0.7,zh-HK;q0.5,en-US;q0.3,en;q0.2, Accept-Encoding: gzip, deflate, br, DNT: 1, Connection: keep-alive, Upgrade-Insecure-Requests: 1, }) # 添加随机延迟 import random time.sleep(random.uniform(1, 3))问题3价格解析错误现象价格解析为0或异常值解决增强价格解析逻辑def parse_price_enhanced(self, price_text): 增强的价格解析方法 if not price_text: return 0.0 import re # 匹配多种价格格式 patterns [ r¥\s*(\d\.?\d*), # ¥100.00 r\s*(\d\.?\d*), # 100.00 r(\d\.?\d*)\s*元, # 100.00元 r(\d\.?\d*), # 纯数字 ] for pattern in patterns: match re.search(pattern, price_text) if match: try: return float(match.group(1)) except ValueError: continue return 0.07. 生产环境部署与优化建议7.1 Docker 容器化部署创建DockerfileFROM python:3.9-slim WORKDIR /app # 安装Chrome浏览器 RUN apt-get update apt-get install -y \ wget \ gnupg \ wget -q -O - https://dl.google.com/linux/linux_signing_key.pub | apt-key add - \ echo deb [archamd64] http://dl.google.com/linux/chrome/deb/ stable main /etc/apt/sources.list.d/google-chrome.list \ apt-get update \ apt-get install -y google-chrome-stable \ rm -rf /var/lib/apt/lists/* # 安装ChromeDriver RUN wget -q https://chromedriver.storage.googleapis.com/$(curl -s https://chromedriver.storage.googleapis.com/LATEST_RELEASE)/chromedriver_linux64.zip \ unzip chromedriver_linux64.zip \ mv chromedriver /usr/local/bin/ \ rm chromedriver_linux64.zip # 复制项目文件 COPY requirements.txt . RUN pip install -r requirements.txt COPY . . # 创建日志目录 RUN mkdir -p logs CMD [python, main.py]创建docker-compose.ymlversion: 3.8 services: shopping-assistant: build: . container_name: smart-shopping-assistant restart: unless-stopped volumes: - ./data:/app/data - ./logs:/app/logs environment: - DATABASE_URLsqlite:///data/products.db - EMAIL_HOSTsmtp.example.com - EMAIL_USERyour_emailexample.com - EMAIL_PASSWORDyour_password logging: driver: json-file options: max-size: 10m max-file: 37.2 性能优化建议数据库优化# 使用连接池 from sqlalchemy.pool import QueuePool engine create_engine( database_url, poolclassQueuePool, pool_size10, max_overflow20, pool_timeout30 )爬虫性能优化# 使用异步请求 import aiohttp import asyncio async def fetch_multiple_products(urls): async with aiohttp.ClientSession() as session: tasks [fetch_single_product(session, url) for url in urls] return await asyncio.gather(*tasks)内存管理优化# 定期清理历史数据 def cleanup_old_data(self, days90): 清理90天前的历史数据 from datetime import datetime, timedelta cutoff_date datetime.now() - timedelta(daysdays) session self.Session() try: deleted_count session.query(PriceHistory).filter( PriceHistory.check_time cutoff_date ).delete() session.commit() logger.info(f清理了 {deleted_count} 条历史数据) except Exception as e: session.rollback() logger.error(f数据清理失败: {e}) finally: session.close()7.3 安全注意事项敏感信息保护# 使用环境变量或配置文件 import os from dotenv import load_dotenv load_dotenv() # 从.env文件加载配置 # 不要硬编码敏感信息 DATABASE_URL os.getenv(DATABASE_URL) API_KEYS os.getenv(API_KEYS, ).split(,)访问频率控制import time from functools import wraps def rate_limit(max_per_minute): 限流装饰器 interval 60.0 / max_per_minute def decorator(func): last_called [0.0] wraps(func) def wrapper(*args, **kwargs): elapsed time.time() - last_called[0] left_to_wait interval - elapsed if left_to_wait 0: time.sleep(left_to_wait) ret func(*args, **kwargs) last_called[0] time
