亚马逊商品信息速查：IPIDEA 企业级 HTTP 服务驱动的可视化采集工具

app/├── app.py              # Flask 后端应用├── amazon_scraper.py   # 数据采集核心逻辑 (包含IPIDEA代理部分)└── templates/    └── index.html      # 前端HTML页面

*   使用 `requests` 库发起 HTTP GET 请求到您的IPIDEA API链接。*   设置 `timeout` 参数，避免长时间等待。*   检查响应状态码是否为 200，确保请求成功。*   从响应文本中提取代理IP字符串，并格式化为 `http://IP:Port` 的形式，这是 Selenium WebDriver 配置代理时所需的格式。*   处理可能的网络错误或 API 返回空值的情况import requests # 用于调用IPIDEA API
IPIDEA_PROXY_API_URL = "替换为自己的" 
def get_ipidea_proxy_url():    """    从IPIDEA API获取一个代理IP并格式化。    此函数调用IPIDEA API，获取一个或多个代理IP，并返回其中一个格式化的URL。    """    try:        print("IPIDEA: 正在请求代理IP...")        response = requests.get(IPIDEA_PROXY_API_URL, timeout=10) # 设置请求超时10秒        if response.status_code == 200:            proxy_list_str = response.text.strip()            if proxy_list_str:                # IPIDEA API返回的是多行IP，每行一个IP:Port。我们只取第一个作为示例。                # 在实际应用中，您可能需要解析所有IP并维护一个IP池。                proxy_ip_port = proxy_list_str.split('\n')[0]                 print(f"IPIDEA: 成功获取代理 {proxy_ip_port}")                return f"http://{proxy_ip_port}" # 格式化为HTTP代理URL            else:                print("IPIDEA: API返回空代理字符串。")                return None        else:            print(f"IPIDEA: 获取代理失败，状态码 {response.status_code}, 响应 {response.text}")            return None    except requests.exceptions.RequestException as e:        print(f"IPIDEA: 请求代理API时出错 {e}")        return None

