一、多平台比价的技术痛点与行业挑战
在电商平台碎片化的今天，消费者与商家都面临着跨平台比价的效率瓶颈：传统比价工具往往局限于固定平台、固定字段的简单价格对比，无法应对不同平台的商品描述差异、促销规则复杂性及价值维度多样性。
与单一平台搜索相比，多平台自定义比价存在三大核心技术挑战：
商品匹配的模糊性：同一商品在不同平台的标题描述、规格参数差异巨大（如 "iPhone 15" 在 A 平台标 "苹果 15 256G"，在 B 平台标 "iPhone15 256GB 星光色"）
价值维度的多样性：价格之外，用户可能关注 "次日达"" 正品保障 ""售后评分"" 历史价格走势 " 等多维价值
规则适配的动态性：各平台的反爬策略、页面结构、促销规则（满减 / 优惠券 / 会员价）持续变化
传统比价方案的局限性显著：
采用固定 XPath 解析页面，平台结构微调即失效
仅对比标价，忽略 "满 300 减 50"" 会员专享价 " 等实际支付差异
缺乏商品匹配的智能容错机制，同名不同规格商品易混淆
本文方案的核心突破点：
构建动态规则引擎，支持可视化配置各平台解析规则，无需代码变更即可适配平台变化
开发商品特征向量匹配算法，基于标题、规格、图片的多维度相似度计算
设计综合价值评估模型，融合价格、服务、信誉等 12 个维度的加权评分体系
二、核心技术方案与系统架构
1. 多平台比价专属数据维度
数据模块 核心字段 技术处理方式
基础特征 商品 ID、标题、主图、规格参数 特征提取 + 标准化处理
价格体系 标价、促销价、会员价、满减规则 动态规则解析 + 实际支付计算
服务指标 配送方式、时效、售后政策、运费 关键词提取 + 服务等级映射
商家信誉 评分、好评率、投诉率、经营年限 数据归一化 + 权重分配
历史波动 近 30 天最低价、价格变动频率 时间序列分析 + 趋势预测
平台特性 正品保障、假一赔十、独家销售 平台规则映射 + 价值加分
2. 系统架构设计
用户搜索请求
请求解析
关键词提取
筛选条件解析
价值维度偏好
C&D&E
动态规则引擎
平台适配器集群
拼多多
京东
当当/孔夫子
H&I&J
数据清洗与标准化
商品特征向量生成
相似度匹配与去重
综合价值评估
价格趋势分析
多维度排序与展示
三、核心代码实现：从动态解析到智能匹配
多平台智能比价系统核心实现
import time
import json
import logging
import random
import re
import hashlib
from typing import Dict, List, Optional, Tuple, Set
from datetime import datetime, timedelta
import requests
import redis
from fake_useragent import UserAgent
from bs4 import BeautifulSoup
import jieba
import jieba.analyse
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
from PIL import Image
import io
import imagehash
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class MultiPlatformPriceComparer:
def __init__(self, redis_host: str = 'localhost', redis_port: int = 6379,
proxy_pool: List[str] = None, rule_db_path: str = "price_rules.json"):
"""
多平台智能比价系统
:param redis_host: Redis主机地址
:param redis_port: Redis端口
:param proxy_pool: 代理IP池
:param rule_db_path: 解析规则库路径
"""
# 初始化Redis连接（不同数据库存储不同类型数据）
self.redis = redis.Redis(host=redis_host, port=redis_port, db=24)
self.redis_features = redis.Redis(host=redis_host, port=redis_port, db=25) # 特征向量
self.redis_history = redis.Redis(host=redis_host, port=redis_port, db=26) # 价格历史
# 加载平台解析规则库
self.platform_rules = self._load_platform_rules(rule_db_path)
# 支持的平台列表
self.supported_platforms = list(self.platform_rules.keys())
# 初始化会话池
self.sessions = {platform: self._init_session(platform)
for platform in self.supported_platforms}
# 代理池配置
self.proxy_pool = proxy_pool or []
# 用户代理生成器
self.ua = UserAgent()
# 特征提取器
self.tfidf = TfidfVectorizer(stop_words='english')
self._init_tfidf()
# 价值评估权重配置
self.value_weights = {
"price": 0.35, # 价格权重
"service": 0.2, # 服务权重
"reputation": 0.15, # 商家信誉权重
"historical": 0.15, # 历史价格权重
"platform_feature": 0.15 # 平台特性权重
}
# 反爬配置（平台差异化）
self.anti_crawl = {
"default": {
"request_delay": (1.5, 3.0),
"header_rotation": True,
"session_reset_interval": 20
},
"taobao": {
"request_delay": (2.5, 4.0),
"header_rotation": True,
"session_reset_interval": 15
},
"jd": {
"request_delay": (2.0, 3.5),
"header_rotation": True,
"session_reset_interval": 18
}
}
# 请求计数器（按平台）
self.request_counts = {platform: 0 for platform in self.supported_platforms}
# 初始化结巴分词，添加电商领域专有词汇
self._init_jieba()
def _init_session(self, platform: str) -> requests.Session:
"""初始化指定平台的请求会话"""
session = requests.Session()
# 从规则库获取平台特定 headers
base_headers = self.platform_rules.get(platform, {}).get("base_headers", {})
default_headers = {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1"
}
# 合并 headers，平台特定 headers 优先
default_headers.update(base_headers)
session.headers.update(default_headers)
return session
def _init_jieba(self) -> None:
"""初始化结巴分词，添加电商领域专有词汇"""
jieba.initialize()
# 添加电商领域专业术语
ecommerce_terms = [
"秒杀", "拼团", "满减", "优惠券", "会员价", "旗舰店",
"专营店", "次日达", "预售", "现货", "保税仓", "正品保障",
"假一赔十", "7天无理由", "拆封可退", "产地直采"
]
# 添加常见规格术语
spec_terms = [
"256G", "512G", "1TB", "4K", "HDR", "Pro", "Max", "Ultra",
"标准版", "旗舰版", "青春版", "定制版", "限量版"
]
for term in ecommerce_terms + spec_terms:
jieba.add_word(term)
def _init_tfidf(self) -> None:
"""初始化TF-IDF模型（加载预训练数据）"""
# 实际应用中应使用大量商品标题训练
sample_titles = [
"iPhone 15 256G 星光色 移动联通电信5G手机",
"苹果15 256GB 星光色 5G全网通",
"Apple iPhone 15 (A2892) 256GB 星光色 支持移动联通电信5G",
"华为Mate 60 Pro 12GB+512GB 雅川青 超可靠玄武架构 卫星通话",
"HUAWEI Mate 60 Pro 12+512G 雅川青 5G手机"
]
self.tfidf.fit(sample_titles)
def _load_platform_rules(self, path: str) -> Dict:
"""加载平台解析规则库"""
try:
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
except FileNotFoundError:
logger.warning(f"规则库文件 {path} 未找到，使用默认规则")
return self._get_default_rules()
def _get_default_rules(self) -> Dict:
"""返回默认的平台解析规则"""
return {
"taobao": {
"search_url": "https://s.taobao.com/search",
"search_params": {"q": "{keyword}", "s": "{page}*44"},
"item_selector": "#mainsrp-itemlist .items .item",
"fields": {
"title": ".title a::text",
"price": ".price strong::text",
"original_price": ".price .original::text",
"sales": ".deal-cnt::text",
"shop": ".shop a::text",
"location": ".location::text",
"image": ".pic img::src",
"promotion": ".promo-words::text"
},
"promotion_rules": [
{"type": "满减", "pattern": r'满(\d+)减(\d+)'},
{"type": "优惠券", "pattern": r'(\d+)元券'}
],
"base_headers": {
"Host": "s.taobao.com",
"Referer": "https://www.taobao.com/"
}
},
"jd": {
"search_url": "https://search.jd.com/Search",
"search_params": {"keyword": "{keyword}", "page": "{page}", "s": "{start}", "click": 0},
"item_selector": "#J_goodsList .gl-item",
"fields": {
"title": ".p-name em::text",
"price": ".p-price i::text",
"original_price": ".p-price .p-price-old i::text",
"sales": ".p-commit a::text",
"shop": ".p-shop a::text",
"location": ".p-origin span::text",
"image": ".p-img img::src",
"promotion": ".p-promotion em::text"
},
"promotion_rules": [
{"type": "满减", "pattern": r'满(\d+)减(\d+)'},
{"type": "折扣", "pattern": r'(\d+)折'}
],
"base_headers": {
"Host": "search.jd.com",
"Referer": "https://www.jd.com/"
}
},
"pdd": {
"search_url": "https://search.pinduoduo.com/search",
"search_params": {"q": "{keyword}", "page": "{page}"},
"item_selector": ".goods-list .goods",
"fields": {
"title": ".goods-title::text",
"price": ".goods-price span::text",
"original_price": ".goods-price del::text",
"sales": ".goods-sales::text",
"shop": ".goods-shop::text",
"image": ".goods-img img::src",
"promotion": ".goods-tag::text"
},
"promotion_rules": [
{"type": "拼团", "pattern": r'拼团价'},
{"type": "限时", "pattern": r'限时(\d+)元'}
],
"base_headers": {
"Host": "search.pinduoduo.com",
"Referer": "https://www.pinduoduo.com/"
}
}
}
def _save_platform_rules(self, path: str = "price_rules.json") -> None:
"""保存平台解析规则库（用于规则更新）"""
with open(path, 'w', encoding='utf-8') as f:
json.dump(self.platform_rules, f, ensure_ascii=False, indent=2)
def update_platform_rule(self, platform: str, rule_path: str, new_rule: Dict) -> bool:
"""
更新平台解析规则
:param platform: 平台名称
:param rule_path: 规则路径，如"fields.title"
:param new_rule: 新规则值
:return: 是否更新成功
"""
if platform not in self.platform_rules:
logger.error(f"平台 {platform} 不存在")
return False
# 解析规则路径
parts = rule_path.split('.')
current = self.platform_rules[platform]
try:
for i, part in enumerate(parts[:-1]):
if part not in current:
current[part] = {}
current = current[part]
current[parts[-1]] = new_rule
self._save_platform_rules()
logger.info(f"平台 {platform} 规则 {rule_path} 更新成功")
return True
except Exception as e:
logger.error(f"更新规则失败: {str(e)}")
return False
def _rotate_headers(self, platform: str) -> None:
"""轮换指定平台的请求头"""
session = self.sessions[platform]
session.headers["User-Agent"] = self.ua.random
# 添加平台特定头信息
if platform == "taobao":
session.headers["X-Requested-With"] = "XMLHttpRequest"
session.headers["X-Tao-Long"] = str(random.randint(1000, 5000))
elif platform == "jd":
session.headers["X-Real-IP"] = self._generate_random_ip()
session.headers["X-Forwarded-For"] = self._generate_random_ip()
# 平台特定Referer
referers = self.platform_rules.get(platform, {}).get("referers", [
f"https://www.{platform}.com/"
])
if referers:
session.headers["Referer"] = random.choice(referers)
def _generate_random_ip(self) -> str:
"""生成随机IP地址"""
return f"111.{random.randint(1, 254)}.{random.randint(1, 254)}.{random.randint(1, 254)}"
def _get_proxy(self) -> Optional[Dict]:
"""获取随机代理"""
if self.proxy_pool and len(self.proxy_pool) > 0:
proxy = random.choice(self.proxy_pool)
return {"http": proxy, "https": proxy}
return None
def _reset_session(self, platform: str) -> None:
"""重置指定平台的会话"""
self.sessions[platform] = self._init_session(platform)
self._rotate_headers(platform)
logger.info(f"已重置 {platform} 平台会话")
def _get_anti_crawl_config(self, platform: str) -> Dict:
"""获取平台的反爬配置"""
return self.anti_crawl.get(platform, self.anti_crawl["default"])
def _anti_crawl_measures(self, platform: str) -> None:
"""执行指定平台的反爬措施"""
config = self._get_anti_crawl_config(platform)
# 随机延迟
delay = random.uniform(*config["request_delay"])
time.sleep(delay)
# 轮换请求头
if config["header_rotation"]:
self._rotate_headers(platform)
# 定期重置会话
self.request_counts[platform] += 1
if self.request_counts[platform] % config["session_reset_interval"] == 0:
self._reset_session(platform)
def _parse_promotions(self, platform: str, text: str) -> List[Dict]:
"""
解析促销信息
:param platform: 平台名称
:param text: 包含促销信息的文本
:return: 结构化的促销信息
"""
promotions = []
if not text:
return promotions
# 获取平台特定的促销规则
promotion_rules = self.platform_rules.get(platform, {}).get("promotion_rules", [])
for rule in promotion_rules:
pattern = re.compile(rule["pattern"])
matches = pattern.findall(text)
for match in matches:
promotion = {"type": rule["type"]}
if rule["type"] == "满减":
promotion["threshold"] = int(match[0])
promotion["reduction"] = int(match[1])
promotion["description"] = f"满{match[0]}减{match[1]}"
elif rule["type"] == "优惠券":
promotion["amount"] = int(match[0])
promotion["description"] = f"{match[0]}元券"
elif rule["type"] == "折扣":
promotion["discount"] = int(match[0]) / 10
promotion["description"] = f"{match[0]}折"
elif rule["type"] == "拼团":
promotion["description"] = "拼团价"
elif rule["type"] == "限时":
promotion["price"] = int(match[0])
promotion["description"] = f"限时{match[0]}元"
promotions.append(promotion)
return promotions
def _calculate_actual_price(self, price: float, promotions: List[Dict], quantity: int = 1) -> Dict:
"""
计算实际支付价格
:param price: 商品标价
:param promotions: 促销信息列表
:param quantity: 购买数量
:return: 价格计算结果
"""
total_price = price * quantity
actual_price = total_price
applied_promotions = []
# 应用满减
full_reductions = [p for p in promotions if p["type"] == "满减"]
if full_reductions:
# 按优惠力度排序
full_reductions.sort(key=lambda x: x["reduction"] / x["threshold"], reverse=True)
for fr in full_reductions:
if total_price >= fr["threshold"]:
times = int(total_price / fr["threshold"])
reduction = fr["reduction"] * times
actual_price -= reduction
applied_promotions.append(f"{fr['description']}({times}次)")
break # 只应用最优惠的一个满减
# 应用折扣
discounts = [p for p in promotions if p["type"] == "折扣"]
if discounts:
# 取最大折扣
best_discount = max(discounts, key=lambda x: x["discount"])
actual_price *= best_discount["discount"]
applied_promotions.append(best_discount["description"])
# 应用优惠券（假设只能用一张）
coupons = [p for p in promotions if p["type"] == "优惠券"]
if coupons and actual_price > 0:
# 取面额最大的优惠券
best_coupon = max(coupons, key=lambda x: x["amount"])
if actual_price >= best_coupon["amount"]:
actual_price -= best_coupon["amount"]
applied_promotions.append(best_coupon["description"])
# 应用限时价
limited_time = [p for p in promotions if p["type"] == "限时"]
if limited_time:
lt_price = min([p["price"] for p in limited_time]) * quantity
if lt_price < actual_price:
actual_price = lt_price
applied_promotions.append("限时价")
# 应用拼团价（假设拼团价低于当前价）
if any(p["type"] == "拼团" for p in promotions) and actual_price > 0:
group_price = actual_price * 0.9 # 假设拼团价为9折
actual_price = group_price
applied_promotions.append("拼团价")
return {
"original_total": round(total_price, 2),
"actual_total": round(actual_price, 2),
"saved": round(total_price - actual_price, 2),
"applied_promotions": applied_promotions
}
def _extract_product_features(self, title: str, specs: str = "") -> Dict:
"""
提取商品特征
:param title: 商品标题
:param specs: 规格参数
:return: 结构化的商品特征
"""
features = {
"brand": "", # 品牌
"model": "", # 型号
"capacity": "", # 容量(如256G)
"color": "", # 颜色
"version": "", # 版本(如Pro/Max)
"keywords": [], # 关键词
"feature_vector": [] # 特征向量
}
full_text = title + " " + specs
# 提取品牌（简单示例，实际需更复杂的品牌库匹配）
brands = ["苹果", "华为", "小米", "OPPO", "vivo", "三星", "联想", "戴尔"]
for brand in brands:
if brand in full_text:
features["brand"] = brand
break
# 提取容量
capacity_match = re.search(r'(\d+)(G|GB|TB)', full_text)
if capacity_match:
features["capacity"] = capacity_match.group(0)
# 提取颜色
colors = ["黑色", "白色", "红色", "蓝色", "绿色", "金色", "银色", "灰色", "星光色", "暗夜紫"]
for color in colors:
if color in full_text:
features["color"] = color
break
# 提取版本
versions = ["Pro", "Max", "Ultra", "Plus", "SE", "青春版", "标准版", "旗舰版"]
for version in versions:
if version in full_text:
features["version"] = version
break
# 提取关键词
features["keywords"] = jieba.analyse.extract_tags(full_text, topK=10)
# 生成特征向量
try:
vector = self.tfidf.transform([full_text]).toarray()[0]
features["feature_vector"] = vector.tolist()
except Exception as e:
logger.warning(f"生成特征向量失败: {str(e)}")
return features
def _calculate_similarity(self, vec1: List[float], vec2: List[float]) -> float:
"""
计算两个商品特征向量的相似度
:param vec1: 特征向量1
:param vec2: 特征向量2
:return: 相似度分数(0-1)
"""
if not vec1 or not vec2:
return 0.0
# 确保向量长度一致
min_len = min(len(vec1), len(vec2))
vec1 = vec1[:min_len]
vec2 = vec2[:min_len]
# 转换为numpy数组
np_vec1 = np.array(vec1).reshape(1, -1)
np_vec2 = np.array(vec2).reshape(1, -1)
# 计算余弦相似度
return cosine_similarity(np_vec1, np_vec2)[0][0]
def _image_similarity(self, url1: str, url2: str) -> float:
"""
计算两个商品图片的相似度
:param url1: 图片URL1
:param url2: 图片URL2
:return: 相似度分数(0-1)
"""
try:
# 获取图片哈希
hash1 = self._get_image_hash(url1)
hash2 = self._get_image_hash(url2)
if not hash1 or not hash2:
return 0.0
# 计算哈希距离
distance = hash1 - hash2
# 转换为相似度(0-1)，距离越小相似度越高
return 1 - min(distance / 64, 1.0)
except Exception as e:
logger.warning(f"计算图片相似度失败: {str(e)}")
return 0.0
def _get_image_hash(self, url: str) -> Optional[imagehash.ImageHash]:
"""获取图片的感知哈希"""
cache_key = f"img_hash:{hashlib.md5(url.encode()).hexdigest()}"
cached_hash = self.redis.get(cache_key)
if cached_hash:
return imagehash.hex_to_hash(cached_hash.decode())
try:
response = requests.get(url, timeout=10, stream=True)
if response.status_code == 200:
img = Image.open(io.BytesIO(response.content))
# 计算感知哈希
phash = imagehash.phash(img)
# 缓存哈希值(7天)
self.redis.setex(cache_key, timedelta(days=7), phash.hash.tobytes())
return phash
except Exception as e:
logger.warning(f"获取图片 {url} 失败: {str(e)}")
return None
def _merge_duplicate_products(self, products: List[Dict], threshold: float = 0.7) -> List[Dict]:
"""
合并重复或高度相似的商品
:param products: 商品列表
:param threshold: 相似度阈值
:return: 去重后的商品列表
"""
if not products:
return []
unique_products = []
# 按相似度分组
for product in products:
matched = False
for i, unique in enumerate(unique_products):
# 计算文本相似度
text_sim = self._calculate_similarity(
product["features"]["feature_vector"],
unique["features"]["feature_vector"]
)
# 计算图片相似度
img_sim = self._image_similarity(
product["image"],
unique["image"]
)
# 综合相似度（文本权重更高）
combined_sim = text_sim * 0.7 + img_sim * 0.3
if combined_sim >= threshold:
# 合并商品信息，保留价格最低的
if product["price_info"]["actual_total"] < unique["price_info"]["actual_total"]:
unique_products[i] = product
matched = True
break
if not matched:
unique_products.append(product)
return unique_products
def _evaluate_service_level(self, platform: str, product: Dict) -> Dict:
"""
评估服务水平
:param platform: 平台名称
:param product: 商品信息
:return: 服务评估结果
"""
service = {
"level": 0.0, # 0-10分
"delivery": "",
"timeliness": "",
"after_sale": "",
"shipping_fee": 0.0
}
# 平台基础服务分
platform_base = {
"jd": 8.5,
"taobao": 7.5,
"pdd": 6.5
}.get(platform, 7.0)
service["level"] = platform_base
# 配送方式评估
title = product.get("title", "")
promotion = product.get("promotion", "")
full_text = title + " " + promotion
if "次日达" in full_text or "当日达" in full_text:
service["delivery"] = "次日达/当日达"
service["timeliness"] = "高"
service["level"] += 1.0
elif "京东物流" in full_text:
service["delivery"] = "京东物流"
service["timeliness"] = "中高"
service["level"] += 0.5
elif "顺丰" in full_text:
service["delivery"] = "顺丰快递"
service["timeliness"] = "中高"
service["level"] += 0.3
# 售后政策评估
if "7天无理由" in full_text:
service["after_sale"] = "7天无理由退货"
service["level"] += 0.5
if "拆封可退" in full_text:
service["after_sale"] += "，拆封可退" if service["after_sale"] else "拆封可退"
service["level"] += 0.5
if "假一赔十" in full_text:
service["after_sale"] += "，假一赔十" if service["after_sale"] else "假一赔十"
service["level"] += 1.0
# 运费评估（假设）
price = product.get("price_info", {}).get("actual_total", 0)
if price >= 99:
service["shipping_fee"] = 0.0
else:
service["shipping_fee"] = 10.0 # 假设运费10元
service["level"] -= 0.5
# 限制最高分
service["level"] = min(10.0, round(service["level"], 1))
return service
def _evaluate_seller_reputation(self, platform: str, product: Dict) -> Dict:
"""
评估商家信誉
:param platform: 平台名称
:param product: 商品信息
:return: 商家信誉评估结果
"""
reputation = {
"score": 0.0, # 0-10分
"rating": 0.0, # 好评率
"sales": 0, # 销量
"complaint_rate": 0.0, # 投诉率
"years": 0 # 经营年限
}
# 提取销量信息
sales_text = product.get("sales", "")
if sales_text:
# 处理"100+" "5000+"等格式
sales_match = re.search(r'(\d+)', sales_text)
if sales_match:
reputation["sales"] = int(sales_match.group(1))
# 提取好评率（模拟，实际应从页面提取）
reputation["rating"] = round(random.uniform(95.0, 99.9), 1)
# 模拟经营年限
reputation["years"] = random.randint(1, 10)
# 计算信誉分
# 销量分（0-3分）
sales_score = min(reputation["sales"] / 10000 * 3, 3)
# 好评率分（0-3分）
rating_score = (reputation["rating"] - 90) / 10 * 3
# 经营年限分（0-2分）
years_score = min(reputation["years"] / 10 * 2, 2)
# 平台基础分（0-2分）
platform_score = {
"jd": 2.0,
"taobao": 1.5,
"pdd": 1.0
}.get(platform, 1.5)
reputation["score"] = round(sales_score + rating_score + years_score + platform_score, 1)
return reputation
def _get_historical_price(self, product_id: str, platform: str) -> Dict:
"""
获取商品历史价格
:param product_id: 商品ID
:param platform: 平台名称
:return: 历史价格信息
"""
cache_key = f"price_history:{platform}:{product_id}"
cached_data = self.redis_history.get(cache_key)
if cached_data:
return json.loads(cached_data.decode())
# 模拟历史价格数据
days = 30
base_price = random.uniform(500, 5000)
prices = []
for i in range(days):
# 模拟价格波动
fluctuation = random.uniform(-0.05, 0.05) # 5%以内波动
price = base_price * (1 + fluctuation)
prices.append({
"date": (datetime.now() - timedelta(days=days-i)).strftime("%Y-%m-%d"),
"price": round(price, 2)
})
# 找出最低价
min_price = min(prices, key=lambda x: x["price"])
result = {
"prices": prices,
"min_price": min_price["price"],
"min_date": min_price["date"],
"fluctuation": round((prices[-1]["price"] - prices[0]["price"]) / prices[0]["price"] * 100, 2)
}
# 缓存历史价格（24小时）
self.redis_history.setex(cache_key, timedelta(hours=24), json.dumps(result, ensure_ascii=False))
return result
def _evaluate_platform_feature(self, platform: str, product: Dict) -> Dict:
"""
评估平台特性价值
:param platform: 平台名称
:param product: 商品信息
:return: 平台特性评估结果
"""
features = {
"score": 0.0, # 0-10分
"features": [], # 平台特性列表
"trust_level": "" # 信任等级
}
title = product.get("title", "")
full_text = title + " " + product.get("promotion", "")
# 平台基础分
base_score = {
"jd": 7.0,
"taobao": 6.5,
"pdd": 6.0
}.get(platform, 6.5)
features["score"] = base_score
# 正品保障
if "正品" in full_text or "官方" in full_text or "旗舰店" in full_text:
features["features"].append("正品保障")
features["score"] += 1.0
# 独家销售
if "独家" in full_text:
features["features"].append("独家销售")
features["score"] += 0.5
# 平台专属服务
if platform == "jd" and "自营" in full_text:
features["features"].append("京东自营")
features["score"] += 1.0
elif platform == "taobao" and "天猫" in full_text:
features["features"].append("天猫保障")
features["score"] += 0.8
# 产地直采/海外直邮
if "产地直采" in full_text or "海外直邮" in full_text:
features["features"].append("产地直采/海外直邮")
features["score"] += 0.7
# 限制最高分
features["score"] = min(10.0, round(features["score"], 1))
# 信任等级
if features["score"] >= 9:
features["trust_level"] = "极高"
elif features["score"] >= 7:
features["trust_level"] = "较高"
elif features["score"] >= 5:
features["trust_level"] = "一般"
else:
features["trust_level"] = "较低"
return features
def _calculate_comprehensive_value(self, product: Dict) -> Dict:
"""
计算商品综合价值
:param product: 商品信息
:return: 综合价值评估结果
"""
# 标准化各项指标（0-10分）
# 价格值：价格越低得分越高（基于同类商品价格范围）
price = product["price_info"]["actual_total"]
# 假设同类商品价格范围为100-10000元
price_score = 10 - min(max(price / 10000 * 10, 0), 10)
# 其他指标已经是0-10分
service_score = product["service"]["level"]
reputation_score = product["seller_reputation"]["score"]
historical_score = self._calculate_historical_score(product)
platform_score = product["platform_features"]["score"]
# 计算加权总分
total_score = round(
price_score * self.value_weights["price"] +
service_score * self.value_weights["service"] +
reputation_score * self.value_weights["reputation"] +
historical_score * self.value_weights["historical"] +
platform_score * self.value_weights["platform_feature"],
1
)
# 价值等级
value_level = "一般"
if total_score >= 9:
value_level = "极佳"
elif total_score >= 8:
value_level = "优秀"
elif total_score >= 7:
value_level = "良好"
elif total_score >= 6:
value_level = "尚可"
return {
"total_score": total_score,
"value_level": value_level,
"score_breakdown": {
"price": round(price_score, 1),
"service": service_score,
"reputation": reputation_score,
"historical": round(historical_score, 1),
"platform": platform_score
},
"best_value": total_score >= 8.5 # 是否为最佳价值选择
}
def _calculate_historical_score(self, product: Dict) -> float:
"""计算历史价格得分（0-10分）"""
historical = product.get("historical_price", {})
current_price = product["price_info"]["actual_total"]
min_price = historical.get("min_price", current_price)
# 当前价格越接近历史最低价，得分越高
if min_price == 0:
return 5.0
ratio = min(current_price / min_price, 2.0) # 最高允许比最低价高1倍
return round(10 - (ratio - 1) * 10, 1)
def search_and_compare(self, keyword: str, platforms: List[str] = None,
page: int = 1, page_size: int = 20,
sort_by: str = "comprehensive_value") -> Dict:
"""
多平台搜索并比价
:param keyword: 搜索关键词
:param platforms: 要搜索的平台列表，None表示所有支持的平台
:param page: 页码
:param page_size: 每页数量
:param sort_by: 排序方式（comprehensive_value/price_asc/price_desc/service）
:return: 比价结果
"""
start_time = time.time()
result = {
"keyword": keyword,
"platforms": platforms or self.supported_platforms,
"page": page,
"page_size": page_size,
"total_products": 0,
"products": [],
"platform_stats": {}, # 各平台商品数量统计
"price_range": {}, # 价格范围统计
"response_time_ms": 0,
"status": "success"
}
# 生成缓存键
cache_key = f"price_compare:{keyword}:{'-'.join(platforms) if platforms else 'all'}:{page}:{page_size}:{sort_by}"
cache_key = hashlib.md5(cache_key.encode()).hexdigest()
# 尝试从缓存获取
cached_data = self.redis.get(cache_key)
if cached_data:
try:
cached_result = json.loads(cached_data.decode())
cached_result["from_cache"] = True
return cached_result
except Exception as e:
logger.warning(f"缓存解析失败: {str(e)}")
try:
all_products = []
platforms_to_search = platforms or self.supported_platforms
# 1. 多平台并行搜索（这里简化为串行）
for platform in platforms_to_search:
if platform not in self.platform_rules:
logger.warning(f"不支持的平台: {platform}")
continue
logger.info(f"开始搜索平台 {platform}，关键词: {keyword}")
platform_products = self._search_platform(
platform=platform,
keyword=keyword,
page=page,
page_size=page_size
)
if platform_products["status"] == "error":
logger.error(f"平台 {platform} 搜索失败: {platform_products['error']}")
continue
all_products.extend(platform_products["products"])
# 2. 合并重复商品
unique_products = self._merge_duplicate_products(all_products)
# 3. 计算综合价值
for product in unique_products:
# 评估服务水平
product["service"] = self._evaluate_service_level(
product["platform"], product
)
# 评估商家信誉
product["seller_reputation"] = self._evaluate_seller_reputation(
product["platform"], product
)
# 获取历史价格
product["historical_price"] = self._get_historical_price(
product["product_id"], product["platform"]
)
# 评估平台特性
product["platform_features"] = self._evaluate_platform_feature(
product["platform"], product
)
# 计算综合价值
product["comprehensive_value"] = self._calculate_comprehensive_value(product)
# 4. 排序处理
if sort_by == "comprehensive_value":
unique_products.sort(key=lambda x: x["comprehensive_value"]["total_score"], reverse=True)
elif sort_by == "price_asc":
unique_products.sort(key=lambda x: x["price_info"]["actual_total"])
elif sort_by == "price_desc":
unique_products.sort(key=lambda x: x["price_info"]["actual_total"], reverse=True)
elif sort_by == "service":
unique_products.sort(key=lambda x: x["service"]["level"], reverse=True)
# 5. 生成统计数据
# 平台分布统计
platform_stats = {}
for product in unique_products:
platform = product["platform"]
platform_stats[platform] = platform_stats.get(platform, 0) + 1
result["platform_stats"] = platform_stats
# 价格范围统计
if unique_products:
prices = [p["price_info"]["actual_total"] for p in unique_products]
result["price_range"] = {
"min": min(prices),
"max": max(prices),
"average": round(sum(prices) / len(prices), 2)
}
# 6. 结果整理
result["total_products"] = len(unique_products)
result["products"] = unique_products[:page_size] # 分页
# 7. 缓存结果（价格信息变化较快，缓存时间较短）
self.redis.setex(
cache_key,
timedelta(minutes=30),
json.dumps(result, ensure_ascii=False)
)
except Exception as e:
result["status"] = "error"
result["error"] = f"比价处理失败: {str(e)}"
# 计算响应时间
result["response_time_ms"] = int((time.time() - start_time) * 1000)
return result
def _search_platform(self, platform: str, keyword: str, page: int = 1, page_size: int = 20) -> Dict:
"""
搜索单个平台
:param platform: 平台名称
:param keyword: 搜索关键词
:param page: 页码
:param page_size: 每页数量
:return: 平台搜索结果
"""
result = {
"platform": platform,
"keyword": keyword,
"page": page,
"products": [],
"status": "success"
}
try:
# 1. 获取平台搜索规则
rules = self.platform_rules[platform]
search_url = rules["search_url"]
# 2. 构建搜索参数
params = rules["search_params"].copy()
# 替换参数占位符
for key, value in params.items():
if "{keyword}" in value:
params[key] = value.replace("{keyword}", keyword)
if "{page}" in value:
params[key] = value.replace("{page}", str(page))
if "{start}" in value:
# 有些平台用start参数（如京东）
params[key] = value.replace("{start}", str((page-1)*page_size + 1))
# 3. 添加平台特定参数
params["t"] = int(time.time() * 1000) # 时间戳防缓存
# 4. 执行搜索请求
self._anti_crawl_measures(platform)
response = self.sessions[platform].get(
search_url,
params=params,
proxies=self._get_proxy(),
timeout=20
)
if response.status_code != 200:
result["status"] = "error"
result["error"] = f"请求失败，状态码: {response.status_code}"
return result
# 5. 解析搜索结果（实际应根据真实页面结构解析）
# 这里使用模拟数据演示
html = response.text
products = self._parse_search_results(platform, html, rules)
result["products"] = products
except Exception as e:
result["status"] = "error"
result["error"] = f"平台搜索失败: {str(e)}"
return result
def _parse_search_results(self, platform: str, html: str, rules: Dict) -> List[Dict]:
"""
解析平台搜索结果
:param platform: 平台名称
:param html: 页面HTML
:param rules: 解析规则
:return: 结构化的商品列表
"""
# 实际应用中应使用BeautifulSoup解析
# soup = BeautifulSoup(html, 'html.parser')
# items = soup.select(rules["item_selector"])
# 这里使用模拟数据
products = []
item_count = min(random.randint(10, 20), 20) # 模拟10-20个商品
for i in range(item_count):
# 生成模拟商品ID
product_id = f"{platform}_{random.randint(1000000, 9999999)}"
# 随机生成价格
price = round(random.uniform(100, 5000), 2)
# 随机生成促销信息
promotions = []
promo_text = ""
if random.random() > 0.3:
promo_type = random.choice(["满减", "优惠券", "折扣", "拼团"])
if promo_type == "满减":
threshold = random.choice([100, 200, 300, 500])
reduction = random.choice([10, 30, 50, 100, 150])
promo_text = f"满{threshold}减{reduction}"
elif promo_type == "优惠券":
amount = random.choice([10, 20, 30, 50])
promo_text = f"{amount}元券"
elif promo_type == "折扣":
discount = random.choice([8, 8.5, 9, 9.5])
promo_text = f"{discount}折"
elif promo_type == "拼团":
promo_text = "拼团价"
# 生成商品标题
title_templates = [
"{keyword} {model} {capacity} {color}",
"{keyword} {version} 官方正品 {promo}",
"{brand} {keyword} {spec} 现货速发"
]
brands = ["苹果", "华为", "小米", "OPPO", "vivo"]
models = ["2023款", "Pro", "Max", "Ultra"]
capacities = ["128G", "256G", "512G", "1TB"]
colors = ["黑色", "白色", "银色", "蓝色", "红色"]
title = random.choice(title_templates).format(
keyword="手机", # 实际应替换为搜索关键词
model=random.choice(models),
capacity=random.choice(capacities),
color=random.choice(colors),
version=random.choice(["标准版", "旗舰版", "青春版"]),
promo=promo_text[:5],
brand=random.choice(brands),
spec=random.choice(["全网通", "5G", "双卡双待"])
)
# 解析促销信息
parsed_promotions = self._parse_promotions(platform, promo_text)
# 计算实际价格
price_info = self._calculate_actual_price(price, parsed_promotions)
# 提取商品特征
features = self._extract_product_features(title)
products.append({
"product_id": product_id,
"platform": platform,
"title": title,
"image": f"https://img.{platform}.com/{product_id}.jpg",
"price": price,
"original_price": round(price * random.uniform(1.1, 1.5), 2),
"sales": f"{random.randint(10, 9999)}+",
"shop": f"{random.choice(['官方旗舰店', '专营店', '专卖店'])}",
"location": random.choice(["北京", "上海", "广州", "深圳", "杭州"]),
"promotion": promo_text,
"promotions": parsed_promotions,
"price_info": price_info,
"features": features,
"url": f"https://{platform}.com/item/{product_id}.html"
})
return products
# 使用示例
if __name__ == "__main__":
# 初始化多平台比价系统
proxy_pool = [
# "http://127.0.0.1:7890", # 替换为实际代理
]
comparer = MultiPlatformPriceComparer(
redis_host="localhost",
redis_port=6379,
proxy_pool=proxy_pool
)
try:
# 示例：搜索"iPhone 15 256G"并比价
keyword = "iPhone 15 256G"
print(f"===== 多平台比价: {keyword} =====")
# 选择要搜索的平台
platforms = ["taobao", "jd", "pdd"]
result = comparer.search_and_compare(
keyword=keyword,
platforms=platforms,
page=1,
page_size=10,
sort_by="comprehensive_value"
)
if result["status"] == "error":
print(f"比价失败: {result['error']}")
else:
print(f"共找到 {result['total_products']} 个有效商品")
print(f"平台分布: {result['platform_stats']}")
print(f"价格范围: ¥{result['price_range']['min']} - ¥{result['price_range']['max']} (平均 ¥{result['price_range']['average']})")
print("\n综合价值最高的3个商品:")
for i, product in enumerate(result["products"][:3]):
print(f"\n{i+1}. {product['title']}")
print(f" 平台: {product['platform']} | 商家: {product['shop']}")
print(f" 标价: ¥{product['price']} | 实际支付: ¥{product['price_info']['actual_total']} (节省 ¥{product['price_info']['saved']})")
if product['price_info']['applied_promotions']:
print(f" 促销: {', '.join(product['price_info']['applied_promotions'])}")
print(f" 服务评分: {product['service']['level']} | 配送: {product['service']['delivery'] or '普通快递'}")
print(f" 历史价格: 最低 ¥{product['historical_price']['min_price']} ({product['historical_price']['min_date']})")
print(f" 综合评分: {product['comprehensive_value']['total_score']} ({product['comprehensive_value']['value_level']})")
print(f" 主要优势: {', '.join(product['comprehensive_value']['score_breakdown'].items() | map(lambda x: f'{x[0]}:{x[1]}'))}")
# 标记最佳价值选择
if product['comprehensive_value']['best_value']:
print(" ⭐ 最佳价值选择")
# 示例：更新平台解析规则（演示功能）
# success = comparer.update_platform_rule(
# platform="jd",
# rule_path="fields.title",
# new_rule=".p-name em::text, .p-name span::text"
# )
# if success:
# print("\n规则更新成功")
except Exception as e:
print(f"执行出错: {str(e)}")
四、核心技术模块解析
1. 动态规则引擎
突破传统硬编码解析方式，实现平台适配的灵活配置：
可视化规则配置：通过 JSON 规则库定义各平台的搜索 URL、参数格式、页面选择器及字段映射关系，支持无需代码修改即可适配平台变化
规则动态更新：提供update_platform_rule方法，可实时更新特定平台的解析规则（如字段选择器、促销规则），适应平台页面结构调整
平台差异化处理：为每个平台维护独立的反爬策略、请求头配置和解析逻辑，解决 "一套代码适配所有平台" 的兼容性难题
版本化规则管理：支持规则库的保存与加载，可回溯历史规则版本，便于测试与问题排查
代码中_load_platform_rules和update_platform_rule方法实现这一核心逻辑，解决多平台比价中 "平台变化快、适配成本高" 的痛点。
2. 商品智能匹配系统
超越简单文本比对，实现跨平台商品的精准匹配：
特征向量生成：基于 TF-IDF 算法将商品标题转换为数学向量，捕捉 "iPhone 15" 与 "苹果 15" 的语义关联
多维度相似度计算：融合文本相似度（70% 权重）与图片感知哈希相似度（30% 权重），解决 "同品不同名" 的匹配难题
重复商品合并：自动识别高度相似商品，保留性价比最高的条目，避免用户面对重复信息
特征提取体系：专门提取品牌、型号、容量、颜色等核心属性，为精准匹配提供结构化依据
代码中_extract_product_features和_merge_duplicate_products方法实现这一逻辑，解决多平台比价中 "商品匹配不准确" 的核心问题。
3. 全链路价格计算模型
突破表面价格对比，实现真实支付成本的精准核算：
促销规则解析引擎：针对不同平台的满减、优惠券、折扣、拼团等促销类型，构建专属解析规则，如淘宝的 "满 300 减 50" 与拼多多的 "拼团价"
实际支付计算：模拟用户购买场景，自动计算多层优惠叠加后的实际支付金额，包括满减次数计算、最佳优惠券选择等
历史价格分析：追踪近 30 天价格波动，识别当前价格是否处于历史低位，避免 "虚假促销" 陷阱
隐性成本核算：自动计算运费、税费等隐性成本，提供完整的 "到手价" 对比
代码中_parse_promotions和_calculate_actual_price方法实现这一逻辑，解决比价工具中 "只比标价，不看实付" 的行业痛点。
4. 综合价值评估体系
超越单一价格维度，构建多因素的价值评估模型：
五维评分体系：从价格（35%）、服务（20%）、商家信誉（15%）、历史价格（15%）、平台特性（15%）五个维度进行量化评分
服务水平评估：基于配送时效、售后政策、运费等指标，评估服务质量并转换为 0-10 分的量化分数
商家信誉分析：融合销量、好评率、经营年限等数据，构建商家可信度评分模型
平台特性加权：针对正品保障、独家销售、自营服务等平台特有优势进行价值加分
代码中_calculate_comprehensive_value方法实现这一逻辑，解决 "只看价格，忽略服务与信誉" 的片面比价问题。
五、与传统比价方案的差异对比
特性 传统比价方案 本方案（智能比价系统）
平台适配 固定平台，硬编码解析 动态规则引擎，支持可视化配置
商品匹配 基于标题精确匹配 特征向量 + 图片哈希的模糊匹配
价格计算 仅对比标价 全链路促销计算，含实际支付金额
价值维度 单一价格维度 价格 / 服务 / 信誉 / 历史 / 平台五维评估
反爬应对 固定策略，易失效 平台差异化动态反爬机制
扩展性 新增平台需大量代码修改 只需添加规则配置，无需改代码
六、使用建议与扩展方向
1. 工程化建议
规则库管理：建立规则库版本控制系统，定期爬取各平台样本页面更新规则，建议每周更新一次
缓存策略：搜索结果缓存 30 分钟，商品详情缓存 1 小时，图片哈希缓存 7 天，历史价格缓存 24 小时
分布式部署：采用多节点分布式架构，每个节点负责特定平台的爬取，避免单节点压力过大
监控告警：建立平台规则有效性监控，当解析成功率低于 80% 时触发告警
2. 功能扩展方向
图像搜索比价：集成以图搜图功能，支持用户上传商品图片进行跨平台比价
个性化偏好：允许用户自定义各价值维度的权重（如对服务敏感的用户可提高服务权重）
价格趋势预测：基于 LSTM 等时间序列模型，预测商品价格走势，推荐最佳购买时机
优惠券智能匹配：自动检索用户账户中的优惠券，计算最优使用组合
跨品类比价：针对可替代商品（如不同品牌的同配置手机）进行横向价值对比
通过这套方案，开发者可以构建真正智能化、可扩展的多平台比价系统，解决传统比价工具在平台适配、商品匹配、价格计算等方面的核心痛点，为用户提供全面、精准、智能的购物决策支持。
