Amazon Rufusの推奨を時間をかけて監視する方法

Michael Lee

Expert Network Defense Engineer

17-Jun-2026

重要なポイント:

Amazon Rufusの推奨は構造化された商品データであり、チャットのトランスクリプトではありません。 type: rufusを指定してscraper.amazonアクターを1回呼び出すことで、各エントリにASIN、タイトル、価格、評価、Rufusが分類したセクションラベルを持つproducts配列が返されます。
推奨のシェアはこのパイプラインが生成する指標です。 Rufusが固定のクエリセットに対してどのASINを提示するかを追跡することで、会話型シェルフは測定可能な可視性信号に変わり、Rufus版のシェア・オブ・ボイスとなります。
Rufusは選択肢をラベル付きセクションに分けます。 レスポンスは「トップピック - ベストANC」や「グレートバリューピック」といったcontent_blocksを通じて商品をグループ化するため、すべての商品には全体的なランクとそれをフレーミングするセクション見出しが付いています。
related_questionsは独自にシードクエリセットを拡張します。 各キャプチャは、Rufusが提案するクエリに対するフォローアップ質問を返し、それらは次回の実行のプロンプトリストに直接フィードバックされます。
すべてのフィールドはヌル可能で、回答はセッションごとに生成されます。 特定の地域に対してRufusの回答がない場合があり、商品スロットがboughtカウントなしで到着する場合もあります。そのため、パイプラインは各実行のスナップショットを保存し、シリーズを読み取りますが、単一の呼び出しではありません。
パイプラインは追加のみのスナップショットと差分に収束します。 各実行は、クエリとキャプチャ時間でキー付けされた1つのJSONLレコードを書き込み、連続するレコードを比較することで、実行間にどのASINが追加または削除されたかを報告します。
始めるのは無料です。 新しいScrapelessアカウントには無料のトライアルクレジットが含まれています — app.scrapeless.comでサインアップできます。

イントロダクション：バイヤーシェルフがアシスタントに移動

Amazon Rufusはショッピング関連の質問に製品のランク付きリストで回答します。買い物客が最高のノイズキャンセリングヘッドフォンを求めると、Rufusは「トップピックス」、「Appleユーザーに最適」、「グレートバリューピック」と題されたグループ化された選択肢を返します。それぞれには価格、評価、および購入リンクが含まれ、アシスタント内で、検索結果ページが読み込まれる前に表示されます。ブランドにとって、質問はもはや結果グリッド上の製品のランクがどこにあるかではありません; Rufusがその製品を名指しするかどうか、どのセクションで、そして何ランクであるかです。

そのシェルフは時間の経過とともに監視するのが難しいです。Rufusはセッションごとに回答を生成し、選択肢はクエリや地域によって変わり、商品カードは自動化と戦う会話型サーフェス内で解決されます。一度目で目で読むだけでは何もわかりません; シグナルは推薦セットが週ごとにどのように動くかです。

このガイドは、Scrapeless Scraping APIの上にモニタリングパイプラインを構築します: 固定クエリセットが入力され、scraper.amazonアクターがRufusの推奨を構造化された製品として返し、パイプラインがASIN、ランク、セクションを抽出するとともに、各ASINをブランドにマッピングし、実行ごとにスナップショットを保存し、スナップショットの差分を報告して推奨のシェアを伝えます。これはGoogle AIオーバービューにおけるブランド可視性と自然に組み合わせられ、検索側で同様の推奨質問を追跡します。

パイプラインの概要

システム全体は6つのステージからなります。

クエリセットの定義 — 固定のバイイング質問リストで、各キャプチャが返すrelated_questionsによって拡張されます。
クエリごとのキャプチャ — type: rufusを指定して各クエリをscraper.amazonアクターにPOSTします。サポートされていない地域はRufusの回答を返さないため、パイプラインは記録してスキップします。
製品の抽出 — content_blocksを辿って、各製品のASIN、全体のランク、セクションラベルを引き出します。
ASIN → ブランドのマッピング — 小さなヒューリスティックを用いて製品タイトルからブランドを解決し、ASINレベル以上でシェアが集計できるようにします。
実行ごとのスナップショットの保存 — 各クエリに対して、クエリとキャプチャ時間でキー付けされた1つのJSONLレコードを追加します；上書きはしません。
時間経過に伴う差分 — 連続したスナップショットを比較して推奨のシェアおよびどのASINが追加または削除されたかを報告します。

