如何随时间监测亚马逊Rufus推荐意见

SEED_QUERIES = [
"最佳降噪耳机",
"最佳旅行无线耳塞",
"最佳预算耳罩耳机",
已知品牌 = ("索尼", "博士", "苹果", "森海塞尔", "Beats", "JBL", "安克")

```python
def 从标题获取品牌(title: str) -> str | None:
    if not title:
        return None
    lowered = title.lower()
    for brand in 已知品牌:
        if brand.lower() in lowered:
            return brand
    return title.split()[0]  # 回退：标题的首个词

import json
import time

def 追加快照(path: str, query: str, rows: list[dict]) -> dict:
    record = {"query": query, "captured_at": int(time.time()), "products": rows}
    with open(path, "a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, ensure_ascii=False) + "\n")
    return record

from collections import Counter

def 推荐份额(rows: list[dict]) -> dict[str, float]:
    counts = Counter(row["brand"] for row in rows if row.get("brand"))
    total = sum(counts.values())
    if not total:
        return {}
    return {brand: round(100 * n / total, 1) for brand, n in counts.most_common()}

def 差异运行(prev_rows: list[dict], curr_rows: list[dict]) -> dict[str, list[str]]:
    prev = {row["asin"] for row in prev_rows}
    curr = {row["asin"] for row in curr_rows}
    return {
        "进入": sorted(curr - prev),
        "丢弃": sorted(prev - curr),
    }

if __name__ == "__main__":
    snapshot_path = "rufus_snapshots.jsonl"
    for query in SEED_QUERIES:
        result = capture_rufus(query)
        rows = extract_products(result)
        if not rows:
            print(f"{query}: 此查询/地区没有 Rufus 答复")
            continue
        append_snapshot(snapshot_path, query, rows)
        share = share_of_recommendation(rows)
        leaders = ", ".join(f"{b} {pct}%" for b, pct in list(share.items())[:3])
        print(f"{query}: {len(rows)} 种产品 — {leaders}")

// Schema 是 scraper.amazon (类型: rufus) 返回的内容；字段值是来自实际运行的示例（部分和产品已修剪）。
{
  "metadata": { "type": "rufus", "rawUrl": "https://…" },
  "result": {
    "user_query": "最佳噪音取消耳机",
    "content_blocks": [
      {
        "type": "product_section",
        "category": "顶部精选 - 最佳 ANC",

"产品": [
          {
            "asin": "B0GN4CFF6H",
            "标题": "索尼 WH-1000XM6/B 无线降噪耳机",
            "价格": "398.00美元",
            "原价": "428.00美元",
            "评分": "4.5",
            "评论数": "238",
            "配送": "免费配送 星期五, 6月19日",
            "url": "https://…"
          }
        ]
      }
    ],
    "产品": [
      { "asin": "B0GN4CFF6H", "标题": "索尼 WH-1000XM6/B …", "类别": "精选 – 最佳降噪" }
    ],
    "相关问题": ["比较索尼 XM6 与博世 QC Ultra 2", "这些产品中有打折吗？"]
  }
}

// 管道输出 — 说明性示例（模式真实，ASIN/品牌/排名值为说明性）：一个仅追加快照行加上推荐份额统计。
{
  "快照": {
    "查询": "最佳降噪耳机",
    "捕获时间": 1781716376,
    "产品": [
      { "asin": "B0GN4CFF6H", "排名": 1, "部分": "精选 – 最佳降噪", "品牌": "索尼" },
      { "asin": "B0FDKR293G", "排名": 2, "部分": "精选 – 最佳降噪", "品牌": "博世" },
      { "asin": "B0GSS4SGZR", "排名": 3, "部分": "适合苹果用户", "品牌": "苹果" }
    ]
  },
  "推荐份额": { "索尼": 33.3, "博世": 33.3, "苹果": 33.3 }
}

两个原因。查询可能不够交易性——将其表述为一个购买问题，并明确产品意图。或者您所针对的区域没有Rufus表面，在这种情况下，参与者报告该商店的区域失败；固定一个支持的区域并将缺失记录为跳过。

**问：我需要代理还是浏览器吗？**
不需要。渲染、区域处理和解析在服务器端运行。您发送一个带有 `x-api-token` 头的POST请求并读取JSON响应；参与者返回已经结构化的推荐集。

**问：我如何获得产品的排名和出现的部分？**
遍历 `result.content_blocks`：每个 `product_section` 块都携带一个 `category` 头和自己的 `products` 数组。在扁平化块时计数产品可以得到总体排名，而块的 `category` 则给出部分——这两者在每个快照中都值得存储。

**问：推荐份额是什么？**
它是一个品牌在查询或查询集中的推荐位置所占的百分比，汇总自捕获的产品。随着时间的推移，它显示品牌在Rufus货架上的存在是增加还是减少——这是对话购物等同于声量份额的指标。

**问：为什么要存储快照而不是单一的当前视图？**
Rufus每个会话生成其答案，因此任何一次捕获都是一个时间点。按查询和捕获时间逐步生成的追加快照为您提供了历史差异对比，因此您可以报告在运行之间哪些ASIN进入或退出，而不是从单个响应中猜测。

**问：我可以同时监控多少个查询？**
保持并发适度——几个正在进行的查询足以进行稳定的监控运行。将范围限定在您采取行动的问题上，并让 `related_questions` 建议下一个要添加的问题，这样每个可计费的调用都能获得其位置。