1.  导入必要的 Selenium 模块： 包括 `webdriver`（用于浏览器控制）、`By`（定位元素）、`WebDriverWait` 和 `expected_conditions`（用于智能等待页面元素加载）、`ChromeOptions`（用于配置 Chrome 浏览器）。
2.  编写 `setup_driver()` 函数：** 这个函数负责启动一个配置好的 Chrome 浏览器实例。
    *   作用： 初始化 Chrome WebDriver，设置浏览器启动选项（如无头模式、用户代理等），并将从 IPIDEA 获取的代理 IP 配置给浏览器。    *   实现细节：        *   创建 `ChromeOptions` 对象，用于定义浏览器的行为和特性。        *   添加反检测参数：`--headless`（让浏览器在后台运行，不显示界面，提高效率）、`--no-sandbox`、`--disable-dev-shm-usage`、`--disable-blink-features=AutomationControlled` 等，这些参数有助于模拟真实用户，减少被识别为自动化程序的可能性。        *   设置一个常见的 `User-Agent`，进一步模拟真实浏览器。        *   关键步骤： 调用第一步中实现的 `get_ipidea_proxy_url()` 函数，获取 IPIDEA企业级HTTP服务。如果成功获取，就通过 `chrome_options.add_argument(f'--proxy-server={proxy_url}')` 将代理配置到 Chrome 浏览器。        *   使用 `webdriver.Chrome(options=chrome_options)` 启动浏览器。        *   设置 `driver.set_page_load_timeout(30)`，防止页面加载时间过长。
 import random # 用于模拟人类行为的随机等待    from selenium import webdriver    from selenium.webdriver.common.by import By    from selenium.webdriver.support.ui import WebDriverWait    from selenium.webdriver.support import expected_conditions as EC    from selenium.webdriver.chrome.options import Options as ChromeOptions    from selenium.common.exceptions import TimeoutException, NoSuchElementException, WebDriverException
    # ... (get_ipidea_proxy_url 函数代码如上) ...
    def setup_driver():        """        设置Chrome驱动并动态配置IPIDEA代理。        每次调用此函数，都会尝试获取一个新的IPIDEA代理来启动浏览器。        """        chrome_options = ChromeOptions()        chrome_options.add_argument("--headless") # 在后台运行，不显示浏览器界面        chrome_options.add_argument("--no-sandbox")        chrome_options.add_argument("--disable-dev-shm-usage")        chrome_options.add_argument("--disable-blink-features=AutomationControlled")        chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])        chrome_options.add_experimental_option('useAutomationExtension', False)        chrome_options.add_argument("--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")        chrome_options.add_argument("--page-load-strategy=normal") # 正常加载策略                proxy_url = get_ipidea_proxy_url() # 调用函数获取IPIDEA代理        if proxy_url:            chrome_options.add_argument(f'--proxy-server={proxy_url}')            print("Driver: 浏览器将通过IPIDEA代理启动。")        else:            print("Driver: 未能获取IPIDEA代理，浏览器将不使用代理启动。")            # 在实际产品中，如果没有代理IP，这里可以抛出异常或进行重试，而不是继续无代理运行
        try:            # 确保您的系统已安装Chrome浏览器，并且chromedriver已在系统PATH中，            # 或者您可以指定 chromedriver 的路径：            # service = Service(executable_path='/path/to/chromedriver')            # driver = webdriver.Chrome(service=service, options=chrome_options)            driver = webdriver.Chrome(options=chrome_options)            driver.set_page_load_timeout(30) # 设置页面加载的最大等待时间            print("Driver: Chrome浏览器启动成功并已配置代理。")            return driver        except Exception as e:            print(f"Driver: 启动浏览器失败 {e}")            raise # 启动失败则抛出异常，停止后续操作

 *   调用 `setup_driver()` 获取一个配置好代理的浏览器实例。    *   构建亚马逊搜索 URL，例如 `https://www.amazon.com/s?k=search_term`。    *   使用 `driver.get(search_url)` 访问目标页面。    *   使用 `time.sleep(random.randint(5, 10))` 模拟人类浏览时的随机等待时间，进一步增强反检测能力。    *   智能等待页面元素： 使用 `WebDriverWait` 和 `EC.presence_of_element_located` 等待关键元素（如搜索结果列表 `[data-component-type='s-search-result']`）出现，确保页面内容已加载完毕。这是防止采集空数据或不完整数据的关键。    *   查找商品元素： 使用 `driver.find_elements(By.CSS_SELECTOR, "[data-component-type='s-search-result']")` 找到所有商品容器元素。    *   遍历并提取信息： 循环遍历每个商品元素，使用 `find_element(By.CSS_SELECTOR, ...)` 提取商品的名称、价格、URL、图片链接和评分。对价格等数据进行细致的解析，处理可能出现的不同显示方式。    *   将提取到的信息封装成字典，并添加到 `products_info` 列表中。    *   使用 `try-except` 块处理元素找不到 (`NoSuchElementException`) 或其他解析错误，保证程序的健壮性。    *   在 `finally` 块中确保 `driver.quit()` 被调用，即使出现错误也能关闭浏览器，释放资源。

