基于 AI 与代理池的智能电商价格监控系统实战

关键配置优化：

# 屏蔽 SSL 证书警告，提高请求成功率
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# 配置智能日志系统
logger.add("price_monitor.log", rotation="1 day", level="INFO")

这些配置能够有效减少网络请求中的干扰因素，提升系统稳定性。

2.2 代理池管理核心实现

代理池是整个系统的网络基础，选择 Decodo 作为代理服务提供商。核心优势是：

• 高匿名度 IP，有效规避识别
• 多地域节点覆盖

这确保了访问的稳定性和隐蔽性，大幅提升抓取成功率。

这些代理节点不仅分布在不同端口，还具备动态切换能力，当某个代理出现响应延迟或连接失败时，系统会自动将其从可用列表中移除，确保始终使用状态最佳的代理进行网络请求。

获取代理的方式需要在控制台配置账户信息，将用户信息输入代码中：

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}")

代理管理的核心设计理念就是轮询机制避免单点过载，同时要达到失效代理实时清理，这样通过智能化的代理调度，就可以确保我们的系统具有高可用性和稳定性。

2.3 AI 内容分析引擎

这是 AI 分析内容的主要部分，负责理解复杂的电商页面结构。**AI 分析的关键优势：**智能识别页面主要内容，同时自动过滤广告和无关信息。AI 模型能够理解页面的语义结构，准确提取商品核心信息。

这里我们给大模型设置了专属的提示词：

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，不要其他文字。

"""

大家在设置的时候可以参考我的提示词给 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}

2.4 智能请求管理与重试机制

结合代理池和 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

智能重试的核心特性：

• 根据 HTTP 状态码智能判断
• 指数退避避免频繁重试

这种机制能够有效应对各种网络异常，提升整体成功率。

2.5 完整的监控流程实现

将所有组件整合，构建完整的商品监控流程：

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

监控流程的优势：

• 网络层和分析层职责分离
• 完整的错误处理和日志记录

每个环节都有详细的状态跟踪，便于问题定位和系统优化。

三、运行结果与性能分析

3.1 完整的工作流程

创建 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)} 个价格预警")

3.2 系统运行效果

通过实际测试，系统表现出了优异的性能，可以快速获取我们需要的商品信息。AI 模型能够准确识别商品名称、价格、库存状态，自动过滤页面中的广告和推荐内容，同时提供稳定的网络访问，智能重试机制应对临时网络问题。

3.3 实际应用效果

在实际应用中，系统的价格预警功能展现出了强大的实用性，其核心设计围绕精准监控与及时提醒展开，形成了一套完整的智能预警机制。依托代理保障网络访问稳定，借助 AI 精准提取商品信息，当监控到商品当前价≤用户设的目标价时，会立即触发预警，还支持 CSV、JSON 格式存储数据，方便后续分析。

四、总结与展望

本项目成功整合了 AI 技术和代理池技术，实现了以下技术突破：

智能化水平显著提升

• AI 理解页面结构，无需预定义解析规则
• 自适应各种电商平台的页面变化

网络访问稳定性优化

• 代理池提供高质量 IP 资源
• 智能重试和故障转移机制

通过 AI 和代理池的深度融合，我们成功构建了一个既智能又稳定的价格监控系统。这种技术组合不仅解决了传统爬虫的痛点，更为未来的智能数据抓取技术发展指明了方向。随着 AI 技术的不断进步和代理服务质量的持续提升，相信这种"AI+ 代理池"的模式将在更多领域得到广泛应用，为数据驱动的商业决策提供更强大的技术支撑。