亚马逊 Coupon 折扣数据监控完整指南：为什么难、怎么做、哪种方案适合你（2026）

import asyncio
import re
from playwright.async_api import async_playwright, Page
from typing import Optional
import logging

logging.basicConfig(level=logging.INFO)

# 亚马逊 Coupon badge 多选择器降级链
COUPON_SELECTORS = [
    "#couponBadgeAsinDetailPageCoupon",    # 主选择器（美国/英国/德国主要布局）
    ".couponBadge",                         # class 备用选择器
    "#vpcButton",                           # Virtual Product Coupon
    "[data-feature-name='couponBadge']",   # 属性选择器（新版布局）
]

# 折扣文本正则解析（支持美/英/德/日等主要站点）
COUPON_PATTERNS = [
    # "Save 15% with coupon" / "Apply 15% coupon"
    (r"(?:save|apply|clip|get)\s+(\d+(?:\.\d+)?)%\s*(?:coupon|off|discount)?",
     "percentage"),
    # "Clip $5 coupon" / "Save £8 with coupon"
    (r"(?:clip|save|get|apply)\s*[$£€¥₹]?\s*(\d+(?:\.\d+)?)\s*(?:coupon|off|discount)?",
     "fixed"),
    # "15% Coupon" (short format)
    (r"(\d+(?:\.\d+)?)%\s*(?:coupon|off)",
     "percentage"),
    # "$5 coupon" / "£8 Coupon"
    (r"[$£€¥₹]\s*(\d+(?:\.\d+)?)\s*(?:coupon|off)?",
     "fixed"),
]


def parse_coupon_text(text: str) -> Optional[dict]:
    """
    解析 Coupon badge 文本，提取折扣类型和面值
    返回 {"coupon_type": "percentage"|"fixed", "coupon_value": float}
    """
    if not text:
        return None
    
    text_lower = text.lower().strip()
    
    for pattern, coupon_type in COUPON_PATTERNS:
        match = re.search(pattern, text_lower, re.IGNORECASE)
        if match:
            try:
                value = float(match.group(1))
                return {
                    "coupon_type": coupon_type,
                    "coupon_value": value,
                    "coupon_text_raw": text.strip()
                }
            except (ValueError, IndexError):
                continue
    
    return None


async def extract_coupon_from_page(page: Page, asin: str) -> Optional[dict]:
    """
    从已加载的 Playwright 页面中提取 Coupon 信息
    使用多选择器降级策略
    """
    for selector in COUPON_SELECTORS:
        try:
            el = await page.query_selector(selector)
            if not el:
                continue
            
            # 检查元素是否可见（不是隐藏的蜜罐节点）
            is_visible = await el.is_visible()
            if not is_visible:
                continue
            
            text = await el.inner_text()
            if not text or len(text.strip()) < 3:
                continue
            
            coupon_data = parse_coupon_text(text)
            if coupon_data:
                coupon_data["selector_used"] = selector
                logging.info(f"  [{asin}] Coupon 找到: {coupon_data} (via {selector})")
                return coupon_data
        except Exception as e:
            logging.debug(f"  [{asin}] 选择器 {selector} 出错: {e}")
            continue
    
    logging.info(f"  [{asin}] 未检测到 Coupon badge")
    return None


async def scrape_coupon(asin: str, marketplace: str = "US",
                         proxy: str = None) -> dict:
    """
    主入口：对单个 ASIN 发起 Playwright 请求并提取 Coupon 数据
    
    Args:
        asin: 目标 ASIN
        marketplace: 站点代码（US/UK/DE/JP/FR 等）
        proxy: 住宅代理地址（格式：http://user:pass@host:port）
    """
    marketplace_domains = {
        "US": "www.amazon.com",
        "UK": "www.amazon.co.uk",
        "DE": "www.amazon.de",
        "JP": "www.amazon.co.jp",
        "FR": "www.amazon.fr",
    }
    domain = marketplace_domains.get(marketplace, "www.amazon.com")
    url = f"https://{domain}/dp/{asin}"
    
    result = {
        "asin": asin,
        "marketplace": marketplace,
        "url": url,
        "coupon": None,
        "error": None
    }
    
    async with async_playwright() as p:
        browser_args = {}
        if proxy:
            browser_args["proxy"] = {"server": proxy}
        
        browser = await p.chromium.launch(
            headless=True,
            args=["--no-sandbox", "--disable-setuid-sandbox"]
        )
        
        context = await browser.new_context(
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/125.0.0.0 Safari/537.36"
            ),
            viewport={"width": 1920, "height": 1080},
            locale="en-US",
        )
        
        page = await context.new_page()
        
        try:
            await page.goto(url, wait_until="domcontentloaded", timeout=30000)
            # 等待主价格区域加载（Coupon 在 price block 渲染后注入）
            await page.wait_for_selector(
                "#corePriceDisplay_desktop_feature_div, #price",
                timeout=10000
            )
            # 额外等待 JS 渲染完成
            await asyncio.sleep(2)
            
            result["coupon"] = await extract_coupon_from_page(page, asin)
            
        except Exception as e:
            result["error"] = str(e)
            logging.error(f"[{asin}] 页面加载失败: {e}")
        finally:
            await browser.close()
    
    return result


