智能电商价格监控系统结合 AI 分析与代理池技术，解决传统爬虫反爬限制与动态页面解析难题。通过代理实现 IP 轮换，利用大模型提取商品结构化数据，集成重试机制保障稳定性。系统支持批量监控与价格预警，适用于多平台电商数据采集场景。
在现代电商环境中，价格监控已成为商家和消费者的刚需。然而传统的网页爬虫面临着反爬虫机制越来越严格、网页结构复杂多变、IP 被封禁等诸多挑战。本文将详细介绍如何结合 AI 智能分析与高质量代理池，构建一个既稳定又智能的电商价格监控系统。
现代电商网站的反爬虫机制日趋完善，传统爬虫面临以下核心挑战：
不同平台的数据呈现方式差异巨大，需要更智能的识别和解析能力。
为了解决这些问题，我们设计了一个"AI + 代理池"的智能抓取架构：
[目标网站] ← [高质量代理池] ← [智能请求管理] ← [AI 内容分析] ← [结构化输出]
核心设计思路：
这种架构将网络访问和内容分析分离，各司其职，大幅提升了系统的稳定性和智能化水平。
首先需通过 pip 安装相关库：
pip install requests beautifulsoup4 openai loguru pandas urllib3 python-dotenv regex
项目构建需要合理的技术栈组合：
import requests
from bs4 import BeautifulSoup
import openai
from loguru import logger
import pandas as pd
import urllib3
from typing import Optional, Dict, List
import re
import json
import random
import time
import datetime
关键配置优化：
# 屏蔽 SSL 证书警告，提高请求成功率
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# 配置智能日志系统
logger.add("price_monitor.log", rotation="1 day", level="INFO")
这些配置能够有效减少网络请求中的干扰因素，提升系统稳定性。
代理池是整个系统的网络基础，选择 Decodo 作为代理服务提供商。核心优势是：
这确保了访问的稳定性和隐蔽性，大幅提升抓取成功率。
这些代理节点不仅分布在不同端口，还具备动态切换能力，当某个代理出现响应延迟或连接失败时，系统会自动将其从可用列表中移除，确保始终使用状态最佳的代理进行网络请求。
获取代理的方式需要在控制台配置账户信息，将用户信息输入代码中：
url = 'https://ip.decodo.com/json'
username = '<YOUR_USERNAME>'
password = '<YOUR_PASSWORD>'
proxy = f"http://{username}:{password}@isp.decodo.com:10001"
class SimpleProxyManager:
"""基于 Decodo 的智能代理管理器"""
def __init__(self):
# Decodo 代理配置 – 多端口负载均衡
base_url = "http://<USERNAME>:<PASSWORD>@isp.decodo.com"
ports = [10001, 10002, 10003, 10004]
self.proxies = [
{"http": f"{base_url}:{port}", "https": f"{base_url}:{port}"}
for port in ports
]
self.current_index = 0
def get_proxy(self) -> Optional[Dict]:
"""智能代理获取 - 轮询算法"""
if not self.proxies:
return None
proxy = self.proxies[self.current_index]
self.current_index = (self.current_index + 1) % len(self.proxies)
return proxy
def remove_proxy(self, proxy: Dict):
"""失效代理自动移除"""
if proxy in self.proxies:
self.proxies.remove(proxy)
logger.warning(f"移除失效代理：{proxy}")
代理管理的核心设计理念就是轮询机制避免单点过载，同时要达到失效代理实时清理，这样通过智能化的代理调度，就可以确保我们的系统具有高可用性和稳定性。
这是 AI 分析内容的主要部分，负责理解复杂的电商页面结构。**AI 分析的关键优势：**智能识别页面主要内容，同时自动过滤广告和无关信息。AI 模型能够理解页面的语义结构，准确提取商品核心信息。
这里我们给大模型设置了专属的提示词：
prompt = f"""
请分析以下电商页面内容，提取商品信息。
返回 JSON 格式：
网页 URL: {url} 网页内容：{text_content}
请提取：
product_name: 商品名称
current_price: 当前价格（只要数字，去掉货币符号）
original_price: 原价（如果有）
stock_status: 库存状态
is_available: 是否有货（true/false）只返回 JSON，不要其他文字。
"""
大家在设置的时候可以参考我的提示词给 AI 明确的指令，这样能让 AI 更精准地理解需求，减少无效输出。下面是完整的工具类代码，已针对 OpenAI API 版本兼容和密钥安全配置做了优化：
import os
import re
import json
from bs4 import BeautifulSoup
from openai import OpenAI
from dotenv import load_dotenv
from loguru import logger
from typing import Dict
load_dotenv()
class AIAnalyzer:
"""基于 GPT 的智能内容分析器，专注电商页面商品信息提取"""
def __init__(self):
"""初始化 OpenAI 客户端，提前校验 API 密钥配置"""
self.openai_api_key = os.getenv("OPENAI_API_KEY")
if not self.openai_api_key:
raise ValueError("请先配置 OpenAI API 密钥！可在.env 文件中添加'OPENAI_API_KEY=你的密钥'")
self.client = OpenAI(api_key=self.openai_api_key)
logger.info("OpenAI 客户端初始化完成，已准备好进行商品信息分析")
def extract_product_info(self, html_content: str, url: str) -> Dict:
"""使用 AI 智能提取商品信息"""
soup = BeautifulSoup(html_content, 'html.parser')
for tag in soup(['script', 'style', 'nav', 'footer', 'header', 'aside']):
tag.decompose()
text_content = soup.get_text(separator=' ', strip=True)
if len(text_content) > 3000:
text_content = text_content[:3000] + "...（内容过长，已截取前 3000 字符）"
logger.debug(f"预处理后待分析文本长度：{len(text_content)}字符")
prompt = f"""
请分析以下电商页面内容，提取商品信息。返回 JSON 格式：
网页 URL: {url}
网页内容：{text_content}
请提取：
1. product_name: 商品名称
2. current_price: 当前价格（只要数字，去掉货币符号）
3. original_price: 原价（如果有）
4. stock_status: 库存状态
5. is_available: 是否有货（true/false）
只返回 JSON，不要其他文字。
"""
try:
response = self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "你是专注于电商数据提取的工具，输出仅 JSON，无多余内容"},
{"role": "user", "content": prompt}
],
temperature=0.1,
max_tokens=300,
timeout=10
)
result_text = response.choices[0].message.content.strip()
json_match = re.search(r'\{[\s\S]*\}', result_text, re.DOTALL)
if json_match:
result = json.loads(json_match.group())
result["source_url"] = url
logger.success(f"成功提取商品信息：{result.get('product_name', '未知商品')}（来自{url}）")
return result
else:
logger.error(f"AI 返回内容格式错误，未匹配到 JSON：{result_text[:150]}...")
return {"error": "AI 返回格式错误", "source_url": url, "raw_response": result_text[:200]}
except Exception as e:
error_msg = f"商品信息提取失败：{str(e)}"
logger.error(f"{error_msg}（URL：{url}）")
return {"error": error_msg, "source_url": url}
结合代理池和 AI 分析，构建智能的网页获取系统：
def fetch_page(self, url: str, max_retries: int = 3) -> tuple[str, bool]:
"""智能网页获取，包含重试和代理切换"""
for attempt in range(max_retries):
proxy = self.proxy_manager.get_proxy()
if not proxy:
logger.warning("没有可用代理，尝试直接连接")
proxy = None
try:
response = self.session.get(url, proxies=proxy, timeout=(5, 20), verify=False)
if response.status_code == 200:
logger.info(f"成功获取页面：{url}")
return response.text, True
elif response.status_code in [403, 429, 503]:
logger.warning(f"访问受限 {response.status_code}, 更换代理重试")
if proxy:
self.proxy_manager.remove_proxy(proxy)
except requests.exceptions.ReadTimeout:
logger.warning(f"读取超时，尝试更换代理 (尝试 {attempt + 1}/{max_retries})")
if proxy:
self.proxy_manager.remove_proxy(proxy)
except Exception as e:
logger.error(f"请求异常：{str(e)}")
if proxy:
self.proxy_manager.remove_proxy(proxy)
delay = random.uniform(2 + attempt * 2, 5 + attempt * 2)
logger.info(f"等待 {delay:.1f} 秒后重试...")
time.sleep(delay)
return "", False
智能重试的核心特性：
这种机制能够有效应对各种网络异常，提升整体成功率。
将所有组件整合，构建完整的商品监控流程：
def monitor_product(self, url: str) -> Dict:
"""单个商品完整监控流程"""
logger.info(f"开始监控商品：{url}")
html_content, success = self.fetch_page(url)
if not success:
return {
"url": url, "success": False, "error": "无法获取页面内容",
"timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
product_info = self.ai_analyzer.extract_product_info(html_content, url)
if "error" in product_info:
return {
"url": url, "success": False, "error": product_info["error"],
"timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
current_price = self._extract_price(product_info.get('current_price'))
original_price = self._extract_price(product_info.get('original_price'))
result = {
"url": url, "success": True,
"product_name": product_info.get('product_name', ''),
"current_price": current_price, "original_price": original_price,
"stock_status": product_info.get('stock_status', ''),
"is_available": product_info.get('is_available', False),
"discount": self._calculate_discount(original_price, current_price),
"timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
logger.success(f"监控成功：{result['product_name']} - ¥{current_price}")
return result
监控流程的优势：
每个环节都有详细的状态跟踪，便于问题定位和系统优化。
创建 PriceMonitor 主类，将代理管理、AI 分析、网页抓取等功能模块有机整合，形成了一个完整的工作流程。让整个监控系统真正可以很简单的运行起来，现在我们只需配置好 Decodo 代理和 OpenAI API 密钥，就能立即开始监控心仪商品的价格变化，实现自动的电商价格追踪：
class PriceMonitor:
"""完整的价格监控系统主类"""
def __init__(self):
self.proxy_manager = SimpleProxyManager()
self.ai_analyzer = AIAnalyzer()
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
})
logger.info("价格监控系统初始化完成")
def fetch_page(self, url: str, max_retries: int = 3) -> tuple[str, bool]:
for attempt in range(max_retries):
proxy = self.proxy_manager.get_proxy()
if not proxy:
logger.warning("没有可用代理，尝试直接连接")
proxy = None
try:
response = self.session.get(url, proxies=proxy, timeout=(5, 20), verify=False)
if response.status_code == 200:
logger.info(f"成功获取页面：{url}")
return response.text, True
elif response.status_code in [403, 429, 503]:
logger.warning(f"访问受限 {response.status_code}, 更换代理重试")
if proxy:
self.proxy_manager.remove_proxy(proxy)
except requests.exceptions.ReadTimeout:
logger.warning(f"读取超时，尝试更换代理 (尝试 {attempt + 1}/{max_retries})")
if proxy:
self.proxy_manager.remove_proxy(proxy)
except Exception as e:
logger.error(f"请求异常：{str(e)}")
if proxy:
self.proxy_manager.remove_proxy(proxy)
delay = random.uniform(2 + attempt * 2, 5 + attempt * 2)
logger.info(f"等待 {delay:.1f} 秒后重试...")
time.sleep(delay)
return "", False
def _extract_price(self, price_str) -> float:
if not price_str:
return 0.0
price_match = re.search(r'(\d+\.?\d*)', str(price_str))
if price_match:
return float(price_match.group(1))
return 0.0
def _calculate_discount(self, original_price: float, current_price: float) -> float:
if original_price > 0 and current_price > 0:
return round((original_price - current_price) / original_price * 100, 2)
return 0.0
def monitor_product(self, url: str) -> Dict:
logger.info(f"开始监控商品：{url}")
html_content, success = self.fetch_page(url)
if not success:
return {"url": url, "success": False, "error": "无法获取页面内容", "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
product_info = self.ai_analyzer.extract_product_info(html_content, url)
if "error" in product_info:
return {"url": url, "success": False, "error": product_info["error"], "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
current_price = self._extract_price(product_info.get('current_price'))
original_price = self._extract_price(product_info.get('original_price'))
result = {
"url": url, "success": True,
"product_name": product_info.get('product_name', ''),
"current_price": current_price, "original_price": original_price,
"stock_status": product_info.get('stock_status', ''),
"is_available": product_info.get('is_available', False),
"discount": self._calculate_discount(original_price, current_price),
"timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
logger.success(f"监控成功：{result['product_name']} - ¥{current_price}")
return result
def batch_monitor(self, urls: List[str]) -> List[Dict]:
logger.info(f"开始批量监控 {len(urls)} 个商品")
results = []
for i, url in enumerate(urls, 1):
logger.info(f"处理第 {i}/{len(urls)} 个商品")
result = self.monitor_product(url)
results.append(result)
if i < len(urls):
delay = random.uniform(3, 8)
logger.info(f"等待 {delay:.1f} 秒...")
time.sleep(delay)
return results
def price_alert(self, results: List[Dict], target_prices: Dict[str, float]):
alerts = []
for result in results:
if not result.get('success'):
continue
url = result['url']
current_price = result['current_price']
product_name = result['product_name']
if url in target_prices and current_price > 0:
target_price = target_prices[url]
if current_price <= target_price:
alert = {
'product_name': product_name, 'current_price': current_price,
'target_price': target_price, 'savings': target_price - current_price,
'url': url, 'timestamp': result['timestamp']
}
alerts.append(alert)
print(f"\n🎉 价格预警触发！")
print(f"商品：{product_name}")
print(f"当前价格：¥{current_price}")
print(f"目标价格：¥{target_price}")
print(f"可节省：¥{alert['savings']:.2f}")
print(f"链接：{url}")
return alerts
if __name__ == "__main__":
monitor = PriceMonitor()
urls_to_monitor = [
"https://item.jd.com/100012043978.html",
"https://detail.tmall.com/item.htm?id=123456789",
]
target_prices = {
"https://item.jd.com/100012043978.html": 299.0,
"https://detail.tmall.com/item.htm?id=123456789": 199.0,
}
results = monitor.batch_monitor(urls_to_monitor)
alerts = monitor.price_alert(results, target_prices)
print("\n=== 监控结果汇总 ===")
for result in results:
if result['success']:
print(f"商品：{result['product_name']}")
print(f"当前价格：¥{result['current_price']}")
print(f"库存状态：{result['stock_status']}")
print(f"监控时间：{result['timestamp']}")
print("-" * 50)
else:
print(f"监控失败：{result['url']} - {result['error']}")
print(f"\n本次监控完成，共处理 {len(results)} 个商品，触发 {len(alerts)} 个价格预警")
通过实际测试，系统表现出了优异的性能，可以快速获取我们需要的商品信息。AI 模型能够准确识别商品名称、价格、库存状态，自动过滤页面中的广告和推荐内容，同时提供稳定的网络访问，智能重试机制应对临时网络问题。
在实际应用中，系统的价格预警功能展现出了强大的实用性，其核心设计围绕精准监控与及时提醒展开，形成了一套完整的智能预警机制。依托代理保障网络访问稳定，借助 AI 精准提取商品信息，当监控到商品当前价≤用户设的目标价时，会立即触发预警，还支持 CSV、JSON 格式存储数据，方便后续分析。
本项目成功整合了 AI 技术和代理池技术，实现了以下技术突破：
智能化水平显著提升
网络访问稳定性优化
通过 AI 和代理池的深度融合，我们成功构建了一个既智能又稳定的价格监控系统。这种技术组合不仅解决了传统爬虫的痛点，更为未来的智能数据抓取技术发展指明了方向。随着 AI 技术的不断进步和代理服务质量的持续提升，相信这种"AI+ 代理池"的模式将在更多领域得到广泛应用，为数据驱动的商业决策提供更强大的技术支撑。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online