def scrape_amazon_search_results(search_term):    """    从亚马逊搜索结果页采集主要商品信息。    每次执行此函数，都会启动一个新的浏览器实例并尝试使用IPIDEA代理。    """    driver = None    products_info = []    try:        driver = setup_driver() # 获取一个配置了IPIDEA代理的浏览器实例        search_url = f"https://www.amazon.com/s?k={search_term}"        print(f"Scraper: 正在访问亚马逊搜索页面: {search_url}")                driver.get(search_url)        time.sleep(random.randint(5, 10)) # 模拟人类浏览的随机等待                # 智能等待搜索结果列表元素出现，最多等待20秒        try:            WebDriverWait(driver, 20).until(                EC.presence_of_element_located((By.CSS_SELECTOR, "[data-component-type='s-search-result']"))            )            print("Scraper: 亚马逊搜索结果页面加载成功。")        except TimeoutException:            print("Scraper: 等待搜索结果超时，可能目标页面未加载或网站访问策略生效。返回空列表。")            return [] # 页面未加载成功，直接返回空数据                # 找到所有商品结果的容器元素        products = driver.find_elements(By.CSS_SELECTOR, "[data-component-type='s-search-result']")                for i, product_element in enumerate(products):            if i >= 20: # 仅采集前20个商品作为示例，避免过度采集                break            try:                # 提取商品名称                title_element = product_element.find_element(By.CSS_SELECTOR, "h2 span.a-size-medium.a-color-base.a-text-normal")                title = title_element.text.strip()                                # 提取商品价格：优先查找屏幕外价格，否则尝试分段价格                price_text = "价格未显示"                price_offscreen_elements = product_element.find_elements(By.CSS_SELECTOR, ".a-price > .a-offscreen")                if price_offscreen_elements:                    price_text = price_offscreen_elements[0].get_attribute("innerText").strip()                else:                     whole_part_elements = product_element.find_elements(By.CSS_SELECTOR, ".a-price-whole")                    fraction_part_elements = product_element.find_elements(By.CSS_SELECTOR, ".a-price-fraction")                    if whole_part_elements and fraction_part_elements:                        price_text = f"${whole_part_elements[0].text.strip()}.{fraction_part_elements[0].text.strip()}"
                # 提取商品详情页链接                url_element = product_element.find_element(By.CSS_SELECTOR, "a[href*='/dp/']")                url = url_element.get_attribute("href")                                # 提取商品图片链接                image_element = product_element.find_element(By.CSS_SELECTOR, "img.s-image")                image = image_element.get_attribute("src")                                # 提取商品评分                rating_elements = product_element.find_elements(By.CSS_SELECTOR, ".a-icon-alt")                rating = rating_elements[0].get_attribute("innerText").strip() if rating_elements else "无评分"                                products_info.append({                    'title': title,                    'price': price_text,                    'url': url,                    'image': image,                    'rating': rating                })            except NoSuchElementException:                # 某个商品缺少部分信息（如无价格），跳过当前商品，继续处理下一个                print(f"Scraper: 警告 - 某个商品元素缺失，跳过。")                continue             except Exception as e:                print(f"Scraper: 解析单个产品时发生错误: {e}。跳过当前商品。")                continue                print(f"Scraper: 成功采集到 {len(products_info)} 条商品信息。")        return products_info            except Exception as e:        print(f"Scraper: 采集过程中发生总体错误: {str(e)}。")        return []    finally:        if driver:            try:                driver.quit()                print("Scraper: 浏览器已关闭。")            except:                pass # 忽略关闭浏览器时可能发生的错误，确保程序不会崩溃

# app/app.py
from flask import Flask, request, jsonify, render_templatefrom flask_cors import CORSimport json# 导入我们新创建的数据采集模块和函数from amazon_scraper import scrape_amazon_search_results 
app = Flask(__name__)CORS(app)  # 允许跨域请求
@app.route('/')def index():    """主页面"""    return render_template('index.html')
@app.route('/api/scrape', methods=['POST'])def scrape_products():    """爬取Amazon产品的API接口"""    try:        data = request.get_json()        search_term = data.get('search_term', 'laptop')                if not search_term:            return jsonify({'success': False, 'error': '搜索词不能为空'}), 400                print(f"API: 开始为搜索词 '{search_term}' 爬取数据...")                # 调用我们新的数据采集函数        products = scrape_amazon_search_results(search_term)                if not products:            return jsonify({'success': False, 'error': '未找到任何产品，或爬取失败。请检查搜索词、代理配置或亚马逊页面结构。'}), 404                print(f"API: 成功采集到 {len(products)} 条产品信息。")        return jsonify({            'success': True,            'search_term': search_term,            'count': len(products),            'products': products        })            except Exception as e:        print(f"API错误: {str(e)}")        return jsonify({'success': False, 'error': f'服务器内部错误: {str(e)}'}), 500
@app.route('/api/health', methods=['GET'])def health_check():    """健康检查接口"""    return jsonify({'status': 'ok', 'message': 'Amazon数据采集API运行正常'})
if __name__ == '__main__':    print("启动Amazon数据采集API服务器...")    # 在生产环境中，请不要使用 debug=True，并且可能需要更稳定的部署方式（如 Gunicorn）    app.run(debug=True, host='0.0.0.0', port=5000)

