![amazon-price-monitoring-tool-guide.png](https://developer.qcloudimg.com/http-save/yehe-11158438/0be2646ac4b2d414c0620bd24cce813d.png)

## 一、业务背景与需求分析

亚马逊关联商品（Related Products）数据是跨境电商数据基础设施的重要组成部分，在以下业务场景中有核心价值：

- **竞品选品分析**：通过 FBT 和 Also Bought 数据构建竞品关联图谱，发现蓝海品类机会
- **关联广告定向**：利用 Also Viewed 数据扩充 ASIN 定向广告的候选池
- **竞争格局研究**：识别 Sponsored 广告位，分析竞争对手的广告投放策略
- **AI Agent 数据供给**：为 Listing 优化 Agent 提供实时竞品关联数据

本文从企业级数据采集架构的角度，解析亚马逊六大关联商品模块的技术实现，适合有一定工程基础的数据团队参考。

---

## 二、六大模块技术特性对比
![Amazon Related Products Modules on Real Product Page.png](https://developer.qcloudimg.com/http-save/yehe-11158438/59263f078a80244eaeb0c3109f2b060b.png)

亚马逊商品详情页包含六个独立的关联商品模块，技术特性差异显著：

```
模块名称                    | 渲染方式        | 容器 ID                          | 数据规模    | 采集难度
─────────────────────────────────────────────────────────────────────────────────────
Frequently Bought Together  | 服务端渲染      | #sims-fbt                        | 2-4 ASIN   | 低
Customers Also Bought       | 服务端+轮播     | #p_dp_sims_btf_1                 | 15-20 ASIN | 中
Customers Also Viewed       | 服务端+轮播     | #p_dp_sims_btf_2                 | 15-20 ASIN | 中
Similar Items               | 混合渲染        | #similarities_feature_div        | 4-8 ASIN   | 中
Sponsored Products          | JavaScript注入  | 穿插在各模块                      | 4-8 ASIN   | 高
Best Sellers Sidebar        | 服务端渲染      | #zg-right-col                    | 3-5 ASIN   | 低
```

**关键技术点**：
- FBT 和 BSR Sidebar：静态 HTML 解析，无需浏览器渲染
- Also Bought / Also Viewed：轮播分页，静态 HTML 只有首屏 4-6 个 ASIN，完整列表需要动态翻页
- Sponsored：JavaScript 动态注入，需要完整浏览器环境
- 所有模块均面临亚马逊的 Cloudflare + ASN 双层拦截，需要住宅代理

---

## 三、架构设计

### 3.1 整体架构

```
┌─────────────────────────────────────────────────────┐
│                   任务调度层                         │
│         (Celery Beat / APScheduler)                  │
│  ASIN 列表管理 → 优先级队列 → 断点续采              │
└──────────────────┬──────────────────────────────────┘
                   │
┌──────────────────▼──────────────────────────────────┐
│                   采集执行层                         │
│   Playwright Cluster (N workers) + 住宅代理池        │
│   - 静态模块：lxml/BeautifulSoup 解析               │
│   - 轮播模块：Playwright 翻页 + DOM 采集            │
│   - Sponsored：Page.on("response") XHR 拦截         │
└──────────────────┬──────────────────────────────────┘
                   │
┌──────────────────▼──────────────────────────────────┐
│                   解析标准化层                       │
│   字段提取 → ASIN 去重 → is_sponsored 标记           │
│   → 数据质量校验 → 结构化输出                       │
└──────────────────┬──────────────────────────────────┘
                   │
┌──────────────────▼──────────────────────────────────┐
│                   存储与服务层                       │
│   PostgreSQL（关联关系图数据）                       │
│   Redis（实时缓存 + 去重状态）                       │
│   ClickHouse（历史变化分析）                        │
└─────────────────────────────────────────────────────┘
```

### 3.2 核心模块实现

```python
# related_products_extractor.py

from bs4 import BeautifulSoup
from dataclasses import dataclass, asdict
from typing import Optional, List, Dict
import re
import asyncio
import logging

logger = logging.getLogger(__name__)


@dataclass
class RelatedProduct:
    asin: str
    title: Optional[str] = None
    price: Optional[str] = None
    rating: Optional[float] = None
    review_count: Optional[int] = None
    image_url: Optional[str] = None
    is_sponsored: bool = False
    module: str = ""
    rank_in_module: int = 0  # 在模块内的排名（第几个展示）


@dataclass
class RelatedProductsResult:
    source_asin: str
    marketplace: str
    scraped_at: str
    fbt: List[RelatedProduct]
    also_bought: List[RelatedProduct]
    also_viewed: List[RelatedProduct]
    similar_items: List[RelatedProduct]
    bsr_sidebar: List[RelatedProduct]
    summary: Dict
    errors: List[str]


class RelatedProductsExtractor:
    """
    亚马逊关联商品六大模块提取器
    支持静态 HTML 解析和 Playwright 动态渲染两种模式
    """

    # 各模块容器选择器配置，按优先级排列
    MODULE_CONTAINERS = {
        "fbt": ["sims-fbt", "frequently-bought-together_feature_div"],
        "also_bought": [
            "p_dp_sims_btf_1",
            "similarities_feature_div",
            "sp-recommendations-feature-div"
        ],
        "also_viewed": [
            "p_dp_sims_btf_2",
            "session-sims-feature-div",
            "customers-also-viewed_feature_div"
        ],
        "similar_items": ["similarities_feature_div", "compare-similar-feature-div"],
        "bsr_sidebar": ["zg-right-col", "browseTree_feature_div"]
    }

    def __init__(self, html: str):
        self.soup = BeautifulSoup(html, "lxml")
        self.errors = []

    def _find_container(self, module: str):
        """按优先级查找模块容器"""
        for cid in self.MODULE_CONTAINERS.get(module, []):
            container = self.soup.find(id=cid)
            if container:
                return container
        return None

    def _extract_asin(self, el) -> Optional[str]:
        """多方式提取 ASIN"""
        # 方式一：data-asin 属性（最可靠）
        asin = el.get("data-asin", "").strip()
        if asin and re.match(r"^[A-Z0-9]{10}$", asin):
            return asin
        # 方式二：href /dp/ASIN 路径
        for link in el.find_all("a", href=True)[:3]:
            m = re.search(r"/dp/([A-Z0-9]{10})", link["href"])
            if m:
                return m.group(1)
        return None

    def _parse_card(self, card_el, module: str, rank: int) -> Optional[RelatedProduct]:
        """通用商品卡片解析"""
        asin = self._extract_asin(card_el)
        if not asin:
            return None

        p = RelatedProduct(asin=asin, module=module, rank_in_module=rank)

        # 标题（多选择器降级）
        for sel in [".a-size-base-plus", ".a-size-medium", ".a-truncate-cut", "[data-rows]"]:
            el = card_el.select_one(sel)
            if el:
                p.title = el.get_text(strip=True)[:200]
                break

        # 价格
        price_w = card_el.select_one(".a-price-whole")
        price_f = card_el.select_one(".a-price-fraction")
        if price_w:
            w = price_w.get_text(strip=True).replace(",", "")
            f = price_f.get_text(strip=True) if price_f else "00"
            p.price = f"${w}.{f}"
        else:
            off = card_el.select_one(".a-price .a-offscreen")
            if off:
                p.price = off.get_text(strip=True)

        # 评分
        r_el = card_el.select_one("[aria-label*='out of 5 stars']")
        if r_el:
            m = re.search(r"([\d.]+)\s+out of 5", r_el.get("aria-label", ""))
            if m:
                try:
                    p.rating = float(m.group(1))
                except ValueError:
                    pass

        # 评论数
        rev = card_el.select_one("[aria-label$='ratings']") or card_el.select_one("[aria-label$='reviews']")
        if rev:
            m = re.search(r"([\d,]+)", rev.get("aria-label", ""))
            if m:
                try:
                    p.review_count = int(m.group(1).replace(",", ""))
                except ValueError:
                    pass

        # 图片
        img = card_el.select_one("img")
        if img:
            p.image_url = img.get("src") or img.get("data-src")

        # Sponsored 检测
        p.is_sponsored = self._is_sponsored(card_el)

        return p

    def _is_sponsored(self, card_el) -> bool:
        """Sponsored 广告四特征联合检测"""
        if card_el.select_one(".s-sponsored-label-info-icon, .puis-sponsored-label-info-icon"):
            return True
        if "sponsored" in card_el.get("data-component-type", "").lower():
            return True
        text_preview = card_el.get_text()[:300].lower()
        if "sponsored" in text_preview:
            return True
        for a in card_el.select("a[href]"):
            href = a.get("href", "")
            if "amazon-adsystem" in href or "aax-us-pdp" in href:
                return True
        return False

    def _extract_from_container(self, module: str) -> List[RelatedProduct]:
        container = self._find_container(module)
        if not container:
            return []

        results, seen = [], set()
        cards = container.select("[data-asin], .a-carousel-card")

        for rank, card in enumerate(cards, 1):
            p = self._parse_card(card, module, rank)
            if p and p.asin not in seen:
                seen.add(p.asin)
                results.append(p)

        return results

    def extract_all(self) -> dict:
        """提取所有模块，统一返回"""
        result = {}
        for module in ["fbt", "also_bought", "also_viewed", "similar_items", "bsr_sidebar"]:
            try:
                result[module] = self._extract_from_container(module)
            except Exception as e:
                logger.error(f"提取模块 {module} 出错: {e}")
                result[module] = []
                self.errors.append(f"{module}: {str(e)}")

        all_products = [p for module_list in result.values() for p in module_list]
        result["summary"] = {
            "total_unique_asins": len({p.asin for p in all_products}),
            "total_sponsored": sum(1 for p in all_products if p.is_sponsored),
            **{f"{k}_count": len(v) for k, v in result.items() if isinstance(v, list)}
        }
        result["errors"] = self.errors
        return result
```

### 3.3 Playwright 动态翻页模块

```python
# carousel_crawler.py

from playwright.async_api import async_playwright, Browser
import asyncio
import re
import logging

logger = logging.getLogger(__name__)


class CarouselCrawler:
    """
    处理亚马逊轮播翻页，获取完整关联列表
    支持异步并发采集，提升吞吐量
    """

    CAROUSEL_CONFIG = {
        "also_bought": {
            "container_ids": ["p_dp_sims_btf_1", "similarities_feature_div"],
            "next_btn": ".a-carousel-goto-nextpage",
            "max_pages": 5
        },
        "also_viewed": {
            "container_ids": ["p_dp_sims_btf_2", "session-sims-feature-div"],
            "next_btn": ".a-carousel-goto-nextpage",
            "max_pages": 5
        }
    }

    def __init__(self, proxy_server: str, marketplace: str = "US"):
        self.proxy_server = proxy_server
        self.base_url = f"https://www.amazon.{'com' if marketplace == 'US' else 'co.uk'}"

    async def extract_carousel_asins(
        self,
        asin: str,
        module: str,
        browser: Browser
    ) -> list[str]:
        """
        异步提取单个商品单个轮播模块的完整 ASIN 列表
        """
        config = self.CAROUSEL_CONFIG.get(module, {})
        all_asins = []

        ctx = await browser.new_context(
            proxy={"server": self.proxy_server},
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
                       "(KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
            viewport={"width": 1440, "height": 900}
        )
        page = await ctx.new_page()

        try:
            await page.goto(f"{self.base_url}/dp/{asin}", timeout=45000, wait_until="domcontentloaded")

            # 等待轮播模块加载
            container_id = None
            for cid in config.get("container_ids", []):
                try:
                    await page.wait_for_selector(f"#{cid}", timeout=8000)
                    container_id = cid
                    break
                except Exception:
                    continue

            if not container_id:
                logger.warning(f"ASIN {asin}: 找不到模块 {module} 的容器")
                return []

            await page.wait_for_timeout(2000)

            for page_num in range(config.get("max_pages", 5)):
                # 采集当前页 ASIN
                cards = await page.query_selector_all(f"#{container_id} [data-asin]")
                for card in cards:
                    a = await card.get_attribute("data-asin")
                    if a and re.match(r"^[A-Z0-9]{10}$", a) and a not in all_asins:
                        all_asins.append(a)

                # 翻页
                next_btn = await page.query_selector(f"#{container_id} {config['next_btn']}")
                if not next_btn:
                    break
                is_visible = await next_btn.is_visible()
                is_enabled = await next_btn.is_enabled()
                if not (is_visible and is_enabled):
                    break

                await next_btn.scroll_into_view_if_needed()
                await next_btn.click()
                await page.wait_for_timeout(1000)

        except Exception as e:
            logger.error(f"轮播采集失败 ASIN={asin}, module={module}: {e}")
        finally:
            await ctx.close()

        return all_asins

    async def batch_extract(self, asins: list[str], module: str, concurrency: int = 3) -> dict[str, list[str]]:
        """
        异步并发采集多个 ASIN 的轮播数据
        """
        results = {}
        semaphore = asyncio.Semaphore(concurrency)

        async with async_playwright() as pw:
            browser = await pw.chromium.launch(headless=True)

            async def process_asin(asin: str):
                async with semaphore:
                    results[asin] = await self.extract_carousel_asins(asin, module, browser)
                    logger.info(f"  {asin}: {len(results[asin])} ASINs")

            await asyncio.gather(*[process_asin(asin) for asin in asins])
            await browser.close()

        return results
```

---

## 四、与 Pangolinfo API 的集成方案
![amazon-price-monitoring-tool-comparison-table.png](https://developer.qcloudimg.com/http-save/yehe-11158438/2bf793fba4992cbc898dfa6cd8845979.png)

对于日均 1000+ ASIN 的采集规模，自建方案的维护成本（住宅代理 + Cloudflare 规则更新 + DOM 结构变化处理）通常超过使用 Pangolinfo Amazon Scraper API的费用。

集成方案（Python）：

```python
import requests
from typing import Optional, Dict

def fetch_related_products_via_api(
    asin: str,
    api_key: str,
    marketplace: str = "US"
) -> Optional[Dict]:
    """
    使用 Pangolinfo API 获取亚马逊关联商品数据
    
    优势：
    - 六大模块统一返回，含轮播完整列表
    - 内置 Sponsored 标记
    - DOM 结构变化时服务端透明更新
    - 无需维护代理池和浏览器集群
    """
    response = requests.post(
        "https://api.pangolinfo.com/v1/amazon/product",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json={
            "asin": asin,
            "marketplace": marketplace,
            "fields": ["related_products", "fbt", "also_bought", "also_viewed", "sponsored_asins"]
        },
        timeout=30
    )
    response.raise_for_status()
    return response.json().get("data", {})
```


---

## 五、监控与告警

生产环境建议监控以下指标：

```python
# 字段提取成功率监控（Prometheus 格式）
# related_products_fbt_success_rate: gauge（目标 > 85%）
# related_products_also_bought_success_rate: gauge（目标 > 75%）
# carousel_pagination_avg_asins: gauge（目标 > 10，低于 6 表示翻页失效）
# sponsored_detection_rate: gauge（监控 Sponsored 识别率变化，突变时告警）
```

字段提取成功率持续低于阈值，通常意味着亚马逊更新了 DOM 结构，需要检查选择器配置。

---

*如有工程问题欢迎在评论区交流，也可关注我们获取亚马逊数据采集的最新动态。*

亚马逊关联商品（Related Products）数据是跨境电商数据基础设施的重要组成部分，在以下业务场景中有核心价值：

企业级亚马逊关联商品数据采集架构方案

开发工具

云计算

后端

亚马逊关联商品数据采集技术解析：涵盖FBT、AlsoBought等六大模块，详细对比服务端渲染与动态加载差异，提供Python代码实现方案，解决轮播翻页、Sponsored识别等核心问题，适合跨境电商数据团队构建高效采集系统。

PostgreSQL

容器

Python

Redis

Agent

2026上云采购 | AI焕新·智启新局

tchousec

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

企业级亚马逊关联商品数据采集架构方案-腾讯云开发者社区-腾讯云

企业级亚马逊关联商品数据采集架构方案

企业级亚马逊关联商品数据采集架构方案

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