动态住宅代理：跨境电商数据抓取的稳定解决方案

# 配置基本的IP获取参数
proxy_config = {
    "country": "US",           # 目标国家
    "state": "CA",            # 目标州/省（可选）
    "city": "Los Angeles",    # 目标城市（可选）
    "rotation_mode": "smart", # 智能轮换模式
    "session_duration": 600,  # 会话持续时间（秒）
    "concurrent_sessions": 5, # 并发会话数
    "retry_failed": True      # 失败自动重试
}

# 获取代理IP
proxy = client.get_proxy(proxy_config)
print(f"分配的IP: {proxy.ip}:{proxy.port}")
print(f"认证信息: {proxy.username}:{proxy.password}")

import requests
import time
import json
import sqlite3
from bs4 import BeautifulSoup
from decodo_sdk import ProxyRotator
import logging

class EcommercePriceMonitor:
    """跨境电商价格监控系统"""
    
    def __init__(self, decodo_api_key, db_path="price_monitor.db"):
        # 初始化Decodo代理轮换器
        self.proxy_rotator = ProxyRotator(
            api_key=decodo_api_key,
            rotation_strategy="adaptive",    # 自适应轮换策略
            max_failures_per_ip=3,          # 单IP最大失败次数
            rotation_interval=300            # 基础轮换间隔5分钟
        )
        
        # 初始化数据库
        self.db_path = db_path
        self.init_database()
        
        # 配置日志
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
        
    def scrape_amazon_price(self, asin, country="US"):
        """抓取Amazon产品价格"""
        # 根据国家选择对应的Amazon域名
        amazon_domains = {
            "US": "amazon.com",
            "UK": "amazon.co.uk", 
            "DE": "amazon.de",
            "JP": "amazon.co.jp",
            "CA": "amazon.ca"
        }
        
        if country not in amazon_domains:
            raise ValueError(f"不支持的国家代码: {country}")
        
        domain = amazon_domains[country]
        url = f"https://www.{domain}/dp/{asin}"
        
        # 获取对应国家的Decodo代理IP
        proxy = self.proxy_rotator.get_proxy(
            country_code=country,
            rotation_trigger="request"
        )
        
        # 构建真实的浏览器请求头
        headers = {
            'User-Agent': self.get_random_user_agent(country),
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': self.get_accept_language(country),
            'Accept-Encoding': 'gzip, deflate, br',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1',
        }
        
        try:
            # 使用Decodo代理发送请求
            response = requests.get(
                url,
                headers=headers,
                proxies={
                    'http': f"http://{proxy.username}:{proxy.password}@{proxy.ip}:{proxy.port}",
                    'https': f"https://{proxy.username}:{proxy.password}@{proxy.ip}:{proxy.port}"
                },
                timeout=15,
                allow_redirects=True
            )
            
            if response.status_code == 200:
                return self.parse_amazon_price(response.text, asin, country, proxy.ip)
            elif response.status_code == 503:
                # 遇到服务不可用，可能是IP被限制，立即轮换
                self.logger.warning(f"Amazon返回503，IP可能被限制: {proxy.ip}")
                self.proxy_rotator.mark_ip_failed(proxy.ip)
                return None
            else:
                self.logger.warning(f"请求失败，状态码: {response.status_code}")
                return None
                
        except requests.RequestException as e:
            self.logger.error(f"请求异常: {e}")
            return None
    
    def parse_amazon_price(self, html_content, asin, country, proxy_ip):
        """解析Amazon价格信息"""
        soup = BeautifulSoup(html_content, 'html.parser')
        
        # Amazon价格元素的多种可能选择器
        price_selectors = [
            'span.a-price-whole',
            'span#price_inside_buybox',
            'span.a-offscreen',
            'span#priceblock_dealprice',
            'span#priceblock_ourprice'
        ]
        
        price_text = None
        for selector in price_selectors:
            price_element = soup.select_one(selector)
            if price_element:
                price_text = price_element.get_text().strip()
                break
        
        if not price_text:
            self.logger.warning(f"无法找到价格信息: {asin}")
            return None
        
        # 提取数字价格
        import re
        price_match = re.search(r'[\d,]+\.?\d*', price_text.replace(',', ''))
        if price_match:
            price = float(price_match.group())
            
            # 获取货币信息
            currency_map = {
                "US": "USD", "UK": "GBP", "DE": "EUR", 
                "JP": "JPY", "CA": "CAD"
            }
            currency = currency_map.get(country, "USD")
            
            # 检查库存状态
            availability = "In Stock"
            if soup.select_one('[data-feature-name="availability"] span'):
                availability_text = soup.select_one('[data-feature-name="availability"] span').get_text()
                if "out of stock" in availability_text.lower():
                    availability = "Out of Stock"
            
            return {
                'product_id': asin,
                'platform': 'Amazon',
                'country': country,
                'price': price,
                'currency': currency,
                'availability': availability,
                'proxy_ip': proxy_ip
            }
        
        return None
    
    def get_random_user_agent(self, country):
        """根据国家获取合适的User-Agent"""
        user_agents = {
            "US": [
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
            ],
        }
        
        import random
        return random.choice(user_agents.get(country, user_agents["US"]))
    
    def get_accept_language(self, country):
        """根据国家获取Accept-Language"""
        languages = {
            "US": "en-US,en;q=0.9",
            "UK": "en-GB,en;q=0.9",
            "DE": "de-DE,de;q=0.9,en;q=0.8",
            "JP": "ja-JP,ja;q=0.9,en;q=0.8",
            "CA": "en-CA,en;q=0.9,fr;q=0.8"
        }
        return languages.get(country, "en-US,en;q=0.9")
    
    def save_price_data(self, price_data):
        """保存价格数据到数据库"""
        if not price_data:
            return
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute("""
            INSERT INTO price_history 
            (product_id, platform, country, price, currency, availability, proxy_ip)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (
            price_data['product_id'],
            price_data['platform'],
            price_data['country'],
            price_data['price'],
            price_data['currency'],
            price_data['availability'],
            price_data['proxy_ip']
        ))
        
        conn.commit()
        conn.close()
        
        self.logger.info(f"保存价格数据: {price_data['product_id']} - {price_data['currency']} {price_data['price']}")
    
 

# 使用示例
if __name__ == "__main__":
    # 初始化监控系统
    monitor = EcommercePriceMonitor(
        decodo_api_key="your-decodo-api-key",
        db_path="ecommerce_monitor.db"
    )
    
    # 设置监控产品列表（Amazon ASIN）
    products_to_monitor = [
        "B08N5WRWNW",  # Echo Dot
        "B07XJ8C8F5",  # Fire TV Stick
        "B09G9FPHY6",  # Kindle Paperwhite
        "B08F7PTF53",  # Echo Show
        "B07HZLHPKP"   # Fire HD Tablet
    ]