ステージ1〜4はすべてのクエリに対して各サイクルで実行され、ステージ5〜6はそれらのキャプチャを時系列に変換します。以下のセクションでは、各ステージを順を追って構築します。

前提条件

Python 3.10以上（以下のコードは標準ライブラリとrequestsのみを使用します）
ScrapelessアカウントおよびAPIキー — app.scrapeless.comでサインアップ
SCRAPELESS_API_KEYとしてエクスポートされたキー
ターミナルとJSONへの基本的な理解

ステージ1 — クエリセットの定義

モニタリングプログラムは、そのクエリセットの質に依存します。追跡するカテゴリにとって重要なバイイング質問から始めます。買い物客がRufusに尋ねるような語句で、明確な購入意図をもって表現します。

python Copy

SEED_QUERIES = [
"最高のノイズキャンセリングヘッドフォン",
    "旅行用の最高のワイヤレスイヤフォン",
    "最高の予算オーバーイヤーヘッドフォン",
]
既知のブランド = ("Sony", "Bose", "Apple", "Sennheiser", "Beats", "JBL", "Anker")


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]  # フォールバック：タイトルの最初の単語

---

## ステージ 4 — 1回のスナップショットをストアする

パイプラインは追記専用です：すべてのキャプチャはクエリとキャプチャ時間でキー付けされた1つのJSONLレコードを書き込み、何も上書きされることはありません。これにより、完全な履歴を比較することができ、悪いまたは空の実行が以前の良い実行を破壊することはありません。