# ── 使用示例 ──────────────────────────────────────
async def main():
    test_asins = ["B0CHP7BPYQ", "B09G9FPHY6"]
    
    for asin in test_asins:
        result = await scrape_coupon(
            asin=asin,
            marketplace="US",
            # proxy="http://user:pass@residential-proxy:port"  # 生产环境需要
        )
        print(f"\n{asin}:")
        print(f"  Coupon: {result['coupon']}")
        print(f"  Error: {result['error']}")

if __name__ == "__main__":
    asyncio.run(main())

{
  "asin": "B0CHP7BPYQ",
  "marketplace": "US",
  "buybox_price": 29.99,
  "list_price": 39.99,
  "coupon_type": "fixed",
  "coupon_value": 5.00,
  "coupon_text": "Clip $5 coupon",
  "coupon_applies_at": "checkout",
  "deal_price": null,
  "subscribe_save_price": 26.99,
  "effective_price": 24.99,
  "buybox_seller_id": "AMAZON",
  "buybox_is_amazon": true,
  "fetched_at": "2026-06-22T09:00:00Z"
}

import requests
import json
import time
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] %(levelname)s: %(message)s"
)

PANGOLINFO_API_KEY = "your_api_key_here"
API_ENDPOINT = "https://api.pangolinfo.com/v1/amazon/product"

# 需要采集的字段（包含所有 Coupon 相关字段）
FIELDS = [
    "buybox_price", "buybox_seller_id", "buybox_is_amazon",
    "list_price", "deal_price", "deal_ends_at",
    "coupon_type", "coupon_value", "coupon_text",
    "coupon_applies_at",
    "subscribe_save_price",
    "effective_price",
    "offer_count"
]


def fetch_coupon_snapshot(asin: str, marketplace: str = "US") -> Optional[dict]:
    """
    通过 Pangolinfo API 获取单个 ASIN 的 Coupon 快照
    返回完整结构化数据，无需处理渲染或选择器问题
    """
    try:
        resp = requests.post(
            API_ENDPOINT,
            headers={
                "Authorization": f"Bearer {PANGOLINFO_API_KEY}",
                "Content-Type": "application/json"
            },
            json={
                "asin": asin,
                "marketplace": marketplace,
                "fields": FIELDS
            },
            timeout=30
        )
        resp.raise_for_status()
        data = resp.json().get("data", {})
        data["_asin"] = asin
        data["_marketplace"] = marketplace
        data["_fetched_at"] = datetime.now(timezone.utc).isoformat()
        return data
    except requests.HTTPError as e:
        logging.error(f"API 错误 [{asin}]: {e}")
        return None
    except Exception as e:
        logging.error(f"未知错误 [{asin}]: {e}")
        return None


