基于AI的智能购物助手开发：打破价格歧视的技术实践 为了打破价格歧视我用AI做了一个智能购物助手【B站AI创造公开赛】在电商购物时你是否遇到过这样的情况同一件商品不同用户看到的价格竟然不一样这就是典型的价格歧视现象。作为开发者我们完全可以用技术手段来应对这个问题。本文将分享如何基于AI技术构建一个智能购物助手帮助消费者打破价格歧视获得更公平的购物体验。1. 智能购物助手的背景与核心概念1.1 什么是价格歧视价格歧视是指商家针对不同消费者群体制定不同价格策略的行为。常见的形式包括基于用户画像的歧视根据用户的浏览历史、购买能力、地理位置等信息差异化定价基于时间的歧视在不同时间段调整商品价格如节假日涨价、深夜降价基于平台的歧视同一商品在不同电商平台定价不同1.2 AI购物助手的技术价值传统比价工具只能简单对比价格而AI购物助手具备更强大的能力智能价格预测基于历史数据预测价格走势多平台实时监控同时监控多个电商平台的商品价格个性化推荐根据用户偏好推荐最优购买时机和平台反爬虫对抗智能应对电商平台的反爬虫机制1.3 技术架构概述整个系统采用微服务架构主要包含以下组件数据采集层负责从各电商平台抓取价格信息AI分析层使用机器学习算法分析价格模式用户交互层提供友好的用户界面和通知服务数据存储层存储历史价格数据和用户配置2. 环境准备与技术选型2.1 开发环境要求基础环境配置操作系统Windows 10/11、macOS 10.15 或 Ubuntu 18.04Python版本3.8或更高版本内存至少8GB RAM存储空间至少10GB可用空间开发工具推荐IDEVS Code 或 PyCharm版本控制Git数据库管理工具MySQL Workbench 或 DBeaver2.2 核心技术栈选择后端技术栈# 核心依赖包 requirements { web_framework: FastAPI, # 高性能Web框架 async_library: aiohttp, # 异步HTTP客户端 html_parsing: BeautifulSoup4, # HTML解析 browser_automation: selenium, # 浏览器自动化 machine_learning: scikit-learn, # 机器学习算法 data_analysis: pandas, # 数据处理 visualization: matplotlib, # 数据可视化 }前端技术栈前端框架Vue.js 3.xUI组件库Element Plus图表库ECharts构建工具Vite2.3 第三方服务集成必要的API服务爬虫代理服务解决IP限制问题消息推送服务实现价格提醒功能云存储服务备份历史数据监控服务确保系统稳定性3. 系统架构设计与实现3.1 数据采集模块设计数据采集是整个系统的基础需要处理各种技术挑战反爬虫策略应对方案import asyncio import random from aiohttp import ClientSession from bs4 import BeautifulSoup import time class PriceCrawler: def __init__(self): self.headers_pool [ { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.8,en-US;q0.5,en;q0.3, }, # 更多Header配置... ] async def crawl_product_price(self, url, platform): 异步爬取商品价格信息 try: # 随机选择Header和延迟时间 headers random.choice(self.headers_pool) delay random.uniform(1, 3) await asyncio.sleep(delay) async with ClientSession() as session: async with session.get(url, headersheaders) as response: if response.status 200: html await response.text() return self.parse_price(html, platform) else: print(f请求失败状态码{response.status}) return None except Exception as e: print(f爬取过程中出现错误{e}) return None def parse_price(self, html, platform): 解析不同平台的HTML页面提取价格 soup BeautifulSoup(html, html.parser) if platform taobao: # 淘宝价格解析逻辑 price_element soup.find(span, class_tb-rmb-num) return float(price_element.text) if price_element else None elif platform jd: # 京东价格解析逻辑 price_element soup.find(span, class_p-price) return float(price_element.text.strip()[1:]) if price_element else None # 其他平台解析逻辑...3.2 AI价格预测模型价格趋势预测算法实现import pandas as pd import numpy as np from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import train_test_split from sklearn.metrics import mean_absolute_error import joblib class PricePredictor: def __init__(self): self.model RandomForestRegressor(n_estimators100, random_state42) self.is_trained False def prepare_features(self, historical_data): 准备机器学习特征 df pd.DataFrame(historical_data) # 时间特征工程 df[day_of_week] df[timestamp].dt.dayofweek df[month] df[timestamp].dt.month df[is_weekend] df[day_of_week].isin([5, 6]).astype(int) # 价格变化特征 df[price_change] df[price].diff() df[price_rolling_mean] df[price].rolling(window7).mean() df[price_rolling_std] df[price].rolling(window7).std() # 节假日特征需要外部节假日数据 df[is_holiday] self._check_holiday(df[timestamp]) return df.dropna() def train_model(self, historical_data): 训练价格预测模型 prepared_data self.prepare_features(historical_data) # 特征和目标变量 features [day_of_week, month, is_weekend, price_rolling_mean, price_rolling_std, is_holiday] X prepared_data[features] y prepared_data[price] # 分割训练测试集 X_train, X_test, y_train, y_test train_test_split(X, y, test_size0.2, random_state42) # 训练模型 self.model.fit(X_train, y_train) # 评估模型 y_pred self.model.predict(X_test) mae mean_absolute_error(y_test, y_pred) print(f模型训练完成测试集MAE: {mae:.2f}) self.is_trained True return self.model def predict_future_prices(self, current_data, days7): 预测未来价格趋势 if not self.is_trained: raise ValueError(模型尚未训练请先调用train_model方法) predictions [] last_data current_data.iloc[-1:].copy() for day in range(days): # 更新时间特征 last_data[timestamp] last_data[timestamp] pd.Timedelta(days1) last_data[day_of_week] last_data[timestamp].dt.dayofweek last_data[month] last_data[timestamp].dt.month last_data[is_weekend] last_data[day_of_week].isin([5, 6]).astype(int) # 预测价格 features last_data[[day_of_week, month, is_weekend, price_rolling_mean, price_rolling_std, is_holiday]] predicted_price self.model.predict(features)[0] predictions.append(predicted_price) # 更新滚动特征用于下一次预测 last_data[price] predicted_price return predictions3.3 用户界面与交互设计Web界面核心组件template div classshopping-assistant header classapp-header h1智能购物助手/h1 div classuser-controls el-input v-modelsearchQuery placeholder搜索商品或粘贴链接 keyup.enterhandleSearch template #append el-button iconsearch clickhandleSearch搜索/el-button /template /el-input /div /header main classmain-content div classproduct-monitor h2价格监控列表/h2 el-table :datamonitoredProducts stylewidth: 100% el-table-column propname label商品名称 width200 template #defaultscope a :hrefscope.row.url target_blank{{ scope.row.name }}/a /template /el-table-column el-table-column propcurrentPrice label当前价格 width100 template #defaultscope span :class{price-drop: scope.row.priceChange 0} ¥{{ scope.row.currentPrice }} /span /template /el-table-column el-table-column proppriceChange label价格变化 width100 template #defaultscope span :class{ positive: scope.row.priceChange 0, negative: scope.row.priceChange 0 } {{ scope.row.priceChange 0 ? : }}{{ scope.row.priceChange }}% /span /template /el-table-column el-table-column propplatform label平台 width100/el-table-column el-table-column label操作 width200 template #defaultscope el-button sizesmall clickviewPriceHistory(scope.row)历史价格/el-button el-button sizesmall typedanger clickremoveProduct(scope.row)移除/el-button /template /el-table-column /el-table /div div classprice-chart h2价格趋势分析/h2 div refchartContainer stylewidth: 100%; height: 400px;/div /div /main /div /template script import { ElMessage } from element-plus import * as echarts from echarts export default { name: ShoppingAssistant, data() { return { searchQuery: , monitoredProducts: [], priceHistory: [] } }, mounted() { this.loadMonitoredProducts() this.initChart() }, methods: { async handleSearch() { if (!this.searchQuery.trim()) { ElMessage.warning(请输入搜索关键词或商品链接) return } try { const response await this.$http.post(/api/products/search, { query: this.searchQuery }) this.monitoredProducts response.data ElMessage.success(搜索完成) } catch (error) { ElMessage.error(搜索失败 error.message) } }, async loadMonitoredProducts() { try { const response await this.$http.get(/api/products/monitored) this.monitoredProducts response.data } catch (error) { console.error(加载监控商品失败, error) } }, initChart() { this.chart echarts.init(this.$refs.chartContainer) // 图表配置和数据处理... } } } /script style scoped .shopping-assistant { max-width: 1200px; margin: 0 auto; padding: 20px; } .price-drop { color: #f56c6c; font-weight: bold; } .positive { color: #f56c6c; } .negative { color: #67c23a; } /style4. 核心功能实现细节4.1 多平台价格监控平台适配器模式实现from abc import ABC, abstractmethod from datetime import datetime import re class PlatformAdapter(ABC): 平台适配器抽象基类 abstractmethod def get_product_info(self, url): 获取商品基本信息 pass abstractmethod def extract_price(self, html): 从HTML中提取价格 pass abstractmethod def extract_title(self, html): 从HTML中提取商品标题 pass class TaobaoAdapter(PlatformAdapter): 淘宝平台适配器 def get_product_info(self, url): 获取淘宝商品信息 # 实现淘宝特定的解析逻辑 pass def extract_price(self, html): 淘宝价格提取 # 多种价格匹配模式 price_patterns [ rprice:([\d.]), rspan classtb-rmb-num([\d.])/span, rem classtb-rmb¥/emspan([\d.])/span ] for pattern in price_patterns: match re.search(pattern, html) if match: return float(match.group(1)) return None def extract_title(self, html): 淘宝标题提取 title_pattern rtitle(.*?)/title match re.search(title_pattern, html) if match: return match.group(1).replace( - 淘宝网, ).strip() return None class JDAdapter(PlatformAdapter): 京东平台适配器 def extract_price(self, html): 京东价格提取 price_patterns [ rspan classp-price¥?([\d.])/span, r\price\:\([\d.])\, rjd\.price\s*\s*[\]?([\d.])[\]? ] for pattern in price_patterns: match re.search(pattern, html) if match: return float(match.group(1)) return None class PlatformFactory: 平台适配器工厂 staticmethod def create_adapter(url): 根据URL创建对应的平台适配器 if taobao.com in url or tmall.com in url: return TaobaoAdapter() elif jd.com in url: return JDAdapter() elif pinduoduo.com in url: return PDDAdapter() else: raise ValueError(f不支持的平台URL: {url})4.2 智能价格提醒系统基于规则的价格提醒引擎class PriceAlertEngine: def __init__(self): self.alert_rules {} self.user_preferences {} def add_alert_rule(self, product_id, rule_type, threshold, user_id): 添加价格提醒规则 rule_id f{product_id}_{rule_type}_{user_id} self.alert_rules[rule_id] { product_id: product_id, rule_type: rule_type, # price_drop, target_price, percentage_change threshold: threshold, user_id: user_id, last_checked: datetime.now(), is_active: True } return rule_id def check_price_alerts(self, current_price, product_id): 检查价格是否触发提醒规则 triggered_alerts [] for rule_id, rule in self.alert_rules.items(): if rule[product_id] product_id and rule[is_active]: if self._evaluate_rule(rule, current_price): triggered_alerts.append(rule_id) return triggered_alerts def _evaluate_rule(self, rule, current_price): 评估单个规则是否触发 rule_type rule[rule_type] threshold rule[threshold] if rule_type price_drop: # 价格下降提醒 historical_avg self._get_historical_average(rule[product_id]) return current_price historical_avg * (1 - threshold) elif rule_type target_price: # 目标价格提醒 return current_price threshold elif rule_type percentage_change: # 百分比变化提醒 last_price self._get_last_price(rule[product_id]) if last_price: change_percentage abs(current_price - last_price) / last_price return change_percentage threshold return False def send_alert_notification(self, rule_id, current_price): 发送价格提醒通知 rule self.alert_rules[rule_id] user_prefs self.user_preferences.get(rule[user_id], {}) message self._format_alert_message(rule, current_price) # 根据用户偏好发送通知 if user_prefs.get(email_alerts, False): self._send_email_alert(user_prefs[email], message) if user_prefs.get(push_alerts, False): self._send_push_notification(rule[user_id], message) if user_prefs.get(sms_alerts, False): self._send_sms_alert(user_prefs[phone], message)4.3 数据存储与缓存策略数据库设计核心表结构-- 商品信息表 CREATE TABLE products ( id VARCHAR(50) PRIMARY KEY, name VARCHAR(255) NOT NULL, url TEXT NOT NULL, platform VARCHAR(50) NOT NULL, image_url TEXT, category VARCHAR(100), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_platform (platform), INDEX idx_category (category) ); -- 价格历史表 CREATE TABLE price_history ( id BIGINT AUTO_INCREMENT PRIMARY KEY, product_id VARCHAR(50) NOT NULL, price DECIMAL(10,2) NOT NULL, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, source VARCHAR(50), FOREIGN KEY (product_id) REFERENCES products(id) ON DELETE CASCADE, INDEX idx_product_timestamp (product_id, timestamp), INDEX idx_timestamp (timestamp) ); -- 用户监控表 CREATE TABLE user_monitors ( id BIGINT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(50) NOT NULL, product_id VARCHAR(50) NOT NULL, alert_rules JSON, is_active BOOLEAN DEFAULT TRUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (product_id) REFERENCES products(id) ON DELETE CASCADE, UNIQUE KEY unique_user_product (user_id, product_id) ); -- 价格预测结果表 CREATE TABLE price_predictions ( id BIGINT AUTO_INCREMENT PRIMARY KEY, product_id VARCHAR(50) NOT NULL, predicted_price DECIMAL(10,2) NOT NULL, prediction_date DATE NOT NULL, confidence_score DECIMAL(5,4), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (product_id) REFERENCES products(id) ON DELETE CASCADE, INDEX idx_product_date (product_id, prediction_date) );Redis缓存实现import redis import json from datetime import timedelta class CacheManager: def __init__(self, hostlocalhost, port6379, db0): self.redis_client redis.Redis(hosthost, portport, dbdb, decode_responsesTrue) def cache_product_price(self, product_id, price_data, expire_minutes30): 缓存商品价格数据 cache_key fprice:{product_id} self.redis_client.setex( cache_key, timedelta(minutesexpire_minutes), json.dumps(price_data) ) def get_cached_price(self, product_id): 获取缓存的商品价格 cache_key fprice:{product_id} cached_data self.redis_client.get(cache_key) if cached_data: return json.loads(cached_data) return None def cache_price_history(self, product_id, history_data, expire_hours24): 缓存价格历史数据 cache_key fhistory:{product_id} self.redis_client.setex( cache_key, timedelta(hoursexpire_hours), json.dumps(history_data) )5. 部署与运维方案5.1 容器化部署配置Dockerfile配置FROM python:3.9-slim # 设置工作目录 WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ g \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建非root用户 RUN useradd -m -u 1000 appuser chown -R appuser:appuser /app USER appuser # 暴露端口 EXPOSE 8000 # 启动命令 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]Docker Compose配置version: 3.8 services: web: build: . ports: - 8000:8000 environment: - DATABASE_URLmysql://user:passworddb:3306/shopping_assistant - REDIS_URLredis://redis:6379/0 depends_on: - db - redis volumes: - ./logs:/app/logs db: image: mysql:8.0 environment: MYSQL_ROOT_PASSWORD: rootpassword MYSQL_DATABASE: shopping_assistant MYSQL_USER: appuser MYSQL_PASSWORD: apppassword volumes: - db_data:/var/lib/mysql ports: - 3306:3306 redis: image: redis:6.2-alpine ports: - 6379:6379 volumes: - redis_data:/data scheduler: build: . command: [python, scheduler.py] environment: - DATABASE_URLmysql://user:passworddb:3306/shopping_assistant - REDIS_URLredis://redis:6379/0 depends_on: - db - redis volumes: db_data: redis_data:5.2 监控与日志管理日志配置实现import logging import logging.config from pathlib import Path def setup_logging(): 配置日志系统 log_dir Path(logs) log_dir.mkdir(exist_okTrue) logging.config.dictConfig({ version: 1, disable_existing_loggers: False, formatters: { detailed: { format: %(asctime)s - %(name)s - %(levelname)s - %(message)s }, simple: { format: %(levelname)s - %(message)s } }, handlers: { file: { class: logging.handlers.RotatingFileHandler, filename: log_dir / shopping_assistant.log, maxBytes: 10485760, # 10MB backupCount: 5, formatter: detailed, level: INFO }, console: { class: logging.StreamHandler, formatter: simple, level: INFO }, error_file: { class: logging.handlers.RotatingFileHandler, filename: log_dir / error.log, maxBytes: 10485760, backupCount: 3, formatter: detailed, level: ERROR } }, loggers: { : { handlers: [file, console, error_file], level: INFO, propagate: True }, price_crawler: { handlers: [file, console], level: INFO, propagate: False } } })6. 常见问题与解决方案6.1 爬虫被封禁问题IP轮换与代理池管理class ProxyManager: def __init__(self): self.proxy_list [] self.current_proxy_index 0 self.failed_proxies set() def add_proxy(self, proxy_url): 添加代理服务器 self.proxy_list.append(proxy_url) def get_next_proxy(self): 获取下一个可用的代理 if not self.proxy_list: return None # 轮询选择代理 proxy self.proxy_list[self.current_proxy_index] self.current_proxy_index (self.current_proxy_index 1) % len(self.proxy_list) if proxy in self.failed_proxies: return self.get_next_proxy() return proxy def mark_proxy_failed(self, proxy_url): 标记代理失败 self.failed_proxies.add(proxy_url) def check_proxy_health(self, proxy_url): 检查代理健康状态 # 实现代理健康检查逻辑 pass6.2 数据准确性保障数据验证与清洗机制class DataValidator: staticmethod def validate_price(price, product_id): 验证价格数据的合理性 if price is None: return False # 获取商品历史价格范围 historical_prices PriceHistory.get_prices(product_id, days30) if historical_prices: avg_price sum(historical_prices) / len(historical_prices) # 价格波动超过50%视为异常 if abs(price - avg_price) / avg_price 0.5: return False # 价格必须为正数且在一定范围内 if price 0 or price 1000000: # 假设最大价格100万 return False return True staticmethod def clean_product_title(title): 清洗商品标题 if not title: return # 移除多余空格和特殊字符 cleaned re.sub(r\s, , title.strip()) cleaned re.sub(r[^\w\s\u4e00-\u9fff-], , cleaned) return cleaned6.3 性能优化策略数据库查询优化-- 为常用查询创建索引 CREATE INDEX idx_price_history_product_time ON price_history(product_id, timestamp DESC); CREATE INDEX idx_products_platform_category ON products(platform, category); CREATE INDEX idx_user_monitors_active ON user_monitors(is_active, user_id); -- 使用覆盖索引优化常见查询 SELECT ph.price, ph.timestamp FROM price_history ph WHERE ph.product_id ? ORDER BY ph.timestamp DESC LIMIT 100;缓存策略优化class OptimizedCacheManager(CacheManager): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.local_cache {} self.local_cache_ttl 300 # 5分钟 def get_price_with_fallback(self, product_id): 多级缓存获取价格 # 首先检查本地内存缓存 if product_id in self.local_cache: cached_data, timestamp self.local_cache[product_id] if time.time() - timestamp self.local_cache_ttl: return cached_data # 然后检查Redis缓存 redis_data self.get_cached_price(product_id) if redis_data: # 更新本地缓存 self.local_cache[product_id] (redis_data, time.time()) return redis_data # 最后从数据库获取 db_data self._get_price_from_db(product_id) if db_data: # 更新所有缓存层 self.cache_product_price(product_id, db_data) self.local_cache[product_id] (db_data, time.time()) return db_data return None7. 最佳实践与工程建议7.1 代码质量保障单元测试示例import pytest from unittest.mock import Mock, patch from price_crawler import PriceCrawler from price_predictor import PricePredictor class TestPriceCrawler: def test_crawl_product_price_success(self): 测试成功爬取商品价格 crawler PriceCrawler() mock_html span classtb-rmb-num299.00/span with patch(aiohttp.ClientSession.get) as mock_get: mock_get.return_value.__aenter__.return_value.status 200 mock_get.return_value.__aenter__.return_value.text.return_value mock_html # 测试代码... def test_price_validation(self): 测试价格数据验证 validator DataValidator() # 测试正常价格 assert validator.validate_price(100.0, test_product) True # 测试异常价格 assert validator.validate_price(-10.0, test_product) False assert validator.validate_price(10000000.0, test_product) False class TestPricePredictor: def test_model_training(self): 测试模型训练功能 predictor PricePredictor() sample_data self._generate_sample_data() model predictor.train_model(sample_data) assert model is not None assert predictor.is_trained True def _generate_sample_data(self): 生成测试数据 # 实现测试数据生成逻辑 pass7.2 安全考虑用户数据保护import hashlib import secrets class SecurityManager: staticmethod def hash_user_identifier(user_id): 哈希化用户标识符 salt shopping_assistant_salt return hashlib.sha256(f{user_id}{salt}.encode()).hexdigest() staticmethod def validate_url_safety(url): 验证URL安全性 allowed_domains [taobao.com, tmall.com, jd.com, pinduoduo.com] return any(domain in url for domain in allowed_domains) staticmethod def sanitize_user_input(input_text): 清理用户输入 import html return html.escape(input_text.strip())7.3 可扩展性设计插件架构支持from abc import ABC, abstractmethod import importlib from pathlib import Path class PlatformPlugin(ABC): 平台插件抽象基类 abstractmethod def get_platform_name(self): pass abstractmethod def extract_price(self, html): pass class PluginManager: def __init__(self, plugin_dirplugins): self.plugin_dir Path(plugin_dir) self.plugins {} self.load_plugins() def load_plugins(self): 动态加载插件 for plugin_file in self.plugin_dir.glob(*.py): if plugin_file.name.startswith(_): continue module_name plugin_file.stem spec importlib.util.spec_from_file_location(module_name, plugin_file) module importlib.util.module_from_spec(spec) spec.loader.exec_module(module) for attr_name in dir(module): attr getattr(module, attr_name) if (isinstance(attr, type) and issubclass(attr, PlatformPlugin) and attr ! PlatformPlugin): plugin_instance attr() self.plugins[plugin_instance.get_platform_name()] plugin_instance通过本文的完整实现方案你可以构建一个功能完善的智能购物助手系统。这个系统不仅能够帮助用户打破价格歧视还具备了良好的可扩展性和维护性。在实际项目中建议根据具体需求调整功能模块并始终关注数据准确性和系统稳定性。