前回の記事はこちら 【連載#24】eBay Sell REST API：Trading APIからREST APIへ：OAuth移行と最初のInventory API呼び出し はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第24回です。 第3回から第23回まで、私たちは Trading API（SOAP形式）と Message API を中心に、出品・在庫管理・注文処理・バルクメッセージ対応まで幅広い実装を積み重ねてきました。今回からは新しいフェーズ【Sell REST - 出品】の幕開けとして、モダンな REST API の世界へと本格的に軸足を移します。 第1回で構築した OAuth 2.0 認証基盤がいよいよ本領を発揮します。Trading API 時代の devID / appID / certID + UserToken という複雑な認証の組み合わせに代わり、すっきりとした Bearer Token でのAPI呼び出しが標準となります。第1回で実装した eBayTokenManager クラスはそのまま活用できます。 この記事で得られること: Inventory API の設計思想（SKU 中心設計）と Trading API（ItemID 中心設計）との根本的な構造の違いを理解し、なぜ3ステップの出品フローになるのかを把握する。 OAuth Bearer Token を使った PUT /inventory_item/{sku} の実際の呼び出し方を、最小構成から生産レベルの実装まで段階的に習得する。 「完全上書き」仕様による意図しないデータ消失、1日250回の修正上限、Inventory API と Trading API 間の非互換性という移行時の3大トラップと、その具体的な回避策を身につける。 背景・なぜこれが重要か (Motivation) 「Trading API がまだ使えるなら、わざわざ REST API に移行する必要ある？」 これは非常に正直な疑問です。確かに現時点では Trading API も動作しており、第3〜17回で構築したシステムは今日でも稼働しています。しかしこの判断を先延ばしにし続けることには、無視できない技術的リスクが潜んでいます。 eBay の API ロードマップは明確に REST と GraphQL API への集中投資を宣言しています。新機能（Promoted Listings の高度な入札制御、次世代の Offer 管理、Dynamic Shipping 等）は REST / GraphQL API でのみ提供されるケースが増え続けており、Trading API 側にはバックポートされません。言い換えれば、Trading API に留まり続けることは「今のシステムを維持できても、eBay マーケットプレイスの進化についていけなくなる」リスクを恒常的に抱えることを意味します。 もう一つの現実的な懸念は「SOAP の学習コスト」です。Trading API が SOAP ベースであるため、新しいチームメンバーのオンボーディングや外部ライブラリとの連携に余分なコストがかかります。REST API への移行は、将来の開発速度を高めるための先行投資でもあります。 本連載では、Trading API の知識を持つ読者が混乱なく Inventory API へ移行できるよう、両者の対応関係を丁寧に解説していきます。第3〜23回で積み上げた知識は無駄になりません。UUIDによる冪等性・バリデーションファースト・エラーハンドリングといったエンジニアリング思想はそのまま REST の世界でも通用します。 基本的な使い方（ベースライン）：createOrReplaceInventoryItemで最初のInventory Itemを作る Inventory API の最初のエンドポイントは PUT /inventory_item/{sku}（operationId: createOrReplaceInventoryItem）です。このメソッドは SKU をキーとして、商品の在庫数・コンディション・商品詳細をeBayのシステムに登録します。Trading API の AddFixedPriceItem（第4回）と大きく異なるのは、この時点では「まだeBayサイトには出品されていない」という点です。まずは最小構成で動作を確認しましょう。 必須ヘッダーは Authorization（Bearer Token）、Content-Type（application/json）、そして Content-Language の3つです。Content-Language は日本向けの出品であっても en-US の指定が基本です。この設定を忘れると 400 エラーで弾かれます。 # inventory_item_basic.py import requests EBAY_INVENTORY_BASE = "https://api.ebay.com/sell/inventory/v1" def create_inventory_item(sku: str, token: str) -> dict: """最小構成でInventory Itemを作成する。 Args: sku: 出品者が定義するSKU（最大50文字、全在庫でユニーク） token: 第1回で取得したOAuth 2.0 Bearer Token Returns: {"status_code": int, "sku": str} 201 = 新規作成成功 / 204 = 既存SKUへの更新成功 """ url = f"{EBAY_INVENTORY_BASE}/inventory_item/{sku}" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Content-Language": "en-US", # 日本向けでも en-US が基本 } payload = { "availability": { "shipToLocationAvailability": { "quantity": 10 } }, "condition": "NEW", "product": { "title": "Classic Denim Jacket - Size M", "description": "<p>A classic vintage denim jacket.</p>", "aspects": { "Brand": ["Levi's"], "Size": ["M"], "Color": ["Blue"] }, "imageUrls": [ "https://i.ebayimg.com/images/g/example/jacket_main.jpg" ] } } resp = requests.put(url, headers=headers, json=payload) resp.raise_for_status() return {"status_code": resp.status_code, "sku": sku} if __name__ == "__main__": TOKEN = "v^1.1#i^1#..." # 第1回のOAuthフローで取得 SKU = "JACKET-DENIM-M-001" result = create_inventory_item(SKU, TOKEN) print(f"HTTP {result['status_code']} - SKU: {result['sku']}") # 201: 新規作成成功（レスポンスボディなし） # 204: 既存SKUへの更新成功（レスポンスボディなし） 補足: SKU 中心設計と ItemID 中心設計の違い Trading API では、出品後に eBay が発行する ItemID がすべての操作の起点でした（ReviseFixedPriceItem・GetItem・EndFixedPriceItem、いずれも ItemID 指定）。一方 Inventory API は、出品者が事前に定義した SKU がすべての操作の起点になります。「在庫は自分のシステムで管理するもの」というパラダイムシフトが根底にあります。SKU は最大50文字で、自分の全在庫を通じてユニークである必要があります。例えばeBayとAmazonで同じ商品を扱う場合、"EBAY-JP-JACKET-001" のようにプラットフォームプレフィックスを付ける設計も実務では有効です。 実務で躓く場面・深いポイント (Core) Inventory API は Trading API と比べてシンプルに見えますが、その設計の根本にある「分離の思想」と仕様上のいくつかの落とし穴を理解しないまま実装を進めると、後で深刻な問題を引き起こします。経験者が必ずぶつかる3大トラップを解説します。 1. Inventory Itemを作っただけでは、商品はまだeBayに出品されていない これが Inventory API を初めて使う Trading API 経験者が最初に混乱するポイントです。Trading API の AddFixedPriceItem（第4回）は「1回のAPIコールで出品完了」するモノリシックな設計でした。しかし Inventory API は責務を分離した設計になっており、eBayサイト上に商品を公開するまでに最低3つのステップが必要です。 ステップ1 — createOrReplaceInventoryItem（本記事の内容）: 「この商品はどんなものか・在庫は何個か」というマスターデータをeBayのシステムに登録する。 ステップ2 — createOffer（次回 #25 の内容）: 「この商品をどのマーケットプレイスに・いくらで・どのポリシーで売るか」というオファー条件を設定する。 ステップ3 — publishOffer（次回 #25 の内容）: オファーを実際に公開し、eBayのサイト上に商品が表示される状態にする。 本記事では Inventory Item の作成（ステップ1）のみを扱います。ステップ1が完了した後、eBay のセラーハブ（Seller Hub）の「Inventory」タブでは商品が登録済みとして確認できますが、「Active Listings」タブには表示されません。これは正常な動作です。あわてて同じ SKU で AddFixedPriceItem を呼び出すと、後述する「非互換性の罠」に陥ります。 2. 「完全上書き」の仕様を知らずに既存データを消してしまう事故 Inventory API の PUT /inventory_item/{sku} はべき等（Idempotent）なメソッドですが、「部分更新」ではなく「完全上書き（Full Replace）」です。送信したペイロードの内容がそのままレコードに保存され、送らなかったフィールドは削除されます。 例えば、最初に product.upc（JANコード）を含めて Inventory Item を登録しておき、後から在庫数（availability.shipToLocationAvailability.quantity）だけを変更しようとして UPC を含まないペイロードを送ると、UPC の情報がまるごと消えてしまいます。eBay の Product Catalog とのマッチング精度が落ち、検索順位への影響も懸念されます。 この事故を防ぐベストプラクティスは、更新の前に必ず GET /inventory_item/{sku} で現在の完全なデータを取得し、変更したい箇所だけをマージした上で PUT を送ることです。後述の「堅牢な実装」セクションの safe_upsert() がこのパターンを実装しています。なお eBay 公式ドキュメントもこの「取得してからマージして更新」のアプローチをベストプラクティスとして明記しています。 注意: 1日250回の修正上限（Daily Modification Cap） Inventory Item の作成・更新には、1日あたり最大250回というレート制限が設けられています。この上限はセラーアカウント単位で適用され、上限に達するとその日はそれ以上の更新がブロックされます。数千件の SKU を持つセラーが一斉に全商品の在庫を更新しようとすると、この上限に引っかかることがあります。 大量更新が必要な場合は、bulkCreateOrReplaceInventoryItem（POST /inventory_item/bulk_create_or_replace）を活用してください。1回のAPIコールで最大25件のInventory Itemを同時に作成・更新できるため、リクエスト消費を大幅に削減できます。また優先度ロジック（在庫残数が少ないものを先に更新するなど）を実装し、重要なアイテムから処理する設計を推奨します。 3. Inventory APIとTrading APIの出品は「別の世界」：混在運用が招くカオス これが移行期に最も危険なトラップです。Inventory API で createOrReplaceInventoryItem → createOffer → publishOffer の手順で作成した出品は、Trading API の ReviseFixedPriceItem・RelistFixedPriceItem・EndFixedPriceItem では操作できません。逆に Trading API の AddFixedPriceItem で作成した出品は、Inventory API の updateOffer・endOffer では操作できません。 この非互換性を知らずに、「既存の Trading API 出品は Trading API で管理しながら、新商品だけ Inventory API で登録する」という並行運用を安易に始めると、数ヶ月後に「どの SKU がどちらのAPIで管理されているか把握できない」という管理の破綻を招きます。ReviseFixedPriceItem を呼んだら 404 が返ってきて初めて気づく、という状況は実務では致命的です。 移行の原則として、1つの SKU は必ずどちらか一方のAPI系統で管理するという鉄則を徹底してください。既存商品を Inventory API へ移行するときは、まず Trading API 側で EndFixedPriceItem を呼び出して出品を終了させてから、Inventory API で新規登録するのが安全な手順です。 頻出エラーコード早見表 HTTPステータス / エラー 主な原因 対処法 400 / 25002 SKU が50文字超、または空文字・禁止文字を含む SKU を英数字・ハイフン・アンダースコアのみ・50文字以内に整形する 400 / 25003 condition の値が ConditionEnum に存在しない（例: "Used" をそのまま送った） CONDITION_MAP で Trading API 値を "USED_EXCELLENT" 等のInventory API値に変換する 400 / 25004 product.title が80文字を超えている title[:80] でトリミングしてから送る（Python スライス） 401 / Unauthorized Bearer Token の有効期限切れ（有効期限は2時間） 第1回実装の eBayTokenManager.get_valid_token() で自動リフレッシュする 404 / 25013 GET 時に指定した SKU の Inventory Item が存在しない get_inventory_item() で None チェックし、新規作成フローへ分岐する 堅牢な実装：Trading APIデータ構造からInventory Item形式への安全な移行アダプター ここでは、Trading API（AddFixedPriceItem）時代のデータ構造を Inventory API のペイロードに変換するアダプター関数と、「完全上書き」問題を回避する safe_upsert() メソッドを実装します。型アノテーション・docstring・入力バリデーション・ログ出力を含む生産レベルの実装です。 TradingItemData は第3〜17回の Trading API 実装で使ってきたデータモデルの典型例を dataclass で表現したものです。adapt_trading_to_inventory_item() がアダプター関数で、TradingItemData を Inventory API のペイロード dict に変換します。InventoryItemManager クラスの safe_upsert() では、更新前に既存データを GET で取得し、ディープマージしてから PUT を送ることで「完全上書き」による意図しないフィールド消失を防ぎます。 # inventory_manager.py from __future__ import annotations import logging from dataclasses import dataclass from typing import Any import requests logger = logging.getLogger(__name__) EBAY_INVENTORY_BASE = "https://api.ebay.com/sell/inventory/v1" # Trading API コンディション文字列 → Inventory API ConditionEnum マッピング CONDITION_MAP: dict[str, str] = { "New": "NEW", "Used": "USED_EXCELLENT", "Very Good": "USED_VERY_GOOD", "Good": "USED_GOOD", "Acceptable": "USED_ACCEPTABLE", "For parts or not working": "FOR_PARTS_OR_NOT_WORKING", } @dataclass class TradingItemData: """Trading API（AddFixedPriceItem）時代のデータ構造を表すデータクラス。 第3〜17回のTrading API実装で使ってきたデータモデルを想定。 """ sku: str title: str description: str condition: str # Trading API 形式 (例: "New") quantity: int brand: str image_urls: list[str] aspects: dict[str, list[str]] | None = None upc: str | None = None ean: str | None = None def adapt_trading_to_inventory_item(data: TradingItemData) -> dict[str, Any]: """Trading API データ構造を Inventory API ペイロードに変換するアダプター。 Args: data: Trading API 形式のアイテムデータ Returns: createOrReplaceInventoryItem に渡すペイロード dict Raises: ValueError: condition が CONDITION_MAP に存在しない場合 """ condition = CONDITION_MAP.get(data.condition) if condition is None: raise ValueError( f"未対応のコンディション: '{data.condition}'. " f"対応値: {list(CONDITION_MAP.keys())}" ) aspects: dict[str, list[str]] = data.aspects or {"Brand": [data.brand]} if "Brand" not in aspects: aspects["Brand"] = [data.brand] product: dict[str, Any] = { "title": data.title[:80], # eBay タイトル最大80文字 "description": data.description, "aspects": aspects, "imageUrls": data.image_urls[:12], # 最大12枚 } if data.upc: product["upc"] = data.upc if data.ean: product["ean"] = data.ean return { "availability": { "shipToLocationAvailability": {"quantity": data.quantity} }, "condition": condition, "product": product, } def _deep_merge(base: dict, override: dict) -> dict: """辞書を再帰的にマージする（override 側の値が優先される）。""" result = dict(base) for key, val in override.items(): if (key in result and isinstance(result[key], dict) and isinstance(val, dict)): result[key] = _deep_merge(result[key], val) else: result[key] = val return result class InventoryItemManager: """Inventory API の /inventory_item を安全に操作するクライアント。""" def __init__(self, token: str) -> None: self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Content-Language": "en-US", }) def get_inventory_item(self, sku: str) -> dict[str, Any] | None: """既存のInventory Itemを取得する。存在しない場合はNoneを返す。 Args: sku: 取得対象のSKU Returns: Inventory Item の dict、存在しない場合は None """ url = f"{EBAY_INVENTORY_BASE}/inventory_item/{sku}" resp = self._session.get(url) if resp.status_code == 404: return None resp.raise_for_status() return resp.json() def safe_upsert( self, sku: str, new_payload: dict[str, Any], ) -> dict[str, Any]: """「完全上書き」仕様に対応した安全な Inventory Item 作成・更新。 既存レコードを取得してディープマージしてから PUT を行うことで、 意図せずフィールドを消してしまう事故を防ぐ。 Args: sku: 作成・更新対象のSKU（最大50文字、全在庫でユニーク） new_payload: adapt_trading_to_inventory_item() で生成したペイロード Returns: {"sku": str, "created": bool, "status_code": int} Raises: ValueError: SKU が50文字を超える場合 requests.HTTPError: API がエラーを返した場合 """ if len(sku) > 50: raise ValueError(f"SKU は50文字以内にしてください: '{sku}'") # 完全上書き対策: 先に既存データを取得してマージする existing = self.get_inventory_item(sku) if existing is not None: logger.info("SKU '%s' 既存データ取得 → マージして更新します", sku) payload = _deep_merge(existing, new_payload) else: logger.info("SKU '%s' → 新規作成します", sku) payload = new_payload url = f"{EBAY_INVENTORY_BASE}/inventory_item/{sku}" resp = self._session.put(url, json=payload) resp.raise_for_status() is_created = (resp.status_code == 201) logger.info( "SKU '%s': %s 完了 (HTTP %d)", sku, "作成" if is_created else "更新", resp.status_code, ) return {"sku": sku, "created": is_created, "status_code": resp.status_code} # ── 使用例 ──────────────────────────────────────────────────────────────── if __name__ == "__main__": from ebay_token_manager import eBayTokenManager # 第1回実装済みクラス tm = eBayTokenManager() token = tm.get_valid_token() manager = InventoryItemManager(token) # Trading API 時代のデータ（移行元） old_data = TradingItemData( sku="JACKET-DENIM-M-001", title="Classic Denim Jacket Size M", description="<p>Vintage denim jacket in mint condition.</p>", condition="New", quantity=5, brand="Levi's", image_urls=["https://i.ebayimg.com/images/g/example/jacket.jpg"], aspects={"Brand": ["Levi's"], "Size": ["M"], "Color": ["Blue"]}, upc="012345678901", ) # アダプターで変換 → 安全にupsert payload = adapt_trading_to_inventory_item(old_data) result = manager.safe_upsert(old_data.sku, payload) print(f"完了: SKU={result['sku']}, 新規作成={result['created']}") 補足: eBayTokenManager との連携とセッション再利用 safe_upsert() 内の GET → PUT の2ステップの間にトークンが失効することは稀ですが、長時間バッチ処理を行う場合はセッション開始時に get_valid_token() でリフレッシュしておくことを推奨します。requests.Session() を使い回すことで TCP 接続のオーバーヘッドを削減し、大量の SKU を処理するバッチのスループットを向上させています。1,000件を超えるバッチ処理では、セッションの再利用だけで処理時間が20〜30%改善するケースもあります。 パフォーマンス・スケーリング視点 (深度) Trading APIとInventory APIの並行運用戦略と段階的移行ロードマップ 数千〜数万件の商品を扱う実務では、一夜にして全商品を Inventory API へ移行するのは現実的ではありません。稼働中のシステムを止めずに段階的に移行するための3フェーズ戦略を解説します。 【フェーズ1: 棚卸しと分類（〜2週間）】まず、現在 Trading API で管理している全 SKU を棚卸しします。「アクティブな出品（現在売れているもの）」「長期間売れていない在庫」「新規追加予定の商品」の3カテゴリに分類します。このフェーズではAPIの変更はまだ行いません。最重要の成果物は、各 SKU に対して api_type（"trading" or "inventory"）と ebay_item_id を記録する管理テーブルです。このテーブルが後のフェーズで生命線になります。 【フェーズ2: 新規商品からInventory APIを適用（〜1ヶ月）】新しく出品する商品はすべて Inventory API（createOrReplaceInventoryItem → createOffer → publishOffer）で登録します。既存の Trading API 出品はそのまま維持します。このフェーズで管理テーブルへの記録ロジックを実装し、「どの SKU がどちらのAPI管轄か」を常に把握できる状態にします。 【フェーズ3: 既存商品の段階的移行（1〜3ヶ月）】優先度の低い商品（長期在庫・売上の少ないもの）から順に移行します。各商品の移行手順は: (1) Trading API 側で EndFixedPriceItem を呼び出して出品を終了 → (2) eBay のシステムが終了を処理するまで数分待機 → (3) adapt_trading_to_inventory_item() でデータ変換 → (4) safe_upsert() で Inventory Item 登録 → (5) createOffer → publishOffer（次回 #25 で実装）で再出品、の順です。1件ずつ完結させ、管理テーブルを更新してから次の商品に進みます。 1日250回の修正上限との兼ね合いも重要です。フェーズ3で大量移行を行う場合、1日に何件移行できるかをあらかじめ計算し、上限の8割程度（200件/日）を目安に処理件数を制限するロジックを実装してください。上限に達した場合は処理を翌日に持ち越すキューイング機構が必要です。 並行運用期間中の最大のリスクは「APIの取り違え」です。管理テーブルを参照せずにコードをハードコードすると、Trading API 管理の商品に対して Inventory API の更新を呼んで 404 が返り、そのエラーを握りつぶした結果として在庫が同期されないという「サイレント障害」が発生します。管理テーブルのルックアップは必ずAPIコールの前に行い、型チェック（api_type のバリデーション）も実装してください。 まとめ 本記事では、連載の新フェーズ【Sell REST - 出品】の幕開けとして、Trading API から Inventory API への移行の第一歩を実装しました。 ベースライン: OAuth Bearer Token を使い PUT /inventory_item/{sku}（createOrReplaceInventoryItem）を呼び出す最小構成を確認し、SKU 中心設計という Inventory API の根本的なパラダイムを理解しました。 深いポイント: Inventory Item を作成しただけではまだ出品されない（Offer が必要な3ステップ設計）、「完全上書き」仕様による意図しないデータ消失事故、1日250回の修正上限と bulk エンドポイントによる効率化、そして Inventory API と Trading API 間の出品非互換性という4つの重要なトラップとその回避策を解説しました。 スケーリング: 数万件規模の既存 Trading API 商品を、サービスを止めずに Inventory API へ移行するための3フェーズ戦略と、並行運用期間中の管理テーブルによる SKU 追跡の重要性を解説しました。 第1〜23回で積み上げたエンジニアリング思想（冪等性・バリデーションファースト・エラーハンドリング）は、Inventory API の実装においても変わらず重要です。むしろそれらの思想がベースとなることで、モダンな REST API への移行がスムーズに進みます。 次のステップ 次回（#25）では、本記事で登録した Inventory Item に対して createOffer と publishOffer を呼び出し、ついに eBay のサイト上に商品を公開する手順を解説します。価格・配送ポリシー・返品ポリシーの設定方法と、Offer のステータス管理（PUBLISHED / UNPUBLISHED）の詳細についても深く掘り下げます。お楽しみに！