```python
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

整数のエポックを captured_at に使用することで、各レコードは自己記述的でソート可能になります。特定のクエリの履歴を読み込むには、ファイルを行ごとに読み取り、queryキーでフィルタリングします — 1回のパスでキャプチャ順にすべてのスナップショットが得られます。

ステージ 5 — 差分と推薦のシェアを報告する

2つの読み取り側関数がスナップショット履歴をメトリクスに変換します。推薦のシェアは、各ブランドが実行中に何回出現したかをカウントし、パーセンテージに正規化します。差分は、2つの実行のASINセットを比較して移動を表示します。

python Copy

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 {
        "entered": sorted(curr - prev),
        "dropped": sorted(prev - curr),
    }

クエリごとに 推薦のシェア を実行して、その質問に対する会話の棚を所有するブランドを確認するか、全体のクエリセットに対してカテゴリー全体のビューを確認します。クエリの最近の2つのスナップショットの間で 実行の差分 を実行して、ブランドが選ばれる瞬間や外れる瞬間をキャッチします — 警告に値する瞬間です。

スケジューリングとスケーリング

キャプチャループがステージをつなぎます：各クエリについて、キャプチャ、抽出、スナップショットを取り、空の結果としてスキップを記録します。スケジュールで実行します — 毎日または毎週 — JSONLファイルは時系列となります。

python Copy

if __name__ == "__main__":
    スナップショットパス = "rufus_snapshots.jsonl"
    for query in SEED_QUERIES:
        result = capture_rufus(query)
        rows = extract_products(result)
        if not rows:
            print(f"{query}: このクエリ/地域に対するRufusの応答はありません")
            continue
        スナップショットを追加(スナップショットパス, query, rows)
        share = 推薦のシェア(rows)
        leaders = ", ".join(f"{b} {pct}%" for b, pct in list(share.items())[:3])
        print(f"{query}: {len(rows)} 商品 — {leaders}")

クエリセットをスケールするときのいくつかの実用的な制約：

同時実行を控えめに保つ — 一度に飛行中のクエリは多少で十分です；モニタリング実行は一定で、バーストではありません。
実行ごとに地域を固定するため、時系列が比較可能になります；ある地域でRufusの応答がないクエリはスキップとして記録され、他の地域の数値に混ざることはありません。
あなたが行動を起こすことにスコープを定める。各クエリは請求可能な呼び出しなので、意思決定を促す質問を監視し、related_questionsに次に追加する質問を提案させます。 Scrapelessの価格ティアに対してペースを計画します。

何が返ってくるか

各キャプチャはフラットな products リストとセクショングループ化された content_blocks を生成します；パイプラインは両方をスナップショット行とシェアテーブルに減らします。以下の形状がアクターによって返されるものであり、1つのセクションにトリムされています。

json Copy

// スキーマはscraper.amazon（タイプ：rufus）が返すものです；フィールド値はライブ実行からの説明的サンプルです（セクションと製品がトリムされています）。
{
  "metadata": { "type": "rufus", "rawUrl": "https://…" },
  "result": {
    "user_query": "最高のノイズキャンセリングヘッドフォン",
    "content_blocks": [
      {
        "type": "product_section",
        "category": "トップピック – 最高のANC",

json Copy

{
  "products": [
    {
      "asin": "B0GN4CFF6H",
      "title": "ソニー WH-1000XM6/B ワイヤレスノイズキャンセリングヘッドフォン",
      "price": "¥53,250",
      "original_price": "¥57,080",
      "rating": "4.5",
      "reviews": "238",
      "delivery": "送料無料 6月19日金曜日到着",
      "url": "https://…"
    }
  ],
  "products": [
    { "asin": "B0GN4CFF6H", "title": "ソニー WH-1000XM6/B …", "category": "おすすめ商品 – ベストANC" }
  ],
  "related_questions": ["ソニー XM6とボーズ QC ウルトラ 2を比較する", "これらの製品はセール中ですか？"]
}

json Copy

{
  "snapshot": {
    "query": "最高のノイズキャンセリングヘッドフォン",
    "captured_at": 1781716376,
    "products": [
      { "asin": "B0GN4CFF6H", "rank": 1, "section": "おすすめ商品 – ベストANC", "brand": "ソニー" },
      { "asin": "B0FDKR293G", "rank": 2, "section": "おすすめ商品 – ベストANC", "brand": "ボーズ" },
      { "asin": "B0GSS4SGZR", "rank": 3, "section": "Appleユーザーに最適", "brand": "Apple" }
    ]
  },
  "share_of_recommendation": { "ソニー": 33.3, "ボーズ": 33.3, "Apple": 33.3 }
}

二つの原因。クエリが十分に取引的ではない可能性があります—購入に関する質問として、明確な商品意図を持って表現してください。または、ターゲットにした地域にルーフスの表面がない場合、その場合はアクターがそのストアの地域の失敗を報告します; サポートされている地域を指定し、ミスをスキップとして記録します。

Q: プロキシやブラウザは必要ですか？
いいえ。レンダリング、地域ハンドリング、パースはサーバー側で実行されます。x-api-tokenヘッダー付きのPOSTを一回送信し、JSONで返されるのを読んでください; アクターはすでに構造化された推薦セットを返します。

Q: 製品が表示されたランクとセクションをどのように取得しますか？
result.content_blocksをたどります：各product_sectionブロックはcategory見出しと独自のproducts配列を持っています。ブロックをフラット化する際に製品をカウントすることで全体のランクが得られ、ブロックのcategoryがセクションを示します—どちらもスナップショットごとに保存する価値があります。

Q: 推薦のシェアとは何ですか？
それは、クエリまたはクエリセット全体でブランドが占める推薦スロットの割合であり、キャプチャされた製品から集計されます。時間を追って追跡すると、ブランドがルーフスの棚で存在感を増しているのか減っているのかを示します—会話型ショッピングのボイスシェアの相当物です。

Q: なぜ単一の最新ビューではなくスナップショットを保存しますか？
ルーフスはセッションごとに回答を生成するため、どのキャプチャも時間のポイントです。クエリとキャプチャ時間でキー付けされた追加のみのスナップショットは、差分を取るための履歴を提供しますので、実行間でどのASINが入ったかまたは減ったかを推測することなく報告できます。

Q: 一度にいくつのクエリを監視できますか？
同時実行は控えめに保ってください — 数件のクエリが進行中であれば、一貫したモニタリング実行に十分です。アクションを行う質問にスコープを設定し、related_questionsが次に追加すべき質問を示唆するのに任せてくださいので、各請求可能な呼び出しがその場所を得ます。

Scrapelessでは、適用される法律、規制、およびWebサイトのプライバシーポリシーを厳密に遵守しながら、公開されているデータのみにアクセスします。このブログのコンテンツは、デモンストレーションのみを目的としており、違法または侵害の活動は含まれません。このブログまたはサードパーティのリンクからの情報の使用に対するすべての責任を保証せず、放棄します。スクレイピング活動に従事する前に、法律顧問に相談し、ターゲットウェブサイトの利用規約を確認するか、必要な許可を取得してください。