<!DOCTYPE html><html lang="zh-CN"><head>    <meta charset="UTF-8">    <meta name="viewport" content="width=device-width, initial-scale=1.0">    <title>Amazon产品数据采集</title>
</head>
<body>    <div class="container">        <div class="header">            <h1>🛒 Amazon产品数据采集</h1>
            <p>快速搜索和获取Amazon产品信息</p>
        </div>
                <div class="content">            <div class="search-section">                <h2>搜索产品</h2>
                <form class="search-form" id="searchForm">                    <input type="text" class="search-input" id="searchInput" placeholder="输入要搜索的产品名称..." value="laptop">                    <button type="submit" class="search-btn" id="searchBtn">搜索</button>
                </form>
            </div>
                        <div id="loading" class="loading" style="display: none;">                <div class="spinner"></div>
                <p>正在搜索产品，请稍候...</p>
            </div>
                        <div id="results" class="results"></div>
                        <div class="api-info">                <h3>API接口信息</h3>
                <p><strong>搜索接口:</strong></p>
                <div class="api-endpoint">POST /api/scrape</div>
                <p><strong>健康检查:</strong></p>
                <div class="api-endpoint">GET /api/health</div>
                <p><strong>请求示例:</strong></p>
                <div class="api-endpoint">{"search_term": "laptop"}</div>
            </div>
        </div>
    </div>
    <script>        document.getElementById('searchForm').addEventListener('submit', async function(e) {            e.preventDefault();                        const searchInput = document.getElementById('searchInput');            const searchBtn = document.getElementById('searchBtn');            const loading = document.getElementById('loading');            const results = document.getElementById('results');                        const searchTerm = searchInput.value.trim();            if (!searchTerm) {                alert('请输入搜索词');                return;            }                        // 显示加载状态            loading.style.display = 'block';            searchBtn.disabled = true;            results.innerHTML = '';                        try {                const response = await fetch('/api/scrape', {                    method: 'POST',                    headers: {                        'Content-Type': 'application/json',                    },                    body: JSON.stringify({                        search_term: searchTerm                    })                });                                const data = await response.json();                                if (data.success) {                    displayResults(data);                } else {                    showError(data.error || '搜索失败');                }            } catch (error) {                showError('网络错误: ' + error.message);            } finally {                loading.style.display = 'none';                searchBtn.disabled = false;            }        });                function displayResults(data) {            const results = document.getElementById('results');                        let html = `                <div class="success">                    <h3>搜索完成！</h3>
                    <p>搜索词: <strong>${data.search_term}</strong></p>
                    <p>找到 <strong>${data.count}</strong> 个产品</p>
                </div>
            `;                        if (data.products && data.products.length > 0) {                html += '<div class="product-grid">';                                data.products.forEach(product => {                    html += `                        <div class="product-card">                            <img src="${product.image || 'https://via.placeholder.com/300x200?text=No+Image'}"                                  alt="${product.title}" class="product-image"                                  onerror="this.src='https://via.placeholder.com/300x200?text=Image+Error'">                            <div class="product-info">                                <div class="product-title">${product.title}</div>
                                <div class="product-price">${product.price || '价格未知'}</div>
                                ${product.rating ? `                                    <div class="product-rating">                                        <span class="stars">${'★'.repeat(Math.floor(product.rating))}</span>
                                        <span>${product.rating}</span>
                                    </div>
                                ` : ''}                                ${product.url ? `                                    <a href="${product.url}" target="_blank" class="product-link">查看详情</a>
                                ` : ''}                            </div>
                        </div>
                    `;                });                                html += '</div>';            }                        results.innerHTML = html;        }                function showError(message) {            const results = document.getElementById('results');            results.innerHTML = `                <div class="error">                    <h3>搜索失败</h3>
                    <p>${message}</p>
                </div>
            `;        }                // 页面加载时检查API状态        window.addEventListener('load', async function() {            try {                const response = await fetch('/api/health');                const data = await response.json();                console.log('API状态:', data);            } catch (error) {                console.error('API健康检查失败:', error);            }        });    </script>
</body>
</html>

pip install Flask Flask-Cors requests selenium

python app.py