前回の記事はこちら 【連載#23】eBay Message API：bulkUpdateConversationで大量の会話ステータスを一括管理する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第23回です。 第21回（getConversationsとgetConversationでバイヤーとのメッセージを取得する）ではMessage API の「読み取り」基盤を、第22回（sendMessageでバイヤーへの自動返信ボットを作る）では「送信」機能をそれぞれ実装しました。本記事は Message API シリーズ（第21〜23回）の総仕上げとして、会話の「ステータス管理」——すなわち既読化・アーカイブ・削除を一括で行う機能を扱います。 自動返信ボットを本番稼働させると、ほどなく別の問題が浮上してきます。それは「対応済みの会話が受信箱に残り続け、まだ未対応の会話が埋もれてしまう」という問題です。件数が少ないうちは手動で既読にできますが、毎日数百件のバイヤーメッセージを処理するスケールになると、ステータス管理そのものが業務ボトルネックになります。本記事では、この課題を解決する POST /bulk_update_conversation エンドポイントを実務レベルで徹底解説します。 この記事で得られること: POST /update_conversation（単件）と POST /bulk_update_conversation（最大10件）の両エンドポイントの仕様を理解し、conversationStatus と read フィールドの排他的挙動という重要な罠を正確に把握する。 大量の会話 ID リストを 10 件ずつチャンク分割して bulkUpdateConversation を呼び出し、各 conversationId の updateStatus を検証して失敗分を再試行キューに積む、本番稼働レベルの Python 実装を手に入れる。 バルク操作の部分成功（partial success）への対処法、定期バッチジョブ化の設計指針、および失敗分の監視ダッシュボード設計など、スケーリング視点の実務ノウハウを習得する。 背景・なぜこれが重要か (Motivation) 「1件ずつ updateConversation を呼べばいいのでは？　どうせ Python の for ループで回せば大量件数も処理できるでしょう？」 これは Message API を使い始めたばかりの開発者が最初に抱く疑問です。技術的には確かに可能です。しかし、この「ループで単件 API を連打する」アプローチは、実務スケールではすぐに限界を露わにします。 まず API コール数の問題です。仮に 1 日に 500 件の会話を対応済みにしたい場合、単件 API を使うと 500 回の HTTP リクエストが発生します。eBay Message API には 1 日あたりのコール数に上限が設定されており、単純なループは貴重なレート制限クォータを浪費します。一方 bulkUpdateConversation は 1 リクエストで最大 10 件を処理できるため、同じ 500 件なら 50 回のリクエストで済みます。コール数を 90% 削減できる計算です。 次にレイテンシの問題です。Python の requests ライブラリで同期的に 500 件のリクエストを直列実行すると、1 リクエストあたり 200〜500 ms のレイテンシがあるため、合計で 100 秒〜250 秒の処理時間が必要になります。これはバッチジョブの許容実行時間を大幅に超えてしまいます。バルク API を使えばリクエスト回数が 1/10 に減り、処理時間も劇的に短縮されます。 さらに重要なのが「エラー管理の複雑さ」です。単件ループでは、途中のリクエストがエラーになった場合に「どこまで成功したか」を自前で追跡しなければなりません。bulkUpdateConversation はレスポンスに各 conversationId ごとの updateStatus を返すため、バッチ単位での成功・失敗の把握が構造化された形で得られます。この「レスポンスを見れば partial success のどれが失敗したか即座にわかる」という設計が、堅牢な再試行ロジックの実装を容易にします。 基本的な使い方（ベースライン）：updateConversationとbulkUpdateConversationの使い分け まず、単件更新の updateConversation と一括更新の bulkUpdateConversation の両方を最小限のコードで確認しましょう。Base URL は getConversations・sendMessage と同様に https://api.ebay.com/commerce/message/v1 です。OAuth 2.0 のアクセストークンを取得済みであることを前提とします（認証フローは第1回を参照してください）。 # message_update_baseline.py import requests from typing import Literal BASE_URL = "https://api.ebay.com/commerce/message/v1" ConversationType = Literal["FROM_MEMBERS", "FROM_EBAY"] ConvStatus = Literal["ACTIVE", "ARCHIVE", "DELETE"] BulkConvStatus = Literal["ACTIVE", "ARCHIVE", "DELETE", "READ", "UNREAD"] # ── 単件更新 ────────────────────────────────────────────────── def update_conversation( access_token: str, conversation_id: str, conversation_type: ConversationType, *, conversation_status: ConvStatus | None = None, read: bool | None = None, ) -> dict: """ POST /update_conversation で1件の会話ステータスを更新する。 重要: conversation_status と read を同時に指定すると read のみが適用され conversation_status は無視される。 """ url = f"{BASE_URL}/update_conversation" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP", } body: dict = { "conversationId": conversation_id, "conversationType": conversation_type, } if conversation_status is not None: body["conversationStatus"] = conversation_status if read is not None: body["read"] = read response = requests.post(url, json=body, headers=headers, timeout=30) response.raise_for_status() return response.json() if response.content else {} # ── 一括更新（最大10件）──────────────────────────────────────── def bulk_update_conversation( access_token: str, conversations: list[dict], ) -> dict: """ POST /bulk_update_conversation で最大10件を一括更新する。 conversations の各要素は: { "conversationId": str, "conversationType": "FROM_MEMBERS" | "FROM_EBAY", "conversationStatus": "ACTIVE" | "ARCHIVE" | "DELETE" | "READ" | "UNREAD" } """ if len(conversations) > 10: raise ValueError(f"bulk APIの上限は10件。実際の件数: {len(conversations)}") url = f"{BASE_URL}/bulk_update_conversation" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP", } body = {"conversations": conversations} response = requests.post(url, json=body, headers=headers, timeout=30) response.raise_for_status() return response.json() # ── 動作確認 ────────────────────────────────────────────────── if __name__ == "__main__": TOKEN = "v^1.1#i^1#..." # ← OAuthトークン # 単件: 会話を既読にする result_single = update_conversation( TOKEN, conversation_id="c_001", conversation_type="FROM_MEMBERS", read=True, ) print("単件更新:", result_single) # 一括: 3件の会話を一度にアーカイブする batch = [ {"conversationId": "c_101", "conversationType": "FROM_MEMBERS", "conversationStatus": "ARCHIVE"}, {"conversationId": "c_102", "conversationType": "FROM_MEMBERS", "conversationStatus": "ARCHIVE"}, {"conversationId": "c_103", "conversationType": "FROM_MEMBERS", "conversationStatus": "ARCHIVE"}, ] result_bulk = bulk_update_conversation(TOKEN, batch) for item in result_bulk.get("conversations", []): print(f" [{item['conversationId']}] → {item.get('updateStatus', '?')}") 補足1: conversationStatus と read の排他的挙動 POST /update_conversation（単件 API）のリクエストボディにはconversationStatus と read の 2 つのステータス制御フィールドがあります。ここで注意すべき重要な仕様があります——この 2 つを同時にリクエストボディに含めると、read のみが適用され、conversationStatus は黙って無視されます。エラーは返りません。つまり「両方指定したつもりが片方しか効いていない」状態に気づかないまま運用してしまうリスクがあります。コードレビューの際は、単一のリクエストで両フィールドを同時指定していないか必ず確認してください。 補足2: conversationType は「更新不可だが必須」という罠 conversationType は更新対象の会話に元々設定されている会話タイプをそのまま指定する必須フィールドです。「この会話を FROM_MEMBERS から FROM_EBAY に変更したい」という操作はできません。conversationType は会話の属性を変更するためではなく、「どの会話を操作するか」を識別するためのキーとして機能します。バイヤーとのメッセージは常に FROM_MEMBERS、eBay からの公式通知は常に FROM_EBAY を指定してください。 補足3: bulkUpdateConversation の conversationStatus 拡張 単件 API の conversationStatus が ACTIVE / ARCHIVE / DELETE の 3 値であるのに対し、バルク API の conversationStatus は READ と UNREAD も追加した 5 値をサポートしています。この非対称な仕様に注意してください。バルク API で大量の会話を一括既読にする場合は conversationStatus=READ を使い、単件 API で同じことをしたい場合は read=True を使うというように、エンドポイントごとにフィールドの使い方が異なります。 実務で躓く場面・深いポイント (Core) ベースライン実装で基本的な動作確認ができたら、次は本番運用で必ず直面する実務上の落とし穴とその解決策を見ていきましょう。 1. bulk API の上限は10件——大量データはチャンク分割が必須 bulkUpdateConversation は 1 リクエストあたり最大 10 件までしか処理できません。11 件以上を一度に送ると API はエラーを返します。実務では対応済み会話が数百件単位で溜まることが珍しくないため、リストを 10 件ずつのチャンク（塊）に分割してから繰り返し API を呼ぶ処理が必須になります。 Python でのチャンク分割は itertools.islice や単純なスライスで実装できますが、本番コードでは以下のようなジェネレータ関数として切り出しておくとテストが書きやすくなります。 # chunk_utils.py from typing import Generator, TypeVar T = TypeVar("T") def chunked(lst: list[T], size: int) -> Generator[list[T], None, None]: """リストを指定サイズのチャンクに分割するジェネレータ""" for i in range(0, len(lst), size): yield lst[i : i + size] # 使用例: 47件のconversationIdリストを10件ずつ処理する BULK_API_LIMIT = 10 conv_ids = [f"c_{n:03d}" for n in range(47)] # 47件のサンプルID for chunk in chunked(conv_ids, BULK_API_LIMIT): print(f"このチャンクで処理する件数: {len(chunk)}") # → 10件 × 4回 + 7件 × 1回 = 合計5リクエストで47件を処理 47 件の場合、10 件 × 4 回 ＋ 7 件 × 1 回の計 5 回のリクエストで処理が完了します。単件 API を使った場合の 47 回と比べ、リクエスト数を約 90% 削減できます。大量件数になるほどこの差は顕著になるため、「どんな件数でも必ず bulkUpdateConversation を使う」方針を設計段階から徹底してください。 2. conversationStatus と read の同時指定——片方が黙って無視される罠 この仕様は実際に踏んだときのデバッグが非常に難しい罠です。単件 API（POST /update_conversation）で「会話をアーカイブしつつ既読にもしたい」と考え、conversationStatus=ARCHIVE と read=True を同時にリクエストボディに含めたとします。 # NG: conversationStatus と read を同時指定 body = { "conversationId": "c_001", "conversationType": "FROM_MEMBERS", "conversationStatus": "ARCHIVE", # ← この値は無視される！ "read": True, # ← こちらだけが適用される } # 結果: 会話は「既読」になるが「アーカイブ」には移動しない # しかもAPIはエラーを返さないため、バグに気づきにくい API はエラーを返さず HTTP 200 を返します。レスポンスボディが空（または updateStatus=SUCCESSFUL）であるため、「成功した」と誤認してしまいます。しかし実際には conversationStatus の変更が適用されておらず、会話はアーカイブされないまま受信箱に残り続けます。 正しい実装は「2 回に分けてリクエストを送る」か、「バルク API（bulkUpdateConversation）の conversationStatus=ARCHIVE を使ってアーカイブし、その後 conversationStatus=READ で既読にする」かのどちらかです。 # OK: 排他性を意識した2ステップ実装（単件APIの場合） # Step 1: 既読にする update_conversation(TOKEN, "c_001", "FROM_MEMBERS", read=True) # Step 2: アーカイブする（別リクエスト） update_conversation(TOKEN, "c_001", "FROM_MEMBERS", conversation_status="ARCHIVE") # OK: バルクAPIで conversationStatus=READ を使う（一括の場合） bulk_update_conversation(TOKEN, [ {"conversationId": "c_001", "conversationType": "FROM_MEMBERS", "conversationStatus": "READ"}, # バルクAPIではREADが使える ]) 「なぜこのような仕様なのか？」という疑問は自然ですが、eBay の公式仕様書にこの挙動は明記されており、ドキュメントを熟読せずに直感で実装すると必ずはまります。コードレビューのチェックリストに「単件 API で conversationStatus と read を同時指定していないか」の項目を追加しておくことを強くお勧めします。 3. Partial success（一部成功・一部失敗）の検出とハンドリング bulkUpdateConversation の重要な特性として、「バッチ内の一部が成功し、一部が失敗する」partial success（部分成功）があります。単件 API であれば HTTP ステータスコード（200 / 4xx / 5xx）だけで成否を判断できますが、バルク API では HTTP 200 が返っても個々の conversationId のレスポンス内に失敗情報が埋め込まれている場合があります。 # バルクAPIのレスポンス例（一部失敗のケース） response_body = { "conversations": [ { "conversationId": "c_101", "updateStatus": "SUCCESSFUL" }, { "conversationId": "c_102", "updateStatus": "FAILED", # ← 失敗 "errors": [ { "errorId": 850050, "domain": "API_MESSAGE", "category": "REQUEST", "message": "Conversation not found or access denied." } ] }, { "conversationId": "c_103", "updateStatus": "SUCCESSFUL" } ] } # ── Partial success の検出ロジック ──────────────────────────── def extract_failed_ids(bulk_response: dict) -> list[str]: """バルク更新レスポンスから失敗したconversationIdを抽出する""" failed = [] for item in bulk_response.get("conversations", []): if item.get("updateStatus") != "SUCCESSFUL": failed.append(item["conversationId"]) errors = item.get("errors", []) for err in errors: print(f" [FAILED] {item['conversationId']} " f"errorId={err.get('errorId')} " f"msg={err.get('message')}") return failed failed_ids = extract_failed_ids(response_body) print(f"失敗件数: {len(failed_ids)}") # → 失敗件数: 1 HTTP レスポンスが 200 であっても updateStatus が FAILED の要素が含まれている場合、それは「API 呼び出し自体は成功したが、個別処理が失敗した」ことを意味します。レスポンスの conversations 配列を必ずループして各要素の updateStatus を確認する処理は、バルク API を使う上で絶対に省いてはなりません。 よくある失敗原因として、「conversationId がすでに削除済み」「アクセス権限のない会話 ID」「conversationType の指定ミス」などが挙げられます。失敗した conversationId は再試行キューに積んで後で処理するか、アラートとして担当者に通知するフローを設計してください。 注意: bulkUpdateConversationに存在しないIDを混ぜると全件が影響を受ける可能性 バルク API のリクエストに 1 件でも不正な conversationId（存在しない・アクセス権限がない・すでに DELETE 済みなど）が含まれている場合、その件だけが FAILED になるのか、バッチ全体がエラーになるのかはエラーの種類によって異なります。特に認証・認可エラー（errorId: 850100 番台）の場合はバッチ全体が HTTP 403 で弾かれることがあります。本番導入前に Sandbox 環境で「不正 ID を意図的に混入させた場合の挙動」を必ずテストしてください。 また、bulkUpdateConversation は DELETE 操作をサポートしており、誤って大量の会話を削除した場合に復元する手段は現時点では提供されていません。DELETE ステータスの一括適用は、テスト環境での十分な検証なしに本番環境で実行しないでください。 頻出エラーコード早見表 Message API の bulkUpdateConversation / updateConversation で実際に遭遇しやすいエラーコードと対処法を以下にまとめます。 errorId カテゴリ メッセージ例 原因 対処 850001 REQUEST Invalid request. Required field 'conversationType' is missing. conversationType が未指定またはスペルミス。 FROM_MEMBERS または FROM_EBAY を必ず明示する。 850010 REQUEST The bulk request exceeds the maximum allowed limit of 10 conversations. bulk API に 11 件以上を送信した。 送信前に len(conversations) <= 10 を検証する。 850050 REQUEST Conversation not found or access denied. 存在しない・削除済み・別セラーの conversationId を指定した。 conversationId を getConversations で事前確認し、partial success ハンドラで失敗 ID を再試行キューから除外する。 850020 REQUEST Invalid value for 'conversationStatus'. Allowed values: ACTIVE, ARCHIVE, DELETE, READ, UNREAD. バルク API で使えない値、またはスペルミス。 bulkUpdateConversation の conversationStatus は ACTIVE / ARCHIVE / DELETE / READ / UNREAD の 5 値のみ。 堅牢な実装：ConversationBulkUpdater クラスによる大量ステータス管理 ここまで解説したすべての実務ポイント——10 件チャンク分割・排他性の検証・partial success のハンドリング・失敗分の再試行キュー——を組み込んだ、プロダクションレベルの完全実装を示します。型アノテーション・docstring・入力バリデーション・例外処理をすべて備えたConversationBulkUpdater クラスとして設計します。 # conversation_bulk_updater.py import logging import time from dataclasses import dataclass, field from typing import Generator, Literal import requests logger = logging.getLogger(__name__) BASE_URL = "https://api.ebay.com/commerce/message/v1" BULK_LIMIT = 10 # bulkUpdateConversation の上限 MAX_RETRY_COUNT = 3 # 失敗時の最大再試行回数 RETRY_DELAY_SEC = 5.0 # 再試行間隔（秒） ConversationType = Literal["FROM_MEMBERS", "FROM_EBAY"] BulkStatus = Literal["ACTIVE", "ARCHIVE", "DELETE", "READ", "UNREAD"] @dataclass class BulkUpdateResult: """バルク更新の実行結果を保持するデータクラス""" total_requested: int = 0 total_successful: int = 0 total_failed: int = 0 failed_ids: list[str] = field(default_factory=list) retry_queue: list[str] = field(default_factory=list) class ConversationBulkUpdater: """ eBay bulkUpdateConversation を使い、大量の会話ステータスを 10 件チャンク分割・Partial success ハンドリング・再試行付きで一括更新するクラス。 Parameters ---------- access_token: OAuth 2.0 アクセストークン conversation_type: 処理対象の会話タイプ（FROM_MEMBERS または FROM_EBAY） target_status: 更新後のステータス（READ / ARCHIVE / ACTIVE 等） """ def __init__( self, access_token: str, conversation_type: ConversationType, target_status: BulkStatus, ) -> None: if not access_token: raise ValueError("access_token は空にできません") if conversation_type not in ("FROM_MEMBERS", "FROM_EBAY"): raise ValueError(f"不正な conversationType: {conversation_type}") if target_status not in ("ACTIVE", "ARCHIVE", "DELETE", "READ", "UNREAD"): raise ValueError(f"不正な target_status: {target_status}") self._token = access_token self._conv_type = conversation_type self._target_status = target_status self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {self._token}", "Content-Type": "application/json", "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP", }) # ── チャンク分割ジェネレータ ──────────────────────────────────── @staticmethod def _chunked( lst: list[str], size: int ) -> Generator[list[str], None, None]: for i in range(0, len(lst), size): yield lst[i : i + size] # ── 単チャンクのバルク呼び出し ───────────────────────────────── def _call_bulk_api(self, chunk: list[str]) -> dict: """ bulkUpdateConversation を 1 チャンク（最大10件）分呼び出す。 HTTP 4xx / 5xx は requests.HTTPError として raise する。 """ body = { "conversations": [ { "conversationId": cid, "conversationType": self._conv_type, "conversationStatus": self._target_status, } for cid in chunk ] } resp = self._session.post( f"{BASE_URL}/bulk_update_conversation", json=body, timeout=30, ) resp.raise_for_status() return resp.json() # ── partial success の解析 ───────────────────────────────────── @staticmethod def _parse_partial_success(response: dict) -> tuple[list[str], list[str]]: """ バルク API レスポンスを解析し (successful_ids, failed_ids) を返す。 """ successful, failed = [], [] for item in response.get("conversations", []): cid = item.get("conversationId", "") if item.get("updateStatus") == "SUCCESSFUL": successful.append(cid) else: failed.append(cid) for err in item.get("errors", []): logger.warning( "bulk update failed: conversationId=%s errorId=%s message=%s", cid, err.get("errorId"), err.get("message"), ) return successful, failed # ── メインの一括更新メソッド ─────────────────────────────────── def update_all( self, conversation_ids: list[str], *, chunk_delay_sec: float = 0.5, ) -> BulkUpdateResult: """ conversation_ids リスト全件を 10 件チャンクで bulkUpdateConversation に送り、 各チャンクの partial success を検証し、失敗分を再試行キューに積む。 Parameters ---------- conversation_ids: 更新対象の conversationId のリスト（件数上限なし） chunk_delay_sec: チャンク間の待機時間（レート制限対策、デフォルト0.5秒） Returns ------- BulkUpdateResult: 処理結果の集計 """ if not conversation_ids: logger.warning("conversation_ids が空です。処理をスキップします。") return BulkUpdateResult() result = BulkUpdateResult(total_requested=len(conversation_ids)) chunks = list(self._chunked(conversation_ids, BULK_LIMIT)) logger.info( "bulkUpdateConversation 開始: 合計 %d 件 / %d チャンク", result.total_requested, len(chunks), ) for idx, chunk in enumerate(chunks, start=1): logger.debug("チャンク %d/%d を処理中 (%d件)", idx, len(chunks), len(chunk)) try: response = self._call_bulk_api(chunk) ok_ids, ng_ids = self._parse_partial_success(response) result.total_successful += len(ok_ids) result.total_failed += len(ng_ids) result.failed_ids.extend(ng_ids) except requests.HTTPError as exc: # チャンク全体が HTTP エラーになった場合（認証エラー等） logger.error( "チャンク %d で HTTP エラー: %s — chunk=%s", idx, exc, chunk ) result.total_failed += len(chunk) result.failed_ids.extend(chunk) # チャンク間のレート制限対策 if idx < len(chunks): time.sleep(chunk_delay_sec) result.retry_queue = result.failed_ids.copy() logger.info( "bulkUpdateConversation 完了: 成功=%d 失敗=%d", result.total_successful, result.total_failed, ) return result # ── 再試行メソッド ───────────────────────────────────────────── def retry_failed( self, result: BulkUpdateResult, max_retries: int = MAX_RETRY_COUNT, ) -> BulkUpdateResult: """ BulkUpdateResult.retry_queue に積まれた失敗分を再試行する。 Exponential backoff で最大 max_retries 回まで再試行する。 """ retry_ids = result.retry_queue.copy() attempt = 0 while retry_ids and attempt < max_retries: attempt += 1 wait = RETRY_DELAY_SEC * (2 ** (attempt - 1)) logger.info( "再試行 %d/%d: 対象=%d件 待機=%gs", attempt, max_retries, len(retry_ids), wait ) time.sleep(wait) sub_result = self.update_all(retry_ids, chunk_delay_sec=1.0) # 成功分を集計に反映 result.total_successful += sub_result.total_successful result.total_failed -= sub_result.total_successful retry_ids = sub_result.failed_ids # 再度失敗した分のみ残す result.retry_queue = retry_ids # 最終的に残った失敗分 return result # ── 使用例 ────────────────────────────────────────────────────── if __name__ == "__main__": import os logging.basicConfig(level=logging.INFO) TOKEN = os.environ["EBAY_ACCESS_TOKEN"] # 環境変数から取得 # 200件の対応済み会話をすべてアーカイブする例 all_ids = [f"c_{n:04d}" for n in range(200)] # 実際はDBやAPIから取得 updater = ConversationBulkUpdater( access_token=TOKEN, conversation_type="FROM_MEMBERS", target_status="ARCHIVE", ) result = updater.update_all(all_ids, chunk_delay_sec=0.5) print(f"成功: {result.total_successful} / 失敗: {result.total_failed}") if result.retry_queue: print(f"再試行キュー: {len(result.retry_queue)} 件") result = updater.retry_failed(result, max_retries=3) print(f"再試行後の残失敗: {len(result.retry_queue)} 件") このクラスのポイントは、update_all メソッドが「何件でも受け付ける」一方で、内部では必ず BULK_LIMIT（10）件ずつのチャンクに分割して API を呼び出す点です。呼び出し元は「件数の上限を気にせず conversationId のリストを渡すだけ」でよく、チャンク分割の複雑さがカプセル化されています。 retry_failed メソッドは Exponential backoff（指数バックオフ）を採用しており、1 回目の再試行は 5 秒後、2 回目は 10 秒後、3 回目は 20 秒後に実行されます。eBay サーバー側の一時的な障害や過負荷が原因の失敗は、この方式でほとんどの場合に自動回復します。3 回再試行しても失敗が残った場合は、result.retry_queue に残留する conversationId がログ・アラート・DB への記録などの後続処理に引き渡されます。 OAuth トークンの安全な管理と有効期限への対応 OAuth トークンは必ず環境変数（EBAY_ACCESS_TOKEN）から読み込んでください。ソースコードに直接ハードコーディングすると、Git リポジトリへの誤コミットによる認証情報の漏洩リスクが生じます。トークンの有効期限は通常 2 時間です。長時間のバッチ処理では途中でトークンが失効することがあるため、401 エラーを検出してトークンを自動更新するリフレッシュロジックをsession のヘッダー更新と組み合わせて実装することを推奨します。 パフォーマンス・スケーリング視点 (深度) 定期バッチジョブ化と失敗分の監視ダッシュボード設計 自動返信ボット（第22回）と会話ステータス管理（本記事）の両方が実装できたら、次のステップはこれらを定期実行バッチジョブとして組み合わせることです。典型的なカスタマーサービス自動化パイプラインは次のような構成になります。 【フェーズ 1: 新着メッセージの取得（第21回）】 getConversations を差分同期（デルタ同期）で定期ポーリングし、前回チェック以降の新着 UNREAD 会話を取得する。実行間隔: 5〜10 分（JST 営業時間内）／ 30 分〜1 時間（営業時間外） 【フェーズ 2: 自動返信処理（第22回）】 キーワード分類エンジンで問い合わせ内容を判定し、対応可能なものは sendMessage で自動返信する。ネガティブメッセージ・クレームは担当者エスカレーションキューに追加する。 【フェーズ 3: ステータス一括更新（本記事）】 フェーズ 2 で自動返信が完了した conversationId を ConversationBulkUpdater で一括 ARCHIVE する。失敗した ID は retry_queue に記録し、次回バッチで再試行する。 # nightly_cs_batch.py —— 夜間バッチジョブの例 import os import logging from datetime import datetime, timezone, timedelta from message_client import EbayMessageClient from auto_reply_bot import AutoReplyBot from conversation_bulk_updater import ConversationBulkUpdater, BulkUpdateResult logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) logger = logging.getLogger("nightly_cs_batch") def run_cs_pipeline(access_token: str) -> None: """ カスタマーサービス自動化の3フェーズパイプラインを実行する。 """ since = datetime.now(timezone.utc) - timedelta(hours=6) # 過去6時間分 # ── Phase 1: 新着会話の取得 ────────────────────────────────── client = EbayMessageClient(access_token) new_convs = client.get_unread_since(since) # conversationId のリスト logger.info("Phase1: 新着会話 %d 件を取得", len(new_convs)) # ── Phase 2: 自動返信処理 ──────────────────────────────────── bot = AutoReplyBot(access_token) replied_ids = [] for conv in new_convs: success = bot.handle(conv) if success: replied_ids.append(conv["conversationId"]) logger.info("Phase2: 自動返信完了 %d 件 / スキップ（要人間対応）%d 件", len(replied_ids), len(new_convs) - len(replied_ids)) # ── Phase 3: 返信済み会話を一括アーカイブ ─────────────────── if not replied_ids: logger.info("Phase3: アーカイブ対象なし。スキップ。") return updater = ConversationBulkUpdater( access_token=access_token, conversation_type="FROM_MEMBERS", target_status="ARCHIVE", ) result: BulkUpdateResult = updater.update_all(replied_ids) if result.retry_queue: # 再試行後も失敗した場合はアラートを発行 logger.error( "Phase3: 最終的に %d 件のアーカイブに失敗。" "監視ダッシュボードに記録します: %s", len(result.retry_queue), result.retry_queue[:5], # 先頭5件のみログ出力 ) # TODO: Slack 通知 / DB への失敗ログ書き込み logger.info( "Phase3 完了: 成功=%d 失敗=%d", result.total_successful, result.total_failed, ) if __name__ == "__main__": TOKEN = os.environ["EBAY_ACCESS_TOKEN"] run_cs_pipeline(TOKEN) このパイプラインを cron（Linux/Mac）や Task Scheduler（Windows）、あるいは GitHub Actions の schedule トリガーで定期実行することで、「新着メッセージの検出 → 自動返信 → 対応済みをアーカイブ」というカスタマーサービスのコアフローが完全に自動化されます。 監視の観点では、result.retry_queue に残留した conversationId をPostgreSQL や BigQuery などのデータストアに書き込み、失敗件数・失敗率・連続失敗 ID などを Grafana や Metabase で可視化する監視ダッシュボードを構築することを推奨します。「毎日 X 件以上失敗が続いている」「特定の conversationId が何度再試行しても失敗する」といったアノマリーの早期発見が、システムの安定運用に直結します。 さらに大規模な運用（数千件 / 日）では、ConversationBulkUpdater の update_all をそのまま同期実行するのではなく、Celery や RQ（Redis Queue）を使ったタスクキューイングアーキテクチャへの移行を検討してください。各チャンク（10 件）を 1 タスクとして非同期にエンキューすることで、ワーカーを水平スケールさせながら API レート制限を守りつつ高スループットな処理が実現できます。 まとめ 本記事では、eBay Message API の会話ステータス一括管理エンドポイント——updateConversation（単件）と bulkUpdateConversation（最大10件一括）——を実務視点で徹底解説しました。 ベースライン: POST /update_conversation では conversationStatus と read の同時指定を避け、POST /bulk_update_conversation ではリクエストごとに最大 10 件の制限を守る。bulkUpdateConversation の conversationStatus は 5 値（ACTIVE / ARCHIVE / DELETE / READ / UNREAD）をサポートしており、単件 API とは仕様が異なる点に注意する。 深いポイント: 大量件数の処理には 10 件チャンク分割が必須。conversationStatus と read の排他的挙動（read のみが適用される）はエラーが返らないため特に注意が必要。バルク API では partial success（部分成功・部分失敗）が発生しうるため、レスポンスの conversations 配列を必ず走査して失敗 ID を再試行キューに積む実装が不可欠。 スケーリング: ConversationBulkUpdater クラスにチャンク分割・Exponential backoff・再試行キューをカプセル化し、3 フェーズパイプライン（取得 → 返信 → アーカイブ）として定期バッチジョブ化する。大規模運用では Celery / RQ によるタスクキューイングへの移行と、失敗件数の監視ダッシュボード構築が安定運用の鍵となる。 これで Message API シリーズ（第21〜23回）が完結しました。第21回で「読む」、第22回で「送る」、そして本記事で「管理する」——この3 つのエンドポイントを組み合わせることで、eBay のカスタマーサービス業務をエンドツーエンドで自動化する完全なシステムを構築できるようになりました。自動返信ボットと一括ステータス管理を実際の本番環境に繋ぎ込み、毎日の受信箱が常にクリーンな状態に保たれる体験を、ぜひ実感してください。 次のステップ 次回（#24）は、本シリーズにとって大きな転換点となる回です。これまで第1〜23回にわたって取り組んできたTrading API / Message API 系のフェーズを卒業し、いよいよ 【Sell REST API フェーズ】 へと突入します。 第24回のテーマは 「Trading APIからREST APIへ：OAuth移行と最初のInventory API呼び出し」です。eBay が次世代の出品管理基盤として推進する Inventory API（REST）の全体像と、Trading API の AddFixedPriceItem から移行する際の思想的・実装的な違いを解説します。Authentication / Authorization の仕組みが XML + AuthToken からOAuth 2.0 Bearer Token へと標準化されるこの移行は、API 設計の近代化において最も重要なステップのひとつです。Message API シリーズで OAuth 2.0 に慣れ親しんだ皆さんなら、この移行をスムーズに進められるはずです。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#22】eBay Message API：sendMessageでバイヤーへの自動返信ボットを作る はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第22回です。 前回（#21）は、getConversations と getConversation を使い、バイヤーからの受信メッセージを一覧取得・詳細取得する方法を解説しました。会話データの「読み込み」基盤が整ったところで、今回はその直接の続編として、POST /send_message エンドポイントを使った「バイヤーへの返信送信」を実装します。 eBay の販売規模が拡大してくると、「発送はいつですか？」「返品できますか？」「在庫はまだありますか？」といった定型的な問い合わせへの個別対応が、人的リソースを大きく圧迫し始めます。特に日本と海外バイヤーのタイムゾーン差によって営業時間外に届くメッセージへの対応は、購買意欲の冷え込みやネガティブフィードバックの温床になります。本記事では、このような課題をコードで解決する自動返信ボットを段階的に構築します。 この記事で得られること: POST /send_message エンドポイントの全パラメータ（conversationId・otherPartyUsername・messageText・emailCopyToSender・messageMedia・reference）の仕様を実際のコードで理解し、既存会話への返信機能の基盤を構築する。 第21回の getConversations と組み合わせた未返信会話の自動検出と、キーワード分類エンジンによる「賢い自動返信ボット」の設計思想と実装パターンを習得する。 二重送信防止・messageText の 2000 文字制限ハンドリング・営業時間外判定ロジック・ネガティブメッセージのエスカレーション処理など、本番稼働に直結する実務ポイントをすべてカバーした完全実装コードを手に入れる。 背景・なぜこれが重要か (Motivation) 「自動返信って、なんでも同じ文面を送ればいいんでしょ？　お問い合わせありがとうございます。担当者が確認し、24時間以内にご返信します、とだけ送っておけば十分では？」 これは多くの初学者が最初に抱く疑問であり、よく見られる実装パターンでもあります。確かに技術的には実現可能ですし、Response Rate（返信率）という指標だけを見れば数字は改善されます。しかし eBay のバイヤーエクスペリエンスという観点から見ると、この「一律返信」戦略は長期的に大きなリスクを孕んでいます。 例えば、バイヤーが「追跡番号を教えてください」と具体的に聞いているのに「担当者が確認します」とだけ返信された場合、バイヤーは「このセラーはボットで適当に返信しているだけだ」と感じ、購買後の不安が解消されません。一方、「ご購入ありがとうございます！商品は2営業日以内に発送し、追跡番号は eBay システム経由でお知らせします」という具体的な回答であれば、バイヤーの心理的な安心感は大きく向上します。つまり自動返信の「内容の質」こそが重要なのです。 自動返信ボットを設計する上で最も重要なのは、「何を自動化し、何を人間に任せるか」の境界線を明確に引くことです。発送予定や一般的な商品質問への回答は自動化に向いていますが、返品・クレーム・商品の欠陥報告などのネガティブな内容は、不適切な自動返信によって状況を悪化させる深刻なリスクがあります。これらは必ず人間がエスカレーション対応すべきカテゴリです。 eBay のセラーパフォーマンス評価では、メッセージへの返信速度（Response Rate）が重要指標のひとつです。適切に設計された自動返信ボットは Response Rate を向上させ、Top Rated Seller ステータスの維持にも直接貢献します。しかし「とにかく何か返信すればいい」という思想で作られたボットは、かえってバイヤーの信頼を損ない逆効果になる点を、最初に認識しておく必要があります。 基本的な使い方（ベースライン）：sendMessageで返信を送る まず最小限の動作確認ができるシンプルな実装から始めましょう。第21回で取得した conversationId を指定して、既存の会話にテキストメッセージを返信する基本形です。 # message_send_baseline.py import requests from typing import Optional EBAY_API_BASE = "https://api.ebay.com/commerce/message/v1" def send_message_to_buyer( access_token: str, conversation_id: str, message_text: str, email_copy_to_sender: bool = False, reference_id: Optional[str] = None, ) -> dict: """ 既存の会話にバイヤーへの返信を送信する最小実装。 Args: access_token : OAuth 2.0 アクセストークン conversation_id : 返信先の会話 ID（getConversations で取得） message_text : 送信するメッセージ本文（最大 2000 文字） email_copy_to_sender : True の場合、送信者にもメールコピーを送付 reference_id : 関連商品の ItemID（任意） Returns: eBay API のレスポンス dict（成功時は空 dict の場合もある） """ url = f"{EBAY_API_BASE}/send_message" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP", } payload: dict = { "conversationId": conversation_id, "messageText": message_text, "emailCopyToSender": email_copy_to_sender, } # 関連商品を紐付ける場合は reference ブロックを追加 # ※ referenceType は現在 LISTING のみサポート if reference_id: payload["reference"] = { "referenceId": reference_id, "referenceType": "LISTING", } response = requests.post(url, headers=headers, json=payload, timeout=10) response.raise_for_status() # 成功時はレスポンスボディが空の場合があるため安全に処理する return response.json() if response.text.strip() else {} # ===== 使用例 ===== if __name__ == "__main__": TOKEN = "v^1.1#i^1#f^0#..." # OAuth アクセストークン CONV_ID = "v1|conv_1234567890|0" # #21 で取得した conversationId result = send_message_to_buyer( access_token=TOKEN, conversation_id=CONV_ID, message_text="ご連絡ありがとうございます。商品は現在発送準備中です。", email_copy_to_sender=False, ) print("返信送信完了:", result) 補足: conversationId と otherPartyUsername の使い分け POST /send_message のリクエストボディで「誰に送るか」を指定するフィールドには2種類あります。conversationId は第21回の getConversations で取得した既存の会話 ID で、バイヤーからのメッセージへの「返信」時に使用します。otherPartyUsername はバイヤーの eBay ユーザー名で、こちらはまったく新規の会話を「起票」する場合（例：取引完了後のフォローアップメッセージ）に使用します。重要なのはこの2つを同時に指定するとバリデーションエラーになる点です。「返信か、新規起票か」でどちらを使うか明確に判断してください。 messageText は必須フィールドで、最大 2000 文字という制限があります。この制限を超えたテキストを送信しようとすると API はエラーを返します。日本語はマルチバイト文字ですが、eBay の Message API では UTF-8 の文字数（コードポイント数）でカウントされるため、Python の len() による事前チェックが有効です。長文テンプレートを動的に生成する場合は、後述の truncate 処理を必ず実装してください。 emailCopyToSender を True に設定すると、送信したメッセージのコピーがセラー自身の登録メールアドレスにも送付されます。手動での確認やログ目的で使用できますが、自動返信ボットで大量送信する場合はメールボックスが溢れる危険があるため、通常は False のままにしておくことを強く推奨します。 messageMedia フィールドは任意で、画像・PDF・ドキュメント・テキストファイルを最大5件添付できます。各要素には mediaName・mediaType（IMAGE / PDF / DOC / TXT）・mediaUrl（HTTPS 必須）を指定します。梱包指示書や商品マニュアルの PDF を自動添付したい場合に活用できますが、mediaUrl に HTTP（非SSL）の URL を指定するとエラーになるため注意が必要です。 実務で躓く場面・深いポイント (Core) ベースライン実装で動作確認ができたら、次は本番運用で必ず直面する実務上の罠とその解決策を見ていきましょう。 1. キーワード分類の誤判定で見当違いの自動返信を送ってしまうリスク 自動返信ボットを実装する際に最初に直面するのが「分類の精度」問題です。例えば「返品」というキーワードをシンプルな文字列マッチで検出しようとすると、「返品は必要ありません、とても満足しています！」というポジティブなフィードバックメッセージも誤って「返品リクエスト」として分類してしまう可能性があります。 もう一つのよくあるパターンは、複数のカテゴリに跨るメッセージです。「発送はいつですか？もし遅れるようなら返品を検討したいのですが」というメッセージは、発送カテゴリにも返品カテゴリにも該当します。このような複合的なメッセージに発送テンプレートだけを返信するのは状況をさらに悪化させる可能性があります。 実務的な対策としては、キーワードの「優先順位」と「ネガティブパターンの先行評価」を設計することです。具体的には、返品・クレームなどのネガティブカテゴリを最高優先度で判定し、いずれかのネガティブキーワードがヒットした場合は自動返信せずに人間エスカレーションフラグを立てる仕組みにします。評価優先順位は NEGATIVE > RETURN > SHIPPING > QUESTION > UNKNOWN の順に設定することを推奨します。 2. 同じ会話に対する二重送信防止（送信済みフラグ管理） 定期実行（例：5分ごとのバッチ処理）で自動返信ボットを動かす場合、あるサイクルで「未返信」として検出・返信した会話が、次のサイクルでも「未返信」として再度検出される問題が発生します。eBay の API 側でステータスが反映されるまでに数十秒〜数分のタイムラグがあることや、getConversations の conversation_status フィルタが想定通りに動作しないエッジケースが原因です。 この問題を防ぐ最もシンプルな実装は、送信済みの conversationId を Python の Set で管理することです。ただし、プロセス再起動後もデータが失われないよう、本番環境では Redis（SADD/SISMEMBER コマンド）や PostgreSQL（sent_messages テーブル）などの永続化ストレージに送信済み ID を保存することを強く推奨します。 また、getConversations API のレスポンスに含まれる conversation_status フィールドも積極的に活用してください。第21回で解説した通り、ANSWERED ステータスの会話はフィルタリングの段階で除外できます。ただし、前述のタイムラグ問題があるため、API フィルタとアプリ側の重複チェックの両方を二重に持つ「多層防御」の設計が最善です。 3. messageTextの文字数制限を超えた場合のtruncate処理 日本語テンプレートは英語と比べて文字数が少なく見えても、バイヤー名・商品名・配送日数などの動的な値を埋め込んでフォーマットした後に 2000 文字を超えることがあります。特に複数の注文詳細を含む長文テンプレートを使う場合は注意が必要です。len() による文字数チェックを怠ると、本番環境で突然 API エラーが返ってきます。 truncate 処理では単純に先頭 2000 文字で切り捨てると文章の途中で切れてしまい、不自然なメッセージがバイヤーに届きます。句読点（。！？）の位置を rfind() で検索し、2000 文字以内の最後の文区切り位置で切り詰めるロジックを実装することを推奨します。切り詰めが発生した場合は必ずログに WARNING レベルで記録し、テンプレートの長さを見直す契機にしてください。 注意: ネガティブメッセージへの自動返信は厳禁 返品要求・商品の欠陥報告・未着の申告・詐欺の疑いを含むメッセージに対して、自動返信ボットが画一的なテンプレートを送ってしまうことは、状況を著しく悪化させる深刻なリスクをはらんでいます。例えば、バイヤーが「商品が届かなかった。返金を要求する」と訴えているのに「ご購入ありがとうございます！発送は2営業日以内です」という返信が届けば、バイヤーは激怒し eBay への Money Back Guarantee 申請や PayPal クレームに発展する可能性があります。 自動返信の対象から必ず除外すべきメッセージカテゴリ: （1）返品・返金・キャンセルの要求、（2）商品の欠陥・破損・未着の報告、（3）詐欺・偽物疑惑を含む内容、（4）複数の深刻な問題が複合したメッセージ。これらを検出した場合は自動返信せず、Slack やメール等で担当者にエスカレーション通知を送り、必ず人間が対応する仕組みを設けてください。 頻出エラーコード早見表 エラーコード エラー内容 発生ケース 対処法 13007 MESSAGE_TEXT_TOO_LONG messageText が 2000 文字を超えている truncate() で事前に切り詰める。ログに WARNING を記録しテンプレートを見直す 2004 FIELD_VALUE_INVALID conversationId と otherPartyUsername を同時に指定 返信時は conversationId のみ、新規起票時は otherPartyUsername のみを指定する 6001 RESOURCE_NOT_FOUND 存在しない・アクセス権のない conversationId、または HTTP の mediaUrl を指定 getConversations で最新の ID を再取得する。mediaUrl は必ず HTTPS にする 1100 INVALID_ACCESS_TOKEN アクセストークンの期限切れ、または message スコープ不足 トークンをリフレッシュし、OAuth スコープに https://api.ebay.com/oauth/api_scope/message が含まれているか確認する 堅牢な実装：getConversations連携と自動返信ボットの完全実装 ここまで解説したすべての実務ポイント（キーワード優先度分類・二重送信防止・文字数制限ハンドリング・営業時間外判定・エスカレーション）を組み込んだ、本番稼働可能な完全実装を示します。 # auto_reply_bot.py import requests import re import logging from datetime import datetime, time from zoneinfo import ZoneInfo from typing import Optional, Dict, List, Set, Tuple, Callable from dataclasses import dataclass, field from enum import Enum, auto logger = logging.getLogger(__name__) EBAY_API_BASE = "https://api.ebay.com/commerce/message/v1" MAX_MESSAGE_LEN = 2000 JST = ZoneInfo("Asia/Tokyo") class MessageCategory(Enum): NEGATIVE = auto() # クレーム・詐欺疑惑等（最高優先度でエスカレーション） RETURN = auto() # 返品・返金・キャンセル（人間対応必須） SHIPPING = auto() # 発送・追跡番号に関する問い合わせ QUESTION = auto() # 在庫・商品スペック等の一般質問 UNKNOWN = auto() # 分類不能（自動返信スキップ） # キーワードルール（優先度降順で定義すること） KEYWORD_RULES: List[Tuple[MessageCategory, List[str]]] = [ (MessageCategory.NEGATIVE, [ r"詐欺", r"偵物", r"クレーム", r"苦情", r"最悪", r"ひどい", r"fraud", r"fake", r"scam", r"complaint", ]), (MessageCategory.RETURN, [ r"返品", r"返金", r"refund", r"return", r"キャンセル", r"cancel", ]), (MessageCategory.SHIPPING, [ r"発送", r"出荷", r"いつ届", r"配送", r"追跡", r"tracking", r"shipping", r"dispatch", ]), (MessageCategory.QUESTION, [ r"在庫", r"サイズ", r"カラー", r"色", r"状態", r"コンディション", r"stock", r"size", r"color", r"condition", ]), ] REPLY_TEMPLATES: Dict[MessageCategory, str] = { MessageCategory.SHIPPING: ( "ご購入・お問い合わせありがとうございます。\n" "ご注文の商品は {shipping_days} 営業日以内に発送いたします。\n" "追跡番号は発送完了後、eBay システムを通じて自動でお知らせします。\n" "ご不明な点がございましたらお気軽にご連絡ください。" ), MessageCategory.QUESTION: ( "お問い合わせありがとうございます。\n" "ご質問内容を確認し、担当者より改めてご回答申し上げます。\n" "通常 24 時間以内にご返信いたします。もうしばらくお待ちください。" ), } @dataclass class AutoReplyBot: """ eBay Message API を使ったバイヤー自動返信ボット。 Attributes: access_token : OAuth 2.0 アクセストークン marketplace_id : eBay マーケットプレイス ID（例: EBAY_JP） business_start : 営業開始時刻（JST） business_end : 営業終了時刻（JST） shipping_days : 発送予定日数（テンプレートに埋め込む） sent_conversation_ids : 送信済み conversationId の Set（二重送信防止） """ access_token: str marketplace_id: str = "EBAY_JP" business_start: time = field(default_factory=lambda: time(9, 0)) business_end: time = field(default_factory=lambda: time(18, 0)) shipping_days: int = 2 sent_conversation_ids: Set[str] = field(default_factory=set) def _headers(self) -> Dict[str, str]: return { "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json", "X-EBAY-C-MARKETPLACE-ID": self.marketplace_id, } # ---------------------------------------------------------- # 1. 未返信会話の取得（第21回 getConversations を活用） # ---------------------------------------------------------- def get_unanswered_conversations(self, limit: int = 50) -> List[Dict]: """UNANSWERED ステータスの会話を最大 limit 件取得する。""" url = f"{EBAY_API_BASE}/get_conversations" params = {"conversation_status": "UNANSWERED", "limit": str(limit)} resp = requests.get(url, headers=self._headers(), params=params, timeout=10) resp.raise_for_status() return resp.json().get("conversations", []) # ---------------------------------------------------------- # 2. キーワード分類エンジン（優先度降順で評価） # ---------------------------------------------------------- def classify(self, text: str) -> MessageCategory: """メッセージ本文をキーワードで分類する（優先度順）。""" for category, patterns in KEYWORD_RULES: for pat in patterns: if re.search(pat, text, re.IGNORECASE): return category return MessageCategory.UNKNOWN # ---------------------------------------------------------- # 3. 文字数制限チェックと truncate（句読点で区切る） # ---------------------------------------------------------- def _truncate(self, text: str) -> str: """2000 文字を超える場合、文の区切りで切り詰める。""" if len(text) <= MAX_MESSAGE_LEN: return text logger.warning("メッセージが %d 文字。%d 文字に切り詰めます。", len(text), MAX_MESSAGE_LEN) truncated = text[:MAX_MESSAGE_LEN] for delim in ("。", "！", "？", ".", "!", "?"): pos = truncated.rfind(delim) if pos > MAX_MESSAGE_LEN // 2: return truncated[: pos + 1] return truncated[:MAX_MESSAGE_LEN - 3] + "..." # ---------------------------------------------------------- # 4. 営業時間判定（JST） # ---------------------------------------------------------- def _is_business_hours(self) -> bool: """現在時刻が JST の営業時間内かを判定する。""" now_jst = datetime.now(JST).time() return self.business_start <= now_jst <= self.business_end # ---------------------------------------------------------- # 5. メッセージ送信（二重送信防止付き） # ---------------------------------------------------------- def send_message( self, conversation_id: str, message_text: str, reference_id: Optional[str] = None, ) -> bool: """ conversationId を指定してバイヤーに返信を送る。 Returns: True: 送信成功 / False: 二重送信スキップまたは送信失敗 """ if conversation_id in self.sent_conversation_ids: logger.info("二重送信防止: %s は送信済みです。", conversation_id) return False safe_text = self._truncate(message_text) payload: Dict = { "conversationId": conversation_id, "messageText": safe_text, "emailCopyToSender": False, } if reference_id: payload["reference"] = { "referenceId": reference_id, "referenceType": "LISTING", } url = f"{EBAY_API_BASE}/send_message" try: resp = requests.post( url, headers=self._headers(), json=payload, timeout=10 ) resp.raise_for_status() self.sent_conversation_ids.add(conversation_id) # 送信成功後に登録 logger.info("返信成功: conversation_id=%s", conversation_id) return True except requests.HTTPError as exc: logger.error( "sendMessage 失敗: status=%d body=%s", exc.response.status_code, exc.response.text, ) return False # ---------------------------------------------------------- # 6. メインループ（バッチ実行のエントリポイント） # ---------------------------------------------------------- def run( self, escalate_callback: Optional[Callable[[str, MessageCategory, str], None]] = None, ) -> None: """ 未返信会話を取得 → 分類 → 自動返信する一連の処理を実行する。 Args: escalate_callback: エスカレーション時に呼ばれる callable。 引数: (conversation_id, category, message_text) """ conversations = self.get_unanswered_conversations() logger.info("%d 件の未返信会話を処理します。", len(conversations)) for conv in conversations: conv_id: str = conv.get("conversationId", "") messages: List[Dict] = conv.get("messages", []) latest_text: str = messages[-1].get("text", "") if messages else "" item_id: Optional[str] = conv.get("itemId") category = self.classify(latest_text) logger.debug("conv_id=%s category=%s", conv_id, category.name) # --- エスカレーション判定（NEGATIVE / RETURN は人間対応必須）--- if category in (MessageCategory.NEGATIVE, MessageCategory.RETURN): logger.warning( "エスカレーション: conv_id=%s category=%s", conv_id, category.name, ) if escalate_callback: escalate_callback(conv_id, category, latest_text) continue # --- 自動返信テンプレートの選択 --- template = REPLY_TEMPLATES.get(category) if template is None: logger.info("自動返信対象外（UNKNOWN）: conv_id=%s", conv_id) continue # --- 返信テキスト生成（営業時間外メッセージの付記）--- reply_text = template.format(shipping_days=self.shipping_days) if not self._is_business_hours(): reply_text += ( "\n\n※ 現在は営業時間外（9:00～18:00 JST）のため、" "翌営業日に改めてご確認いたします。" ) # --- 送信実行 --- self.send_message(conv_id, reply_text, reference_id=item_id) # ===== エントリポイント ===== if __name__ == "__main__": import os logging.basicConfig(level=logging.INFO) def slack_escalation( conv_id: str, category: MessageCategory, text: str ) -> None: """Slack への通知（実装は slack_sdk 等を使用）""" print(f"[ESCALATION] {category.name}: conv={conv_id[:20]} msg={text[:50]}") bot = AutoReplyBot( access_token=os.environ["EBAY_ACCESS_TOKEN"], shipping_days=2, ) bot.run(escalate_callback=slack_escalation) このコードのポイントは、エスカレーション処理を escalate_callback という関数オブジェクトで外部から注入できる設計にしていることです。本番環境では slack_sdk を使った Slack 通知、smtplib を使ったメール送信、あるいは Jira への Issue 自動起票など、チームの運用体制に合わせた実装を引数として渡すことができます。ロジックとI/Oが明確に分離されているため、ユニットテストも容易です。 また、sent_conversation_ids を dataclass のフィールドとして保持することで、同一プロセス実行サイクル内の二重送信を防ぎます。プロセス再起動後も永続化するには、Set の追加・参照部分を Redis（pipeline で SADD + SISMEMBER）または PostgreSQL（sent_messages テーブルへの INSERT ON CONFLICT DO NOTHING）と同期させる拡張が必要です。実運用ではこの永続化層を最初から組み込んでおくことを強く推奨します。 パフォーマンス・スケーリング視点 (深度) レート制限対策とキューイングによる非同期送信 eBay Message API にはレート制限（Rate Limit）が設定されています。大規模セラーが数百〜数千件の未返信会話を一括処理しようとすると、連続した POST /send_message リクエストが短時間に集中し、HTTP 429 Too Many Requests が返り始めます。この状態で単純な for ループを回しているだけでは、多くのメッセージ送信がサイレントに失敗してしまいます。 実務的な対策として、以下の2つのアプローチを組み合わせることを推奨します。 【アプローチ 1：Exponential Backoff（指数バックオフ）による自動リトライ】 429 エラーを受け取った際に即座にリトライせず、待機時間を指数関数的に延ばしながら再試行する方式です。tenacity ライブラリを使うことで、バックオフロジックをデコレータ1行で追加できます。 # rate_limited_send.py import requests import logging from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, before_sleep_log, ) logger = logging.getLogger(__name__) class RateLimitError(Exception): """eBay API から HTTP 429 を受け取った際に送出するカスタム例外。""" pass def _check_rate_limit(resp: requests.Response) -> None: """HTTP 429 を受け取ったら RateLimitError を送出する。""" if resp.status_code == 429: retry_after = int(resp.headers.get("Retry-After", 60)) raise RateLimitError(f"Rate limited. Retry-After: {retry_after}s") resp.raise_for_status() @retry( retry=retry_if_exception_type(RateLimitError), wait=wait_exponential(multiplier=2, min=5, max=120), # 5s -> 10s -> 20s -> ... stop=stop_after_attempt(5), before_sleep=before_sleep_log(logger, logging.WARNING), ) def send_with_backoff( session: requests.Session, url: str, headers: dict, payload: dict, ) -> dict: """指数バックオフ付きの sendMessage 呼び出し（最大 5 回リトライ）。""" resp = session.post(url, headers=headers, json=payload, timeout=10) _check_rate_limit(resp) return resp.json() if resp.text.strip() else {} 【アプローチ 2：キューイングによる非同期送信】 数百件を超える規模になると、同期的なバッチ処理では実行時間そのものが問題になります。Python の asyncio と aiohttp を使った非同期処理、または Redis Queue（RQ）や Celery を使ったタスクキューイングへの移行を検討してください。キューイングのアーキテクチャでは「会話の検出ジョブ（ポーリング）」と「メッセージ送信ジョブ」を分離します。検出ジョブは5分ごとに getConversations を呼んで未返信会話を Redis キューに積み、送信ワーカーはキューからタスクを順に取り出して sendMessage を実行します。この設計により送信処理がピーク時間に集中するのを防ぎ、レート制限に引っかかるリスクを大幅に低減できます。 requests.Session オブジェクトを AutoReplyBot インスタンス内で保持して複数の API 呼び出し間で共有することで、コネクションプールが有効になりリクエストのオーバーヘッドを削減できます。大量処理を行う場合は HTTPAdapter の pool_maxsize パラメータをデフォルトの 10 から 20〜30 程度に引き上げることを検討してください。また、クラウド環境（AWS Lambda 等）でボットを動かす際は、ウォームスタート時に Session を再利用するグローバルインスタンスパターンを採用すると、コールドスタートのオーバーヘッドを回避できます。 まとめ 本記事では、eBay Message API の POST /send_message エンドポイントを使ったバイヤー自動返信ボットを、ベースライン実装から本番稼働レベルまで段階的に構築しました。 ベースライン: conversationId と messageText を指定した最小構成の sendMessage 呼び出しで既存会話への返信機能を実装し、emailCopyToSender・messageMedia・reference などのオプションパラメータの仕様と使い分けを理解する。 深いポイント: キーワード分類の優先度設計（NEGATIVE > RETURN > SHIPPING > QUESTION）、Set による二重送信防止、messageText の 2000 文字制限に対応した句読点ベースの truncate 処理、そしてネガティブ・返品メッセージの人間エスカレーション。これらを組み合わせた dataclass ベースの完全実装コードで、実運用に直結するボットを構築する。 スケーリング: tenacity による Exponential Backoff で HTTP 429 に堅牢に対応し、大規模処理ではキューイングアーキテクチャ（RQ / Celery）と非同期処理（asyncio + aiohttp）への移行でシステムの安定性とスループットを確保する。 次のステップ 次回（#23）は、Message API シリーズの最終回として bulkUpdateConversation エンドポイントを解説します。個別に会話ステータスを更新するのではなく、一度のリクエストで大量の会話を「既読」「アーカイブ」「スター付き」などに一括変更する方法と、カスタマーサポートのワークフローへの組み込み方を紹介します。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#21】eBay Message API：getConversationsとgetConversationでバイヤーとのメッセージを取得する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第21回です。 今回から新しい API カテゴリ【Message API】に突入します。これは eBay が従来の Trading API 版 CS 管理機能（GetMemberMessages）を全面的に REST 化した、新しい会話管理 API「M2M Public API Service」です。バイヤーからの問い合わせへの対応は、カスタマーサービス業務の中核であり、応答速度が出品者評価（Feedback）やアカウントヘルスに直結します。本記事では、REST 版 Message API の入口となる 2 つのエンドポイント——getConversations と getConversation——を実務視点で徹底解説します。 前回（#20）は Inventory Mapping API シリーズの締めくくりとして、AI 推奨結果を Inventory API に渡して高品質な出品を自動生成する方法を解説しました。今回からは、出品後に発生するバイヤーとのコミュニケーション管理——すなわちメッセージ API の世界に踏み込みます。 この記事で得られること: GET /conversation（getConversations）を使い、conversation_type・conversation_status・reference_id などのパラメータを組み合わせてバイヤーからの未読メッセージ一覧を効率よく取得する方法。 GET /conversation/{conversation_id}（getConversation）で特定の会話スレッド内の全メッセージをページネーションしながら取得する方法。 start_time / end_time による差分同期の設計方針と、生産環境で躓きやすいパラメータ制約・エラーコードの対処法。 背景・なぜこれが重要か (Motivation) 「第14回で Trading API の GetMemberMessages を実装したのに、なぜまた別の API が必要なのでしょうか？」 これは Message API を初めて目にした開発者の多くが抱く、ごく自然な疑問です。結論から言うと、eBay は Trading API の段階的廃止を正式にアナウンスしており、全 API を REST / GraphQL ベースの新世代アーキテクチャへ移行する計画を進めています。GetMemberMessages もその対象であり、将来的に REST 版 Message API（M2M Public API Service）に完全置き換えられます。今から REST 版に移行しておくことは、長期的な保守コスト削減に直結します。 Trading API 版との最大の違いは、会話タイプの明確な分離です。GetMemberMessages では eBay からの公式通知もバイヤーからの問い合わせも混在したまま返却されていました。REST 版では conversation_type パラメータにより、FROM_EBAY（eBay からの公式通知・アラート）と FROM_MEMBERS（バイヤーとの直接メッセージ）を最初のリクエスト時点で明確に分離します。これにより「返信が必要なバイヤーメッセージだけを抽出する」処理が格段にシンプルになり、不要なデータを取得する無駄がなくなります。 ページネーション仕様も標準化されました。Trading API が独自の EntriesPerPage / PageNumber 方式を採用していたのに対し、REST 版は limit と offset によるシンプルなオフセット方式を採用しており、他の eBay REST API と同じ感覚で実装できます。また、FROM_MEMBERS に限り start_time / end_time による時刻フィルタリングが可能になり、「前回ポーリング以降の新着メッセージだけを差分取得する」設計が容易になっています。 以下の表に Trading API 版と REST 版の主要な違いをまとめます。 項目 Trading API: GetMemberMessages REST: Message API 認証方式 XML + Auth Token OAuth 2.0 Bearer Token 会話種別の分離 なし（全メッセージ混在） FROM_EBAY / FROM_MEMBERS で明示分離 ページネーション EntriesPerPage + PageNumber limit + offset 時刻フィルタ StartCreationTime + EndCreationTime start_time / end_time（FROM_MEMBERS のみ） 廃止予定 廃止予定あり 現行推奨 API 基本的な使い方（ベースライン）：getConversationsで未読会話一覧を取得する まず最小限の実装で動かしてみましょう。GET /conversation を叩いて、バイヤーからの未読会話一覧を取得します。conversation_type=FROM_MEMBERS と conversation_status=UNREAD の組み合わせが、カスタマーサービス自動化の第一歩です。OAuth 2.0 のアクセストークンを取得済みであることを前提とします（認証フローは第1回を参照してください）。 # get_conversations_baseline.py import requests BASE_URL = "https://api.ebay.com/commerce/message/v1" def get_unread_conversations(access_token: str) -> dict: """バイヤーからの未読会話一覧を取得する（最小実装）""" url = f"{BASE_URL}/conversation" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP", } params = { "conversation_type": "FROM_MEMBERS", # 必須: バイヤーとのメッセージ "conversation_status": "UNREAD", # 任意: 未読のみ絞り込み "limit": 25, # 1回あたりの取得件数 (最大50) "offset": 0, # 開始位置 } response = requests.get(url, headers=headers, params=params, timeout=30) response.raise_for_status() return response.json() if __name__ == "__main__": TOKEN = "v^1.1#i^1#..." # ← ここに実際のOAuthトークンをセット result = get_unread_conversations(TOKEN) total = result.get("total", 0) print(f"未読会話総数: {total}") for conv in result.get("conversations", []): conv_id = conv.get("conversationId", "") subject = conv.get("subject", "(件名なし)") buyer = conv.get("buyer", {}).get("username", "") print(f" [{conv_id}] {subject} (バイヤー: {buyer})") レスポンスの conversations 配列には、各会話のメタデータ（conversationId・subject・buyer・creationDate・messageStatus など）が含まれます。ここで取得した conversationId を使って、次の getConversation でスレッド内の全メッセージを取得します。 補足: conversation_type の 2 種類について FROM_MEMBERS はバイヤーとセラー間の直接メッセージ（購入前の質問・交渉・クレームなど）に使用します。FROM_EBAY は eBay が送信する公式通知（ポリシー違反通知・セキュリティアラート・システムメッセージ等）に使用します。カスタマーサービス自動化においては FROM_MEMBERS が主要ターゲットです。FROM_EBAY のメッセージは自動返信の対象ではなく、モニタリング・アーカイブ目的での取得が中心になります。 特定の商品（Listing）に紐付いたメッセージだけを絞り込みたい場合は、reference_id と reference_type を組み合わせます。例えば、item_id が「123456789012」の商品に関する未読メッセージだけを取得したい場合は、params に reference_id="123456789012" と reference_type="LISTING" を追加します。これにより、複数商品を管理するセラーが「この SKU に関する問い合わせだけを優先処理する」ワークフローを実装できます。 # 特定Listing（item_id）のメッセージに絞り込む例 params_with_listing = { "conversation_type": "FROM_MEMBERS", "conversation_status": "UNREAD", "reference_id": "123456789012", # eBay の item_id "reference_type": "LISTING", "limit": 50, "offset": 0, } reference_type には現在 LISTING が主要な値ですが、eBay の将来の拡張に備えて文字列として定数管理することを推奨します。 実務で躓く場面・深いポイント (Core) ベースライン実装は単純に見えますが、実務で使うとすぐにいくつかの「罠」に気づきます。以下では、実際に頻繁に発生するトラブルとその回避策を具体的に解説します。 1. conversation_type は必須パラメータ——指定し忘れると即 400 エラー getConversations のドキュメントを斜め読みして「とりあえず全件取ってみよう」と conversation_type を省略してリクエストを送ると、eBay はすぐさま HTTP 400 Bad Request を返します。このパラメータは仕様上 required（必須）と明記されており、省略した場合はリクエスト自体が受け付けられません。 エラーレスポンスの例は次のようになります。 { "errors": [ { "errorId": 850001, "domain": "API_MESSAGE", "category": "REQUEST", "message": "Invalid request. The 'conversation_type' field is required.", "parameters": [ { "name": "fieldName", "value": "conversation_type" } ] } ] } 注意: conversation_type は getConversation（個別取得）でも必須です。 パスパラメータ conversation_id を指定しているからといって省略できません。ページネーション関連のラッパー関数を作る際は、必ず conversation_type を引数として受け取り、常にリクエストに含める設計にしてください。 2. limit の最大値は 50——大量取得時は offset ページネーションが必須 limit パラメータのデフォルト値は 25 で、上限は 50 です。「一括取得したい」と limit=100 や limit=200 を指定しても、API は 400 エラーを返すか、silently に上限値の 50 に丸め込んで返します。未読メッセージが大量に溜まっているアカウント（例えば数日間放置した場合）では、1 回のリクエストでは全件取得できないケースがほとんどです。 正しいアプローチは offset によるページネーションです。レスポンスに含まれる total フィールドが全件数を示しており、「offset + 取得件数 >= total」になるまでループを回します。以下にシンプルなページネーションループのスニペットを示します。 def get_all_conversations(access_token: str, conv_type: str) -> list[dict]: """全会話をページネーションで取得する""" url = f"{BASE_URL}/conversation" headers = {"Authorization": f"Bearer {access_token}", "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP"} all_convs = [] offset = 0 limit = 50 # 常に最大値を使う while True: params = {"conversation_type": conv_type, "limit": limit, "offset": offset} resp = requests.get(url, headers=headers, params=params, timeout=30) resp.raise_for_status() data = resp.json() convs = data.get("conversations", []) all_convs.extend(convs) total = data.get("total", 0) offset += len(convs) if offset >= total or not convs: break # 全件取得完了 return all_convs 補足: offset ページネーションの既知の問題 offset ベースのページネーションは「取得中に新着が追加された場合に重複・欠落が発生しうる」という既知の問題を抱えています。リアルタイム性が重要な本番環境では、取得後に重複排除（conversationId をキーとした dedup）をかけることを推奨します。 3. start_time / end_time は FROM_MEMBERS 専用——FROM_EBAY で使うとエラー 差分同期で「前回チェック以降の新着メッセージだけを取得したい」という要件は非常に一般的です。start_time と end_time パラメータはまさにそのためにありますが、これらは conversation_type=FROM_MEMBERS の場合にしか利用できません。FROM_EBAY に対して start_time を指定すると、エラーが返るか、パラメータが無視されて全件が返却される不安定な挙動を示します。 FROM_EBAY の時刻フィルタリングが必要な場合は、クライアント側でレスポンスを受け取った後に creationDate フィールドで絞り込む後処理フィルタリングを実装してください。ただし FROM_EBAY のメッセージ量は一般に少ないため、全件取得してクライアント側でフィルタする方式でも実用上の問題は少ないでしょう。 start_time / end_time のフォーマットは ISO 8601 形式（UTC）です。Python での生成例を示します。 from datetime import datetime, timezone, timedelta # 過去24時間のメッセージを取得する場合 now = datetime.now(timezone.utc) start = now - timedelta(hours=24) params = { "conversation_type": "FROM_MEMBERS", # 必須: FROM_MEMBERSのみ有効 "start_time": start.strftime("%Y-%m-%dT%H:%M:%S.000Z"), "end_time": now.strftime("%Y-%m-%dT%H:%M:%S.000Z"), "limit": 50, "offset": 0, } 頻出エラーコード早見表 Message API で実際に遭遇しやすいエラーコードと対処法を以下にまとめます。 HTTPステータス / errorId エラーメッセージ（抜粋） 原因と対処法 400 / 850001 field 'conversation_type' is required conversation_type を省略した。必ず FROM_MEMBERS か FROM_EBAY を指定すること。 400 / 850002 Invalid value for 'limit'. Maximum allowed value is 50. limit に 50 超の値を指定した。limit=50 に修正し offset でページネーションする。 400 / 850010 start_time is not supported for conversation_type FROM_EBAY FROM_EBAY に start_time を指定した。FROM_MEMBERS のみ対応。クライアント側でフィルタする。 401 / 1001 Invalid access token. Token has expired. OAuth トークンが失効。トークンを再取得して Bearer ヘッダを更新する。 404 / 850100 Conversation not found. conversation_id が存在しないか、自分のアカウントに紐付いていない。getConversations で取得した ID のみ getConversation に渡すこと。 429 / — Too many requests. レート制限に到達。Retry-After ヘッダの秒数だけ待機してからリトライする。 堅牢な実装：全会話の一括取得と個別メッセージ収集クラス ここまでのポイントをすべて盛り込んだ、プロダクションレベルの実装を示します。EbayMessageClient クラスとして実装し、型アノテーション・docstring・入力バリデーション・例外処理・レート制限対応を備えています。getConversations で全会話を offset ページネーションで取得し、各会話に対して getConversation で全メッセージを取得後、JSON ファイルに保存するところまでをワンクラスで完結させます。 # message_client.py import json import time import logging from dataclasses import dataclass, field from typing import Optional, Iterator import requests logger = logging.getLogger(__name__) BASE_URL = "https://api.ebay.com/commerce/message/v1" @dataclass class ConversationFilter: """会話取得フィルタ設定。 Attributes: conversation_type: 必須。"FROM_MEMBERS" または "FROM_EBAY"。 status: 任意。"UNREAD" / "READ" / "ACTIVE" / "ARCHIVE" / "DELETE"。 other_party_username: 任意。特定バイヤーのユーザー名でフィルタ。 reference_id: 任意。特定 Listing の item_id。 reference_type: 任意。reference_id と合わせて使用（例: "LISTING"）。 start_time: 任意。ISO 8601形式（FROM_MEMBERS のみ有効）。 end_time: 任意。ISO 8601形式（FROM_MEMBERS のみ有効）。 """ conversation_type: str status: Optional[str] = None other_party_username: Optional[str] = None reference_id: Optional[str] = None reference_type: Optional[str] = None start_time: Optional[str] = None end_time: Optional[str] = None class EbayMessageClient: """eBay Message API クライアント（プロダクションレベル）。""" MAX_LIMIT = 50 DEFAULT_TIMEOUT = 30 VALID_TYPES = {"FROM_MEMBERS", "FROM_EBAY"} VALID_STATUSES = {"ACTIVE", "ARCHIVE", "DELETE", "READ", "UNREAD"} def __init__(self, access_token: str, marketplace_id: str = "EBAY_JP") -> None: """ Args: access_token: OAuth 2.0 アクセストークン（必須）。 marketplace_id: マーケットプレイス ID（デフォルト: EBAY_JP）。 Raises: ValueError: access_token が空の場合。 """ if not access_token or not access_token.strip(): raise ValueError("access_token は空にできません。") self._token = access_token.strip() self._marketplace_id = marketplace_id self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {self._token}", "Content-Type": "application/json", "X-EBAY-C-MARKETPLACE-ID": self._marketplace_id, }) def _validate_filter(self, f: ConversationFilter) -> None: """ConversationFilter の入力値を事前検証する。""" if f.conversation_type not in self.VALID_TYPES: raise ValueError( f"conversation_type は {self.VALID_TYPES} のいずれかを指定してください。" f"受け取った値: {f.conversation_type!r}" ) if f.status and f.status not in self.VALID_STATUSES: raise ValueError( f"status は {self.VALID_STATUSES} のいずれかを指定してください。" f"受け取った値: {f.status!r}" ) if (f.start_time or f.end_time) and f.conversation_type != "FROM_MEMBERS": raise ValueError( "start_time / end_time は conversation_type='FROM_MEMBERS' の場合のみ" "使用できます。FROM_EBAY では利用不可です。" ) def _get(self, path: str, params: dict) -> dict: """内部 GET リクエスト（レート制限・タイムアウト対応）。""" url = f"{BASE_URL}{path}" try: resp = self._session.get( url, params=params, timeout=self.DEFAULT_TIMEOUT ) if resp.status_code == 429: retry_after = int(resp.headers.get("Retry-After", 60)) logger.warning( f"レート制限に到達しました。{retry_after} 秒待機します..." ) time.sleep(retry_after) resp = self._session.get( url, params=params, timeout=self.DEFAULT_TIMEOUT ) resp.raise_for_status() return resp.json() except requests.exceptions.HTTPError as exc: logger.error( f"HTTP エラー: {exc.response.status_code} | " f"Body: {exc.response.text[:300]}" ) raise except requests.exceptions.Timeout: logger.error( f"リクエストタイムアウト ({self.DEFAULT_TIMEOUT}s): {url}" ) raise def iter_conversations( self, f: ConversationFilter ) -> Iterator[dict]: """全会話をページネーションで逐次返すジェネレータ。 Args: f: 取得条件を指定した ConversationFilter。 Yields: 各会話のメタデータ辞書。 """ self._validate_filter(f) params: dict = {"conversation_type": f.conversation_type} if f.status: params["conversation_status"] = f.status if f.other_party_username: params["other_party_username"] = f.other_party_username if f.reference_id: params["reference_id"] = f.reference_id params["reference_type"] = f.reference_type or "LISTING" if f.start_time: params["start_time"] = f.start_time if f.end_time: params["end_time"] = f.end_time offset = 0 while True: params["limit"] = self.MAX_LIMIT params["offset"] = offset data = self._get("/conversation", params) conversations = data.get("conversations", []) if not conversations: break yield from conversations total = data.get("total", 0) offset += len(conversations) if offset >= total: break def get_conversation_messages( self, conversation_id: str, conversation_type: str ) -> list[dict]: """特定会話の全メッセージをページネーションで取得する。 Args: conversation_id: getConversations で取得した会話 ID。 conversation_type: "FROM_MEMBERS" または "FROM_EBAY"。 Returns: 全メッセージのリスト（時系列順）。 Raises: ValueError: conversation_id が空の場合。 """ if not conversation_id or not conversation_id.strip(): raise ValueError("conversation_id は空にできません。") if conversation_type not in self.VALID_TYPES: raise ValueError( f"conversation_type は {self.VALID_TYPES} のいずれかを指定してください。" ) messages: list[dict] = [] offset = 0 while True: params = { "conversation_type": conversation_type, "limit": self.MAX_LIMIT, "offset": offset, } data = self._get(f"/conversation/{conversation_id}", params) msgs = data.get("messages", []) messages.extend(msgs) total = data.get("total", 0) offset += len(msgs) if not msgs or offset >= total: break return messages def fetch_and_save( self, f: ConversationFilter, output_path: str, ) -> int: """全会話とメッセージを取得し JSON ファイルに保存する。 Args: f: 取得条件フィルタ。 output_path: 保存先ファイルパス（.json）。 Returns: 保存した会話件数。 """ result: list[dict] = [] for conv in self.iter_conversations(f): conv_id = conv.get("conversationId", "") logger.info(f"会話取得中: {conv_id}") messages = self.get_conversation_messages( conv_id, f.conversation_type ) result.append({ "conversation": conv, "messages": messages, }) with open(output_path, "w", encoding="utf-8") as fp: json.dump(result, fp, ensure_ascii=False, indent=2) logger.info( f"{len(result)} 件の会話を {output_path} に保存しました。" ) return len(result) # ─── 実行例 ───────────────────────────────────────────────────────────── if __name__ == "__main__": import os from datetime import datetime, timezone, timedelta logging.basicConfig(level=logging.INFO) TOKEN = os.environ["EBAY_ACCESS_TOKEN"] client = EbayMessageClient(access_token=TOKEN) # 過去72時間の未読バイヤーメッセージを全件取得 now = datetime.now(timezone.utc) start = now - timedelta(hours=72) flt = ConversationFilter( conversation_type="FROM_MEMBERS", status="UNREAD", start_time=start.strftime("%Y-%m-%dT%H:%M:%S.000Z"), end_time=now.strftime("%Y-%m-%dT%H:%M:%S.000Z"), ) saved = client.fetch_and_save(flt, "unread_conversations.json") print(f"保存完了: {saved} 件") fetch_and_save メソッドはジェネレータ（iter_conversations）を使って会話を 1 件ずつ処理するため、大量の会話がある場合でもメモリ使用量を一定に保てます。各会話のメッセージ取得（get_conversation_messages）も内部でページネーションを行うため、長大なスレッドでも安全に全件取得できます。 注意: OAuth トークンの管理について OAuth トークンは環境変数（EBAY_ACCESS_TOKEN）から読み込む設計にしてください。ソースコードにトークンをハードコーディングすると、Git リポジトリへの誤コミットによる情報漏洩リスクが生じます。トークンの有効期限は通常 2 時間ですので、長時間バッチ処理を行う場合はリフレッシュトークンを使ったトークン自動更新機能（第1回参照）と組み合わせてください。 パフォーマンス・スケーリング視点 (深度) 大量会話を効率的に処理する差分同期（デルタ同期）戦略 規模が大きいセラーアカウントでは、未読メッセージが数千件に達することもあります。毎回全件取得（フルスキャン）するのは API のレート制限を消費するだけでなく、処理時間も長くなります。REST 版 Message API が提供する start_time パラメータを活用した差分同期（デルタ同期）が、スケーラブルな設計の鍵です。 差分同期の基本戦略は次の通りです。まず、前回のポーリング成功時刻を永続ストレージ（データベースやファイル）に記録します。次回ポーリング時は、その時刻を start_time として指定することで、新着分のみを取得できます。この方式により、API コール数とデータ転送量を大幅に削減できます。 # delta_sync.py import json import os from datetime import datetime, timezone from message_client import EbayMessageClient, ConversationFilter CHECKPOINT_FILE = "last_sync_time.json" def load_checkpoint() -> str: """前回同期時刻を ISO 8601 形式で返す（初回は7日前）。""" if os.path.exists(CHECKPOINT_FILE): with open(CHECKPOINT_FILE) as fp: data = json.load(fp) return data.get("last_sync_time", "") # 初回実行: 過去7日分を取得 from datetime import timedelta start = datetime.now(timezone.utc) - timedelta(days=7) return start.strftime("%Y-%m-%dT%H:%M:%S.000Z") def save_checkpoint(sync_time: str) -> None: """同期完了時刻を保存する。""" with open(CHECKPOINT_FILE, "w") as fp: json.dump({"last_sync_time": sync_time}, fp) def run_delta_sync(access_token: str) -> None: now = datetime.now(timezone.utc) now_str = now.strftime("%Y-%m-%dT%H:%M:%S.000Z") start_str = load_checkpoint() client = EbayMessageClient(access_token=access_token) flt = ConversationFilter( conversation_type="FROM_MEMBERS", start_time=start_str, end_time=now_str, ) saved = client.fetch_and_save( flt, f"delta_{now.strftime('%Y%m%d_%H%M%S')}.json" ) print(f"差分取得完了: {saved} 件の新着会話") # 成功した場合のみチェックポイントを更新 save_checkpoint(now_str) チェックポイントファイルの更新は、fetch_and_save が例外なく完了した後に行うことが重要です。途中でエラーが発生した場合は古いチェックポイントを維持することで、次回実行時に該当期間を再取得（冪等な再試行）できます。 ポーリング間隔と API レート制限の設計 Message API のレート制限は eBay の利用規約・API キーのプランによって異なりますが、一般的には 1 日あたりのコール数と 1 秒あたりのコール数の両方に上限があります。カスタマーサービスのポーリング間隔として、以下の指針を推奨します。 営業時間内（JST 9:00〜21:00）: 5〜10 分間隔でポーリング。バイヤーからの問い合わせに迅速に応答するため、短い間隔が望ましいですが、レート制限を考慮して最短でも 5 分以上とすることを推奨します。 営業時間外（JST 21:00〜翌 9:00）: 30 分〜1 時間間隔でポーリング。自動返信ボット（次回 #22 で解説）と組み合わせることで、営業時間外でもバイヤーへの初期応答を自動化できます。 HTTP 429（レート制限超過）を受け取った場合は、Retry-After ヘッダに指定された秒数だけ待機してからリトライします。前掲の EbayMessageClient._get メソッドにはこの処理が組み込まれています。また、複数アカウントを管理する場合は API キーをアカウントごとに分離することで、レート制限の消費を独立させることができます。 まとめ 本記事では、eBay Message API（M2M Public API Service）の入口となる 2 つのエンドポイント——getConversations と getConversation——を解説しました。 ベースライン: conversation_type=FROM_MEMBERS + conversation_status=UNREAD の組み合わせでバイヤーからの未読メッセージ一覧を取得する基本フロー。reference_id + reference_type で特定 Listing への絞り込みができる仕組みも確認しました。 深いポイント: conversation_type は getConversations・getConversation の両方で必須パラメータであること。limit の上限は 50 であり大量取得には offset ページネーションが必須であること。start_time / end_time は FROM_MEMBERS 専用であり FROM_EBAY では利用できない制約があること。これらを知らずに実装すると即座に 400 エラーに直面します。 スケーリング: EbayMessageClient クラスによる型安全な実装と、チェックポイントファイルを使った差分同期（デルタ同期）戦略。ポーリング間隔の設計とレート制限への対応により、本番環境での安定した運用が可能になります。 Trading API の GetMemberMessages と比較すると、REST 版はパラメータが整理されており、他の eBay REST API と同じ作法で実装できる点が大きな利点です。また OAuth 2.0 の標準的な認証フローを使用するため、既存のトークン管理基盤（第1回で構築済み）をそのまま流用できます。 次のステップ バイヤーのメッセージを取得できるようになったら、次の目標は「自動で返信する」ことです。メッセージを読むだけでは CS 自動化の半分しか完成していません。 次回（#22）では、Message API の sendMessage エンドポイントを使い、よくある問い合わせパターン（在庫確認・発送状況・返品手順など）をテンプレートベースで自動返信するボットを構築します。今回実装した EbayMessageClient クラスを拡張して、読み取り→分類→返信の完全な自動化パイプラインを完成させましょう。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#20】eBay Inventory Mapping API + Inventory API：AI推奨結果をInventory APIに渡して高品質な出品を自動作成する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第20回です。 前回（#19）では、Inventory Mapping APIにサブミットしたタスクをポーリングで監視し、タスク完了後にListingPreviewオブジェクト群（AI推奨のカテゴリ・タイトル・アスペクト・説明文・SKU・mappingReferenceId）を取得しました。これでAIの推奨結果が手元に揃った状態です。 本記事では、その「プレビュー結果を眺めるだけ」で終わらせず、Inventory APIの createOrReplaceInventoryItem へ実際に渡して出品アイテムレコードを自動作成する——Inventory Mapping API三部作の「仕上げ」を実装します。 この記事で得られること: ListingPreview（GraphQL形式）のaspectsをInventory API REST形式の辞書へ変換するデータマッピング層の設計と実装。 mappingReferenceIdをSKUと紐付けてDBに永続化し、後日AI推薦効果を定量評価できるトレーサビリティ設計。 COMPLETED_WITH_ERROR商品の混入を防ぐ入力バリデーションと、bulkCreateOrReplaceInventoryItemを使ったスケーラブルなE2Eパイプラインの実装。 背景・なぜこれが重要か (Motivation) 「推奨結果をコピペして手動で出品すればいいのでは？」 Inventory Mapping APIを初めて使う開発者が必ず抱く疑問です。数件であれば確かに手動コピーも現実的ですが、数百・数千件のカタログを毎日処理するECオペレーションでは、すぐに壁に当たります。 プログラム化が必要な理由は3つあります。「スケーラビリティ」——Mapping APIは1タスクで最大500件をバッチ処理できますが、その結果を手動で出品しては処理能力を活かせません。「データ一貫性」——コピペによるタイプミスや項目の抜けが出品エラーや検索品質低下に直結します。「トレーサビリティ」——mappingReferenceIdをどのSKUに紐付けたか記録しなければ、AI推薦がビジネスにどれだけ貢献したかを後から検証できません。 この3課題を同時解決するのが、本記事で実装する「AI推奨→Inventory API自動連携パイプライン」です。 基本的な使い方（ベースライン）：ListingPreviewからcreateOrReplaceInventoryItemへ まず最小限の動作確認コードから始めます。Inventory APIの createOrReplaceInventoryItem はPUT /sell/inventory/v1/inventory_item/{sku} として呼び出します。リクエストボディの主要フィールドを確認しましょう。 product.title: 商品タイトル（最大80文字）。 product.description: 商品説明文（基本的なHTMLタグ可）。 product.aspects: Item Specificsの辞書形式。キー＝アスペクト名（最大40文字）、値＝文字列の配列（各値最大50文字）。 product.imageUrls: 商品画像URLの配列（HTTPS必須、offer公開前に最低1枚必要）。 condition: 商品コンディションの列挙値（例: NEW, USED_EXCELLENT など）。 availability.shipToLocationAvailability.quantity: 利用可能在庫数。 なお、createOrReplaceInventoryItemはPUTリクエストで、新規作成・更新ともにHTTP 204（No Content）を返します。このAPIだけでは出品は完了せず、その後createOrReplaceOffer → publishOfferが必要です（Offer APIシリーズは第21回以降で解説予定）。 以下は最小限の実装例です。 # baseline_inventory_upload.py import requests from typing import Any def _convert_aspects_to_rest(gql_aspects: list[dict]) -> dict[str, list[str]]: """ GraphQL ListingPreviewProductAspect配列をInventory API REST形式の辞書に変換する。 GraphQL形式（第19回で取得した構造）: [{"localizedAspectName": "Brand", "value": [{"localizedValue": "Nike"}]}, ...] REST形式（createOrReplaceInventoryItemが期待する構造）: {"Brand": ["Nike"], "Color": ["Black"], ...} """ result: dict[str, list[str]] = {} for aspect in (gql_aspects or []): name = aspect.get("localizedAspectName") or aspect.get("name", "") if not name: continue raw_values = aspect.get("value") or aspect.get("values") or [] if raw_values and isinstance(raw_values[0], dict): values = [v.get("localizedValue", "") for v in raw_values if v.get("localizedValue")] else: values = [str(v) for v in raw_values if v] if values: result[name] = values return result def create_inventory_item_from_preview( preview: dict[str, Any], access_token: str, condition: str = "NEW", quantity: int = 10, ) -> dict: """ListingPreviewデータをInventory APIに渡して在庫アイテムを作成する（最小実装）。""" sku = preview["sku"] images = preview.get("images", []) image_urls = [img if isinstance(img, str) else img.get("value", "") for img in images] body = { "product": { "title": preview.get("title", ""), "description": preview.get("description", ""), "aspects": _convert_aspects_to_rest(preview.get("aspects", [])), "imageUrls": image_urls, }, "condition": condition, "availability": { "shipToLocationAvailability": {"quantity": quantity} }, } url = f"https://api.ebay.com/sell/inventory/v1/inventory_item/{sku}" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "Content-Language": "en-US", } resp = requests.put(url, json=body, headers=headers, timeout=30) resp.raise_for_status() return {"sku": sku, "http_status": resp.status_code} 補足: conditionはListingPreviewに含まれない createOrReplaceInventoryItemのconditionフィールドはoffer公開前に必須ですが、Inventory Mapping APIのListingPreviewには含まれていません。「商品コンディションはセラー自身が判断する情報」という設計思想によるものです。パイプラインでは外部（商品マスターDB等）からSKUに対応するconditionを注入する設計にしてください。 実務で躓く場面・深いポイント (Core) ベースラインコードは動きますが、本番に乗せると必ず遭遇する3つの落とし穴を解説します。 1. GraphQLのaspects形式とRESTのaspects形式——データ構造の違いと安全な変換 第19回で取得したListingPreview.aspectsはGraphQLの「配列形式」です。一方、Inventory API REST側が期待するproduct.aspectsは「辞書（オブジェクト）形式」です。この変換を誤ると全商品でHTTP 400が返り続けます。 GraphQL形式（ListingPreviewProductAspect配列）の例: # GraphQL レスポンス（第19回で取得済み） gql_aspects = [ {"localizedAspectName": "Brand", "value": [{"localizedValue": "Nike"}]}, {"localizedAspectName": "Color", "value": [{"localizedValue": "Black"}, {"localizedValue": "White"}]}, {"localizedAspectName": "Size", "value": [{"localizedValue": "M"}]}, ] REST API形式（createOrReplaceInventoryItemが受け付ける形式）の例: # Inventory API REST リクエストボディ内の product.aspects rest_aspects = { "Brand": ["Nike"], "Color": ["Black", "White"], "Size": ["M"], } 変換自体は単純ですが、見落としやすい制限が2点あります。アスペクト名の最大長は40文字、値の最大長は1つにつき50文字です。Mapping APIの推奨値がこの制限を超えることは稀ですが、超えた場合はInventory APIがerrorId: 25044または25045を返します。また、localizedValueが空文字のエントリをそのまま送るとバリデーションエラーになるため、変換時に除外処理を入れておくことが重要です。 2. mappingReferenceIdのトレーサビリティ管理を怠ると後でAI推薦効果を検証できなくなる eBay公式ドキュメントは「Inventory Mapping APIの推奨結果を使って出品を作成・更新する場合、mappingReferenceIdをその出品に含めること」と明記しています。これはトレーサビリティとAI推薦精度の効果測定のためです。 本記事執筆時点（Inventory API v1.18.5）において、createOrReplaceInventoryItemのリクエストボディにmappingReferenceId専用のフィールドは確認できませんでした。そのため、SKUとmappingReferenceIdの対応を内部DB（SQLite等）に永続化する設計を推奨します。憶測で存在しないフィールド名を指定すると、APIが400エラーを返す原因になるため注意してください。 # sku_mapping_store.py — mappingReferenceId を SQLite で管理する import sqlite3 from datetime import datetime def init_mapping_store(db_path: str = "ebay_mapping.db") -> None: conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS sku_mapping_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT NOT NULL, mapping_ref_id TEXT NOT NULL, category_id TEXT, created_at TEXT NOT NULL, UNIQUE(sku, mapping_ref_id) ) """) conn.commit() conn.close() def save_mapping_reference( sku: str, mapping_ref_id: str, category_id: str | None = None, db_path: str = "ebay_mapping.db", ) -> None: """SKUとmappingReferenceIdの対応をDBに保存する。""" conn = sqlite3.connect(db_path) conn.execute( "INSERT OR IGNORE INTO sku_mapping_log" " (sku, mapping_ref_id, category_id, created_at)" " VALUES (?, ?, ?, ?)", (sku, mapping_ref_id, category_id, datetime.utcnow().isoformat()), ) conn.commit() conn.close() このテーブルを持つことで、後日「AI推薦カテゴリで出品したSKUとそうでないSKUの売上CVRを比較する」といった分析が可能になります。mappingReferenceIdはListingPreview1件ごとに発行される個別IDなので、SKUと1対1で紐付けて管理するのが正しい使い方です。 補足: 将来のAPI変更への備え eBayがInventory APIにmappingReferenceId専用フィールドを追加する可能性もあります。その際は、既存のSQLiteテーブルを参照しながら移行できるため、DBへの永続化は将来の変更にも柔軟に対応できる設計です。 3. COMPLETED_WITH_ERRORの商品を誤って自動出品してしまうリスク 第19回で解説した通り、タスク結果にはCOMPLETED（全成功）とCOMPLETED_WITH_ERROR（一部エラーあり）の2ステータスがあります。COMPLETED_WITH_ERRORのタスクには成功した商品と失敗した商品が混在しています。 よくある実装ミスは、タスクステータスを確認せずlistingPreviewsが空でなければ全件をInventory APIに渡してしまうパターンです。エラーとなった商品のListingPreviewは推奨データが不完全な状態（titleが空・aspectsが0件など）になっている場合があり、品質の低い出品が自動生成されるリスクがあります。 安全策は各ListingPreviewに対して「最低品質チェック」を実施してからAPIを呼ぶことです。チェック項目: title が非空か、aspects が1件以上あるか、imageUrls が1件以上あるか、mappingReferenceId が存在するか。不備があればスキップして警告ログを残し、人手確認に回します。 注意: conditionの列挙値はカテゴリによって異なる conditionフィールドに渡す値はConditionEnumの文字列（NEW、USED_EXCELLENT等）ですが、有効な値はeBayサイトとカテゴリによって異なります。例えばあるカテゴリではNEWのみ有効で、USED_EXCELLENTを渡すとerrorId: 25013が返ります。 事前にMetadata APIのgetItemConditionPoliciesを呼び出し、対象カテゴリで有効なconditionIDを取得してからConditionEnumへマッピングすることを推奨します。特にeBay JPとeBay USでは同じカテゴリでもサポートされるconditionが異なる場合があります。 頻出エラーコード早見表 createOrReplaceInventoryItemで遭遇しやすいエラーと対処法のまとめです。 # ────────────────────────────────────────────────── # エラーコード早見表 (createOrReplaceInventoryItem) # ────────────────────────────────────────────────── # HTTP 400 | errorId: 25002 # 内容: SKU値が無効（使用禁止文字を含む、または長すぎる） # 対処: SKUは英数字・ハイフン・アンダースコアのみ。最大50文字に収める。 # HTTP 400 | errorId: 25013 # 内容: conditionの値が対象カテゴリで無効 # 対処: Metadata API getItemConditionPoliciesで # カテゴリ別の有効なcondition一覧を事前取得して照合する。 # HTTP 400 | errorId: 25044 / 25045 # 内容: アスペクト名が40文字超 / アスペクト値が50文字超 # 対処: _convert_aspects内でname[:40]・value[:50]に切り詰めるか、 # 超過したアスペクトをスキップしてログに残す。 # HTTP 400 | errorId: 25709 # 内容: カテゴリで必須のItem Specificが欠落している # 対処: Taxonomy API getItemAspectsForCategoryで必須アスペクトを確認し、 # Mapping API推奨結果に含まれない場合はデフォルト値で補完する。 # HTTP 401 | errorId: 1001 # 内容: アクセストークンが無効または期限切れ # 対処: OAuthリフレッシュロジックを実装し、401時に自動リフレッシュ&リトライする。 堅牢な実装：E2Eパイプライン（ポーリング結果取得→変換→Inventory API登録→トレーサビリティ保存） ここまでの知識を統合した、本番運用に耐えるE2Eパイプラインを示します。型アノテーション・docstring・例外処理・入力バリデーションをすべて含みます。 # inventory_mapping_pipeline.py # Inventory Mapping API 三部作（#18-#20）の仕上げ実装 import logging import requests from dataclasses import dataclass, field from typing import Any from sku_mapping_store import init_mapping_store, save_mapping_reference logger = logging.getLogger(__name__) @dataclass class ListingPreviewValidationError(Exception): """ListingPreviewの品質チェック失敗を表す例外。""" sku: str reasons: list[str] = field(default_factory=list) def __str__(self) -> str: return f"SKU={self.sku} バリデーション失敗: {self.reasons}" class InventoryMappingPipeline: """ Inventory Mapping APIの推奨結果をInventory APIに投入するE2Eパイプライン。 Attributes: access_token: eBay OAuthアクセストークン。 default_condition: conditionのデフォルト値。SKUごとに上書き可能。 default_quantity: 初回在庫数のデフォルト値。 db_path: mappingReferenceId管理用SQLiteのパス。 """ BASE_URL = "https://api.ebay.com/sell/inventory/v1" def __init__( self, access_token: str, default_condition: str = "NEW", default_quantity: int = 10, db_path: str = "ebay_mapping.db", ) -> None: self.access_token = access_token self.default_condition = default_condition self.default_quantity = default_quantity self.db_path = db_path init_mapping_store(db_path) # ---------------------------------------------------------------- # パブリックメソッド # ---------------------------------------------------------------- def process_listing_previews( self, previews: list[dict[str, Any]], condition_map: dict[str, str] | None = None, quantity_map: dict[str, int] | None = None, ) -> dict[str, list[str]]: """ ListingPreviewのリストを受け取り、一括でInventory APIに登録する。 Args: previews: 第19回ポーリング結果のListingPreview配列。 condition_map: SKU→conditionの上書き辞書。 quantity_map: SKU→在庫数の上書き辞書。 Returns: {"succeeded": [sku,...], "failed": [sku,...], "skipped": [sku,...]} """ results: dict[str, list[str]] = {"succeeded": [], "failed": [], "skipped": []} for preview in previews: sku = preview.get("sku", "UNKNOWN_SKU") try: self._validate_preview(preview) condition = (condition_map or {}).get(sku, self.default_condition) quantity = (quantity_map or {}).get(sku, self.default_quantity) self._upload_inventory_item(preview, condition, quantity) self._save_traceability(preview) results["succeeded"].append(sku) logger.info("Inventory item created: sku=%s", sku) except ListingPreviewValidationError as e: logger.warning("品質チェック不合格のためスキップ: %s", e) results["skipped"].append(sku) except requests.HTTPError as e: logger.error("HTTP error sku=%s: %s", sku, e.response.text) results["failed"].append(sku) except Exception as e: logger.exception("予期しないエラー sku=%s: %s", sku, e) results["failed"].append(sku) return results # ---------------------------------------------------------------- # プライベートメソッド # ---------------------------------------------------------------- def _validate_preview(self, preview: dict[str, Any]) -> None: """COMPLETED_WITH_ERROR商品の混入を防ぐ最低品質チェック。""" sku = preview.get("sku", "UNKNOWN_SKU") reasons: list[str] = [] if not preview.get("title", "").strip(): reasons.append("title が空") if not preview.get("aspects"): reasons.append("aspects が空（Item Specificsなし）") if not preview.get("images"): reasons.append("images が空（画像URLなし）") if not preview.get("mappingReferenceId"): reasons.append("mappingReferenceId が欠落") if reasons: raise ListingPreviewValidationError(sku=sku, reasons=reasons) def _convert_aspects(self, gql_aspects: list[dict]) -> dict[str, list[str]]: """GraphQL aspects配列 → REST辞書形式。名前40文字・値50文字上限を強制。""" result: dict[str, list[str]] = {} for aspect in (gql_aspects or []): name = (aspect.get("localizedAspectName") or aspect.get("name", "")).strip() if not name: continue if len(name) > 40: logger.warning("アスペクト名が40文字超のためスキップ: %r", name) continue raw_values = aspect.get("value") or aspect.get("values") or [] if raw_values and isinstance(raw_values[0], dict): values = [ v.get("localizedValue", "")[:50] for v in raw_values if v.get("localizedValue") ] else: values = [str(v)[:50] for v in raw_values if v] if values: result[name] = values return result def _upload_inventory_item( self, preview: dict[str, Any], condition: str, quantity: int ) -> None: """createOrReplaceInventoryItemを呼び出す。""" sku = preview["sku"] images = preview.get("images", []) image_urls = [ img if isinstance(img, str) else img.get("value", "") for img in images if img ] body = { "product": { "title": preview.get("title", ""), "description": preview.get("description", ""), "aspects": self._convert_aspects(preview.get("aspects", [])), "imageUrls": image_urls, }, "condition": condition, "availability": { "shipToLocationAvailability": {"quantity": quantity} }, } url = f"{self.BASE_URL}/inventory_item/{sku}" headers = { "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json", "Content-Language": "en-US", } resp = requests.put(url, json=body, headers=headers, timeout=30) resp.raise_for_status() def _save_traceability(self, preview: dict[str, Any]) -> None: """SKUとmappingReferenceIdの対応をDBに永続化する。""" cat = preview.get("category") or {} category_id = cat.get("categoryId") or cat.get("id") if isinstance(cat, dict) else None save_mapping_reference( sku=preview.get("sku", ""), mapping_ref_id=preview.get("mappingReferenceId", ""), category_id=category_id, db_path=self.db_path, ) # ==================================================== # 使用例 # ==================================================== if __name__ == "__main__": import json # 第19回ポーリングスクリプトが出力したJSONを読み込む with open("listing_previews.json") as f: previews = json.load(f) pipeline = InventoryMappingPipeline( access_token="v^1.1#i^1#...", # OAuthトークン default_condition="NEW", default_quantity=5, ) result = pipeline.process_listing_previews( previews=previews, condition_map={"SKU-12345": "USED_EXCELLENT"}, # SKUごとの上書き ) print(f"成功: {len(result['succeeded'])}件") print(f"スキップ: {len(result['skipped'])}件") print(f"失敗: {len(result['failed'])}件") このパイプラインは3つの責務を明確に分離しています。_validate_preview（品質チェック）、_convert_aspects（データ変換）、_save_traceability（トレーサビリティ）。各メソッドをPytestで個別にユニットテストできるため、CI組み込みも容易です。 timeoutを必ず設定する requests.put()にtimeout=30を設定しています。大量バッチ処理中にeBay側の応答が遅延した場合、timeoutなしだとスレッドが無限にブロックされます。SandboxとProductionでそれぞれ適切な値を設定してください。 パフォーマンス・スケーリング視点 (深度) 数百〜数千件の商品を定期バッチで処理する場合、1件ずつcreateOrReplaceInventoryItemを呼ぶアーキテクチャではすぐに限界が来ます。eBayが提供する一括投入エンドポイントと再処理キューを組み合わせて対処します。 bulkCreateOrReplaceInventoryItemで最大25件を一括投入する Inventory APIには POST /sell/inventory/v1/bulk_create_or_replace_inventory_item という一括エンドポイントがあり、1リクエストで最大25件を処理できます。単発PUTと比較してAPI呼び出し回数を最大1/25に削減でき、スループットが大幅に向上します。 # bulk_upload.py — 25件単位のバッチ投入 import requests from itertools import islice BULK_URL = "https://api.ebay.com/sell/inventory/v1/bulk_create_or_replace_inventory_item" CHUNK_SIZE = 25 def _chunk(lst: list, size: int): it = iter(lst) while chunk := list(islice(it, size)): yield chunk def bulk_create_inventory_items( inventory_bodies: list[dict], access_token: str, ) -> list[dict]: """ skuキーを含む各SKU分のリクエストボディを25件ずつ一括投入する。 Returns: 各チャンクのAPIレスポンスのリスト。 responses[i]["responses"][j]["statusCode"] で個別の成否を確認すること。 """ headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "Content-Language": "en-US", } all_responses = [] for chunk in _chunk(inventory_bodies, CHUNK_SIZE): resp = requests.post( BULK_URL, json={"requests": chunk}, headers=headers, timeout=60, ) resp.raise_for_status() all_responses.append(resp.json()) return all_responses 補足: bulkレスポンスのHTTP 207（Multi-Status）に注意 25件中一部が成功・一部が失敗した場合でも、HTTPレスポンス自体は200系になります。resp.raise_for_status()だけではエラーを検知できないため、responses[i]["responses"][j]["statusCode"]を個別に確認し、204以外（400系等）のエントリを抽出して再処理キューに積む実装が必要です。 失敗SKUの再処理キュー設計 バッチ処理中の部分失敗に対して、「その場でリトライ」よりも「失敗キューに積んで後で処理する」アーキテクチャが堅牢です。以下はJSONLファイルベースの軽量なキュー実装例です。 # retry_queue.py — 失敗SKUをJSONLファイルで管理する import json from pathlib import Path from datetime import datetime RETRY_QUEUE = Path("retry_queue.jsonl") def enqueue_failed(sku: str, error_msg: str, body: dict) -> None: """失敗したSKUを再処理キューに追記する。""" entry = { "sku": sku, "error": error_msg, "body": body, "queued_at": datetime.utcnow().isoformat(), "retry_count": 0, } with RETRY_QUEUE.open("a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") def drain_retry_queue(access_token: str, max_retries: int = 3) -> None: """キューの失敗SKUを順次再処理し、成功分はキューから削除する。""" if not RETRY_QUEUE.exists(): return remaining = [] for line in RETRY_QUEUE.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue entry = json.loads(line) if entry["retry_count"] >= max_retries: # 上限超過 → 人手確認リストに転記（実装省略） continue # ここで createOrReplaceInventoryItem を再呼び出し（実装省略） # 失敗した場合のみ remaining に積む entry["retry_count"] += 1 remaining.append(entry) RETRY_QUEUE.write_text( "\n".join(json.dumps(e, ensure_ascii=False) for e in remaining), encoding="utf-8", ) JSONLファイルは行ごとに1件のJSONを格納する形式で、追記が容易かつ部分読み込みが可能なため、ジョブキューの簡易実装に適しています。スループット要件が高まった際はこのキューをAWS SQSやRedisに差し替えるだけで、パイプラインの主要ロジックを変えずにスケールアップできます。 並列処理の観点では、concurrent.futures.ThreadPoolExecutorを使って複数SKUのAPI呼び出しを同時実行することも有効です。ただし、eBay APIには1セラーあたりのレートリミットがあるため、max_workers=5程度から始めて429（Too Many Requests）が出ないか確認しながら調整してください。 まとめ 本記事では、Inventory Mapping APIが出力したAI推奨結果（ListingPreview）をInventory APIのcreateOrReplaceInventoryItemに連携するE2Eパイプラインを実装しました。 ベースライン: ListingPreviewのGraphQL aspects配列（[{localizedAspectName, value: [...]}]）をInventory API REST形式の辞書（{"Brand": ["Nike"]}）に変換する_convert_aspects関数と、title・description・imageUrlsのシンプルなフィールドマッピングを確立しました。 深いポイント: ①mappingReferenceIdはREST API専用フィールドがないため内部SQLiteでSKUと1対1に永続化するトレーサビリティ設計を採用。②COMPLETED_WITH_ERRORの商品が品質不足のまま出品されるリスクを_validate_previewの事前チェックで排除。③conditionはInventory Mapping APIが推奨しないため外部から注入する設計が必須。 スケーリング: bulkCreateOrReplaceInventoryItemで25件単位の一括投入を実現し、部分失敗はJSONLキューで管理して後回し再処理することで、大量カタログにも対応できるアーキテクチャを構築しました。 これでInventory Mapping API三部作（第18-20回）が完結しました。第18回のタスク生成・第19回のポーリングと結果取得・第20回の実出品連携——この3ステップを繋ぐことで、商品メタデータを入力するだけでAIがカテゴリ・アスペクトを決定し、Inventory APIへ自動登録される完全自動化パイプラインが完成しました。 次のステップ 次回（#21）からは新しいAPIカテゴリ「Message API」に入ります。 「Message API入門：getConversationsとgetConversationでバイヤーとのメッセージを取得する」をテーマに、ECオペレーションで欠かせないカスタマーコミュニケーション自動化の第一歩を解説します。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#19】eBay Inventory Mapping API②：タスク完了をポーリングしてプレビュー結果を安全に取得する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第19回です。 前回（#18）では、GraphQL mutation の startListingPreviewsCreation を呼び出し、商品データを投入してタスク ID を取得するところまでを実装しました。しかし、ここで多くの開発者が立ち止まります。「タスク ID をもらったはいいが、結果はどうやって受け取るのか？ いつ完了するのか？」——本記事はまさにその問いに答えます。 Inventory Mapping API の処理は AI による推論を伴うため、同期的に結果が返ってきません。listingPreviewsCreationTaskById という Query を一定間隔で呼び出し（ポーリング）、タスクが完了するまで待つ設計が必要です。ただし「ただ繰り返し呼べばいい」というものでもなく、レート制限・タイムアウト・エラー判別という3つの壁が待ち構えています。 この記事で得られること: listingPreviewsCreationTaskById の result フィールドを使った「null 判定パターン」の完全理解——IN_PROGRESS のような enum は存在しない、という eBay API 特有の設計思想を把握できます。 COMPLETED / COMPLETED_WITH_ERROR / FAILED という 3 種類の completionStatus と、それに付随する invalidProducts・unmappedProducts・unprocessedProducts の違いを実務レベルで理解できます。 指数バックオフ付きポーリング・タイムアウト制御・エラー別ハンドリングを備えた、本番稼働に耐えるプロダクションレベルの Python 実装を習得できます。 背景・なぜこれが重要か (Motivation) 「GraphQL の mutation を呼んだら、その場で結果も返ってくるんじゃないの？」 Inventory Mapping API を初めて触る開発者のほぼ全員が、最初にこの疑問を抱きます。確かに、多くの GraphQL API はリクエストに対してそのまま結果を同期的に返します。しかし eBay の Inventory Mapping API は根本的に異なる設計を採用しています。 なぜ非同期なのでしょうか。Inventory Mapping API が内部で行っていることを考えると納得できます。あなたが投入した商品データ（タイトル・説明・外部商品 ID など）に対して、eBay の AI エンジンが「この商品は eBay カタログの何に対応するか」「最適なカテゴリはどこか」「Item Specifics は何を付けるべきか」という推論処理を走らせています。数十件であれば数秒で終わることもありますが、数百件のバッチや複雑な商品では数分から最長 10 分程度かかることもあります。これを同期的に処理しようとすると、HTTP 接続がタイムアウトしてしまいます。 そのため、eBay は「タスクを受け付けた」という証明としてタスク ID（前回取得）を即時返却し、実際の処理は非同期で進める設計を採用しています。クライアント側はそのタスク ID を使って「もう終わったか？」と定期的に問い合わせる——これがポーリングパターンです。 補足: eBay の「null 判定パターン」について 処理中かどうかを判定するために、eBay は IN_PROGRESS のような明示的な enum 値を使いません。その代わり、result フィールドそのものが null かどうかで状態を表現します。result が null → まだ処理中。result に値が入っている → 処理完了（成否は completionStatus で判断）。このパターンを知らずにコードを書くと、null チェックを忘れてデータをパースしようとして AttributeError や TypeError に悩まされることになります。 基本的な使い方（ベースライン）：listingPreviewsCreationTaskById で結果を取得する まず最小限のポーリングループを実装します。GraphQL の Query を requests で送り、result フィールドが null かどうかをチェックして、値が返ってくるまでループします。 以下が最小実装のコードです。 # polling_baseline.py import time import requests GRAPHQL_URL = "https://api.ebay.com/sell/listing_preview/graphql" QUERY = """ query GetTask($id: ID!) { listingPreviewsCreationTaskById(input: { id: $id }) { requestedId listingPreviewsCreationTask { id result { completionStatus listingPreviews { sku title mappingReferenceId category { id } } invalidProducts { externalProductId errors { message } } unmappedProducts { externalProductId } unprocessedProducts { externalProductId } } } } } """ def poll_task(task_id: str, access_token: str, interval: int = 10) -> dict: """ タスクが完了するまでポーリングし、result を返す（最小実装）。 interval: ポーリング間隔（秒） """ headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", } variables = {"id": task_id} while True: resp = requests.post( GRAPHQL_URL, json={"query": QUERY, "variables": variables}, headers=headers, timeout=30, ) resp.raise_for_status() data = resp.json() task = ( data.get("data", {}) .get("listingPreviewsCreationTaskById", {}) .get("listingPreviewsCreationTask") ) if task is None: raise ValueError(f"タスクが見つかりません: {task_id}") # null 判定パターン: result が None なら処理中 result = task.get("result") if result is not None: return result # 完了（status は呼び出し側で判定） print(f"[polling] タスク {task_id} は処理中です。{interval}秒後に再確認します...") time.sleep(interval) # 使用例 if __name__ == "__main__": TASK_ID = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" # 第18回で取得したタスクID TOKEN = "v^1.1.xxxx..." result = poll_task(TASK_ID, TOKEN) print(f"completionStatus: {result['completionStatus']}") print(f"listingPreviews 件数: {len(result['listingPreviews'])}") 補足: completionStatus の 3 種類の値 result が返ってきたら、その中の completionStatus を確認します。取り得る値は 3 種類です。 【COMPLETED】すべての商品が正常にマッピングされ、listingPreviews にプレビューが格納されています。 【COMPLETED_WITH_ERROR】一部の商品は正常にマッピングされましたが、一部の商品に問題がありました。listingPreviews と問題商品リスト（invalidProducts / unmappedProducts / unprocessedProducts）が混在します。 【FAILED】タスク全体が失敗しました。listingPreviews は空です。 この上記の最小実装には、実務で致命的になる欠陥がいくつかあります。次のセクションで詳しく解説します。 実務で躓く場面・深いポイント (Core) ベースライン実装を本番環境に持ち込むと、必ずいくつかの壁にぶつかります。ここでは、現場で頻出する3つの落とし穴とその回避策を解説します。 1. 固定間隔の無限ループはレート制限を引き起こす 上記の最小実装では、interval=10（秒）で固定間隔のポーリングを行っています。一見問題なさそうですが、次のシナリオを考えてみてください。「10件の商品を並行して処理する5本のポーリングループを、1サーバーで同時実行している」——この場合、1分間に 5本 × 6回 = 30 回の API コールが発生します。商品数が増えてバッチサイズが大きくなると、あっという間に eBay の API レート制限に抵触します。 さらに深刻な問題は、「タスクが高速で完了しそうな場合」です。商品数が少なければ数秒で完了することもありますが、固定間隔だと最大 interval 秒のロスが生じます。逆に「処理が長引いているとき」に同じ短い間隔でポーリングし続けるのは明らかな無駄です。 解決策は「指数バックオフ（Exponential Backoff）」です。最初は短い間隔（例: 5秒）でポーリングを開始し、完了していなければ間隔を徐々に延ばしていきます（10秒 → 20秒 → 最大 60秒）。これにより、高速完了のタスクへの応答性を保ちながら、長時間かかるタスクへの無駄なコールを削減できます。 2. COMPLETED_WITH_ERROR を見落として不完全なデータで後続処理を進める罠 Inventory Mapping API 実装で最も多いバグは「completionStatus の確認漏れ」です。具体的には、result が返ってきた瞬間に「完了 = 成功」と判断して listingPreviews をそのまま処理してしまうパターンです。 例えば 100件の商品を投入し、80件は正常にマッピングされたが 20件は問題があった場合、completionStatus は COMPLETED_WITH_ERROR になります。この状態で listingPreviews だけを見ると 80件は問題なく見えます。しかし残りの 20件がどうなったのかを確認しないまま Inventory API に渡すと、「投入した 100件のうち 80件しか出品されていない」というサイレントバグが生まれます。このようなバグは検知が難しく、実務では非常に発見が遅れます。 必ず completionStatus を明示的に確認し、COMPLETED_WITH_ERROR の場合は問題商品を別途ログ・再試行キューに振り分ける処理を実装してください。 3. invalidProducts / unmappedProducts / unprocessedProducts の区別を誤ると原因調査が長引く COMPLETED_WITH_ERROR のとき、問題のあった商品は 3 種類のリストに振り分けられます。それぞれの意味を正確に理解しないと、「なぜこの商品がマッピングされなかったのか」の原因調査に無駄な時間がかかります。 【invalidProducts】：投入したデータ自体に不備がある商品です。例として、externalProductId が空、必須フィールドが欠落している、フォーマットが不正などが該当します。この場合の対処は「入力データの修正」であり、同じデータを再送しても同じ結果になります。errors フィールドに具体的なエラーメッセージが入っているため、必ず参照してください。 【unmappedProducts】：入力データ自体は有効だが、eBay カタログに対応する商品が見つからなかった商品です。つまり「データは問題ないが、AI が eBay の商品カタログと照合できなかった」状態です。対処としては、タイトルや説明を補強して再試行する、または手動でカテゴリ・Item Specifics を指定する方法があります。 【unprocessedProducts】：システム的な理由（タイムアウト、内部エラーなど）で処理が完了しなかった商品です。データに問題があるわけではないため、そのまま再試行（リトライ）することで成功する可能性が高いです。これを invalidProducts と混同して「データを修正」しようとすると、無駄な作業が発生します。 注意: タイムアウト設計について AI による推論処理は、商品数・複雑さによって処理時間が大きく変動します。eBay の公式ドキュメントでは最長 10 分程度かかる可能性が示唆されています。そのため、ポーリングには必ず上限時間（タイムアウト）を設定してください。推奨は 15〜20 分（余裕を持って設定）。タイムアウトした場合は、タスク ID を記録した上でエラーとして扱い、アラートを発報して運用チームが確認できるようにしてください。タイムアウト後のタスクが実は完了していた場合でも、タスク ID は有効なため後から結果を取得できます。 頻出エラーコード早見表 ポーリング中に発生しうる HTTP / GraphQL エラーと対処法をまとめます。 HTTP ステータス / エラー内容 原因と対処法 HTTP 401 Unauthorized アクセストークンの有効期限切れ。OAuth 2.0 のトークンリフレッシュを行い、新しいトークンで再試行してください。 HTTP 429 Too Many Requests API レート制限に抵触。ポーリング間隔を延ばす（指数バックオフ）か、並行ポーリング数を削減してください。 HTTP 500 / 503 eBay 側の一時的なサーバーエラー。最大3回まで指数バックオフでリトライ。継続する場合は eBay Developer Support に連絡。 GraphQL: "Task not found" タスク ID が無効または別の OAuth スコープのトークンを使用している。スコープを確認し、正しいトークンを使用してください。 GraphQL: "Access denied" 必要な OAuth スコープ（sell.listing_preview）がトークンに含まれていない。 堅牢な実装：指数バックオフとエラー別ハンドリングを備えたポーリングエンジン 上記の3つの落とし穴をすべてクリアした、本番稼働に耐えるポーリング実装を示します。型アノテーション・docstring・例外処理・入力バリデーション・指数バックオフ・タイムアウト・completionStatus 別の結果ハンドリングを完備しています。 # inventory_mapping_poller.py """ Inventory Mapping API タスクポーリングエンジン。 指数バックオフ・タイムアウト・completionStatus 別ハンドリングを実装。 Usage: poller = InventoryMappingPoller(access_token=os.environ["EBAY_ACCESS_TOKEN"]) outcome = poller.poll(task_id="xxxxxx-xxxx-xxxx") if outcome.is_success: for preview in outcome.listing_previews: print(preview["sku"], preview["title"]) """ from __future__ import annotations import logging import time from dataclasses import dataclass, field from typing import Any import requests logger = logging.getLogger(__name__) GRAPHQL_URL = "https://api.ebay.com/sell/listing_preview/graphql" QUERY = """ query GetTaskById($id: ID!) { listingPreviewsCreationTaskById(input: { id: $id }) { requestedId listingPreviewsCreationTask { id result { completionStatus listingPreviews { sku title mappingReferenceId description category { id } images { value } aspects { name values } } invalidProducts { externalProductId errors { message errorId } } unmappedProducts { externalProductId title } unprocessedProducts { externalProductId title } } } } } """ @dataclass class PollOutcome: """ポーリング結果を格納するデータクラス。""" task_id: str completion_status: str # COMPLETED / COMPLETED_WITH_ERROR / FAILED listing_previews: list[dict[str, Any]] # 正常にマッピングされた商品 invalid_products: list[dict[str, Any]] # 入力データ不備 unmapped_products: list[dict[str, Any]] # カタログ照合失敗 unprocessed_products: list[dict[str, Any]] # システムエラーで未処理 timed_out: bool = False error_message: str = "" @property def is_success(self) -> bool: """COMPLETED のみ True（COMPLETED_WITH_ERROR は False）。""" return self.completion_status == "COMPLETED" @property def has_partial_results(self) -> bool: """一部成功・一部失敗の場合 True。""" return self.completion_status == "COMPLETED_WITH_ERROR" @property def problem_count(self) -> int: """問題のあった商品の合計件数。""" return ( len(self.invalid_products) + len(self.unmapped_products) + len(self.unprocessed_products) ) @dataclass class PollerConfig: """ポーリング設定。""" initial_interval: float = 5.0 # 初回待機時間（秒） backoff_factor: float = 1.8 # 間隔の乗数 max_interval: float = 60.0 # 最大ポーリング間隔（秒） timeout_seconds: float = 900.0 # タイムアウト（15分） max_server_retries: int = 3 # HTTP 5xx 連続エラーの最大リトライ数 class InventoryMappingPoller: """ listingPreviewsCreationTaskById を指数バックオフでポーリングし、 タスク完了後に PollOutcome を返すポーリングエンジン。 """ def __init__(self, access_token: str, config: PollerConfig | None = None) -> None: if not access_token: raise ValueError("access_token が空です。OAuth 2.0 トークンを設定してください。") self._token = access_token self._config = config or PollerConfig() def poll(self, task_id: str) -> PollOutcome: """ タスクが完了するまでポーリングを実行し、PollOutcome を返す。 Args: task_id: 第18回で取得した startListingPreviewsCreation のタスク ID。 Returns: PollOutcome: completionStatus 別の結果データ。 Raises: ValueError: task_id が空の場合。 RuntimeError: 最大リトライ数を超えたサーバーエラーが発生した場合。 """ if not task_id or not task_id.strip(): raise ValueError("task_id が空です。") cfg = self._config interval = cfg.initial_interval start_time = time.monotonic() server_error_count = 0 logger.info("ポーリング開始: task_id=%s, timeout=%ss", task_id, cfg.timeout_seconds) while True: # タイムアウト判定 elapsed = time.monotonic() - start_time if elapsed >= cfg.timeout_seconds: logger.warning("タイムアウト: task_id=%s (%.0fs経過)", task_id, elapsed) return PollOutcome( task_id=task_id, completion_status="", listing_previews=[], invalid_products=[], unmapped_products=[], unprocessed_products=[], timed_out=True, error_message=f"タイムアウト（{cfg.timeout_seconds}秒経過）。タスクIDを記録して後から再確認してください。", ) try: result = self._call_api(task_id) server_error_count = 0 # 成功したらリセット except requests.HTTPError as exc: status_code = exc.response.status_code if exc.response else 0 if status_code == 401: # トークン切れは再試行しても意味がないのですぐ終了 raise RuntimeError( "HTTP 401: アクセストークンが無効または期限切れです。" "トークンをリフレッシュして再実行してください。" ) from exc if status_code == 429: # レート制限: 最大 interval の 2 倍待つ wait = min(interval * 2, cfg.max_interval) logger.warning("HTTP 429 レート制限。%.0f秒待機します...", wait) time.sleep(wait) continue if status_code >= 500: server_error_count += 1 if server_error_count > cfg.max_server_retries: raise RuntimeError( f"HTTP {status_code} が {cfg.max_server_retries} 回連続しました。" "eBay サーバー側の問題の可能性があります。" ) from exc logger.warning( "HTTP %s サーバーエラー（%d/%d回目）。リトライします...", status_code, server_error_count, cfg.max_server_retries, ) time.sleep(interval) continue raise # その他の HTTP エラーは再スロー # GraphQL エラーチェック if "errors" in result: messages = [e.get("message", "") for e in result["errors"]] raise RuntimeError(f"GraphQL エラー: {'; '.join(messages)}") task_data = ( result.get("data", {}) .get("listingPreviewsCreationTaskById", {}) .get("listingPreviewsCreationTask") ) if task_data is None: raise RuntimeError( f"タスクが見つかりません（task_id={task_id}）。" "IDが正しいか、同じ OAuth スコープのトークンを使っているか確認してください。" ) # null 判定パターン: result が None なら処理中 task_result = task_data.get("result") if task_result is None: logger.info( "処理中: task_id=%s (経過 %.0fs) → %.0f秒後に再確認", task_id, elapsed, interval, ) time.sleep(interval) # 指数バックオフで interval を更新 interval = min(interval * cfg.backoff_factor, cfg.max_interval) continue # --- タスク完了 --- status = task_result.get("completionStatus", "FAILED") previews = task_result.get("listingPreviews", []) or [] invalid = task_result.get("invalidProducts", []) or [] unmapped = task_result.get("unmappedProducts", []) or [] unprocessed = task_result.get("unprocessedProducts", []) or [] outcome = PollOutcome( task_id=task_id, completion_status=status, listing_previews=previews, invalid_products=invalid, unmapped_products=unmapped, unprocessed_products=unprocessed, ) self._handle_outcome(outcome) return outcome def _call_api(self, task_id: str) -> dict[str, Any]: """GraphQL API を 1 回呼び出して生のレスポンス dict を返す。""" headers = { "Authorization": f"Bearer {self._token}", "Content-Type": "application/json", } response = requests.post( GRAPHQL_URL, json={"query": QUERY, "variables": {"id": task_id}}, headers=headers, timeout=30, ) response.raise_for_status() return response.json() def _handle_outcome(self, outcome: PollOutcome) -> None: """completionStatus に応じてログ出力・問題商品の分類を行う。""" status = outcome.completion_status if status == "COMPLETED": logger.info( "COMPLETED: task_id=%s | listingPreviews=%d件", outcome.task_id, len(outcome.listing_previews), ) elif status == "COMPLETED_WITH_ERROR": logger.warning( "COMPLETED_WITH_ERROR: task_id=%s | 正常=%d件 / 問題=%d件", outcome.task_id, len(outcome.listing_previews), outcome.problem_count, ) # invalidProducts: 入力データ不備 → 修正が必要 for item in outcome.invalid_products: errors = [e.get("message", "") for e in (item.get("errors") or [])] logger.error( " [INVALID] externalProductId=%s | errors=%s", item.get("externalProductId"), errors, ) # unmappedProducts: カタログ照合失敗 → タイトル補強 or 手動設定 for item in outcome.unmapped_products: logger.warning( " [UNMAPPED] externalProductId=%s | title=%s", item.get("externalProductId"), item.get("title", ""), ) # unprocessedProducts: システムエラーで未処理 → そのまま再試行 for item in outcome.unprocessed_products: logger.warning( " [UNPROCESSED] externalProductId=%s → リトライキューへ追加", item.get("externalProductId"), ) elif status == "FAILED": logger.error( "FAILED: task_id=%s | listingPreviews は空です。" "投入データを確認して再実行してください。", outcome.task_id, ) else: logger.error("不明な completionStatus: %s (task_id=%s)", status, outcome.task_id) # ============================================================ # 実行例 # ============================================================ if __name__ == "__main__": import os import json logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) TOKEN = os.environ["EBAY_ACCESS_TOKEN"] TASK_ID = os.environ["TASK_ID"] # 第18回で取得したタスクID poller = InventoryMappingPoller(access_token=TOKEN) outcome = poller.poll(task_id=TASK_ID) if outcome.timed_out: print(f"タイムアウト: {outcome.error_message}") elif outcome.is_success: print(f"完全成功: {len(outcome.listing_previews)} 件のプレビューを取得しました。") # 第20回: Inventory API へ渡す処理をここに追加 elif outcome.has_partial_results: print(f"部分成功: {len(outcome.listing_previews)} 件成功 / {outcome.problem_count} 件に問題あり") # unprocessed を再試行キューへ retry_ids = [p.get("externalProductId") for p in outcome.unprocessed_products] print(f"再試行対象: {json.dumps(retry_ids, ensure_ascii=False)}") else: print("タスク全体が失敗しました。ログを確認してください。") このポーリングエンジンの設計上の重要ポイントをまとめます。 【1】指数バックオフの実装: initial_interval=5秒から開始し、backoff_factor=1.8 で間隔を拡大（5 → 9 → 16 → 29 → 52 → 60秒でキャップ）。処理が速く終わるタスクへの応答性を保ちつつ、長時間処理時の無駄なコールを抑制します。 【2】タイムアウトのハードコードを避ける: PollerConfig の timeout_seconds をデフォルト 900秒（15分）に設定しつつ、呼び出し元が上書きできる設計にしています。処理規模に応じて柔軟に調整できます。 【3】completionStatus 別の分岐処理: _handle_outcome メソッドが各種問題商品を種別ごとに記録します。invalidProducts は修正対象、unprocessedProducts は即時リトライ対象、unmappedProducts は人手レビュー対象——それぞれの対処方針が異なることをコード上で明示しています。 【4】PollOutcome データクラスの活用: タスク ID・ステータス・プレビューリスト・問題商品リストをひとつのオブジェクトにまとめることで、後続処理（第20回の Inventory API への受け渡し）がシンプルに書けるようになります。 パフォーマンス・スケーリング視点 (深度) 1日数十件の商品を単発で処理するフェーズでは、上記のシンプルな同期ポーリングで十分です。しかし、商品カタログが数千件規模になり、複数のタスクを並行処理する必要が出てくると、設計を見直す必要があります。 複数タスクの並行ポーリング設計 例えば、1000件の商品を 100件ずつ 10個のタスクに分割して startListingPreviewsCreation に投入するケースを考えます。各タスクを順番にポーリングすると、最後のタスクの結果を受け取るまでに、最初のタスクの処理待ち時間 × タスク数 だけの余分な時間がかかってしまいます。 解決策は concurrent.futures.ThreadPoolExecutor を使った並行ポーリングです。以下のコードで、複数タスクを同時にポーリングできます。 # concurrent_polling.py import concurrent.futures import os import logging from inventory_mapping_poller import InventoryMappingPoller, PollerConfig, PollOutcome logger = logging.getLogger(__name__) # 同時ポーリング数は3〜5が実務的な上限。 # 多すぎると HTTP 429 が頻発し、かえって遅くなります。 MAX_WORKERS = 4 def poll_all_tasks( task_ids: list[str], access_token: str, ) -> dict[str, PollOutcome]: """ 複数のタスクを並行してポーリングし、task_id -> PollOutcome の dict を返す。 Args: task_ids: 第18回で取得したタスク ID のリスト。 access_token: eBay OAuth 2.0 アクセストークン。 Returns: dict[str, PollOutcome]: タスクIDをキーとした結果の辞書。 """ poller = InventoryMappingPoller( access_token=access_token, config=PollerConfig( initial_interval=8.0, # 並行時は間隔を少し広げる backoff_factor=2.0, max_interval=60.0, timeout_seconds=1200.0, # 並行タスクが多い場合はタイムアウトを延ばす ), ) results: dict[str, PollOutcome] = {} with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: future_to_id = { executor.submit(poller.poll, task_id): task_id for task_id in task_ids } for future in concurrent.futures.as_completed(future_to_id): task_id = future_to_id[future] try: outcome = future.result() results[task_id] = outcome logger.info("完了: %s → %s", task_id, outcome.completion_status) except Exception as exc: logger.error("エラー: %s → %s", task_id, exc) return results if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") TASK_IDS = [ "task-id-001", "task-id-002", "task-id-003", ] TOKEN = os.environ["EBAY_ACCESS_TOKEN"] all_results = poll_all_tasks(TASK_IDS, TOKEN) total_previews = sum(len(r.listing_previews) for r in all_results.values()) print(f"合計 {len(all_results)} タスク完了 | 取得プレビュー数: {total_previews}") 注意: MAX_WORKERS を大きくしすぎないでください。 4〜5 を超えると HTTP 429（Too Many Requests）が頻発し、かえってスループットが低下します。実務では MAX_WORKERS=3〜4 から始めて、エラーレートを監視しながら調整してください。 補足: Notification API 連携による「プッシュ型」への移行展望 現状のポーリング設計はシンプルですが、大規模になると「無駄なコール」が増えます。将来的には eBay の Notification API（SetNotificationPreferences / 第15回参照）と組み合わせることで、タスク完了時に eBay 側からコールバックを受け取る「プッシュ型」アーキテクチャに移行できます。プッシュ型では、ポーリングコール自体が不要になるため、API コール数を大幅に削減できます。大規模オペレーション（1日 1,000 件超）を目指す場合は、この移行を検討してください。 まとめ 本記事では、Inventory Mapping API の非同期タスク処理において、最も重要な「結果の取得」を実装しました。 ベースライン: listingPreviewsCreationTaskById の result フィールドを使った null 判定パターンを理解しました。result が null なら処理中、値が入ったら completionStatus で成否を判断——この設計思想が eBay 非同期 API の基本です。 深いポイント: 固定間隔ポーリングによるレート制限の罠、COMPLETED_WITH_ERROR の見落とし、そして invalidProducts / unmappedProducts / unprocessedProducts という3種類の問題商品リストの区別——これらを正確に理解することで、原因不明のサイレントバグを予防できます。 スケーリング: 指数バックオフ付きのポーリングエンジンに加え、ThreadPoolExecutor を使った複数タスクの並行ポーリング設計を習得しました。将来的には Notification API 連携によるプッシュ型への移行で、さらなる効率化が可能です。 これで、Inventory Mapping API からプレビュー結果を安全かつ効率的に取得する基盤が整いました。次はいよいよ、取得したプレビューデータを活用して、実際に eBay に商品を出品するステップです。 次のステップ 今回取得した listingPreviews には、AI が推奨するカテゴリ・Item Specifics・タイトルが格納されています。次回（#20）は、このプレビューデータを Inventory API（createOrReplaceInventoryItem / publishOffer）に渡し、高品質な出品を自動作成する完全な自動出品パイプラインを構築します。Inventory Mapping API → ポーリング → Inventory API という一連のフローが完成する、この連載のひとつの到達点となります。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#17】eBay Trading API：GetFeedback と LeaveFeedback でフィードバックデータ取得と自動評価送信を実装する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第17回です。 前回（#16）は、GetCategories と GetCategoryFeatures を用いて eBay のカテゴリ構造とバリデーションルールをプログラムから取得する方法を解説しました。これにより、出品する前の「カテゴリ選定・入力チェック」フェーズが自動化でき、出品システムとしての精度が大きく高まりました。 今回のテーマは 「フィードバック（評価）管理」 です。eBay においてフィードバックは、セラーのアカウント健康度と検索ランキングに直結する極めて重要な指標でありながら、多くのセラーが手動対応のまま放置しています。本記事では、GetFeedback と LeaveFeedback という2つの Trading API メソッドを活用し、評価の取得・分析から発送完了後の自動ポジティブ返信まで、Python で完結する運用システムを構築します。 この記事で得られること: GetFeedback を使ったセラー評価一覧の全件取得と、Positive / Neutral / Negative 別の集計・スコア分析の実装方法。 評価自動化で必ず踏む「タイミング問題」「二重送信」「ポリシー制約」という3大落とし穴とその回避策。 発送完了を検知してから自動的に Positive フィードバックを送る、本番稼働レベルの Python スクリプト。 背景・なぜこれが重要か (Motivation) 「好評率が高ければ万事大吉？　フィードバックなんて自然に集まるものでは？」 これは eBay 初心者が陥りがちな認識です。実際には、フィードバックは意識的に管理しなければ確実に劣化していく指標です。 eBay において Top Rated Seller（TRS）の資格を得るには、直近12ヶ月で Positive Feedback Percentage が 98% 以上、かつ一定件数以上の評価が必要です。TRS であることは単なるステータスシンボルではありません。検索結果の Best Match アルゴリズムにおいてポジティブな重み付けが行われるため、商品の可視性（Visibility）が向上し、実質的に「広告費ゼロの SEO 効果」をもたらします。 逆に、Negative（ネガティブ）評価が数件積み重なると、Positive Percentage が急落し、TRS 資格を失うのは想像以上に早いです。例えば、1,000件の評価のうち20件が Negative になると Positive Percentage は 98.0% を下回り、TRS ラインを割り込みます。一度失った TRS 資格を回復するには数ヶ月単位の時間がかかります。 一方、バイヤー目線では、購入後にセラーへのフィードバックを送ることを忘れているケースが多くあります。セラー側から LeaveFeedback で先に Positive を送ることで、バイヤーが「そういえば評価しなきゃ」と思い出し、返礼として評価を付けてくれる確率が統計的に上がります。これは評価件数の増加、ひいてはアカウント信頼性の向上につながります。 つまり、フィードバック管理の自動化は「事後処理」ではなく、アカウント健康度を維持するための「能動的なアカウント経営」です。手動管理ではスケールに限界があり、数百件・数千件の注文を処理するセラーには自動化は必須要件です。 基本的な使い方（ベースライン）：GetFeedback で評価一覧を取得する まず、zeep ライブラリを使った SOAP 呼び出しで、セラーのフィードバック一覧を取得する最小構成の実装から始めます。GetFeedback はページネーションをサポートしており、大量の評価がある場合は複数ページに分けて取得する必要があります。 # get_feedback_baseline.py import zeep import zeep.transports from dataclasses import dataclass from typing import List, Optional WSDL_URL = "https://api.ebay.com/wsapi?WSDL" # 本番環境 # Sandbox: "https://api.sandbox.ebay.com/wsapi?WSDL" @dataclass class FeedbackEntry: feedback_id: str item_id: str transaction_id: str comment_type: str # Positive / Neutral / Negative comment_text: str commenter_user_id: str role: str # Seller / Buyer comment_time: Optional[str] = None def get_feedback_list( auth_token: str, user_id: str, feedback_type: str = "FeedbackReceivedAsSeller", entries_per_page: int = 200, ) -> List[FeedbackEntry]: # GetFeedback API を呼び出し、指定ユーザーの評価一覧を全ページ取得する。 # feedback_type: FeedbackReceivedAsSeller / FeedbackReceivedAsBuyer / # FeedbackLeft / FeedbackReceived transport = zeep.transports.Transport(timeout=30) client = zeep.Client(wsdl=WSDL_URL, transport=transport) all_entries: List[FeedbackEntry] = [] page_number = 1 while True: request_body = { "RequesterCredentials": {"eBayAuthToken": auth_token}, "UserID": user_id, "FeedbackType": feedback_type, "Pagination": { "EntriesPerPage": entries_per_page, "PageNumber": page_number, }, } response = client.service.GetFeedback(**request_body) ack = getattr(response, "Ack", "Failure") if ack not in ("Success", "Warning"): errors = getattr(response, "Errors", []) raise RuntimeError( f"GetFeedback failed (Ack={ack}): " f"{[getattr(e, 'LongMessage', '') for e in errors]}" ) feedback_detail_array = getattr(response, "FeedbackDetailArray", None) if not feedback_detail_array: break # 評価なし、または全ページ取得完了 details = getattr(feedback_detail_array, "FeedbackDetail", []) for detail in details: all_entries.append( FeedbackEntry( feedback_id=str(getattr(detail, "FeedbackID", "")), item_id=str(getattr(detail, "ItemID", "")), transaction_id=str(getattr(detail, "TransactionID", "")), comment_type=str(getattr(detail, "CommentType", "")), comment_text=str(getattr(detail, "CommentText", "")), commenter_user_id=str(getattr(detail, "CommentingUser", "")), role=str(getattr(detail, "Role", "")), comment_time=str(getattr(detail, "CommentTime", "")), ) ) # ページネーション: 次のページが存在するか確認 pagination_result = getattr(response, "PaginationResult", None) if not pagination_result: break total_pages = getattr(pagination_result, "TotalNumberOfPages", 1) if page_number >= total_pages: break page_number += 1 return all_entries def analyze_feedback_score(entries: List[FeedbackEntry]) -> dict: # 取得した評価一覧を集計し、スコアを分析する。 positive = sum(1 for e in entries if e.comment_type == "Positive") neutral = sum(1 for e in entries if e.comment_type == "Neutral") negative = sum(1 for e in entries if e.comment_type == "Negative") total = positive + neutral + negative positive_percentage = (positive / total * 100) if total > 0 else 0.0 return { "positive_count": positive, "neutral_count": neutral, "negative_count": negative, "total_count": total, "positive_percentage": round(positive_percentage, 2), "is_top_rated_eligible": positive_percentage >= 98.0 and total >= 100, } # ── 動作確認 ── if __name__ == "__main__": TOKEN = "YOUR_AUTH_TOKEN_HERE" USER_ID = "your_seller_id" entries = get_feedback_list(TOKEN, USER_ID) analysis = analyze_feedback_score(entries) print(f"総評価数 : {analysis['total_count']}") print(f"Positive : {analysis['positive_count']}") print(f"Neutral : {analysis['neutral_count']}") print(f"Negative : {analysis['negative_count']}") print(f"好評率 : {analysis['positive_percentage']}%") print(f"TRS 資格候補: {analysis['is_top_rated_eligible']}") 補足: FeedbackType パラメータの選択指針 GetFeedback の FeedbackType パラメータには4種類があり、用途によって使い分けます。「FeedbackReceivedAsSeller」は、バイヤーからセラーへの評価を取得します。アカウント健康度の監視や TRS チェックはこれを使います。「FeedbackLeft」は、自分がバイヤーに対して過去に残した評価の履歴を確認できます。二重送信チェックなどに利用します。「FeedbackReceived」は全タイプを混在取得しますが、フィルタリングコストが増えるため目的が明確な場合は専用タイプを選ぶことを推奨します。 補足: CommentType の3種類とスコアへの影響 FeedbackDetail の CommentType には Positive（好評）、Neutral（中立）、Negative（否定）の3種類があります。Feedback Score（累計スコア数）の計算式は「Positive件数 − Negative件数」であり、Neutral は加算も減算もされません。一方、Positive Feedback Percentage（好評率）は「Positive ÷ (Positive + Neutral + Negative) × 100」で算出されます。好評率においては Neutral も分母に含まれるため、Neutral が増えると好評率が下がる点に注意が必要です。 実務で躓く場面・深いポイント (Core) ここからは、フィードバック自動化システムを実稼働させる上で、多くのエンジニアが必ず数時間単位でハマる3つの大きな落とし穴と、その回避策を解説します。 1. フィードバックを残せる「期限」と「タイミング」の落とし穴 eBay のフィードバックには、取引完了から 60日以内 という有効期限があります。これを超えた取引に対して LeaveFeedback を呼び出すと、Error 819（"This transaction is not eligible for Feedback"）が返ります。長い出品期間を持つ商品や、国際配送で時間がかかる商品を扱うセラーは、取引完了のタイムスタンプを必ず記録し、定期的に60日期限をチェックする仕組みが不可欠です。 しかし、さらに重要な問題があります。「取引完了 = 即座にフィードバックを送ってよい」という勘違いです。 eBay には、取引に対してケース（Case）や申請（Request）が開かれている場合があります。バイヤーが「商品が届かない（Item Not Received）」や「商品が説明と異なる（Not as Described）」という申請をオープンしている最中に Positive フィードバックを送ってしまうと、バイヤーにとって「セラーが一方的に問題を解決済みとみなそうとしている」という印象を与えかねません。これはむしろ Negative 評価を引き起こすリスクを高めます。 注意: 自動フィードバック送信のタイミングポリシー eBay の公式ガイドラインでも、オープン中のケース・リクエスト・申請がある取引へのフィードバックは推奨されていません。堅牢な実装では、GetOrders API で取引のステータスを確認するか、発送確認（Shipped ステータス）から一定日数のバッファを設けてからフィードバックを送信するロジックが必要です。目安は国内配送で発送後7日、国際配送で発送後21日程度です。この「待機バッファ」のパラメータは設定値として外部化し、商品カテゴリや配送先国ごとに調整できる設計にしておくと実運用で柔軟に対応できます。 2. セラーはネガティブ評価をバイヤーに対して残せない（2008年ポリシー変更） これは2008年の eBay ポリシー変更ですが、現在でも意外と知らないエンジニアが多い重要な仕様です。 2008年以降、セラーはバイヤーに対して Negative または Neutral の評価を残すことができません。LeaveFeedback の CommentType に "Negative" や "Neutral" を指定して送信すると、Error 821（"You cannot leave a negative or neutral comment for a buyer"）が返ります。 これは eBay が「セラーからの報復評価（Retaliatory Feedback）」を防ぐために意図的に設けた制約です。バイヤーがクレームを付けた後にセラーが Negative を返すことを恐れてバイヤーがクレームを躊躇するという悪循環を断ち切るための仕組みです。 この仕様を知らずに CommentType を動的に設定するスクリプトを組むと、バグの温床になります。セラーがバイヤーに送るフィードバックは、常に CommentType="Positive" にハードコードするのが正解です。CommentText（本文）のみをビジネスロジックに応じて動的に生成するアーキテクチャにすることで、このポリシー制約を構造的に守ることができます。 3. LeaveFeedback の二重送信ガードと eBay レート制限 LeaveFeedback には eBay 側のレート制限があります。同一ユーザーへのフィードバックは特定の時間窓内で回数制限があり、大量の注文を一括処理する際に無視すると途中から Error 21916588 が返り始めます。これは大量注文を一度にバルク処理するセラーが特に踏みやすい罠です。 同じく重大なのが二重送信です。同一取引（ItemID + TransactionID の組み合わせ）に対してフィードバックを2回送信しようとすると Error 819 が返り、2回目の送信は失敗します。eBay 側の保護機能があるとはいえ、「なぜ失敗したのか」をログで追跡できないシステムでは、単純な API エラーと区別がつかなくなります。 送信済みフィードバックを追跡するために、ローカル DB（SQLite、PostgreSQL、RDS 等）に(item_id, transaction_id, feedback_sent_at, status) のレコードを記録し、送信前にこのテーブルを参照する冪等性チェックが必須です。DB を挟むことで、API 呼び出し失敗時のリトライも安全に行えます。 頻出エラーコード早見表 LeaveFeedback / GetFeedback を実装する際に頻繁に遭遇するエラーコードをまとめます。 エラーコード メッセージ（略） 対処法 819 Transaction is not eligible for Feedback 取引完了から60日以上経過、または同一取引に送信済み。送信済みDBを確認し、期限切れはスキップする。 821 Cannot leave negative/neutral for a buyer セラーはバイヤーへ Negative/Neutral を送信不可。CommentType を常に "Positive" にハードコードする。 21916588 Rate limit exceeded for LeaveFeedback 短時間に大量送信でレート制限に到達。time.sleep() によるスロットリングを実装し、再試行は指数バックオフで行う。 1030 The specified user is suspended バイヤーのアカウントが停止状態。GetUser でユーザーステータスを事前確認してスキップ処理を追加する。 21917248 You cannot leave feedback for yourself テスト環境でセラーとバイヤーが同一 UserID のケース。Sandbox では別アカウントを用意する。 堅牢な実装：発送完了後に自動 Positive フィードバックを送るスクリプト 上記の落とし穴をすべてクリアした、本番稼働レベルの自動フィードバック送信システムを実装します。設計の要点は以下の3点です。 ① ローカル DB（SQLite）による送信済みチェックで二重送信を防ぐ冪等性の確保。 ② 発送完了からの経過日数バッファチェックにより、紛争中取引への誤送信を防ぐ。 ③ レート制限対策のスロットリングと、指数バックオフによるリトライロジック。 # auto_feedback_sender.py import sqlite3 import time import logging import zeep import zeep.transports from datetime import datetime, timezone, timedelta from dataclasses import dataclass from typing import List, Optional logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s" ) logger = logging.getLogger(__name__) WSDL_URL = "https://api.ebay.com/wsapi?WSDL" # ── データモデル ── @dataclass class ShippedOrder: # 発送済み注文情報（GetOrders などから取得したデータを想定） item_id: str transaction_id: str buyer_user_id: str shipped_at: datetime # 発送完了のタイムスタンプ（UTC） order_id: str = "" # ── DB管理: 送信済みフィードバック追跡 ── def init_db(db_path: str = "feedback_tracker.db") -> sqlite3.Connection: # フィードバック送信履歴を管理する SQLite テーブルを初期化する。 # ItemID + TransactionID をユニーク制約で管理し冪等性を保証する。 conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS sent_feedback ( id INTEGER PRIMARY KEY AUTOINCREMENT, item_id TEXT NOT NULL, transaction_id TEXT NOT NULL, buyer_user_id TEXT NOT NULL, sent_at TEXT NOT NULL, status TEXT NOT NULL, error_message TEXT, UNIQUE (item_id, transaction_id) ) """) conn.commit() return conn def is_feedback_already_sent( conn: sqlite3.Connection, item_id: str, transaction_id: str ) -> bool: # 指定取引のフィードバック送信済みかどうかをDBで確認する。 cursor = conn.execute( "SELECT 1 FROM sent_feedback " "WHERE item_id = ? AND transaction_id = ? AND status = 'success'", (item_id, transaction_id), ) return cursor.fetchone() is not None def record_feedback_result( conn: sqlite3.Connection, item_id: str, transaction_id: str, buyer_user_id: str, status: str, error_message: Optional[str] = None, ) -> None: # フィードバック送信結果をDBに記録する。 conn.execute( "INSERT OR REPLACE INTO sent_feedback " "(item_id, transaction_id, buyer_user_id, sent_at, status, error_message) " "VALUES (?, ?, ?, ?, ?, ?)", ( item_id, transaction_id, buyer_user_id, datetime.now(timezone.utc).isoformat(), status, error_message, ), ) conn.commit() # ── フィードバック送信コア ── def _build_comment_text(buyer_user_id: str) -> str: # 送信するフィードバックコメントを生成する。 # eBay のガイドラインでは過度に定型的なコメントはフィルタされる場合があるため、 # 簡潔で誠実なコメントを推奨する。 return "Quick payment and smooth transaction. Highly recommended buyer! A++++" def leave_positive_feedback( client: zeep.Client, auth_token: str, item_id: str, transaction_id: str, buyer_user_id: str, max_retries: int = 3, ) -> None: # 指定取引に対して Positive フィードバックを送信する。 # 指数バックオフによるリトライロジックを内蔵する。 # 注意: CommentType は常に "Positive" にハードコード # eBay ポリシー（2008年〜）: セラーはバイヤーへ Negative/Neutral を送信不可 request_body = { "RequesterCredentials": {"eBayAuthToken": auth_token}, "ItemID": item_id, "TransactionID": transaction_id, "TargetUser": buyer_user_id, "CommentType": "Positive", "Comment": _build_comment_text(buyer_user_id), } last_error: Optional[Exception] = None for attempt in range(1, max_retries + 1): try: response = client.service.LeaveFeedback(**request_body) ack = getattr(response, "Ack", "Failure") if ack in ("Success", "Warning"): if ack == "Warning": logger.warning( f"LeaveFeedback succeeded with warnings " f"(item={item_id}, tx={transaction_id})" ) logger.info( f"Feedback sent: item={item_id}, tx={transaction_id}, " f"buyer={buyer_user_id}" ) return # エラーの詳細を解析 errors = getattr(response, "Errors", []) error_codes = [str(getattr(e, "ErrorCode", "")) for e in errors] error_msgs = [str(getattr(e, "LongMessage", "")) for e in errors] # 819: 既に送信済みまたは対象外 → リトライ不要 if "819" in error_codes: raise RuntimeError( f"Transaction not eligible (code=819): {error_msgs}. Skipping." ) # 821: ポリシー違反 → リトライ不要 if "821" in error_codes: raise RuntimeError( "Policy violation (code=821): Cannot leave negative for buyer." ) last_error = RuntimeError( f"LeaveFeedback failed (codes={error_codes}): {error_msgs}" ) except RuntimeError: raise # リトライ不要なエラーはそのまま再送出 except Exception as exc: last_error = exc wait_sec = 2 ** attempt logger.warning( f"Attempt {attempt}/{max_retries} failed: {exc}. " f"Retrying in {wait_sec}s..." ) time.sleep(wait_sec) # 指数バックオフ: 2s, 4s, 8s raise RuntimeError( f"Max retries ({max_retries}) exceeded for item={item_id}, " f"tx={transaction_id}. Last error: {last_error}" ) # ── メインループ: バッチ処理 ── def run_feedback_batch( auth_token: str, shipped_orders: List[ShippedOrder], shipping_buffer_days: int = 7, inter_request_sleep: float = 1.0, db_path: str = "feedback_tracker.db", ) -> dict: # 発送済み注文リストに対して、条件を満たすものに一括で Positive フィードバックを送る。 # shipping_buffer_days: 発送完了から待機する日数（紛争ケース対策） # inter_request_sleep: API リクエスト間のスリープ秒数（レート制限対策） conn = init_db(db_path) transport = zeep.transports.Transport(timeout=30) client = zeep.Client(wsdl=WSDL_URL, transport=transport) now_utc = datetime.now(timezone.utc) buffer_threshold = timedelta(days=shipping_buffer_days) # 60日制限に2日の安全マージンを設ける feedback_expiry = timedelta(days=58) results = { "success": 0, "skipped_buffer": 0, "skipped_sent": 0, "skipped_expired": 0, "failed": 0, } for order in shipped_orders: item_id = order.item_id tx_id = order.transaction_id buyer = order.buyer_user_id shipped_at = order.shipped_at if shipped_at.tzinfo is None: shipped_at = shipped_at.replace(tzinfo=timezone.utc) elapsed = now_utc - shipped_at # 1. バッファ期間内: 発送直後はスキップ（紛争ケース対策） if elapsed < buffer_threshold: logger.debug( f"Skipping (buffer): item={item_id}, elapsed={elapsed.days}d" ) results["skipped_buffer"] += 1 continue # 2. 期限切れ: 58日超はスキップ if elapsed > feedback_expiry: logger.warning( f"Skipping (expired): item={item_id}, elapsed={elapsed.days}d" ) results["skipped_expired"] += 1 continue # 3. DB に送信済み記録あり: スキップ（冪等性チェック） if is_feedback_already_sent(conn, item_id, tx_id): logger.debug(f"Skipping (already sent): item={item_id}, tx={tx_id}") results["skipped_sent"] += 1 continue # フィードバック送信 try: leave_positive_feedback(client, auth_token, item_id, tx_id, buyer) record_feedback_result(conn, item_id, tx_id, buyer, "success") results["success"] += 1 except Exception as exc: error_msg = str(exc) logger.error( f"Failed: item={item_id}, tx={tx_id}. Error: {error_msg}" ) record_feedback_result(conn, item_id, tx_id, buyer, "failed", error_msg) results["failed"] += 1 finally: # レート制限対策: リクエスト間に必ずスリープを挟む time.sleep(inter_request_sleep) conn.close() logger.info(f"Feedback batch completed: {results}") return results # ── エントリーポイント（動作確認用）── if __name__ == "__main__": TOKEN = "YOUR_AUTH_TOKEN_HERE" # 実際には GetOrders API から取得した発送済み注文リストを使用する sample_orders = [ ShippedOrder( item_id="123456789012", transaction_id="9876543210", buyer_user_id="sample_buyer_01", shipped_at=datetime.now(timezone.utc) - timedelta(days=10), ), ShippedOrder( item_id="987654321098", transaction_id="1234567890", buyer_user_id="sample_buyer_02", shipped_at=datetime.now(timezone.utc) - timedelta(days=3), # バッファ内 ), ] summary = run_feedback_batch( auth_token=TOKEN, shipped_orders=sample_orders, shipping_buffer_days=7, inter_request_sleep=1.5, ) print(f"実行結果: {summary}") 補足: GetOrders API との連携 run_feedback_batch 関数は ShippedOrder のリストを引数に受け取ります。実際のシステムでは、GetOrders API（または REST の Orders API）で発送完了（Shipped）ステータスの注文を定期取得し、ShippedOrder オブジェクトに変換してこの関数に渡します。スケジューラー（cron、Celery Beat、AWS EventBridge 等）で1日1〜2回このバッチを実行するのが一般的な運用パターンです。DB の sent_feedback テーブルが冪等性を保証するため、複数回実行しても二重送信は発生しません。 パフォーマンス・スケーリング視点 (深度) 大量注文のスロットリング戦略と定期フィードバックレポート 月間数千件以上の注文を処理する大規模セラーにとって、フィードバック管理のアーキテクチャには追加の考慮が必要です。 【スロットリングの設計】 eBay の LeaveFeedback は1日あたりのコール数制限があります。大量の注文が一度に発生した場合（セール期間など）、単純なループ処理ではレート制限に達してしまいます。実務では以下のアーキテクチャを推奨します。 まず、対象注文をキュー（RabbitMQ、AWS SQS、Redis Queue 等）に投入します。コンシューマーはキューからメッセージを取り出し、各リクエスト間に必ず inter_request_sleep 秒を確保して順次処理します。キューの深さ（メッセージ数）をモニタリングすることで、バックログの蓄積状況をリアルタイムに把握できます。突発的な大量注文でも、キューがバッファとして機能するためレート制限エラーを回避できます。 【定期フィードバックレポートと健康度監視】 GetFeedback で週次または月次に評価データを取得し、Positive Percentage のトレンドを DB に蓄積していくことで、アカウント健康度の変化を早期に検知できます。Positive Percentage が 98.5% を下回ったらアラート通知（Slack、メール等）を送る仕組みを設けることで、TRS ラインの 98.0% を割り込む前に対処できます。 # feedback_health_monitor.py # 週次レポート生成と健康度アラートの実装例 import sqlite3 from datetime import datetime, timezone def record_weekly_score( db_path: str, positive_count: int, neutral_count: int, negative_count: int, ) -> None: # 週次フィードバックスコアを時系列DBに記録する。 # 蓄積データにより Positive Percentage のトレンドグラフを描画可能にする。 conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS feedback_history ( recorded_at TEXT NOT NULL, positive_count INTEGER, neutral_count INTEGER, negative_count INTEGER, positive_percentage REAL ) """) total = positive_count + neutral_count + negative_count pct = round(positive_count / total * 100, 2) if total > 0 else 0.0 conn.execute( "INSERT INTO feedback_history VALUES (?, ?, ?, ?, ?)", ( datetime.now(timezone.utc).isoformat(), positive_count, neutral_count, negative_count, pct, ), ) conn.commit() conn.close() def check_health_alert( positive_percentage: float, threshold: float = 98.5 ) -> bool: # Positive Percentage が閾値を下回った場合にアラートを返す。 # 閾値を TRS 基準（98.0%）より高めに設定することで早期警告を実現する。 if positive_percentage < threshold: print( f"[ALERT] Positive Percentage {positive_percentage}% " f"has dropped below threshold {threshold}%. " "Immediate review of recent Negative feedbacks is recommended." ) # 本番では Slack Webhook / SNS / PagerDuty 等に通知を送る return True return False # ── 使用例 ── if __name__ == "__main__": TOKEN = "YOUR_AUTH_TOKEN_HERE" USER_ID = "your_seller_id" # 1. GetFeedback で現在のスコアを取得 from get_feedback_baseline import get_feedback_list, analyze_feedback_score entries = get_feedback_list(TOKEN, USER_ID) analysis = analyze_feedback_score(entries) # 2. 週次スコアを時系列DBに記録 record_weekly_score( db_path="health_history.db", positive_count=analysis["positive_count"], neutral_count=analysis["neutral_count"], negative_count=analysis["negative_count"], ) # 3. 健康度アラートを確認 is_alert = check_health_alert(analysis["positive_percentage"], threshold=98.5) if is_alert: print("Action required: review recent Negative feedbacks in Seller Hub.") else: print(f"Account health is good: {analysis['positive_percentage']}%") GetFeedback の FeedbackSummary フィールドには、FeedbackScore（累計スコア）、PositiveFeedbackPercent（好評率）、そして直近1ヶ月・6ヶ月・12ヶ月の期間別サマリーが含まれています。これらを週次で記録することで、SQLite または PostgreSQL による時系列トレンド分析が可能になり、Negative 評価のスパイクを可視化・早期検知できます。 大規模運用における最終的なアーキテクチャとしては、①GetOrders（日次バッチ）→ ②キューイング → ③スロットリング付き LeaveFeedback → ④健康度スコアの時系列記録 → ⑤閾値アラートという5段構成が理想的です。この構成により、人手を介さずともアカウント健康度を高水準に維持し続けられます。 まとめ 本記事では、eBay セラーのアカウント健康度管理において核心的な役割を果たすフィードバック API の全体像を実装しました。 ベースライン: GetFeedback で評価一覧を全件ページネーション取得し、Positive / Neutral / Negative 別に集計して Positive Percentage を算出する Python 実装を構築しました。FeedbackType と CommentType の正確な理解が出発点です。 深いポイント: 60日有効期限と紛争中取引への誤送信リスク、2008年ポリシー変更によるセラーの CommentType 制約（Positive のみ可）、そして DB 管理による二重送信ガードの3点が、自動化システムの信頼性を決定づける要因です。これらを見落とすと本番環境でサイレントにエラーが積み重なります。 スケーリング: キューイング＋スロットリングによるレート制限対策と、週次スコアの時系列記録による早期健康度アラートを組み合わせることで、数千件規模の注文をこなす大規模セラーでも安定運用できるアーキテクチャが完成します。 本記事をもって、第2回から始まった Trading API 連載（全16回）が完結します。GeteBayDetails でのメタデータ取得（#2, #3）に始まり、画像アップロード（#3）、単一商品・バリエーション出品（#4, #6）、在庫管理（#7〜#9）、注文管理（#10〜#12）、返品・クレーム対応（#13〜#15）、カテゴリ情報取得（#16）、そして本記事のフィードバック管理（#17）まで、Trading API の実務的な全領域をカバーしました。これらを組み合わせることで、出品から販売・アフターケアまでを Python で完全自動化する基盤が整いました。 次のステップ 次回（#18）から、本連載は大きな技術的転換点を迎えます。「Inventory Mapping API 入門：Python と GraphQL で AI 推奨の出品プレビューを作成する」と題し、連載42回シリーズで初めて GraphQL API が登場します。これまでの Trading API（SOAP/XML）から、よりモダンで型安全な GraphQL ベースの eBay 新世代 API へと舞台が移ります。 GraphQL のクエリ構造、Python からの呼び出し方、そして AI 推奨カテゴリのプレビュー取得まで、新しいパラダイムをゼロから丁寧に解説します。Trading API で培った実装力を武器に、次のステージへ進みましょう。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#18】eBay Inventory Mapping API：PythonとGraphQLでAI推奨の出品プレビューを作成する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第18回です。 第2回から第17回まで、実に16回にわたって Trading API（SOAP）の世界を歩んできました。GeteBayDetails によるメタデータ取得、EPS 画像アップロード、AddFixedPriceItem による出品、在庫管理、注文処理、フィードバック自動化——これらすべてが SOAP プロトコルと XML の上に成り立つ、長年の実績ある技術でした。そして前回（#17）の GetFeedback / LeaveFeedback を最後に、Trading API 連載はひとつの完結を迎えます。 本回は、連載にとっての重要な技術的転換点となります。ここからは、eBay が次世代の API 基盤として推進する GraphQL API の世界に入ります。その第一弾が、本記事で扱う「Inventory Mapping API（インベントリマッピング API）」です。これは 2025 年に正式リリースされた eBay 初の GraphQL API であり、SOAP や REST とは根本的に異なるプロトコルを採用しています。既存の商品データ（GTIN・UPC・EAN・ISBN などの標準商品コード）を AI が解析し、最適な eBay カテゴリや Item Specifics（商品の詳細スペック）を自動推薦する、次世代の出品支援機能です。 大量の商品を eBay に出品する際、最大のボトルネックのひとつが「カテゴリの選定」と「Item Specifics（ブランド・カラー・サイズ等）の手入力」です。これらをすべて人手で行うと 1 商品あたり数分のコストがかかり、数千件規模になると現実的ではありません。Inventory Mapping API の AI 推薦機能を活用することで、この作業を大幅に削減できます。 この記事で得られること: eBay 初の GraphQL API（Inventory Mapping API）の全体構造と、従来の SOAP・REST との根本的な違い——「なぜ POST 1 本だけで動くのか」を正確に理解する。 startListingPreviewsCreation mutation を Python + requests ライブラリで呼び出し、AI 推薦プレビュータスクを起動してタスク ID を確実に取得する実装手順。 Sandbox 環境で「本当にテストできること」と「できないこと」の正確な線引き——モックデータの落とし穴を理解し、テスト戦略を正しく設計する方法。 背景・なぜこれが重要か (Motivation) 「REST API と何が違うの？ GraphQL って難しそう……」 GraphQL という言葉を初めて聞いたとき、多くの開発者がこう感じます。確かに、SOAP でも REST でもない第三の選択肢であり、構文も独特です。しかし、実際に触れてみると、eBay の Inventory Mapping API における GraphQL の使い方は非常にシンプルです。エンドポイントが 1 つ（https://api.ebay.com/graphql）に統一され、HTTP メソッドも POST のみ。リクエストボディに「何をしたいか（mutation）」と「何を返してほしいか（フィールド指定）」を同時に書くだけです。REST のように「エンドポイントをどのパスにするか」を設計する必要がなく、最初のハードルを越えれば、むしろすっきりした構造に感じるはずです。 では、なぜ eBay はこの機能を GraphQL で提供するのでしょうか。答えは「柔軟なフィールド選択」にあります。出品プレビューの結果には、カテゴリ情報・推薦 Item Specifics・タイトル・画像など多くのフィールドが含まれます。REST では全フィールドが固定のレスポンスとして返り、不要なデータも転送されます。GraphQL ならクライアントが「今必要なフィールドだけ」を指定できるため、通信効率が高く、将来 API がフィールドを追加しても既存クライアントコードへの影響が最小化されます。 そして、この機能が解決する実務課題は明確です。たとえば、電子機器を 500 件出品するシナリオを考えてみてください。 【課題1: カテゴリ選定の手間】 eBay のカテゴリ構造は数万件に及び、「Sony ヘッドフォン」を出品するためだけでも Consumer Electronics > Portable Audio & Headphones > Headphones という深い階層を探索する必要があります。これを 500 件分、人手でやることは現実的ではありません。 【課題2: Item Specifics 入力の品質】 eBay では各カテゴリに「必須の Item Specifics」があり、これが欠落すると出品クオリティスコアが下がり、検索順位に悪影響が出ます。ブランド・タイプ・接続方式・インピーダンスなど、カテゴリごとに要求されるスペックは異なり、網羅的に入力するには専門知識が必要です。 Inventory Mapping API はこの両方を、商品の UPC や EAN をキーにした AI 解析で自動的に推薦します。人間はその推薦結果をレビューして承認するだけという、「人間は最終判断のみ」のワークフローを実現できます。これが、Trading API 時代の手動カテゴリ選定から脱却するための、新世代のアプローチです。 基本的な使い方（ベースライン）：GraphQLでstartListingPreviewsCreationを呼び出す まず、最小限の動作するコードから始めます。ここでは、UPC コードを持つ商品 1 件を Inventory Mapping API に送り、AI 推薦タスクを起動してタスク ID を取得するまでの流れを示します。 SOAP や REST と異なり、GraphQL では以下の 3 点が重要です。まず、エンドポイントは常に 1 つ（https://api.ebay.com/graphql）で固定です。次に、HTTP メソッドは常に POST を使用します。そして、リクエストボディには query（または mutation）文字列とvariables（変数）を JSON として渡します。 # inventory_mapping_basic.py import os import requests GRAPHQL_ENDPOINT = "https://api.ebay.com/graphql" MUTATION = """ mutation StartListingPreviews($input: StartListingPreviewsCreationInput!) { startListingPreviewsCreation(input: $input) { errors { errorDescription } listingPreviewsCreationTask { id } } } """ def start_listing_previews(access_token: str) -> str | None: """ startListingPreviewsCreation mutation を呼び出してタスクIDを返す（最小実装）。 必須ヘッダー: Authorization: Bearer <token> Content-Type: application/json X-EBAY-C-MARKETPLACE-ID: EBAY_US # 現時点で US のみ対応 必須 OAuth scope: https://api.ebay.com/oauth/api_scope/sell.inventory.mapping """ headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "X-EBAY-C-MARKETPLACE-ID": "EBAY_US", } variables = { "input": { "externalProducts": [ { "sku": "HEADPHONE-SONY-XM5", "title": "Sony WH-1000XM5 Wireless Noise Canceling Headphones Black", "images": [ "https://example.com/images/sony-wh1000xm5-black.jpg" ], "externalProductIdentifierInput": { "productType": "UPC", # EAN / ISBN / MPN も指定可能 "productId": "027242920972" } } ] } } payload = {"query": MUTATION, "variables": variables} resp = requests.post(GRAPHQL_ENDPOINT, json=payload, headers=headers, timeout=30) resp.raise_for_status() body = resp.json() # GraphQL は HTTP 200 でも errors フィールドにエラーが入ることがある if "errors" in body: print(f"GraphQL エラー: {body['errors']}") return None result = body["data"]["startListingPreviewsCreation"] # mutation レベルのビジネスエラー（errors 配列） if result.get("errors"): print(f"API ビジネスエラー: {result['errors']}") return None task_id = result["listingPreviewsCreationTask"]["id"] print(f"タスクID取得成功: {task_id}") return task_id if __name__ == "__main__": token = os.environ["EBAY_ACCESS_TOKEN"] start_listing_previews(token) 補足: GraphQL と REST の違い——なぜ POST 1 本なのか REST API では、エンドポイントの URL パス（例: /sell/inventory/v1/inventory_item/{sku}）とHTTP メソッド（GET / POST / PUT / DELETE）の組み合わせで「何をしたいか」を表現します。一方、GraphQL では URL は常に同じ（/graphql）で、「何をしたいか」はリクエストボディの mutation 文字列の中に書きます。そのため、ネットワーク監視ツールから見ると「すべてのリクエストが POST /graphql に見える」という点が、REST に慣れた開発者にとっては最初は戸惑いますが、慣れると非常にシンプルに感じられます。 また、GraphQL では「返してほしいフィールドだけをリクエストに書く」という設計思想があります。上記のコードで listingPreviewsCreationTask { id } とだけ書いているのはそのためです。将来、タスクの result フィールドも一緒に取得したくなったら { id result { completionStatus } } と追記するだけで対応できます——サーバー側の変更は不要です。 実務で躓く場面・深いポイント (Core) ここでは、実際に Inventory Mapping API を使い始めた際にエンジニアが必ずといっていいほど踏む落とし穴と、その回避策を解説します。Trading API との設計の違いに起因するものが多く、特に GraphQL 初体験のエンジニアには重要なポイントです。 1. US Marketplace 限定であることを見落として失敗する Inventory Mapping API は、現時点（2025 年）では EBAY_US（米国サイト）のみに対応しています。他のマーケットプレイスのデータを処理しようとしても、API はエラーを返します。 具体的には、X-EBAY-C-MARKETPLACE-ID ヘッダーに EBAY_JP（日本）や EBAY_DE（ドイツ）を指定すると、HTTP 403 が返るか、mutation の errors フィールドにエラーが格納されます。日本の eBay セラーが出品支援に使う場合でも、このヘッダーは必ず EBAY_US に固定する必要があります。また、eBay Motors（米国の自動車サイト）およびそのサブカテゴリも非対応です。車・バイク関連の商品カテゴリには現時点では使用できません。 補足: なぜ US 限定なのか？ Inventory Mapping API の AI エンジンは eBay US の商品カタログをベースに学習されています。AI の推薦精度が保証されるのは EBAY_US のカテゴリ・スペック体系に対してのみであるため、現時点では US に限定されています。将来的には他のマーケットプレイスへの展開が予定されていますが、2025 年のリリース時点では US のみです。 2. Sandbox のモックデータと本番結果の違いを理解していないと検証を誤る Inventory Mapping API には Sandbox 環境（https://api.sandbox.ebay.com/graphql）が存在します。「Sandbox で試してから Production 投入」という通常の開発フローが使えるという点では、他の eBay API と同じです。しかし、Sandbox における重要な制限を理解していないと、間違った前提でテストしてしまいます。 Sandbox でできること: API の呼び出しフロー全体（認証 → mutation 実行 → タスク ID 取得）をコードで確認できる。 エラーハンドリング（HTTP エラー、GraphQL エラー、ビジネスエラー）の実装を検証できる。 後続の結果取得コード（第19回で解説するポーリング処理）との連携フローをテストできる。 Sandbox でできないこと: 返却されるプレビューデータは本物の AI 推薦結果ではなく、ランダムなモックデータ（ダミーカテゴリ・ダミースペック）です。 処理時間も実際の AI 推論コストを反映しておらず、Sandbox では最長 10 分程度のランダムな遅延がシミュレートされます（本番環境では通常数十秒〜数分程度）。 AI 推薦精度（正しいカテゴリが推薦されるか等）は Sandbox では一切検証できません。これは必ず本番環境（Production）で少量のサンプルを用いて確認してください。 つまり、Sandbox は「コードの動作確認」には使えますが、「API が正しい結果を返すかの確認」には使えません。この線引きを明確に理解した上でテスト戦略を立てることが、後の手戻りを防ぐ鍵です。 3. externalProducts の入力データ品質が AI 推薦精度を左右する startListingPreviewsCreation に渡す externalProducts の各フィールドの品質は、AI が返す推薦結果の精度に直結します。「とりあえず title だけ渡せばいいだろう」という安易な実装は、推薦精度の著しい低下を招きます。 images フィールドの重要性: AI は画像からも商品カテゴリや属性を認識します。HTTPS で直接アクセスできる高解像度の商品画像 URL を少なくとも 1 枚は渡すことを強く推奨します。リダイレクトを含む URL や、認証が必要な URL は AI が解析できません。 externalProductIdentifierInput（GTIN 等）の効果: UPC / EAN / ISBN / MPN などの標準商品コードがある場合、必ず指定してください。AI は GTIN をキーにして eBay の商品カタログデータベースを参照でき、カテゴリ推薦と Item Specifics 補完の精度が大幅に向上します。GTIN がない場合は sku と title のみでも動作しますが、推薦精度は低下します。 title の最適化: title には商品の本質的な属性（ブランド・モデル名・主要スペック）を英語で簡潔に記述することが理想です。80 文字以内に重要な情報を凝縮してください。日本語タイトルも受け付けますが、US マーケット向け AI の学習データは英語商品が中心のため、英語タイトルの方が精度は高くなります。 注意: OAuth Scope の設定漏れ Inventory Mapping API の呼び出しには、OAuth アクセストークンに https://api.ebay.com/oauth/api_scope/sell.inventory.mapping スコープが必要です。このスコープを付与せずにリクエストすると、HTTP 401 Unauthorized が返ります。 第1回〜第17回で使用してきた Trading API 用のトークンには、このスコープは含まれていません。新たに eBay Developer Portal でアプリケーションの設定を確認し、このスコープを追加した上で、ユーザーに再認証（OAuth フロー）を実行させる必要があります。既存のシステムにこの API を組み込む際に最も多い初歩的ミスのひとつです。 頻出エラーコード早見表 GraphQL API では REST とはエラーの表現方法が異なります。HTTP ステータスコードだけを見ていると、エラーを見逃すケースがあります。 エラー種別 発生状況 対策 HTTP 401 Unauthorized OAuth トークンが無効、または sell.inventory.mapping スコープが付与されていない Developer Portal で scope を確認し、ユーザーに再認証を促す HTTP 403 Forbidden X-EBAY-C-MARKETPLACE-ID が EBAY_US 以外（または未指定）、あるいは Motors カテゴリ ヘッダーを "EBAY_US" に固定する。Motors サブカテゴリは非対応 GraphQL errors（HTTP 200） mutation 文字列の構文エラー、または変数の型不一致（例: productType に無効な値を指定） errors[].message を確認。productType は "UPC" "EAN" "ISBN" "MPN" のいずれか mutation.errors（ビジネスエラー、HTTP 200） HTTP は 200 だが startListingPreviewsCreation.errors 配列にエラーが含まれる。空の externalProducts を渡した場合など errors[].errorDescription を確認して原因を特定する。externalProducts は 1 件以上必須 堅牢な実装：型安全・入力検証・エラー処理を備えたInventoryMappingClientクラス ここまでの落とし穴をすべて踏まえた上で、プロダクションレベルの Python クラスを実装します。型アノテーション・docstring・入力バリデーション・エラーハンドリングを完備した、実運用に耐えうる設計です。 # inventory_mapping_client.py """ eBay Inventory Mapping API クライアント。 startListingPreviewsCreation mutation を呼び出し、タスクIDを確実に取得する。 依存: requests>=2.28.0 必須 OAuth scope: https://api.ebay.com/oauth/api_scope/sell.inventory.mapping """ from __future__ import annotations import logging from dataclasses import dataclass from typing import Optional import requests logger = logging.getLogger(__name__) GRAPHQL_ENDPOINT = "https://api.ebay.com/graphql" SANDBOX_ENDPOINT = "https://api.sandbox.ebay.com/graphql" # ExternalProductIdentifierEnum で許可される値 VALID_PRODUCT_TYPES = frozenset({"UPC", "EAN", "ISBN", "MPN"}) START_MUTATION = """ mutation StartListingPreviews($input: StartListingPreviewsCreationInput!) { startListingPreviewsCreation(input: $input) { errors { errorDescription } listingPreviewsCreationTask { id } } } """ @dataclass class ExternalProductIdentifier: """標準商品コード（GTIN 等）を表すデータクラス。""" product_type: str # "UPC" | "EAN" | "ISBN" | "MPN" product_id: str def __post_init__(self) -> None: if self.product_type not in VALID_PRODUCT_TYPES: raise ValueError( f"product_type は {VALID_PRODUCT_TYPES} のいずれかを指定してください: " f"'{self.product_type}'" ) if not self.product_id.strip(): raise ValueError("product_id は空にできません") @dataclass class ExternalProduct: """ Inventory Mapping API に送る商品データを表すデータクラス。 AI 推薦精度を高めるには: 1. images を必ず 1 枚以上含める（HTTPS の公開 URL） 2. identifier（UPC/EAN/ISBN/MPN）を指定する 3. title を英語・80 文字以内で記述する """ sku: str title: str images: list[str] identifier: Optional[ExternalProductIdentifier] = None def __post_init__(self) -> None: if not self.sku.strip(): raise ValueError("sku は空にできません") if not self.title.strip(): raise ValueError("title は空にできません") if len(self.title) > 80: logger.warning( "SKU '%s': title が 80 文字を超えています（%d 文字）。" "AI 推薦精度に影響する可能性があります。", self.sku, len(self.title), ) if not self.images: logger.warning( "SKU '%s': images が未指定です。AI 推薦精度が低下する可能性があります。", self.sku, ) for url in self.images: if not url.startswith("https://"): raise ValueError( f"SKU '{self.sku}': 画像 URL は HTTPS である必要があります: {url}" ) def to_graphql_input(self) -> dict: """GraphQL mutation の variables 用 dict に変換する。""" payload: dict = { "sku": self.sku, "title": self.title, "images": self.images, } if self.identifier: payload["externalProductIdentifierInput"] = { "productType": self.identifier.product_type, "productId": self.identifier.product_id, } return payload class InventoryMappingClient: """ eBay Inventory Mapping API クライアント。 GraphQL エンドポイントに startListingPreviewsCreation mutation を送信し、 AI 推薦プレビュータスクの ID を返す。 結果の取得は非同期（別途ポーリングまたは Notification API が必要）。 Sandbox 環境: use_sandbox=True を指定すると api.sandbox.ebay.com に接続する。 Sandbox ではモックデータが返るため、AI 推薦精度の検証には使用できない。 コードフローの動作確認（認証・エラーハンドリング等）に限り使用すること。 Raises: ValueError: 引数が不正な場合（空トークン等） requests.HTTPError: HTTP エラー（401/403 等）が発生した場合 RuntimeError: GraphQL エラーまたは API ビジネスエラーが発生した場合 """ def __init__( self, access_token: str, use_sandbox: bool = False, timeout: int = 30, ) -> None: if not access_token.strip(): raise ValueError("access_token は空にできません") self._endpoint = SANDBOX_ENDPOINT if use_sandbox else GRAPHQL_ENDPOINT self._timeout = timeout self._session = requests.Session() self._session.headers.update( { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", # 現時点では EBAY_US 以外は非対応（Motors も非対応） "X-EBAY-C-MARKETPLACE-ID": "EBAY_US", } ) logger.info("InventoryMappingClient 初期化完了 [endpoint=%s]", self._endpoint) def start_listing_previews_creation( self, products: list[ExternalProduct], ) -> str: """ startListingPreviewsCreation mutation を実行してタスク ID を返す。 Args: products: AI 推薦プレビューを作成する商品リスト（1 件以上） Returns: タスク ID（str）。後続のポーリング処理や Notification API で使用する。 Raises: ValueError: products が空の場合 requests.HTTPError: HTTP 401/403 等のエラー RuntimeError: GraphQL 実行エラーまたは API ビジネスエラー """ if not products: raise ValueError("products は 1 件以上指定してください") variables = { "input": { "externalProducts": [p.to_graphql_input() for p in products], } } logger.info( "startListingPreviewsCreation 呼び出し: %d 件の商品", len(products) ) try: response = self._session.post( self._endpoint, json={"query": START_MUTATION, "variables": variables}, timeout=self._timeout, ) response.raise_for_status() except requests.Timeout: logger.error("API タイムアウト（%d 秒）", self._timeout) raise except requests.HTTPError as exc: # 401: scope 不足、403: marketplace 非対応 など logger.error( "HTTP エラー %s: %s", exc.response.status_code, exc.response.text[:500], ) raise body = response.json() # GraphQL プロトコルレベルのエラー（構文エラー・型不一致 等） # HTTP は 200 だが body["errors"] にエラーが含まれる if "errors" in body: logger.error("GraphQL 実行エラー: %s", body["errors"]) raise RuntimeError(f"GraphQL エラー: {body['errors']}") mutation_data = (body.get("data") or {}).get( "startListingPreviewsCreation", {} ) # API ビジネスレベルのエラー（HTTP 200、mutation.errors に格納） biz_errors = mutation_data.get("errors") or [] if biz_errors: descriptions = [e["errorDescription"] for e in biz_errors] logger.error("API ビジネスエラー: %s", descriptions) raise RuntimeError(f"Inventory Mapping API エラー: {descriptions}") task = (mutation_data.get("listingPreviewsCreationTask")) or {} task_id: Optional[str] = task.get("id") if not task_id: raise RuntimeError( "レスポンスにタスク ID が含まれていません。" f"予期しないレスポンス形式: {body}" ) logger.info("タスク ID 取得成功: %s", task_id) return task_id 上記のクライアントを実際に使う呼び出し例を示します。 # main.py import os import logging from inventory_mapping_client import ( InventoryMappingClient, ExternalProduct, ExternalProductIdentifier, ) logging.basicConfig(level=logging.INFO) def main() -> None: access_token = os.environ["EBAY_ACCESS_TOKEN"] # Sandbox でテストする場合は use_sandbox=True を指定する # ただし Sandbox の結果はモックデータ——AI 推薦精度の検証には使えない client = InventoryMappingClient(access_token=access_token, use_sandbox=False) products = [ ExternalProduct( sku="SONY-WH1000XM5-BLK", title="Sony WH-1000XM5 Wireless Noise Canceling Headphones Black", images=["https://example.com/images/wh1000xm5-black.jpg"], identifier=ExternalProductIdentifier( product_type="UPC", product_id="027242920972", ), ), ExternalProduct( sku="APPLE-AIRPODS-PRO2", title="Apple AirPods Pro 2nd Generation with MagSafe Case USB-C", images=["https://example.com/images/airpods-pro2.jpg"], identifier=ExternalProductIdentifier( product_type="UPC", product_id="195949206399", ), ), ] task_id = client.start_listing_previews_creation(products) print(f"タスク投入完了。タスクID: {task_id}") print("次のステップ: このIDを使って結果をポーリングしてください（第19回参照）") if __name__ == "__main__": main() パフォーマンス・スケーリング視点 (深度) 大量商品を一括投入する際のバッチ設計と非同期化の展望 実務で数千件の商品を Inventory Mapping API に投入する場合、単純にループで 1 件ずつ呼び出すのは非効率であり、レート制限（Rate Limit）に引っかかる可能性もあります。ここでは、大量商品を安全にバッチ処理するための設計ポイントを解説します。 【バッチサイズの設計】 startListingPreviewsCreation の 1 リクエストに含めることができる externalProducts の件数は eBay の公式ドキュメントで確認してください。実務上は 1 リクエストあたり 50 件程度を目安にバッチ分割するのが安定した運用につながります。バッチが大きすぎると AI 処理に時間がかかりタスクが FAILED になるリスクが上がります。バッチ間には 1 秒程度のウェイトを入れてレート制限を回避しましょう。 # batch_submission.py """ 大量商品を一定サイズのバッチに分割して Inventory Mapping API に投入する。 """ import time import logging from itertools import islice from inventory_mapping_client import InventoryMappingClient, ExternalProduct logger = logging.getLogger(__name__) BATCH_SIZE = 50 # 1 リクエストあたりの推奨件数 DELAY_SECONDS = 1.0 # バッチ間のウェイト（レート制限対策） def chunked(iterable, size: int): """iterable を size 件ごとのチャンクに分割するジェネレータ。""" it = iter(iterable) while chunk := list(islice(it, size)): yield chunk def submit_in_batches( client: InventoryMappingClient, products: list[ExternalProduct], ) -> list[str]: """ 商品リストをバッチに分割してタスクを投入し、タスク ID リストを返す。 Args: client : InventoryMappingClient インスタンス products: 全商品リスト Returns: タスク ID のリスト（バッチ数 == len(返り値)） Raises: RuntimeError: いずれかのバッチで投入に失敗した場合 """ batches = list(chunked(products, BATCH_SIZE)) task_ids: list[str] = [] logger.info( "合計 %d 件を %d バッチ（各最大 %d 件）で投入します", len(products), len(batches), BATCH_SIZE, ) for idx, batch in enumerate(batches, start=1): logger.info( "バッチ %d/%d を投入中 (%d 件)...", idx, len(batches), len(batch) ) try: task_id = client.start_listing_previews_creation(batch) task_ids.append(task_id) logger.info("バッチ %d 完了: タスクID=%s", idx, task_id) except (RuntimeError, Exception) as exc: logger.error("バッチ %d 投入失敗: %s", idx, exc) raise if idx < len(batches): time.sleep(DELAY_SECONDS) logger.info("全バッチ投入完了。タスクID一覧: %s", task_ids) return task_ids 補足: Notification API との組み合わせによる非同期化 上記のバッチ投入で取得したタスク ID を使って「処理が完了したか」を確認する方法は大きく 2 つあります。第19回で解説する「ポーリング（定期的な問い合わせ）」と、eBay の Notification API を使った「プッシュ通知（イベント駆動）」です。 Notification API では LISTING_PREVIEW_CREATION_TASK_STATUS というイベントトピックを購読することで、タスクが完了した瞬間に Webhook で通知を受け取ることができます。数十バッチを同時に投入する大規模システムでは、ポーリングよりも Notification API との組み合わせの方がサーバーリソースの無駄遣いが少なく、リアルタイム性も高くなります。本連載では第19回でポーリング実装を詳しく解説した後、将来の回で Notification API との統合についても触れる予定です。 まとめ 本記事では、連載第18回として Trading API（SOAP）から eBay 初の GraphQL API への転換点を迎え、Inventory Mapping API の入門から実践的な実装までを解説しました。 ベースライン: GraphQL は POST 1 本・エンドポイント固定のシンプルな構造。startListingPreviewsCreation mutation に externalProducts（sku / title / images / 標準商品コード）を渡すと、非同期で AI 推薦タスクが起動してタスク ID が返る。 深いポイント: US Marketplace 限定（X-EBAY-C-MARKETPLACE-ID: EBAY_US 必須）、sell.inventory.mapping スコープの取得、Sandbox はコード動作確認に使えるが AI 推薦精度の検証にはモックデータのため使えない、入力データ品質（HTTPS 画像・GTIN・英語タイトル）が推薦精度を左右する——という 4 つのポイントを押さえた。 スケーリング: バッチサイズ 50 件程度を目安に商品をチャンク分割して投入し、Notification API（LISTING_PREVIEW_CREATION_TASK_STATUS）との組み合わせでイベント駆動型の非同期アーキテクチャへ発展させる展望を描いた。 Trading API の SOAP の世界から GraphQL の世界への第一歩は、思ったよりシンプルです。最初のリクエストが成功してタスク ID を受け取った瞬間——「GraphQL、案外やれる」と感じるはずです。 次のステップ タスクを投入しても、肝心の AI 推薦結果（推薦カテゴリ・Item Specifics・タイトル）はまだ取得できていません。Inventory Mapping API はタスクが非同期で処理されるため、完了するまで待って結果を取得する仕組みが別途必要です。 次回（#19）「Inventory Mapping API②：タスク完了をポーリングしてプレビュー結果を取得する」では、listingPreviewsCreationTaskById query を使って定期的にタスクの completionStatus を確認し、COMPLETED になったタイミングで推薦カテゴリ・Item Specifics・タイトルを取得・活用する実装を詳しく解説します。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#16】eBay Trading API：GetCategoriesとGetCategoryFeaturesでカテゴリ構造とItem Specificsルールを自動取得する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第16回です。 前回（#15）は、SetNotificationPreferencesを用いてeBayからのイベント通知を設定し、PythonサーバーでWebhookとして受信する仕組みを構築しました。これにより、注文や支払いの変化をリアルタイムに検知できるようになりました。 しかし実務でよくある落とし穴として、「通知を受け取る前の段階」、すなわち出品（Listing）時にカテゴリ選択を誤り、後から修正が効かない状況に陥ることがあります。間違ったカテゴリに出品すると、検索露出が著しく低下するだけでなく、「そのカテゴリで必須のItem Specifics（商品の詳細属性）」が不明なまま出品してしまい、eBayから出品差し止めのエラーを受けることもあります。 この記事で得られること: GetCategoriesを呼び出してeBayのカテゴリツリー全体を取得し、ローカルにJSONとして保存する実装パターン。 GetCategoryFeaturesを使って特定カテゴリの必須Item Specifics・推奨Item Specifics・バリエーション対応可否を自動判定するスクリプト。 カテゴリデータの大量件数に起因するパフォーマンス問題と、本番運用のための定期同期・ローカルキャッシュ戦略。 背景・なぜこれが重要か (Motivation) 「だいたい合ってそうなカテゴリに入れておけばいいじゃないか？」 eBay開発を始めたエンジニアから、この言葉を何度か聞いたことがあります。その気持ちは理解できます。カテゴリIDは単なる整数値であり、「6000」と「6001」のどちらが正解なのかは、eBayのサイトを実際に確認しなければわかりません。 しかし、eBayのカテゴリは単なる「分類ラベル」ではありません。カテゴリには以下の情報が紐付いています。 必須Item Specifics: そのカテゴリで出品するために必ず入力しなければならない属性（例: 衣類カテゴリなら「Brand」「Size Type」「Size」など）。欠損するとAPIレベルでエラー、もしくはeBay品質スコアが低下し検索下位に沈みます。 バリエーション対応可否 (VariationsEnabled): カテゴリによってはバリエーション出品（色・サイズ展開）が禁止されています。連載#6で実装したバリエーションXMLを送っても、このフラグがfalseのカテゴリでは容赦なくエラーが返ります。 コンディション制約 (ConditionEnabled): 「新品」「中古」などのコンディション指定が必須か、または特定コンディションが禁止されているかどうか。 これらの情報をプログラムから取得するためのAPIがGetCategories（カテゴリツリーの取得）とGetCategoryFeatures（カテゴリ別ルールの取得）です。出品システムを本番稼働させるなら、この二つのAPIを使いこなすことは避けて通れません。 基本的な使い方（ベースライン）：GetCategoriesでカテゴリツリーを取得する まずはGetCategoriesの最小構成から始めます。このAPIはeBayのサイト（US、JPなど）全体のカテゴリ階層をSOAPレスポンスとして返します。zeepライブラリを使ったPython実装は以下の通りです。 # get_categories_baseline.py import json import zeep from zeep import Client from zeep.transports import Transport import requests WSDL_URL = "https://api.ebay.com/wsapi?callname=GetCategories&siteid=0&version=1265" APP_TOKEN = "YOUR_OAUTH_TOKEN" # User tokenまたはApp token def get_categories_raw(site_id: int = 0, level_limit: int = 2) -> dict: """ GetCategoriesを呼び出してカテゴリツリーを取得する（最小実装）。 site_id: 0=US, 101=Italy, 189=Switzerland など level_limit: 取得する階層の深さ。None指定で全階層（注意：レスポンスが巨大になる） """ transport = Transport(session=requests.Session()) client = Client( "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl", transport=transport ) header_data = { "RequesterCredentials": { "eBayAuthToken": APP_TOKEN } } header = client.get_element("ns0:RequesterCredentials")(**header_data) request_body = { "CategorySiteID": site_id, "DetailLevel": "ReturnAll", "LevelLimit": level_limit, "ViewAllNodes": True, } response = client.service.GetCategories( _soapheaders={"RequesterCredentials": {"eBayAuthToken": APP_TOKEN}}, **request_body ) return zeep.helpers.serialize_object(response, target_cls=dict) if __name__ == "__main__": result = get_categories_raw(site_id=0, level_limit=2) categories = result.get("CategoryArray", {}).get("Category", []) print(f"取得カテゴリ数: {len(categories)}") for cat in categories[:5]: print(f" ID={cat['CategoryID']}, Name={cat['CategoryName']}, Leaf={cat.get('LeafCategory', False)}") 補足: LevelLimitとLeafCategoryについて LevelLimitを指定しない（もしくはNone）場合、eBayはカテゴリツリーの全階層を返します。USサイトの場合、カテゴリ数は数万件に及ぶため、レスポンスXMLのサイズは数MB〜十数MBになります。開発初期はLevelLimit=2程度で動作確認するのが賢明です。 LeafCategoryフィールドがTrueのカテゴリだけが実際に商品を出品できる末端カテゴリです。中間ノード（親カテゴリ）に出品しようとするとエラーになります。「カテゴリIDを自動推薦するロジックを組む場合、LeafCategory=Trueのみを候補とする」という絞り込みは必ず実装してください。 実務で躓く場面・深いポイント (Core) ここからは、ベースラインのコードを実運用に乗せる際に必ずぶつかる落とし穴を3つ解説します。 1. カテゴリツリーの巨大さとキャッシュ戦略 LevelLimitを指定せずにGetCategoriesを呼ぶと、USサイトでは20,000件を超えるカテゴリが返ってきます。このAPIを商品1件出品するたびに都度呼び出すのは、パフォーマンス的にも費用的にも論外です。 eBayのカテゴリツリーは頻繁に変わるものではありませんが、四半期ごとに新カテゴリが追加・廃止されることがあります。推奨される実装パターンは次の通りです。 まず初回起動時または週次バッチでGetCategoriesを呼び出し、全カテゴリをローカルのJSONファイルまたはSQLiteに保存します。アプリケーション起動時はそのキャッシュをメモリに読み込み、カテゴリ検索はすべてローカルで処理します。eBayはGetCategoriesのレスポンスにCategoryVersionという整数値を含めており、前回取得時のバージョンと比較することで「差分更新が必要かどうか」を事前判定できます。 # category_cache.py import json import os from datetime import datetime CACHE_FILE = "ebay_categories_us.json" def load_or_refresh_categories(fetcher_func, force_refresh: bool = False) -> list[dict]: """ ローカルキャッシュがあればそれを返し、なければAPIを呼んで保存する。 force_refresh=Trueで強制的にAPIから再取得する。 """ if not force_refresh and os.path.exists(CACHE_FILE): with open(CACHE_FILE, "r", encoding="utf-8") as f: data = json.load(f) print(f"キャッシュから読み込み: {len(data['categories'])}件 (更新日時: {data['fetched_at']})") return data["categories"] print("APIからカテゴリ取得中...") raw = fetcher_func() categories = raw.get("CategoryArray", {}).get("Category", []) cache_data = { "fetched_at": datetime.utcnow().isoformat(), "category_version": raw.get("CategoryVersion"), "categories": [ { "id": c["CategoryID"], "name": c["CategoryName"], "parent_id": c.get("CategoryParentID"), "level": c["CategoryLevel"], "is_leaf": c.get("LeafCategory", False), "virtual": c.get("Virtual", False), } for c in categories ] } with open(CACHE_FILE, "w", encoding="utf-8") as f: json.dump(cache_data, f, ensure_ascii=False, indent=2) print(f"保存完了: {len(categories)}件") return cache_data["categories"] 注意: カテゴリIDが廃止された場合の挙動 eBayがカテゴリを廃止する際、旧カテゴリIDで出品しようとするとError 21916585（Invalid category ID）が返ります。キャッシュが古いと、プログラムからは存在するように見えても実際には使えないカテゴリIDで出品リクエストを投げてしまいます。本番環境では少なくとも月次でキャッシュの強制リフレッシュを組み込んでください。 2. GetCategoryFeaturesでVariationsEnabledを判定する 連載#6で実装したバリエーション出品（Multi-SKU）を特定カテゴリで行う前に、そのカテゴリがバリエーションをサポートしているか確認しなければなりません。これを怠ると、精巧に組み上げたVariations XMLがError 21916616（Variations not supported for this category）で弾かれます。 GetCategoryFeaturesのリクエストでFeatureIDに「VariationsEnabled」を指定することで、その情報だけをピンポイントで取得できます。 # check_variations_enabled.py import zeep import requests from zeep.transports import Transport def check_variations_enabled(category_id: str, token: str) -> bool: """ 指定カテゴリがバリエーション出品をサポートしているか確認する。 Returns: True=バリエーション対応, False=非対応 """ transport = Transport(session=requests.Session()) client = zeep.Client( "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl", transport=transport ) response = client.service.GetCategoryFeatures( _soapheaders={"RequesterCredentials": {"eBayAuthToken": token}}, CategoryID=category_id, FeatureID=["VariationsEnabled"], DetailLevel="ReturnAll", ViewDataOfAllDescendents=False, ) result = zeep.helpers.serialize_object(response, target_cls=dict) # SiteDefaultsはサイト全体のデフォルト値 # Category[0].VariationsEnabledはカテゴリ固有の値（上書き） site_default = result.get("SiteDefaults", {}).get("VariationsEnabled", False) categories = result.get("Category", []) or [] if categories: cat_val = categories[0].get("VariationsEnabled") if cat_val is not None: return bool(cat_val) return bool(site_default) if __name__ == "__main__": TOKEN = "YOUR_TOKEN" test_cats = [ ("11450", "Clothing, Shoes & Accessories"), ("99", "Everything Else"), ("619", "Computers/Tablets & Networking"), ] for cat_id, name in test_cats: enabled = check_variations_enabled(cat_id, TOKEN) print(f"カテゴリ {cat_id} ({name}): VariationsEnabled={enabled}") 補足: SiteDefaultsとCategory固有値の優先順位 GetCategoryFeaturesのレスポンスには「SiteDefaults」と個別の「Category」の二層構造があります。SiteDefaultsはサイト全体のデフォルト値を表し、CategoryがNoneまたは返ってこない場合はSiteDefaultsの値を採用します。Category固有の値が存在する場合はそちらが優先されます。このロジックを実装しないと、デフォルトで対応しているカテゴリについてVariationsEnabled=Falseという誤判定が起きることがあります。 連載#6で触れたMaxGranularFitmentCountフィールドも同様にGetCategoryFeaturesから取得できます。このフィールドはカテゴリごとのSKU上限数を示しており、大量バリエーションを持つ商品の出品前チェックとして活用できます。 3. 必須Item Specificsと推奨Item Specificsの見分け方 GetCategoryFeaturesのFeatureIDに「ItemSpecifics」を指定すると、そのカテゴリで定義されているItem Specificsの一覧と、それぞれが必須（Required）か推奨（Recommended）かオプション（Optional）かが返ってきます。 ここで注意が必要なのは「SelectionMode」と「MinValues」の組み合わせです。 SelectionMode=FreeText かつ MinValues=0 → 任意入力（Optional） SelectionMode=SelectionOnly かつ MinValues=1 → リストから選択必須（Required） SelectionMode=SelectionOrFreeText かつ MinValues=1 → リストか自由入力で必須（Required） この三つを判定して「必須フィールドが全部埋まっているか」を出品前にチェックするロジックを組むことで、APIエラーになる前に問題を検知できます。 頻出エラーコード早見表 エラーコード メッセージ 対処法 Error 21916585 Invalid category ID specified カテゴリIDが存在しない、または廃止済み。キャッシュを最新化して再確認する。 Error 21916616 Variations are not supported for this category バリエーション非対応カテゴリにVariationsブロックを送信した。GetCategoryFeaturesでVariationsEnabled=Trueを事前確認する。 Error 21916587 The feature VariationPictures is not supported VariationPictures非対応カテゴリで画像の軸指定を行った。 Error 878 Category ID is required CategoryIDフィールドが空またはnull。LeafCategoryのIDのみ指定可能。 堅牢な実装：カテゴリツリーJSON保存と必須フィールド自動チェックスクリプト ここまでの知識をまとめ、以下の二つの責務を持つ本番品質のスクリプトを実装します。 (1) CategoryCacheManager: GetCategoriesを呼び出し、全カテゴリをJSONに永続化・読み込みする管理クラス (2) CategoryFeatureChecker: GetCategoryFeaturesを呼び出し、必須Item Specifics・VariationsEnabled・ConditionEnabledを検査し、出品前の可否判定レポートを返す関数 # ebay_category_tools.py from __future__ import annotations import json import os import logging from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Optional import zeep import zeep.helpers import requests from zeep.transports import Transport logger = logging.getLogger(__name__) @dataclass class CategoryInfo: """単一カテゴリの基本情報。""" id: str name: str parent_id: Optional[str] level: int is_leaf: bool virtual: bool = False @dataclass class FeatureCheckResult: """GetCategoryFeaturesの解析結果。""" category_id: str variations_enabled: bool condition_enabled: bool required_item_specifics: list[str] = field(default_factory=list) recommended_item_specifics: list[str] = field(default_factory=list) errors: list[str] = field(default_factory=list) def is_listing_safe(self, provided_specifics: set[str]) -> tuple[bool, list[str]]: """ 出品時に提供したItem Specificsキー一覧を受け取り、 必須フィールドが揃っているか検証する。 Returns: (ok: bool, missing_fields: list[str]) """ missing = [s for s in self.required_item_specifics if s not in provided_specifics] return (len(missing) == 0), missing class CategoryCacheManager: """ GetCategoriesのレスポンスをJSONファイルにキャッシュし、 ローカル検索・ID解決を提供するマネージャー。 """ def __init__(self, token: str, cache_path: str = "ebay_categories.json", site_id: int = 0, cache_ttl_days: int = 7): if not token or not token.strip(): raise ValueError("eBay OAuth tokenは必須です") self.token = token self.cache_path = cache_path self.site_id = site_id self.cache_ttl = timedelta(days=cache_ttl_days) self._categories: dict[str, CategoryInfo] = {} self._client: Optional[zeep.Client] = None def _get_client(self) -> zeep.Client: if self._client is None: transport = Transport(session=requests.Session()) self._client = zeep.Client( "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl", transport=transport ) return self._client def _is_cache_fresh(self) -> bool: """キャッシュファイルが存在し、TTL内であればTrueを返す。""" if not os.path.exists(self.cache_path): return False try: with open(self.cache_path, "r", encoding="utf-8") as f: data = json.load(f) fetched_at = datetime.fromisoformat(data["fetched_at"]) return datetime.utcnow() - fetched_at < self.cache_ttl except (KeyError, ValueError, json.JSONDecodeError): return False def _fetch_from_api(self) -> list[dict]: """GetCategories APIを呼び出して全カテゴリを取得する。""" client = self._get_client() logger.info("GetCategories APIを呼び出し中 (site_id=%d)...", self.site_id) try: response = client.service.GetCategories( _soapheaders={"RequesterCredentials": {"eBayAuthToken": self.token}}, CategorySiteID=self.site_id, DetailLevel="ReturnAll", ViewAllNodes=True, ) except zeep.exceptions.Fault as e: raise RuntimeError(f"GetCategories SOAPエラー: {e.message}") from e result = zeep.helpers.serialize_object(response, target_cls=dict) categories = result.get("CategoryArray", {}).get("Category", []) or [] version = result.get("CategoryVersion") logger.info("取得完了: %d件 (CategoryVersion=%s)", len(categories), version) return categories, version def load(self, force_refresh: bool = False) -> None: """ カテゴリデータをロードする。 キャッシュが新鮮であればファイルから、そうでなければAPIから取得する。 """ if not force_refresh and self._is_cache_fresh(): logger.info("キャッシュから読み込み中: %s", self.cache_path) with open(self.cache_path, "r", encoding="utf-8") as f: data = json.load(f) raw_cats = data["categories"] else: raw_cats, version = self._fetch_from_api() cache_data = { "fetched_at": datetime.utcnow().isoformat(), "category_version": version, "site_id": self.site_id, "categories": [ { "id": str(c["CategoryID"]), "name": c["CategoryName"], "parent_id": str(c.get("CategoryParentID", "")), "level": int(c.get("CategoryLevel", 1)), "is_leaf": bool(c.get("LeafCategory", False)), "virtual": bool(c.get("Virtual", False)), } for c in raw_cats ], } with open(self.cache_path, "w", encoding="utf-8") as f: json.dump(cache_data, f, ensure_ascii=False, indent=2) logger.info("キャッシュ保存完了: %s", self.cache_path) raw_cats = cache_data["categories"] self._categories = { c["id"]: CategoryInfo(**c) for c in raw_cats } logger.info("メモリにロード完了: %d件のカテゴリ", len(self._categories)) def find_by_id(self, category_id: str) -> Optional[CategoryInfo]: """IDでカテゴリを検索する。""" return self._categories.get(str(category_id)) def search_by_name(self, keyword: str, leaf_only: bool = True) -> list[CategoryInfo]: """カテゴリ名（部分一致）でカテゴリを検索する。""" kw = keyword.lower() results = [ cat for cat in self._categories.values() if kw in cat.name.lower() and (not leaf_only or cat.is_leaf) ] return sorted(results, key=lambda c: c.name) def get_ancestors(self, category_id: str) -> list[CategoryInfo]: """指定カテゴリの祖先カテゴリを根から順に返す。""" path = [] current = self.find_by_id(category_id) visited = set() while current and current.id not in visited: path.append(current) visited.add(current.id) if current.parent_id and current.parent_id != current.id: current = self.find_by_id(current.parent_id) else: break return list(reversed(path)) class CategoryFeatureChecker: """ GetCategoryFeaturesを呼び出して、 カテゴリの出品ルールを解析するクラス。 """ def __init__(self, token: str): if not token or not token.strip(): raise ValueError("eBay OAuth tokenは必須です") self.token = token self._client: Optional[zeep.Client] = None def _get_client(self) -> zeep.Client: if self._client is None: transport = Transport(session=requests.Session()) self._client = zeep.Client( "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl", transport=transport ) return self._client def check(self, category_id: str) -> FeatureCheckResult: """ 指定カテゴリの機能フラグとItem Specifics要件を取得・解析する。 Args: category_id: LeafカテゴリのID（文字列） Returns: FeatureCheckResult Raises: ValueError: category_idが空の場合 RuntimeError: APIエラーが発生した場合 """ if not category_id or not str(category_id).strip(): raise ValueError("category_idは必須です") client = self._get_client() try: response = client.service.GetCategoryFeatures( _soapheaders={"RequesterCredentials": {"eBayAuthToken": self.token}}, CategoryID=str(category_id), FeatureID=["VariationsEnabled", "ConditionEnabled", "ItemSpecifics"], DetailLevel="ReturnAll", ViewDataOfAllDescendents=False, ) except zeep.exceptions.Fault as e: raise RuntimeError(f"GetCategoryFeatures SOAPエラー (category={category_id}): {e.message}") from e result = zeep.helpers.serialize_object(response, target_cls=dict) return self._parse_result(category_id, result) def _parse_result(self, category_id: str, result: dict) -> FeatureCheckResult: """レスポンスdictをFeatureCheckResultに変換する内部メソッド。""" site_defaults = result.get("SiteDefaults", {}) or {} cats = result.get("Category", []) or [] cat_data = cats[0] if cats else {} def resolve(key: str, default=None): """カテゴリ固有値 > SiteDefaults の優先順位で値を解決する。""" val = cat_data.get(key) if val is not None: return val return site_defaults.get(key, default) variations_enabled = bool(resolve("VariationsEnabled", False)) condition_enabled = bool(resolve("ConditionEnabled", False)) # Item Specificsの解析 required_fields: list[str] = [] recommended_fields: list[str] = [] item_specifics_data = resolve("ItemSpecificsEnabled") specifics_list = [] if isinstance(item_specifics_data, dict): specifics_list = item_specifics_data.get("ItemSpecific", []) or [] elif isinstance(item_specifics_data, list): specifics_list = item_specifics_data for spec in specifics_list: name = spec.get("Name", "") min_values = int(spec.get("MinValues", 0) or 0) if min_values >= 1: required_fields.append(name) else: recommended_fields.append(name) return FeatureCheckResult( category_id=category_id, variations_enabled=variations_enabled, condition_enabled=condition_enabled, required_item_specifics=required_fields, recommended_item_specifics=recommended_fields, ) def print_feature_report(result: FeatureCheckResult) -> None: """FeatureCheckResultを見やすく標準出力する。""" print(f"\n=== カテゴリ {result.category_id} 出品前チェックレポート ===") print(f" VariationsEnabled : {result.variations_enabled}") print(f" ConditionEnabled : {result.condition_enabled}") if result.required_item_specifics: print(f" 必須Item Specifics ({len(result.required_item_specifics)}件):") for s in result.required_item_specifics: print(f" - {s}") else: print(" 必須Item Specifics: なし") if result.recommended_item_specifics: print(f" 推奨Item Specifics ({len(result.recommended_item_specifics)}件):") for s in result.recommended_item_specifics[:10]: # 長い場合は先頭10件 print(f" - {s}") if __name__ == "__main__": import sys logging.basicConfig(level=logging.INFO) TOKEN = os.environ.get("EBAY_TOKEN", "") if not TOKEN: print("環境変数 EBAY_TOKEN を設定してください") sys.exit(1) # Step 1: カテゴリツリーをロード（キャッシュ優先） cache_mgr = CategoryCacheManager(token=TOKEN, site_id=0) cache_mgr.load() # Step 2: キーワードでカテゴリ検索 results = cache_mgr.search_by_name("T-Shirt", leaf_only=True) print(f"\n'T-Shirt'検索結果: {len(results)}件") for cat in results[:5]: ancestors = cache_mgr.get_ancestors(cat.id) path = " > ".join(a.name for a in ancestors) print(f" [{cat.id}] {path}") # Step 3: 特定カテゴリの出品ルールをチェック checker = CategoryFeatureChecker(token=TOKEN) target_cat_id = "15687" # 例: Men's T-Shirts feature_result = checker.check(target_cat_id) print_feature_report(feature_result) # Step 4: 出品前の必須Item Specifics検証 my_specifics = {"Brand", "Size", "Color"} # 自社システムが用意した属性 ok, missing = feature_result.is_listing_safe(my_specifics) if ok: print("\n必須Item Specificsは全て揃っています。出品可能です。") else: print(f"\n不足している必須Item Specifics: {missing}") print(" 出品前にこれらの属性を追加してください。") パフォーマンス・スケーリング視点 (深度) 本番稼働のeBay出品システムにおいて、カテゴリデータの管理はそれ単体で一つのサブシステムとして設計する価値があります。 カテゴリDBキャッシュと定期同期タスクの設計 JSONファイルへのキャッシュは開発・小規模運用では十分ですが、複数サーバーにスケールアウトする場合、各サーバーがそれぞれ別のJSONを持つと整合性が取れません。このフェーズに入ったら、カテゴリデータをPostgreSQLやMySQLなどの共有RDBMSに格納する設計に移行します。 テーブル設計のポイントは、category_idをプライマリキー（VARCHAR）、parent_idを外部キー（自己参照）とし、is_leaf・levelをインデックス付きカラムとして持つことです。カテゴリ名のキーワード検索にはFULL TEXT INDEXを活用すると、数万件のカテゴリ名から数ミリ秒でヒットします。 定期同期タスクは、Celery Beat（またはcron）で週に一度GetCategoriesを呼び出し、DBの内容と差分を取って更新・追加・廃止を反映します。廃止カテゴリはDBから削除するのではなく、is_active=Falseフラグを立てる「論理削除」が安全です。既にそのカテゴリで出品している商品への影響を追跡できるからです。 # sync_categories_task.py（Celeryタスクのイメージ） from celery import shared_task from ebay_category_tools import CategoryCacheManager import logging logger = logging.getLogger(__name__) @shared_task(name="sync_ebay_categories") def sync_ebay_categories() -> dict: """ 週次で実行するeBayカテゴリ同期タスク。 キャッシュTTLを0に設定することでAPIから強制再取得する。 """ import os token = os.environ["EBAY_TOKEN"] manager = CategoryCacheManager( token=token, cache_path="/shared/cache/ebay_categories.json", site_id=0, cache_ttl_days=0, # TTL=0 → 常にAPIから取得 ) manager.load(force_refresh=True) count = len(manager._categories) logger.info("カテゴリ同期完了: %d件", count) return {"synced_count": count} GetCategoryFeaturesについても同様に、頻繁に出品するカテゴリのFeatureCheckResultをRedisにシリアライズしてキャッシュすることを推奨します。TTLは7〜30日程度が現実的です。カテゴリのルールはGetCategoriesほど頻繁には変わりませんが、eBayのポリシー変更シーズン（毎年春・秋）前後は要注意です。 注意: GetCategoryFeaturesのAPI呼び出しコスト GetCategoryFeaturesはGetCategoriesよりも重いAPIです。1カテゴリ1リクエストであり、10,000カテゴリ分を一括で取得しようとすると当然10,000回のAPI呼び出しが必要です。実務では「実際に出品に使うカテゴリだけを都度オンデマンドで取得し、Redisにキャッシュする」というLazy-Loadingパターンが最も効率的です。全カテゴリを事前に取得しようとしないでください。 まとめ 本記事では、eBay Trading APIのメタデータ系エンドポイントであるGetCategoriesとGetCategoryFeaturesを使い、出品カテゴリの構造とルールをプログラムから取得・活用する方法を解説しました。 ベースライン: zeepを使ったGetCategories呼び出しと、カテゴリツリーのJSON永続化。LevelLimitとLeafCategoryフィールドの基本的な扱い方。 深いポイント: カテゴリツリーの大量データに対するキャッシュ戦略と強制リフレッシュの実装。GetCategoryFeaturesのSiteDefaults/Category優先順位ロジック。必須Item Specificsと推奨Item Specificsの判定と、VariationsEnabled・MaxGranularFitmentCountを使った出品前バリデーション。 スケーリング: JSONキャッシュから共有RDBMSへの移行設計、Celery Beatによる週次同期タスク、RedisによるGetCategoryFeaturesのLazy-Loadingキャッシュパターン。 これにより、カテゴリIDを勘で入力することなく、プログラムが自動的に正しいLeafカテゴリを特定し、そのカテゴリに必要なItem Specificsが揃っているかを出品前に検証できる、堅牢な出品パイプラインの基盤が整いました。 次のステップ 次回（#17）は、【Trading API - フィードバック管理】フェーズに進み、GetFeedbackで自分・相手のフィードバック履歴を取得し、LeaveFeedbackで条件に基づく自動評価送信を実装します。 取引完了後のフィードバック自動化は、セラーの評価スコアを維持するための重要な運用タスクです。どうぞお楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#15】eBay Trading API：SetNotificationPreferencesでeBayイベント通知を設定してPythonサーバーで受信する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第15回です。 前回（#14）は、GetMemberMessages / AddMemberMessageAAQToPartnerを使い、バイヤーからの問い合わせに自動返信するCS（カスタマーサポート）システムを構築しました。メッセージ管理の自動化により応答時間を劇的に短縮できましたが、「注文が確定したか」「入金が完了したか」を知るためには、依然として定期的なAPIポーリングが必要でした。 本記事では、このポーリング問題を根本から解決します。eBay の Platform Notifications 機能を使い、注文確定・入金・フィードバックなどのビジネスイベントが発生した瞬間に eBay から自分のサーバーへ通知を Push させるイベント駆動アーキテクチャを構築します。Trading API の SetNotificationPreferences で通知先 URL とイベント種別を登録し、FastAPI で受信エンドポイントを実装します。 この記事で得られること: SetNotificationPreferences と GetNotificationPreferences を使い、通知 URL と購読イベント種別を zeep（SOAP クライアント）経由で登録・確認する Python コードの実装。 eBay がエンドポイントの正当性を確認する チャレンジ・レスポンス検証（SHA-256 署名）の仕組みと、FastAPI による完全実装。 FastAPI 受信エンドポイントの構築：イベント種別（AuctionCheckoutComplete / FixedPriceTransaction / FeedbackLeft / ItemSold）ごとの処理振り分け、べき等性の確保、Celery + Redis によるスケールアウト設計。 背景・なぜこれが重要か (Motivation) 「定期的に GetOrders を叩けば十分じゃないのか？」 eBay 開発を始めたエンジニアの多くが最初にこう考えます。実際、5分おきに GetOrders を実行すれば新規注文を概ね検知できます。しかしこれは「動く」だけであって、「正しいアーキテクチャ」ではありません。 eBay の Trading API には「1日あたりの API 呼び出し数上限（API Call Limit）」があります。GetOrders を5分おきに実行すると1日288回のコールを消費します。複数のセラーアカウントを管理していたり、出品・在庫更新などの API 操作も並行して行う場合、この上限はあっという間に枯渇します。ポーリング間隔を短くするほど消費が速くなるという本質的なジレンマがあり、「もっとリアルタイムに」という要求を満たすほどコスト（API 消費）が跳ね上がります。 一方、Platform Notifications はイベント駆動型（Event-Driven）アーキテクチャです。eBay 側でイベントが発生したタイミングで、あなたのサーバーへ HTTPS POST リクエストが飛んできます。ポーリングのような API 呼び出し消費はゼロです。注文確定から数秒以内に通知が届くため、発送処理や在庫更新などの後続処理を即座に起動できます。 規模が拡大するほどこの差は顕著になります。月間 500 注文のセラーにとってはポーリングでも許容範囲ですが、月間 5,000 注文を超えてくると、通知ベースアーキテクチャは「あると便利」から「絶対に必要」へと変わります。 基本的な使い方（ベースライン）：SetNotificationPreferencesで通知URLを登録する まず zeep ライブラリを使って SetNotificationPreferences SOAP API を呼び出し、通知 URL と購読したいイベント種別を登録します。その後 GetNotificationPreferences で登録内容を確認する流れも合わせて示します。事前に pip install zeep requests を実行してください。 SetNotificationPreferences には大きく2つの設定ブロックがあります。ApplicationDeliveryPreferences は通知先 URL やペイロード形式といったアプリケーション全体の設定、UserDeliveryPreferenceArray は購読する個別イベント種別の有効/無効リストです。 # set_notification_prefs.py（ベースライン） from zeep import Client, Settings from zeep.transports import Transport import requests WSDL_URL = "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl" API_VERSION = "1311" SANDBOX_EP = "https://api.sandbox.ebay.com/ws/api.dll" PROD_EP = "https://api.ebay.com/ws/api.dll" SITE_ID_JP = "101" # eBay Japan def _make_session(config: dict, call_name: str, is_sandbox: bool) -> requests.Session: """Trading API 呼び出し用の HTTPヘッダー付き Session を生成する""" session = requests.Session() ep = SANDBOX_EP if is_sandbox else PROD_EP session.headers.update({ "X-EBAY-API-CALL-NAME" : call_name, "X-EBAY-API-SITEID" : SITE_ID_JP, "X-EBAY-API-COMPATIBILITY-LEVEL" : API_VERSION, "X-EBAY-API-APP-NAME" : config["app_id"], "X-EBAY-API-DEV-NAME" : config["dev_id"], "X-EBAY-API-CERT-NAME" : config["cert_id"], }) return session def register_notification_url( config: dict, notification_url: str, events: list, is_sandbox: bool = False ) -> None: """ SetNotificationPreferences を呼び出し、通知URLとイベント種別を登録する。 Args: config : app_id / dev_id / cert_id / user_token を含む辞書 notification_url: eBay からの通知を受け取る HTTPS URL events : 購読するイベント種別リスト is_sandbox : Sandbox 環境の場合は True """ session = _make_session(config, "SetNotificationPreferences", is_sandbox) settings = Settings(strict=False, xml_huge_tree=True) client = Client(wsdl=WSDL_URL, settings=settings, transport=Transport(session=session)) notification_enables = [ {"EventType": ev, "EventEnable": "Enable"} for ev in events ] response = client.service.SetNotificationPreferences( RequesterCredentials={"eBayAuthToken": config["user_token"]}, ApplicationDeliveryPreferences={ "ApplicationURL" : notification_url, "ApplicationEnable" : "Enable", "NotificationPayloadType" : "eBLSchemaSOAP", "DeviceType" : "Platform", }, UserDeliveryPreferenceArray={ "NotificationEnable": notification_enables }, ) if response.Ack not in ("Success", "Warning"): for err in (response.Errors or []): raise RuntimeError(f"[{err.ErrorCode}] {err.LongMessage}") print(f"通知URL登録完了: {notification_url}") def check_notification_preferences(config: dict, is_sandbox: bool = False) -> None: """GetNotificationPreferences で現在の通知設定を確認する""" session = _make_session(config, "GetNotificationPreferences", is_sandbox) settings = Settings(strict=False, xml_huge_tree=True) client = Client(wsdl=WSDL_URL, settings=settings, transport=Transport(session=session)) response = client.service.GetNotificationPreferences( RequesterCredentials={"eBayAuthToken": config["user_token"]}, PreferenceLevel="UserData", ) adp = response.ApplicationDeliveryPreferences print(f" 登録URL : {adp.ApplicationURL}") print(f" 有効状態 : {adp.ApplicationEnable}") udpa = response.UserDeliveryPreferenceArray if udpa and udpa.NotificationEnable: for ne in udpa.NotificationEnable: print(f" イベント : {ne.EventType} -> {ne.EventEnable}") # ===== 実行例 ===== if __name__ == "__main__": config = { "app_id" : "YourApp-XXXX", "dev_id" : "your-dev-id-xxxx", "cert_id" : "your-cert-id-xxxx", "user_token" : "AgAAAA**...", } EVENTS = [ "AuctionCheckoutComplete", "FixedPriceTransaction", "FeedbackLeft", "ItemSold", ] register_notification_url( config=config, notification_url="https://your-server.example.com/ebay/notifications", events=EVENTS, is_sandbox=True, # まずは Sandbox でテスト ) check_notification_preferences(config, is_sandbox=True) 補足: ApplicationDeliveryPreferences の各フィールド ApplicationURL は eBay が通知を POST する先の HTTPS エンドポイント URL です。ApplicationEnable を "Enable" に設定することで通知が有効になります。NotificationPayloadType は "eBLSchemaSOAP"（推奨）を指定すると SOAP 形式のリッチなペイロードが届きます。DeviceType は常に "Platform" を指定します。 購読可能な主要イベント種別は次の通りです。AuctionCheckoutComplete（オークション落札後のチェックアウト完了）、FixedPriceTransaction（固定価格取引完了）、FeedbackLeft（バイヤーによるフィードバック投稿）、ItemSold（商品売却）は特に頻繁に使用されます。他にも ItemEndedBySeller（出品終了）、BidReceived（入札受付）など多数のイベントがサポートされています。GetNotificationPreferences の PreferenceLevel="Application" を指定すると、利用可能なイベント一覧が取得できます。 実務で躓く場面・深いポイント (Core) Platform Notifications の設定は一見シンプルですが、実務に投入すると必ず以下の壁にぶつかります。特に「なぜ通知が届かないのか」というデバッグは複数の原因が絡み合うため、数時間を費やす罠になりがちです。 1. 通知URLのHTTPS必須とチャレンジ・レスポンス検証 最も重要な制約は、通知 URL は本番環境では必ず HTTPS でなければならない点です（自己署名証明書は受け付けません。Let's Encrypt などの正規証明書が必要です）。Sandbox では http://localhost が一時的に許容される場合もありますが、本番では HTTPS のみです。 さらに、URL を登録しただけでは通知は届きません。eBay はまず GET リクエストを送信し、エンドポイントの正当性を確認する「チャレンジ・レスポンス検証」を行います。eBay は challenge_code というクエリパラメータ付きの GET を送り、サーバーは SHA-256(challenge_code + verificationToken + endpointURL) を計算した16進ハッシュを JSON で返さなければなりません。このレスポンスが正しくないと、SetNotificationPreferences の呼び出し自体は成功しても実際の通知は一切届きません。 さらに本番では、eBay が実際に通知を POST する際に X-EBAY-SIGNATURE ヘッダーが付与されます。このシグネチャを検証しないシステムは、悪意のある第三者がエンドポイントを叩いて注文処理を誤作動させるリスクを抱えます。本番システムでは署名検証を必ず実装してください。 2. 重複配信への対応（べき等性キーの設計） eBay の通知は「At-Least-Once（少なくとも1回）配信」です。ネットワーク障害やサーバーの応答遅延が発生した場合、同一イベントが2回・3回と配信されることがあります。この前提を無視して「通知が来たら即座に注文レコードを作成する」実装をすると、同じ注文がデータベースに複数回書き込まれるという深刻なバグが発生します。 対策はべき等性（Idempotency）の確保です。eBay が送信する SOAP メッセージには Timestamp や ItemID / TransactionID が含まれており、これらを組み合わせた一意キーで「処理済みかどうか」を管理します。Redis の SET NX（Not eXists）コマンドを使い、処理開始時にキーを立て、成功後にそのキーを維持するパターンが堅牢です。 ポイントは、べき等性チェックを「処理完了後に記録する」のではなく「処理開始時にアトミックに取得する」ことです。処理中にクラッシュした場合に通知IDが「処理済み」として残ると、再送時にスキップされて注文が永遠に処理されない「処理漏れ」が発生します。失敗時はキーを削除してリトライを許可する設計が必要です。 3. SandboxとProductionで異なる環境設定の管理 eBay Sandbox と本番環境では API 呼び出し先が異なります（api.sandbox.ebay.com vs api.ebay.com）。通知も同様で、Sandbox 用の通知 URL と Production 用の通知 URL を環境変数で明確に分離して管理することが重要です。 よくある失敗が、「Sandbox 用に登録した URL に本番通知が届いてしまう（またはその逆）」という混線です。URL のパスに /sandbox/ や /production/ を含める命名規則を採用し、環境を判別しやすくするのが実務的なベストプラクティスです。設定をコードにハードコードせず、環境変数（.env や AWS Secrets Manager）から読み込む設計にしてください。 注意: Sandbox 環境での通知遅延と動作差異 Sandbox 環境では eBay の通知配信が本番より大幅に遅延する場合があります（場合によっては数時間後）。また Sandbox ではすべてのイベント種別がサポートされているわけではありません。開発中に通知が届かない場合は、まず GetNotificationPreferences で設定が正しく反映されているかを確認し、次に Sandbox 特有の遅延を疑ってください。ポーリングで動作確認してから通知に切り替える段階的アプローチも有効です。 頻出エラーコード早見表 エラーコード 概要 対処法 21916667 DeliveryURL が無効な形式 URL の形式を確認。http:// は本番では不可。正規ドメインの https:// を使用する。 21916669 ApplicationURL must be https http:// を https:// に変更。Let's Encrypt 等の正規 SSL 証明書が必要。 21916672 指定の EventType がサポート外 GetNotificationPreferences（PreferenceLevel=Application）で利用可能なイベント一覧を確認し、正確な文字列を指定する。 21916680 ApplicationURL の最大登録数を超過 GetNotificationPreferences で現在の登録数を確認し、不要なURLを削除してから再登録する。 堅牢な実装：FastAPIによるイベント振り分けエンドポイント ここでは実務に耐える FastAPI ベースの通知受信サーバーを実装します。チャレンジ・レスポンス検証、イベント種別ごとの処理振り分け、エラーハンドリング、BackgroundTasks を使った即時応答を含む完全な実装です。事前に pip install fastapi uvicorn を実行してください。起動コマンドは uvicorn notification_receiver:app --host 0.0.0.0 --port 8000 です。 デコレータ方式の @notification_handler レジストリを採用することで、新しいイベント種別への対応を関数1つ追加するだけで拡張でき、コアのディスパッチロジックに変更を加えずに済みます。オープン・クローズド原則に従った設計です。 # notification_receiver.py import hashlib import logging import os import xml.etree.ElementTree as ET from typing import Optional from fastapi import FastAPI, Request, HTTPException, Query, BackgroundTasks from fastapi.responses import JSONResponse logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") # 環境変数から設定を読み込む（本番は Secrets Manager 等を使用） VERIFICATION_TOKEN: str = os.environ["EBAY_VERIFICATION_TOKEN"] # 32文字以上推奨 ENDPOINT_URL: str = os.environ["EBAY_NOTIFICATION_ENDPOINT_URL"] app = FastAPI(title="eBay Platform Notification Receiver") # ── イベントハンドラーレジストリ ────────────────────────────── _HANDLERS: dict = {} def notification_handler(event_type: str): """@notification_handler("AuctionCheckoutComplete") デコレータ""" def decorator(fn): _HANDLERS[event_type] = fn return fn return decorator # ── GET /ebay/notifications ← チャレンジ・レスポンス検証 ────── @app.get("/ebay/notifications") async def handle_challenge(challenge_code: Optional[str] = Query(default=None)): """ eBay エンドポイント検証（チャレンジ・レスポンス）に応答する。 SetNotificationPreferences で URL を登録すると、eBay はまず GET リクエストを送信してエンドポイントの正当性を確認する。 challengeResponse = SHA-256(challenge_code + verificationToken + endpointUrl) """ if challenge_code is None: raise HTTPException(status_code=400, detail="challenge_code が必要です") hash_input = f"{challenge_code}{VERIFICATION_TOKEN}{ENDPOINT_URL}" challenge_resp = hashlib.sha256(hash_input.encode("utf-8")).hexdigest() logger.info(f"チャレンジ検証 code={challenge_code[:8]}... resp={challenge_resp[:8]}...") return JSONResponse(content={"challengeResponse": challenge_resp}) # ── POST /ebay/notifications ← 実際の通知受信 ───────────────── @app.post("/ebay/notifications") async def receive_notification(request: Request, bg: BackgroundTasks): """ eBay からのプラットフォーム通知を受信し、イベント種別で振り分ける。 重要: eBay は 30 秒以内に 200 応答を受け取れない場合に再送を行う。 重い処理はすべて BackgroundTasks で非同期実行し、即座に 200 を返す。 """ body = await request.body() bg.add_task(_dispatch, body) return JSONResponse(content={"status": "accepted"}, status_code=200) # ── 内部: SOAP XML 解析 → イベント振り分け ──────────────────── SOAP_NS = "http://schemas.xmlsoap.org/soap/envelope/" async def _dispatch(body: bytes) -> None: """SOAPボディを解析し、登録済みハンドラーへ振り分ける""" try: root = ET.fromstring(body) soap_body = root.find(f"{{{SOAP_NS}}}Body") if soap_body is None: logger.error("SOAPボディの解析失敗") return for child in soap_body: local_tag = child.tag.split("}")[-1] if "}" in child.tag else child.tag handler = _HANDLERS.get(local_tag) if handler: logger.info(f"通知受信: {local_tag}") await handler(child) else: logger.warning(f"未定義の通知タイプ: {local_tag}") except ET.ParseError as exc: logger.error(f"XML解析エラー: {exc} | body先頭={body[:120]}") def _text(elem, path: str, default: str = "") -> str: """XML要素から安全にテキストを取得するヘルパー""" node = elem.find(path) return node.text if node is not None and node.text else default # ── イベントハンドラー定義 ──────────────────────────────────── @notification_handler("AuctionCheckoutComplete") async def handle_auction_checkout(elem) -> None: """オークション落札確定（AuctionCheckoutComplete）を処理する""" item_id = _text(elem, "Item/ItemID") buyer_id = _text(elem, "Transaction/Buyer/UserID") txn_id = _text(elem, "Transaction/TransactionID") logger.info(f"[AuctionCheckout] item={item_id} txn={txn_id} buyer={buyer_id}") # TODO: 注文DBへの保存、梱包・発送フロー起動 @notification_handler("FixedPriceTransaction") async def handle_fixed_price_txn(elem) -> None: """固定価格取引完了（FixedPriceTransaction）を処理する""" item_id = _text(elem, "Item/ItemID") txn_id = _text(elem, "Transaction/TransactionID") amount = _text(elem, "Transaction/TransactionPrice", "0.00") logger.info(f"[FixedPriceTxn] item={item_id} txn={txn_id} amount=JPY{amount}") # TODO: 在庫数の減算、注文確認メール送信 @notification_handler("FeedbackLeft") async def handle_feedback_left(elem) -> None: """フィードバック受信（FeedbackLeft）を処理する""" comment_type = _text(elem, "FeedbackDetail/CommentType") commenter = _text(elem, "FeedbackDetail/CommentingUser") logger.info(f"[Feedback] type={comment_type} from={commenter}") # TODO: Negative/Neutral の場合は緊急アラートを発報 @notification_handler("ItemSold") async def handle_item_sold(elem) -> None: """商品売却（ItemSold）を処理する""" item_id = _text(elem, "ItemID") logger.info(f"[ItemSold] item={item_id}") # TODO: 出品リストの更新、補充アラート _HANDLERS レジストリとデコレータパターンにより、新しいイベント種別への対応は @notification_handler("新イベント名") を付けた async 関数を追加するだけです。コアの _dispatch ロジックに変更を加える必要がないため、機能拡張時の影響範囲を最小化できます。 receive_notification エンドポイントが即座に 200 を返し、重い処理を BackgroundTasks に委ねる設計は非常に重要です。eBay は 30 秒以内に 200 応答を受信できない場合に同一通知を再送します。重いハンドラーが同期で実行されると再送ループに入り、「重複通知の嵐」を引き起こします。さらに、ハンドラー内で例外が発生しても 500 を eBay に返してはいけません。500 は再送トリガーになるからです。すべての例外を内部で catch し、eBay には常に 200 を返す設計にしてください。 パフォーマンス・スケーリング視点 (深度) 非同期キューとべき等性キー設計によるスケールアウト FastAPI の BackgroundTasks はプロセス内での非同期処理であり、サーバーが1台の間は十分に機能します。しかし、月間1万件を超える注文を処理する規模になると単一プロセスでの処理はボトルネックになります。また、サーバーが突然クラッシュした場合、BackgroundTasks に積まれた未処理の通知が失われるリスクもあります。 本番規模のシステムでは、受信（FastAPI）と処理（ワーカー）を分離し、Celery + Redis（または Amazon SQS）を使ったメッセージキューアーキテクチャへの移行を強く推奨します。FastAPI はリクエストを受け取ったらキューに積んで即座に 200 を返し、複数の Celery ワーカーが並列でキューを消化するパターンです。これによりワーカーを水平スケールアウトして処理能力を動的に調整できます。 # tasks.py ─ Celery + Redis によるべき等処理ワーカー import logging from celery import Celery import redis logger = logging.getLogger(__name__) redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True) celery_app = Celery("ebay_notify", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1") IDEMPOTENCY_TTL = 86_400 # 24時間（秒） @celery_app.task(bind=True, max_retries=3, default_retry_delay=10) def process_fixed_price_transaction(self, notification_id: str, payload: dict) -> None: """ 固定価格取引通知をべき等性を保証しながら非同期処理する。 Args: notification_id : 通知の一意ID（Timestamp + ItemID + TransactionID のハッシュ等） payload : 通知 XML から抽出した取引データ """ # ── Step1: べき等性チェック（SET NX = Not eXists） ────────── idem_key = f"ebay:processed:{notification_id}" acquired = redis_client.set(idem_key, "1", ex=IDEMPOTENCY_TTL, nx=True) if not acquired: logger.info(f"重複通知をスキップ: {notification_id}") return # 既に処理済み → 安全に終了 try: # ── Step2: ビジネスロジック ────────────────────────────── decrement_inventory(payload["item_id"], qty=1) order_id = create_order(payload) send_confirmation_email(payload["buyer_email"], order_id) logger.info(f"処理完了: notification={notification_id} order={order_id}") except Exception as exc: # 処理失敗時はべき等性キーを削除してリトライを許可 redis_client.delete(idem_key) logger.error(f"処理失敗（リトライ予定）: {exc}") raise self.retry(exc=exc) # FastAPI 側からはこう呼び出す: # process_fixed_price_transaction.delay(notification_id, payload) SET NX（Not eXists）は Redis のアトミック操作であるため、複数ワーカーが同じ notification_id を同時に処理しようとしてもどちらか一方だけが処理を進められることを保証します。分散システムにおける「競合状態（Race Condition）」の回避に不可欠なパターンです。 notification_id の生成方法も重要です。eBay の SOAP メッセージには固定の通知 ID フィールドが常に存在するわけではありません。Timestamp + ItemID + TransactionID を結合した文字列の SHA-256 ハッシュをキーとして使う設計が、重複排除において最も堅牢です。TTL を 24 時間に設定することで、Redis のメモリ消費を抑えつつ実運用上の再送ウィンドウをカバーできます。 まとめ 本記事では、定期ポーリングからイベント駆動アーキテクチャへの移行を実現する eBay Platform Notifications の全体像を解説しました。 ベースライン: zeep を使った SetNotificationPreferences / GetNotificationPreferences の呼び出しにより、通知 URL とイベント種別（AuctionCheckoutComplete / FixedPriceTransaction / FeedbackLeft / ItemSold）を登録・確認する基本実装。 深いポイント: チャレンジ・レスポンス検証（SHA-256）によるエンドポイント認証、At-Least-Once 配信に対応したべき等性設計、Sandbox / Production の環境分離の重要性、および頻出エラーコードへの対処法。 スケーリング: FastAPI の BackgroundTasks による即時応答パターンから、Celery + Redis を使ったキューベースのスケールアウト設計、および Redis SET NX を用いた分散環境でのべき等性保証。 Platform Notifications を導入することで、API コール消費をゼロに抑えながら注文確定から数秒以内に在庫更新・出荷指示・確認メールを起動できるリアルタイムシステムが実現します。 次のステップ 通知システムが整ったことで、eBay セラー業務における「リアクティブな処理」の基盤が完成しました。次は「プロアクティブな出品戦略」を支える基礎知識に目を向けます。 次回（#16）は、「GetCategoriesとGetCategoryFeaturesで出品カテゴリの構造とルールを取得する」です。eBay の膨大なカテゴリツリーをプログラムから走査し、対象カテゴリで必須となる Item Specifics や出品ルールを API で動的に取得する方法を解説します。お楽しみに！ 次の記事はこちら