class CouponChangeDetector:
    """
    有状态的 Coupon 变化检测器
    对比相邻两次快照，生成告警事件
    """

    def __init__(self):
        self._state: dict[str, dict] = {}

    def process(self, snapshot: dict) -> list[dict]:
        """
        处理新快照，返回检测到的 Coupon 相关变化事件列表
        """
        asin = snapshot.get("_asin", "?")
        events = []
        prev = self._state.get(asin, {})
        ts = snapshot.get("_fetched_at", "")

        # 事件 1：新 Coupon 上线
        had_coupon = prev.get("has_coupon", False)
        has_coupon = bool(snapshot.get("coupon_type"))

        if has_coupon and not had_coupon:
            events.append({
                "event_type": "COUPON_ADDED",
                "asin": asin,
                "coupon_type": snapshot.get("coupon_type"),
                "coupon_value": snapshot.get("coupon_value"),
                "coupon_text": snapshot.get("coupon_text"),
                "effective_price": snapshot.get("effective_price"),
                "timestamp": ts
            })
            logging.warning(
                f"🎫 新 Coupon 上线: {asin} | "
                f"{snapshot.get('coupon_type')}={snapshot.get('coupon_value')} | "
                f"实际最低价: ${snapshot.get('effective_price', 'N/A')}"
            )

        # 事件 2：Coupon 下架
        elif not has_coupon and had_coupon:
            events.append({
                "event_type": "COUPON_REMOVED",
                "asin": asin,
                "prev_coupon_value": prev.get("coupon_value"),
                "timestamp": ts
            })
            logging.warning(f"❌ Coupon 下架: {asin} | 原 Coupon: {prev.get('coupon_value')}")

        # 事件 3：Coupon 面值变化
        elif has_coupon and had_coupon:
            prev_val = prev.get("coupon_value", 0)
            curr_val = snapshot.get("coupon_value", 0)
            if curr_val != prev_val:
                events.append({
                    "event_type": "COUPON_CHANGED",
                    "asin": asin,
                    "old_value": prev_val,
                    "new_value": curr_val,
                    "coupon_type": snapshot.get("coupon_type"),
                    "timestamp": ts
                })
                logging.warning(
                    f"🔄 Coupon 面值变化: {asin} | "
                    f"{prev_val} → {curr_val} ({snapshot.get('coupon_type')})"
                )

        # 事件 4：Coupon + Deal 叠加出现（高优先级告警）
        if has_coupon and snapshot.get("deal_price"):
            events.append({
                "event_type": "COUPON_PLUS_DEAL",
                "asin": asin,
                "deal_price": snapshot.get("deal_price"),
                "coupon_value": snapshot.get("coupon_value"),
                "effective_price": snapshot.get("effective_price"),
                "timestamp": ts
            })
            logging.warning(
                f"🔥 Coupon + Deal 叠加: {asin} | "
                f"Deal=${snapshot.get('deal_price')} + Coupon={snapshot.get('coupon_value')} | "
                f"最终价: ${snapshot.get('effective_price')}"
            )

        # 更新状态
        self._state[asin] = {
            "has_coupon": has_coupon,
            "coupon_type": snapshot.get("coupon_type"),
            "coupon_value": snapshot.get("coupon_value"),
            "effective_price": snapshot.get("effective_price")
        }

        return events


def run_coupon_monitor(
    asins: list[str],
    marketplace: str = "US",
    interval_minutes: int = 60,
    output_file: str = "coupon_snapshots.jsonl",
    events_file: str = "coupon_events.jsonl",
    max_cycles: Optional[int] = None
):
    """
    批量 Coupon 监控主循环
    - 按指定间隔轮询目标 ASIN 列表
    - 检测 Coupon 上线/下架/面值变化/叠加活动事件
    - 数据 JSONL 格式断点续存
    """
    output = Path(output_file)
    events_out = Path(events_file)
    detector = CouponChangeDetector()
    cycle = 0

    logging.info(
        f"启动 Coupon 监控: {len(asins)} 个 ASIN | "
        f"站点: {marketplace} | 间隔: {interval_minutes}min"
    )

    while max_cycles is None or cycle < max_cycles:
        cycle += 1
        logging.info(f"\n=== 第 {cycle} 轮 | {datetime.now().strftime('%Y-%m-%d %H:%M')} ===")

        cycle_snapshots = []
        cycle_events = []

        for i, asin in enumerate(asins, 1):
            snapshot = fetch_coupon_snapshot(asin, marketplace)
            if not snapshot:
                continue

            events = detector.process(snapshot)
            cycle_snapshots.append(snapshot)
            cycle_events.extend(events)

            coupon = snapshot.get("coupon_type")
            logging.info(
                f"  [{i}/{len(asins)}] {asin}: "
                f"BB=${snapshot.get('buybox_price', 'N/A')} | "
                f"Coupon={'✓ ' + str(snapshot.get('coupon_value')) if coupon else '✗'} | "
                f"Deal={'✓' if snapshot.get('deal_price') else '✗'} | "
                f"实际最低=${snapshot.get('effective_price', 'N/A')}"
            )
            time.sleep(0.5)

        # 写入快照和事件
        with open(output, "a", encoding="utf-8") as f:
            for s in cycle_snapshots:
                f.write(json.dumps(s, ensure_ascii=False) + "\n")

        if cycle_events:
            with open(events_out, "a", encoding="utf-8") as f:
                for e in cycle_events:
                    f.write(json.dumps(e, ensure_ascii=False) + "\n")

        logging.info(
            f"本轮: {len(cycle_snapshots)}/{len(asins)} 成功 | "
            f"{len(cycle_events)} 个变化事件"
        )

        if max_cycles is None or cycle < max_cycles:
            time.sleep(interval_minutes * 60)

    logging.info("监控任务结束")


# ── 使用示例 ──────────────────────────────────────
if __name__ == "__main__":
    competitor_asins = [
        "B0CHP7BPYQ",
        "B09G9FPHY6",
        "B08N5WRWNW",
        # 添加更多竞品 ASIN...
    ]

    run_coupon_monitor(
        asins=competitor_asins,
        marketplace="US",
        interval_minutes=60,         # 每小时一次
        output_file="coupon_snapshots.jsonl",
        events_file="coupon_events.jsonl",
        max_cycles=None              # 持续运行
    )