前回の記事はこちら 【連載#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 で動的に取得する方法を解説します。お楽しみに！ 次の記事はこちら

【重要】EUエネルギーラベル規制におけるPDFドキュメントサポート開始のお知らせ (EU Energy Labelling PDF Support) 開発者の皆様、 EUエネルギーラベル規制（EU Energy Labelling regulations）に準拠するため、EU圏内で電化製品、光源、スマートフォン、タブレット、タイヤなどの対象商品を出品するセラー（販売者）は、出品内にエネルギーラベルおよび製品パフォーマンス情報を提供することが義務付けられています。この情報は、バイヤー（購入者）が十分な情報に基づいて購入決定を行うために不可欠です。 新機能: PDFドキュメントのサポート (Support for PDF documents) eBayでは従来、エネルギーラベル（Energy Label）および製品情報シート（Product Information Sheet）の提供において画像フォーマットのみをサポートしていましたが、この度PDFドキュメントのアップロードに対応いたしました。 この機能は Media API を通じて利用可能であり、開発者はPDFまたは画像ドキュメントをアップロードし、それらを出品に関連付けることができます。これにより、セラーおよびサードパーティツールプロバイダーは、サポートされている画像形式に加えて、PDF形式でもドキュメントをアップロードできるようになります。 エネルギーラベルおよび製品情報シートPDFのアップロード手順 これらのドキュメントをアップロードするには、Media API を使用します。 ドキュメントリソースの作成 (Create the document resource) createDocument メソッドを呼び出してドキュメントをステージング（準備）し、documentId を取得します。 ドキュメントファイルのアップロード (Upload the document file) 返された documentId を使用して uploadDocument メソッドを呼び出し、PDFまたは画像ファイルをアップロードします。 サポートされているドキュメントタイプ（documentType）は以下の通りです： EEK_ENERGY_LABEL （エネルギー効率ラベル） EEK_PRODUCT_INFORMATION_SHEET （製品情報シート） アップロードしたドキュメントと出品の関連付け ドキュメントのアップロードが完了したら、取得した documentId を出品リクエストに含めて送信します。 APIごとの指定方法: Inventory API: 出品の作成または更新時（例: createOffer または updateOffer）に、regulatory.documents コンテナ内にドキュメントIDを指定します。 Trading API: AddItem または ReviseItem を呼び出す際に、Regulatory.Documents コンテナ内にドキュメントIDを指定します。 重要（注意事項）: PDFファイルのサイズは 10MB を超えてはなりません。 システム連携においてエネルギーラベルの全フィールドセットをサポートすることで、セラーが出品上の表示内容を最も適切にコントロールできるようになります。エネルギー効率情報が欠落している出品は非表示となり、バイヤーが閲覧または購入できなくなる可能性があります。 詳細情報については、eBayセラーセンターの Energy Efficiency Information ヘルプページをご参照ください。 今後ともよろしくお願い申し上げます。

前回の記事はこちら 【連載#14】eBay Trading API：GetMemberMessages / AddMemberMessageAAQToPartnerでバイヤーへの自動返信を実装する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第14回です。 前回（#13）は CompleteSale API を使い、発送済みマークと追跡番号をプログラムから一括登録する方法を解説しました。荷物が旅立ったその瞬間から、今度はバイヤーからのメッセージが届き始めます。「荷物はどこ？」「これ返品できる？」「このアイテムの使い方は？」――CS（カスタマーサポート）対応は、越境 EC における最も労働集約的な業務の一つです。 本記事では、Trading API の GetMemberMessages と AddMemberMessageAAQToPartner を組み合わせ、未返信メッセージを自動取得・分類し、テンプレートで返信する「CSボット」を Python で実装します。 この記事で得られること: GetMemberMessages のパラメータ体系と、未返信メッセージのみを効率的に絞り込むフィルタリング手法。 AddMemberMessageAAQToPartner の必須三点セット（ItemID・RecipientID・ParentMessageID）と、その一つでも欠けると発生するエラーの回避策。 日英混在メッセージを「発送・返品・質問」に分類し、テンプレートで自動返信するキーワードベースの分類ロジックの実装と、その限界・注意事項。 背景・なぜこれが重要か (Motivation) 「メッセージなんて、受け取ったら手動でその都度返せばいい。」 これは初学者が抱く最も自然な感想ですが、eBay のビジネスルールとセラー評価制度を知ると、その認識がいかに危険かがわかります。 eBay は各セラーに「Seller Level」を付与し、その評価指標の一つに Response Rate（返信率）と Response Time（返信時間）があります。eBay 公式の基準によれば、バイヤーからの問い合わせに 24時間以内に返信できなかった場合、Response Rate がカウントされます。この数字が一定水準を下回ると、出品が Best Match（eBay の検索ランキングアルゴリズム）で不利な扱いを受け、さらには「Below Standard」セラーへと格下げされるリスクがあります。 出品数が数十件のうちは手動対応でも回せます。しかし SKU 数が数百・数千件に増え、時差のある海外バイヤーから深夜にメッセージが届くようになると、手動対応は物理的に破綻します。API による自動化は、スケールする越境 EC ビジネスにとって避けられないステップです。 また、自動返信は「とりあえず受け取ったことを知らせる」だけでも大きな効果があります。バイヤーは返信があると安心し、ネガティブフィードバック（Negative Feedback）を残す確率が著しく下がります。CS 品質の向上は、Defect Rate（欠陥率）の改善にも直結します。 基本的な使い方（ベースライン）：GetMemberMessagesで未返信メッセージを取得する まず zeep を使った最小限の実装で、直近 24 時間の未返信 ASQ メッセージ（Ask Seller Question: 商品ページの「Contact seller」から届くメッセージ）を取得してみます。 # cs_bot_baseline.py import zeep from datetime import datetime, timedelta, timezone EBAY_WSDL = "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl" def get_member_messages(user_token: str) -> dict: """未返信のバイヤーメッセージを取得する（直近24時間分）。""" transport = zeep.Transport() client = zeep.Client(wsdl=EBAY_WSDL, transport=transport) now = datetime.now(timezone.utc) start = now - timedelta(hours=24) response = client.service.GetMemberMessages( _soapheaders={ 'RequesterCredentials': {'eBayAuthToken': user_token} }, ErrorLanguage='ja_JP', WarningLevel='High', MailMessageType='AskSellerQuestion', MessageStatus='Unanswered', StartCreationTime=start.strftime('%Y-%m-%dT%H:%M:%S.000Z'), EndCreationTime=now.strftime('%Y-%m-%dT%H:%M:%S.000Z'), Pagination={'EntriesPerPage': 25, 'PageNumber': 1}, ) return response # --- 動作確認 --- if __name__ == "__main__": import os TOKEN = os.environ["EBAY_USER_TOKEN"] resp = get_member_messages(TOKEN) print(f"Ack: {resp.Ack}") exchanges = ( getattr(resp.MemberMessage, "MemberMessageExchange", []) or [] ) print(f"未返信メッセージ数: {len(exchanges)}") for ex in exchanges: q = ex.Question print(f" MessageID={q.MessageID}, Sender={q.SenderID}") print(f" QuestionType={q.QuestionType}") print(f" Subject: {q.Subject}") 補足: GetMemberMessages の主要パラメータ MailMessageType には AskSellerQuestion（ASQ）、All（全種）などを指定できます。本記事では ASQ のみを対象とします。AllMessages（セラーが送信したメッセージを含む全スレッド）と混同しないよう注意してください。 MessageStatus には Unanswered（未返信）と Answered（返信済み）の2値を指定できます。Unanswered を指定することで、返信が必要なメッセージだけを効率的に取得できます。 レスポンスの MemberMessageExchange は「1往復の会話スレッド」を表すオブジェクトです。その中の Question がバイヤーからの問い合わせ本体であり、MessageID、SenderID、Body、ItemID（商品ページから問い合わせた場合）などが含まれます。QuestionType フィールドには General（一般的な質問）、Shipping（配送関連）、CustomizedOrder（カスタム注文）などが自動付与されることがありますが、常にセットされるとは限りません。そのため、本記事では本文のキーワード分類を主な判定ロジックとして採用します。 実務で躓く場面・深いポイント (Core) GetMemberMessages で取得まではできた。では返信 API を呼ぶ――ここで多くのエンジニアが数時間を溶かします。AddMemberMessageAAQToPartner は一見シンプルに見えて、非常に厳格なパラメータ検証を行います。 1. AddMemberMessageAAQToPartner の「必須三点セット」 AddMemberMessageAAQToPartner の呼び出しには、以下の3つのパラメータが必ず揃っていなければなりません。 ItemID: 元の問い合わせが紐付いている商品の ItemID。GetMemberMessages の MemberMessageExchange.Item.ItemID から取得します。商品ページを経由しない一般的なメッセージ（たとえばセラーページの Contact ボタンから来たもの）は ItemID が空の場合があり、その場合は AddMemberMessageAAQToPartner ではなく AddMemberMessageRTQ を使う必要があります。 MemberMessage.RecipientID: 返信先のユーザー ID（Question.SenderID）。eBay アカウントの UserID を文字列で指定します。大文字小文字が一致しないと Error 21915815 が発生します。 MemberMessage.ParentMessageID: 元の問い合わせの MessageID。返信がどのスレッドに属するかを eBay が紐付けるためのキーです。これを誤ると、全く別の会話に返信が付く、あるいは Error 21915814 で弾かれます。 最も踏みやすい落とし穴は、GetMemberMessages の結果から ItemID を取得する際に None チェックを怠ることです。Item オブジェクト自体が None の場合があり、そのまま .ItemID を参照すると AttributeError がスローされます。必ず以下のように安全にアクセスしてください。 # 安全な ItemID 取得パターン item_id = getattr(getattr(exchange, "Item", None), "ItemID", None) if not item_id: # ItemID がない = RTQ または一般メッセージ → 手動対応キューへ logger.warning("ItemID なし: message_id=%s", message_id) continue 2. 日英混在メッセージのキーワード分類の落とし穴 グローバルな eBay では、日本のバイヤーが日本語で書いてきたり、英語で書いてきたりします。さらに「トラッキングナンバーはいつ届く？」のように日英混在のメッセージも珍しくありません。 キーワードリストを英語だけで組むと、日本語メッセージを UNKNOWN（分類不能）と判定してしまい、手動対応キューが溢れます。逆に日本語キーワードだけを使うと英語バイヤーへの対応が漏れます。実務では、分類ロジックに両言語のキーワードセットを含めることが必須です。 また、単純な部分文字列マッチングでは誤判定が起きやすい点にも注意が必要です。たとえば「return address（返送先住所）」は配送関連であって返品関連ではありませんが、"return" というキーワードに引っかかります。ビジネスの規模が大きくなったら、形態素解析（fugashi / MeCab）や軽量な分類モデル（scikit-learn の TF-IDF + ロジスティック回帰など）への移行を検討してください。 3. 自動返信してはいけない場面と重複送信ループ防止 注意: 以下のケースには絶対に自動返信を送らないでください。誤った自動返信が状況を悪化させ、eBay からのペナルティを受ける可能性があります。 クレーム・不正申告（SNAD: Significantly Not As Described / Item Not Received）: これらは eBay Money Back Guarantee（eBay バイヤー保護）の対象ケースです。機械的な返信テンプレートはバイヤーの怒りを増幅させます。必ず人間が対応してください。QuestionType が "INR" や "SNAD" の場合は即座に手動対応フラグを立てます。 詐欺疑いメッセージ（フィッシング・外部決済誘導）: "PayPal only", "pay outside eBay" などの文言が含まれるメッセージへの自動返信は、詐欺行為への加担とみなされるリスクがあります。 連続自動返信ループ: バイヤーが自動返信に再返信し、再度 Unanswered として取得されてしまう場合、ループが発生します。ParentMessageID で既に自分が返信したスレッドを追跡し、重複返信を防ぐ制御が必要です。実装例として、返信済み MessageID を Redis や DB に保存し、run_once() の冒頭でチェックするパターンが一般的です。 頻出エラーコード早見表 エラーコード 説明と対処 Error 21915814 ParentMessageID に対応するメッセージが存在しない、または既に別のメッセージが返信として紐付き済み。MessageID の取得元（GetMemberMessages の最新呼び出し）が古くなっていないか確認すること。 Error 21915815 RecipientID に指定した UserID が eBay に存在しない、または大文字小文字が一致していない。SenderID をそのまま流用し、独自加工（toLowerCase 等）をかけないこと。 Error 21915414 ItemID が eBay に存在しない、または出品が終了・削除されている。商品ページから来た問い合わせでも、出品が終了後に届いたメッセージは ItemID が無効化されていることがある。このエラーは自動返信をスキップして手動対応キューへ回す。 Error 55012 OAuth トークンの有効期限切れ、またはスコープ不足。eBay Trading API は User Token（Require Token）スコープが必須。Application Token のみでは GetMemberMessages・AddMemberMessageAAQToPartner は呼べない。トークンのリフレッシュ処理を実装すること。 堅牢な実装：キーワード分類とテンプレート自動返信 CSボット 上記の実務的な課題をすべて踏まえた、型アノテーション・docstring・例外処理つきの完全実装を示します。未返信メッセージの取得、キーワードによる3カテゴリ分類、テンプレート返信の一連のフローを CSBot クラスにまとめます。 # cs_bot_production.py import zeep import logging from datetime import datetime, timedelta, timezone from dataclasses import dataclass from typing import Optional from enum import Enum, auto logger = logging.getLogger(__name__) class MessageCategory(Enum): """メッセージの分類カテゴリ。""" SHIPPING = auto() # 発送・配送関連 RETURN = auto() # 返品・返金関連 QUESTION = auto() # 商品に関する一般質問 UNKNOWN = auto() # 分類不能 -> 手動対応キューへ KEYWORDS: dict[MessageCategory, list[str]] = { MessageCategory.SHIPPING: [ 'ship', 'tracking', 'deliver', 'delivery', 'dispatch', 'sent', 'transit', '発送', '追跡', '配送', '届', '送り', 'トラッキング', '輸送', ], MessageCategory.RETURN: [ 'return', 'refund', 'broken', 'damaged', 'wrong item', 'cancel', '返品', '返金', '壊れ', '破損', '違う', '間違い', 'キャンセル', ], MessageCategory.QUESTION: [ 'question', 'condition', 'compatible', 'include', 'color', 'size', '質問', '状態', '対応', '含ま', '色', 'サイズ', '詳細', '仙5様', ], } TEMPLATES: dict[MessageCategory, str] = { MessageCategory.SHIPPING: ( "お問い合わせありがとうございます。\n" "ご注文の商品は発送済みです。\n" "追跡番号: {tracking_number}\n" "お届けまでしばらくお待ちください。" ), MessageCategory.RETURN: ( "ご不便をおかけして申し訳ございません。\n" "返品・返金については eBay のリターンプロセスをご利用ください。\n" "ご不明な点があればお気軽にご連絡ください。" ), MessageCategory.QUESTION: ( "お問い合わせありがとうございます。\n" "ご質問の件について確認し、改めてご返答いたします。\n" "しばらくお時間をいただけますと幸いです。" ), } @dataclass class EbayMessage: """GetMemberMessages から取得した1件のメッセージを表すデータクラス。""" message_id: str item_id: str # 空文字列の場合は ItemID 未紐付けメッセージ sender_id: str subject: str body: str question_type: str message_status: str class CSBot: """eBay CS 自動返信ボット。""" EBAY_WSDL = "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl" def __init__(self, user_token: str) -> None: self.user_token = user_token transport = zeep.Transport() self.client = zeep.Client(wsdl=self.EBAY_WSDL, transport=transport) self._headers = { 'RequesterCredentials': {'eBayAuthToken': self.user_token} } # ---------------------------------------------------------- # Step 1: 未返信メッセージ取得 # ---------------------------------------------------------- def get_unanswered_messages( self, hours_back: int = 24, page: int = 1 ) -> list[EbayMessage]: """ 直近 hours_back 時間内の未返信 ASQ メッセージを取得する。 Args: hours_back: 遡る時間数（デフォルト 24 時間）。 page: ページ番号（デフォルト 1、1ページ最大 25 件）。 Returns: EbayMessage のリスト。 Raises: RuntimeError: eBay がエラーを返した場合。 """ now = datetime.now(timezone.utc) start = now - timedelta(hours=hours_back) try: resp = self.client.service.GetMemberMessages( _soapheaders=self._headers, ErrorLanguage='ja_JP', WarningLevel='High', MailMessageType='AskSellerQuestion', MessageStatus='Unanswered', StartCreationTime=start.strftime('%Y-%m-%dT%H:%M:%S.000Z'), EndCreationTime=now.strftime('%Y-%m-%dT%H:%M:%S.000Z'), Pagination={'EntriesPerPage': 25, 'PageNumber': page}, ) except zeep.exceptions.Fault as exc: logger.error("GetMemberMessages SOAP Fault: %s", exc) raise if resp.Ack not in ('Success', 'Warning'): errors = resp.Errors or [] codes = [f"{e.ErrorCode}: {e.ShortMessage}" for e in errors] raise RuntimeError(f"GetMemberMessages failed: {codes}") exchanges = ( getattr(resp.MemberMessage, "MemberMessageExchange", []) or [] ) result: list[EbayMessage] = [] for ex in exchanges: msg = ex.Question # ItemID は None の場合があるため安全に取得する item_id = ( getattr(getattr(ex, "Item", None), "ItemID", None) or "" ) result.append(EbayMessage( message_id=msg.MessageID, item_id=item_id, sender_id=msg.SenderID, subject=msg.Subject or "", body=msg.Body or "", question_type=msg.QuestionType or "", message_status=ex.MessageStatus or "", )) logger.info("取得メッセージ数: %d (page=%d)", len(result), page) return result # ---------------------------------------------------------- # Step 2: キーワード分類 # ---------------------------------------------------------- def classify_message(self, message: EbayMessage) -> MessageCategory: """ 件名 + 本文をキーワードで分類する。 優先順位: RETURN > SHIPPING > QUESTION > UNKNOWN。 Args: message: 分類対象の EbayMessage。 Returns: MessageCategory。 """ text = (message.subject + " " + message.body).lower() # 返品・クレーム系を最優先で判定 for kw in KEYWORDS[MessageCategory.RETURN]: if kw.lower() in text: return MessageCategory.RETURN for kw in KEYWORDS[MessageCategory.SHIPPING]: if kw.lower() in text: return MessageCategory.SHIPPING for kw in KEYWORDS[MessageCategory.QUESTION]: if kw.lower() in text: return MessageCategory.QUESTION return MessageCategory.UNKNOWN # ---------------------------------------------------------- # Step 3: テンプレート自動返信 # ---------------------------------------------------------- def reply_with_template( self, message: EbayMessage, category: MessageCategory, template_vars: Optional[dict] = None, ) -> bool: """ 分類結果に応じたテンプレートで AddMemberMessageAAQToPartner を呼び出す。 UNKNOWN / ItemID なし の場合はスキップして False を返す。 Args: message: 返信対象の EbayMessage。 category: classify_message が返した MessageCategory。 template_vars: テンプレート埋め込み変数（例: tracking_number）。 Returns: True = 送信成功、False = スキップ（手動対応要）。 Raises: RuntimeError: API 呼び出し失敗時。 """ if category == MessageCategory.UNKNOWN: logger.warning( "分類不能 -> 手動対応キューへ: message_id=%s, sender=%s", message.message_id, message.sender_id, ) return False if not message.item_id: # ItemID なし = RTQ または一般メッセージ。API が対応不可。 logger.warning( "ItemID なし -> スキップ: message_id=%s", message.message_id ) return False body = TEMPLATES[category].format(**(template_vars or {})) try: resp = self.client.service.AddMemberMessageAAQToPartner( _soapheaders=self._headers, ErrorLanguage="ja_JP", ItemID=message.item_id, # 必須: 商品 ItemID MemberMessage={ 'Body': body, 'RecipientID': {'value': [message.sender_id]}, 'ParentMessageID': message.message_id, }, ) except zeep.exceptions.Fault as exc: logger.error("AddMemberMessageAAQToPartner Fault: %s", exc) raise if resp.Ack not in ('Success', 'Warning'): errors = resp.Errors or [] codes = [f"{e.ErrorCode}: {e.ShortMessage}" for e in errors] raise RuntimeError( f"Reply failed for {message.message_id}: {codes}" ) logger.info( "返信成功: message_id=%s, category=%s", message.message_id, category.name, ) return True # ---------------------------------------------------------- # 1サイクル実行 # ---------------------------------------------------------- def run_once(self, template_vars: Optional[dict] = None) -> dict: """ 未返信メッセージ取得 -> 分類 -> 返信 を1サイクル実行する。 Returns: {'replied': int, 'skipped': int, 'errors': int} """ stats = {'replied': 0, 'skipped': 0, 'errors': 0} messages = self.get_unanswered_messages(hours_back=24) for msg in messages: category = self.classify_message(msg) try: sent = self.reply_with_template(msg, category, template_vars) stats['replied' if sent else 'skipped'] += 1 except RuntimeError as exc: logger.error("返信エラー: %s", exc) stats['errors'] += 1 logger.info("CSBot run 完了: %s", stats) return stats # ---- エントリポイント ---- if __name__ == "__main__": import os logging.basicConfig(level=logging.INFO) bot = CSBot(user_token=os.environ["EBAY_USER_TOKEN"]) result = bot.run_once( template_vars={"tracking_number": "JPN123456789"} ) print(result) このクラス設計において特に注意が必要なのは、RecipientID の渡し方です。zeep の型マッピングによっては RecipientID を StringArray 型としてリスト形式でラップする必要があります。zeep が生成した型定義（client.get_type）を事前に確認し、StringArray 型であれば value キーでリストを包む形式で渡してください。もし単純な文字列で渡してしまうと、zeep の内部で型変換が失敗しランタイムエラーとなります。 また、TEMPLATES 内の返信文を実際の eBay メッセージとして送信すると、文字数制限（概ね 2,000 文字以内）があります。テンプレートを長くしすぎると Error 21915104（Body が長すぎる）が発生します。各テンプレートは 500 文字以内に収めることを推奨します。 パフォーマンス・スケーリング視点 (深度) ポーリング設計と大量メッセージのページネーション CSBot を定期実行するには、run_once() を cron ジョブや APScheduler・Celery Beat などのタスクスケジューラから呼び出すのが基本パターンです。ポーリング間隔については、30分〜1時間ごとが現実的です。それより短い間隔で GetMemberMessages を高頻度に呼び出すと、eBay の API コールリミット（1日あたりのコール数上限）を消費するリスクがあります。特に Sandbox 環境では上限が本番より低いため、テスト中でも過負荷にならないよう注意してください。 出品数が多く大量のメッセージが届くセラーの場合、1ページ（25件）では取りきれないことがあります。GetMemberMessages のレスポンスには PaginationResult.TotalNumberOfPages が含まれるため、これを使ってすべてのページを取得する全件ループを実装します。 def get_all_unanswered_messages( self, hours_back: int = 24 ) -> list[EbayMessage]: """全ページを走査して未返信メッセージを全件取得する。""" all_messages: list[EbayMessage] = [] page = 1 while True: batch = self.get_unanswered_messages( hours_back=hours_back, page=page ) all_messages.extend(batch) if len(batch) < 25: break # 最終ページ（25件未満）に到達 page += 1 logger.info("全件取得完了: 計 %d 件", len(all_messages)) return all_messages REST版 Messaging API への将来的な移行と Webhook アーキテクチャ 本記事で解説した GetMemberMessages / AddMemberMessageAAQToPartner は Trading API（SOAP）の機能です。eBay はモダン化の一環として、REST ベースの各種 API の整備を進めており、メッセージング機能についても将来的に REST 版の充実が見込まれます。 REST 版では、JSON ペイロードによるシンプルな HTTP リクエストで同等の機能を実現でき、zeep のような SOAP クライアントライブラリへの依存がなくなります。また、OAuth 2.0 のアプリケーショントークンとユーザートークンの使い分けも REST 版ではより明確に整理されています。 現時点（2026年）では Trading API の GetMemberMessages が実稼働環境での信頼性が高く、本番システムの主力として問題ありません。ただし、将来の移行を見越して実装を CSBot クラスに閉じ込め（カプセル化）、外部から SOAP / REST の実装の違いを意識させないアーキテクチャにしておくことを強く推奨します。インターフェース（get_unanswered_messages / reply_with_template）を共通化しておけば、実装を差し替える際の変更影響範囲を最小化できます。 補足: Webhook アーキテクチャへの移行 次回（#15）で解説する SetNotificationPreferences を活用すると、eBay のイベント（新着メッセージ通知など）を Push 型でサーバーに受信することができ、ポーリング方式から Webhook 型アーキテクチャへの移行が可能になります。これにより、API コール数を大幅に削減しつつリアルタイム性を向上させることができます。「メッセージが届いた瞬間にボットが動く」という理想的な CS 自動化が実現します。 まとめ 本記事では、eBay CS 対応を自動化する核心となる2つの Trading API メソッドを実装しました。 ベースライン: GetMemberMessages で未返信の ASQ メッセージを取得し、MessageID・SenderID・ItemID の三点情報を正しく抽出する手法。 深いポイント: AddMemberMessageAAQToPartner の必須三点セット（ItemID・RecipientID・ParentMessageID）を揃えることの重要性と、日英混在キーワード分類の設計、および UNKNOWN・ItemID なし・クレーム系メッセージへの自動返信禁止ルール。 スケーリング: ページネーションによる全件取得ループの実装と、将来的な REST 版 Messaging API への移行を見越したカプセル化設計のアプローチ。 CS の自動化は「完全自動」を目指すのではなく、「確実に返信すべきものを自動化し、人間の判断が必要なケースを漏れなく手動キューに回す」という設計思想が成功の鍵です。UNKNOWN 分類と ItemID なし・クレーム系の手動フォールバックを必ず実装してください。 次のステップ CSボットをポーリング（定期実行）から Webhook（Push 通知）型に進化させるには、eBay が送信するイベント通知の受信基盤が必要です。 次回（#15）は「SetNotificationPreferences で eBay イベント通知を設定して Python サーバーで受信する」をテーマに、通知エンドポイントの登録から Python（FastAPI）での受信・署名検証まで、エンドツーエンドの実装を解説します。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#13】eBay Trading API：CompleteSaleで発送済みマークと追跡番号をAPIから一括登録する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第13回です。 前回（#12）では、GetOrders を使って注文一覧を取得し、CSV へのエクスポートや出荷管理ダッシュボードの基盤を構築しました。商品を梱包して運送業者に渡し、追跡番号（Tracking Number）を手に入れた瞬間——あなたのプログラムはそこで止まっていませんか？実は eBay は、発送が完了したという事実を API 経由で明示的に通知しなければ、「発送済み」とはみなしてくれません。 本記事で取り組む課題は、「CSV に記入した追跡番号を CompleteSale API でまとめて eBay に登録し、注文ステータスを Shipped（発送済み）に自動更新するスクリプト」の実装です。第12回のツールで取得した注文データをそのまま活用できる設計にします。 この記事で得られること: CompleteSale API の構造——ItemID・TransactionID・OrderID の正しい使い分けと、zeep（Python SOAP クライアント）を使った最小実装。 実務で必ずハマる罠——キャリアコードの厳密な指定、重複呼び出し時のエラー処理、発送済みに変更できない注文ステータスの落とし穴。 CSV ファイルから複数注文の追跡番号を一括読み込みし、API レート制限・エラーハンドリング・リトライを考慮したプロダクションレベルのバッチスクリプト。 背景・なぜこれが重要か (Motivation) 「発送したなら、それで終わりじゃないの？」 Trading API を初めて使う開発者が最初に抱く素朴な疑問です。実際に荷物を送ったのだから、eBay も自動的に「発送済み」と判断してくれる——そう思いたいのは自然なことです。しかし現実は違います。eBay の注文管理システムは、あくまでも API やセラーハブ経由で「発送した」という通知を受け取るまで、ステータスを「Awaiting Shipment（発送待ち）」のまま保持し続けます。 補足: CompleteSale が内部的に行うこと CompleteSale を呼び出すと、eBay システム内部で以下が一連に発生します。（1）注文ステータスを Awaiting Shipment → Shipped に更新。（2）バイヤーへ「出品者があなたの注文を発送しました」というメール通知を自動送信。（3）追跡番号が付帯されている場合は、eBay の注文詳細ページに追跡リンクが表示される。（4）バイヤーが自分でステータスを確認できる eBay の配送トラッカー（Delivery Status）が有効化される。 この通知を怠った場合の影響は、想像以上に深刻です。 【1】バイヤー満足度の低下: バイヤーは注文確認メールを受け取った後、配送の進捗を心配します。「発送された」という通知が届かないと、不安から「商品はいつ届くの？」というメッセージが来たり、最悪の場合 Item Not Received（INR）の紛争（Case）を申請されてしまいます。 【2】eBay のセラーパフォーマンス指標への悪影響: eBay は「発送通知の迅速性（Tracking Upload）」をセラー評価の一部として計測しています。特に Top Rated Seller（TRS）ステータスを維持しているセラーにとって、発送通知の遅延が積み重なると、TRS バッジを失うリスクがあります。 【3】資金の解放遅延: eBay Managed Payments 環境では、セラーへの支払いが「発送確認後」に解放される仕組みになっています。CompleteSale を叩かないと、資金の受け取りが遅れることがあります。 これらの理由から、「出荷したらすぐに CompleteSale を叩く」ことを、バッチ処理として自動化することが生産性向上の必須要件となるのです。 基本的な使い方（ベースライン）：CompleteSale の最小実装 まず、1件の注文に対して zeep で CompleteSale を呼び出す最小限のコードを示します。zeep は Python の SOAP クライアントライブラリで、eBay Trading API の WSDL を読み込み、Python のオブジェクトとして API を扱えるようにしてくれます。 インストールは以下のコマンドで行います。 pip install zeep requests 以下が最小実装のコードです。 # complete_sale_basic.py import os import requests from zeep import Client, Settings from zeep.transports import Transport WSDL_URL = "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl" SITEID = "0" # eBay.com (US). 日本は 101 だが Trading API SiteID は 0 のまま def complete_sale_basic( token: str, dev_id: str, app_id: str, cert_id: str, item_id: str, transaction_id: str, tracking_number: str, carrier_code: str, ) -> dict: """ 1件の注文を発送済みにマークし、追跡番号を登録する（最小実装） Args: token: eBay User Token（OAuth 認証済み） item_id: 出品 ItemID（例: "110123456789"） transaction_id: 取引 TransactionID（例: "1234567890"） tracking_number: 追跡番号（例: "JD000012345678901"） carrier_code: eBay 規定のキャリアコード（例: "JP_POST"） Returns: zeep レスポンスオブジェクト（Ack, Errors等を含む） """ session = requests.Session() session.headers.update({ "X-EBAY-API-COMPATIBILITY-LEVEL": "1155", "X-EBAY-API-DEV-NAME": dev_id, "X-EBAY-API-APP-NAME": app_id, "X-EBAY-API-CERT-NAME": cert_id, "X-EBAY-API-SITEID": SITEID, "X-EBAY-API-CALL-NAME": "CompleteSale", "Content-Type": "text/xml", }) transport = Transport(session=session) settings = Settings(strict=False, xml_huge_tree=True) client = Client(wsdl=WSDL_URL, transport=transport, settings=settings) response = client.service.CompleteSale( RequesterCredentials={"eBayAuthToken": token}, ItemID=item_id, TransactionID=transaction_id, Shipped=True, Shipment={ "ShipmentTrackingDetails": [{ "ShipmentTrackingNumber": tracking_number, "ShippingCarrierUsed": carrier_code, }] }, ) ack = getattr(response, "Ack", "Unknown") if ack in ("Success", "Warning"): print(f"[OK] ItemID={item_id}, TransactionID={transaction_id}, Ack={ack}") else: errors = getattr(response, "Errors", []) print(f"[NG] Ack={ack}, Errors={errors}") return response if __name__ == "__main__": complete_sale_basic( token = os.environ["EBAY_USER_TOKEN"], dev_id = os.environ["EBAY_DEV_ID"], app_id = os.environ["EBAY_APP_ID"], cert_id = os.environ["EBAY_CERT_ID"], item_id = "110123456789", transaction_id = "1234567890", tracking_number= "JD000012345678901", carrier_code = "JP_POST", ) 補足: Shipped と Paid の違い CompleteSale には Shipped と Paid という 2 種類のフラグが存在します。Shipped=True は「物理的な発送完了を通知する」フラグ、Paid=True は「支払いを受け取ったことを確認する」フラグです。現在の eBay Managed Payments 環境では、支払いは eBay が自動管理するため、Paid を手動で変更する必要はほとんどありません。本記事では Shipped=True のみを扱います。 補足: OrderID ではなく ItemID + TransactionID を使う理由 CompleteSale は、1つの注文ラインアイテム（Order Line Item）を単位として処理します。1つのバイヤーが同じカートで複数商品を購入した場合でも、CompleteSale の呼び出しは各ラインアイテム（ItemID + TransactionID のペア）ごとに行います。OrderID はまとめて管理する際の識別子であり、CompleteSale では直接受け付けません。※ バージョンによっては OrderLineItemID（ItemID-TransactionID 形式）もサポートされていますが、本記事では最もシンプルな ItemID + TransactionID の組み合わせを使用します。 実務で躓く場面・深いポイント (Core) ベースライン実装を本番環境で走らせると、必ずいくつかの壁にぶつかります。ここでは、実際の開発現場で頻出するエラーと落とし穴を解説します。 1. ItemID と TransactionID の対応関係の罠 第12回の GetOrders では、1件の注文（Order）の中に複数の OrderLineItem が含まれることがあります。そして、それぞれのラインアイテムには独自の ItemID と TransactionID が割り当てられています。「OrderID さえわかれば大丈夫」という考えは危険です——CompleteSale は OrderID を受け付けず、必ず ItemID と TransactionID のペアが必要です。 前回の GetOrders レスポンスでは、以下の階層でこれらの識別子を取得できます: Order └─ OrderID: "28-12345-67890" └─ TransactionArray └─ Transaction ├─ Item │ └─ ItemID: "110123456789" ← CompleteSale に使う └─ TransactionID: "9876543210" ← CompleteSale に使う GetOrders の Python 処理コードでは、以下のように取得します: # GetOrders のレスポンスから ItemID と TransactionID を抽出する for order in orders: for txn in order.TransactionArray.Transaction: item_id = txn.Item.ItemID transaction_id = txn.TransactionID # この 2 つを CSV に保存しておく 注意 第12回で CSV に保存した際に OrderID だけを記録していた場合は、もう一度 GetOrders を叩いて TransactionID と ItemID を取得し直す必要があります。この設計ミスは非常によく見られます——最初から「ItemID + TransactionID + 追跡番号」の3列を CSV に含める設計にしてください。 2. キャリアコードは自由記述ではない——eBay 規定のコードを使う CompleteSale の ShippingCarrierUsed フィールドには、自由なテキストを入力できるように見えますが、eBay が内部で認識してトラッキングリンクを生成できるキャリアコードは決まっています。「ヤマト運輸」「佐川急便」「日本郵便」という日本語や英語の正式名称をそのまま送ると、エラーにはならず Warning で通過してしまう一方で、バイヤーの注文ページに追跡リンクが表示されません。 eBay が認識する主要な日本関連キャリアコードは以下の通りです。GeteBayDetails API の ShippingCarrierDetails で取得することもできます。 VALID_CARRIER_CODES_JP = { "JP_POST": "日本郵便（ゆうパック、EMS、国際eパケット等）", "YAMATO": "ヤマト運輸（クロネコヤマト）", "SAGAWA": "佐川急便", "SEINO": "西濃運輸", "NITTSU": "日本通運（ペリカン便）", "DHL": "DHL Express", "FEDEX": "FedEx", "UPS": "UPS", "USPS": "米国郵政公社（米国発送のみ）", "TNT": "TNT Express", "OTHER": "上記以外（追跡リンク非生成）", } 注意 "OTHER" を使うと追跡番号はシステムに保存されますが、バイヤーの注文ページに追跡リンクが生成されません。バイヤー体験を最大化するため、実際のキャリアに対応する正確なコードを使用してください。また、"YAMATO"・"JP_POST" など、コードの大文字小文字は eBay API が通常正規化してくれますが、念のため常に大文字で送信するのがベストプラクティスです。 3. 重複呼び出しと冪等性——すでに Shipped の注文を再送したらどうなる？ バッチ処理では、ネットワーク障害やタイムアウトによってスクリプトが途中で停止し、再実行が必要になることがあります。この時、すでに CompleteSale で Shipped にした注文をもう一度送信しようとすると何が起きるでしょうか。 結論としては、CompleteSale は同一の ItemID + TransactionID に対して Shipped=True を再送しても、eBay 側は基本的にエラーを返さず「Warning」として処理を続けます（Ack="Warning", ErrorCode=21916867 相当）。追跡番号が既に登録されている場合は、新しい追跡番号として追記される動作になります。 これは一見安全に見えますが、重複した追跡番号がバイヤーの注文ページに複数表示されてしまうという問題があります。実装上のベストプラクティスとしては、バッチ処理の結果（成功した OrderID のリスト）を CSV や DB に記録しておき、再実行時にはすでに処理済みのレコードをスキップする「べき等性（Idempotency）の確保」を行うことです。 import json from pathlib import Path PROCESSED_FILE = Path("processed_orders.json") def load_processed_orders() -> set: if PROCESSED_FILE.exists(): return set(json.loads(PROCESSED_FILE.read_text())) return set() def save_processed_order(order_id: str) -> None: processed = load_processed_orders() processed.add(order_id) PROCESSED_FILE.write_text(json.dumps(list(processed))) # バッチ処理内での使い方 processed = load_processed_orders() for record in records: if record.order_id in processed: logger.info(f"スキップ（処理済み）: {record.order_id}") continue # ... CompleteSale を呼び出す ... save_processed_order(record.order_id) 頻出エラーコード早見表 以下は CompleteSale を実装する際に実際に遭遇する頻出エラーコードとその対処法です。 エラーコード Severity 原因 対処法 788 Error ItemID または TransactionID が存在しない GetOrders で再取得して確認する 21916867 Warning 注文ステータスが Shipped に変更できない状態 注文の現在ステータスを GetOrders で確認 21917053 Error ShippingCarrierUsed が無効なキャリアコード VALID_CARRIER_CODES から正しいコードを選択 21916588 Error OrderLineItemID のフォーマットが不正 "ItemID-TransactionID" 形式か確認 37 Error eBay Auth Token が無効または期限切れ トークンを再生成して環境変数を更新 エラーコード 37 は認証エラーです。User Token の有効期限は約 18 ヶ月ですが、Sandbox のトークンは 5 年のケースもあります。本番環境でのエラーコード 37 は、ほとんどの場合トークンの更新漏れが原因です。 堅牢な実装：CSV 一括追跡番号登録スクリプト ここでは、第12回の GetOrders で生成した注文 CSV に追跡番号を追記したファイルを入力として受け取り、CompleteSale で一括処理するプロダクションレベルのスクリプトを実装します。 まず、入力 CSV のフォーマットを確認しておきましょう。第12回のスクリプトで出力した CSV に tracking_number 列と carrier_code 列を追加したものを想定します。 order_id,item_id,transaction_id,tracking_number,carrier_code 1234567890-9876543210,110123456789,9876543210,JD000012345678901,JP_POST 2345678901-8765432109,110987654321,8765432109,604123456789,YAMATO 3456789012-7654321098,111234567890,7654321098,1234567890123456789,FEDEX 以下が完全なバッチ処理スクリプトです。型アノテーション・docstring・バリデーション・エラーハンドリング・ログ出力を完備しています。 # complete_sale_batch.py """ CSVから追跡番号を一括読み込みし、CompleteSaleで発送済みマークを登録する。 Usage: export EBAY_USER_TOKEN="v^1.1..." export EBAY_DEV_ID="xxxxxxxx-xxxx-..." export EBAY_APP_ID="YourApp-..." export EBAY_CERT_ID="xxxxxxxx-xxxx-..." python complete_sale_batch.py --csv shipments.csv """ import csv import logging import os import time import argparse from dataclasses import dataclass from typing import List, Optional import requests from zeep import Client, Settings from zeep.transports import Transport from zeep.exceptions import Fault logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger(__name__) WSDL_URL = "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl" SITEID = "0" # eBay が受け付けるキャリアコード（主要なもの） VALID_CARRIER_CODES = { "JP_POST", "YAMATO", "SAGAWA", "SEINO", "NITTSU", "DHL", "FEDEX", "UPS", "USPS", "TNT", "OTHER", } # ─── データクラス ───────────────────────────── @dataclass class ShipmentRecord: order_id: str item_id: str transaction_id: str tracking_number: str carrier_code: str def validate(self) -> None: """入力値の整合性を API 呼び出し前に検証する""" if not self.item_id.isdigit(): raise ValueError(f"item_id が数値でありません: '{self.item_id}'") if not self.transaction_id.isdigit(): raise ValueError(f"transaction_id が数値でありません: '{self.transaction_id}'") if not self.tracking_number.strip(): raise ValueError(f"tracking_number が空です (order_id={self.order_id})") if self.carrier_code not in VALID_CARRIER_CODES: raise ValueError( f"無効な carrier_code: '{self.carrier_code}'. " f"有効なコード: {sorted(VALID_CARRIER_CODES)}" ) # ─── CompleteSale クライアント ──────────────── class CompleteSaleClient: """zeep を使った CompleteSale の薄いラッパー""" def __init__(self, token: str, dev_id: str, app_id: str, cert_id: str): self.token = token session = requests.Session() session.headers.update({ "X-EBAY-API-COMPATIBILITY-LEVEL": "1155", "X-EBAY-API-DEV-NAME": dev_id, "X-EBAY-API-APP-NAME": app_id, "X-EBAY-API-CERT-NAME": cert_id, "X-EBAY-API-SITEID": SITEID, "X-EBAY-API-CALL-NAME": "CompleteSale", "Content-Type": "text/xml", }) transport = Transport(session=session, timeout=30) settings = Settings(strict=False, xml_huge_tree=True) self._client = Client(wsdl=WSDL_URL, transport=transport, settings=settings) def complete_sale(self, record: ShipmentRecord) -> Optional[str]: """ 1件の注文を CompleteSale で処理する。 Returns: 成功時は "Success" または "Warning"、失敗時は None。 Raises: Fault: SOAP レベルの障害 ValueError: 入力値バリデーションエラー """ record.validate() # ← API 呼び出し前に必ず検証 response = self._client.service.CompleteSale( RequesterCredentials={"eBayAuthToken": self.token}, ItemID=record.item_id, TransactionID=record.transaction_id, Shipped=True, Shipment={ "ShipmentTrackingDetails": [{ "ShipmentTrackingNumber": record.tracking_number, "ShippingCarrierUsed": record.carrier_code, }] }, ) ack = getattr(response, "Ack", "Failure") if ack == "Warning": warnings = getattr(response, "Errors", []) for w in warnings: code = getattr(w, "ErrorCode", "?") message = getattr(w, "LongMessage", "?") logger.warning(f" [WARNING] Code={code}: {message}") if ack not in ("Success", "Warning"): errors = getattr(response, "Errors", []) err_msgs = [ f"Code={getattr(e, 'ErrorCode', '?')}: {getattr(e, 'LongMessage', '?')}" for e in errors ] raise RuntimeError(f"CompleteSale 失敗 [{record.order_id}]: {err_msgs}") return ack # ─── CSV 読み込み ───────────────────────────── def load_shipments_from_csv(csv_path: str) -> List[ShipmentRecord]: """ CSV ファイルから ShipmentRecord のリストを生成する。 期待するヘッダ: order_id, item_id, transaction_id, tracking_number, carrier_code """ records = [] with open(csv_path, newline="", encoding="utf-8-sig") as f: reader = csv.DictReader(f) required_cols = {"order_id", "item_id", "transaction_id", "tracking_number", "carrier_code"} if not required_cols.issubset(set(reader.fieldnames or [])): missing = required_cols - set(reader.fieldnames or []) raise ValueError(f"CSV に必要な列が不足しています: {missing}") for row_num, row in enumerate(reader, start=2): # header=行1 records.append(ShipmentRecord( order_id = row["order_id"].strip(), item_id = row["item_id"].strip(), transaction_id = row["transaction_id"].strip(), tracking_number = row["tracking_number"].strip(), carrier_code = row["carrier_code"].strip().upper(), )) return records # ─── 一括処理メイン ─────────────────────────── def run_batch(csv_path: str, delay_sec: float = 0.5) -> None: """ CSV を読み込み、全注文に対して CompleteSale を実行する。 失敗した注文は最後にサマリ表示する。 """ client = CompleteSaleClient( token = os.environ["EBAY_USER_TOKEN"], dev_id = os.environ["EBAY_DEV_ID"], app_id = os.environ["EBAY_APP_ID"], cert_id = os.environ["EBAY_CERT_ID"], ) records = load_shipments_from_csv(csv_path) total = len(records) success = [] failures = [] logger.info(f"処理開始: {total} 件の注文を CompleteSale に送信します。") for i, record in enumerate(records, start=1): prefix = f"[{i:>4}/{total}] OrderID={record.order_id}" try: ack = client.complete_sale(record) logger.info(f"{prefix} → 成功 (Ack={ack})") success.append(record.order_id) except (Fault, RuntimeError, ValueError) as e: logger.error(f"{prefix} → 失敗: {e}") failures.append({"order_id": record.order_id, "reason": str(e)}) except Exception as e: logger.error(f"{prefix} → 予期しないエラー: {e}", exc_info=True) failures.append({"order_id": record.order_id, "reason": str(e)}) finally: # API レート制限対策: 連続呼び出し間に必ずウエイトを挟む if i < total: time.sleep(delay_sec) # ─── サマリ出力 ─────────────────────────── logger.info("=" * 60) logger.info(f"処理完了: 成功={len(success)} 件 / 失敗={len(failures)} 件 / 合計={total} 件") if failures: logger.error("以下の注文は失敗しました（手動確認が必要です）:") for f in failures: logger.error(f" - OrderID={f['order_id']}: {f['reason']}") # ─── エントリポイント ───────────────────────── if __name__ == "__main__": parser = argparse.ArgumentParser(description="CompleteSale 一括処理スクリプト") parser.add_argument("--csv", required=True, help="入力CSVファイルのパス") parser.add_argument("--delay", type=float, default=0.5, help="API呼び出し間隔（秒）。デフォルト: 0.5") args = parser.parse_args() run_batch(csv_path=args.csv, delay_sec=args.delay) このスクリプトの重要な設計ポイントを整理します。 【1】validate() メソッドによる事前検証: API を呼び出す前に、item_id が数値であるか、carrier_code が有効なコードであるか、tracking_number が空でないかを確認します。これにより、明らかに不正なデータでの API コールを防ぎ、不要な API コール数を節約します。 【2】@dataclass による型安全: ShipmentRecord を dataclass として定義することで、フィールドの存在と型が保証されます。CSV の列名変更による KeyError をコンストラクタ呼び出し時に早期検知できます。 【3】詳細なログ出力: logging モジュールを使い、各注文の処理結果をタイムスタンプ付きで記録します。100件以上の一括処理では、どの注文で問題が起きたかを素早く特定するために詳細ログが不可欠です。 【4】最終サマリの出力: 全処理後に成功・失敗の件数と失敗した OrderID のリストを明示します。失敗したレコードは手動で再処理や確認が必要なため、このサマリが運用上の重要な起点となります。 実行コマンド例は以下の通りです。 # 環境変数をセット export EBAY_USER_TOKEN="v^1.1.xxxxxx..." export EBAY_DEV_ID="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" export EBAY_APP_ID="YourApp-xxxx-xxxx-xxxx-xxxxxxxxxxxx" export EBAY_CERT_ID="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" # バッチ処理を実行（API 呼び出し間隔 0.5 秒） python complete_sale_batch.py --csv shipments.csv --delay 0.5 パフォーマンス・スケーリング視点 (深度) 1日に数十件の注文を処理するフェーズでは、上記のシンプルなシーケンシャル処理で十分です。しかし、売上が伸びて 1日あたり 200〜500 件を超えるようになると、処理速度と API レート制限の両方の観点で設計を見直す必要が出てきます。 大量注文処理の並列化と API レート制限の管理 eBay Trading API には、アカウントあたり 1 日に呼び出せるコール数の上限があります。デフォルトでは CompleteSale を含む多くの API で 1 日あたり 5,000 コールが上限です（アカウントの認定状況によって異なり、最大 150,000 コールまで申請で引き上げ可能です）。 シーケンシャル処理（delay=0.5 秒）では、1 時間あたり最大 7,200 件を処理できます。多くの場合これで十分ですが、さらに大量の注文を短時間で処理したい場合は concurrent.futures.ThreadPoolExecutor を使った並列処理が有効です。ただし、並列処理では API レート制限を超過しないよう、スレッドセーフなレートリミッターが必要です。 # complete_sale_concurrent.py（スケーリング版） import concurrent.futures import threading import time from typing import List, Tuple # 同時実行スレッド数。Trading API の 1日あたり上限 5,000 コールを # 考慮し、ピーク時でも安全なレート（例: 最大 3 並列）に抑える。 MAX_WORKERS = 3 DELAY_PER_REQ = 0.3 # 秒 _rate_lock = threading.Lock() _last_call_ts = 0.0 def _throttled_complete_sale( client: "CompleteSaleClient", record: "ShipmentRecord", ) -> Tuple[str, str]: """スロットル付きの CompleteSale 呼び出し""" global _last_call_ts with _rate_lock: now = time.monotonic() wait = DELAY_PER_REQ - (now - _last_call_ts) if wait > 0: time.sleep(wait) _last_call_ts = time.monotonic() try: ack = client.complete_sale(record) return (record.order_id, "ok") except Exception as e: return (record.order_id, f"error: {e}") def run_batch_concurrent(records: List["ShipmentRecord"], client: "CompleteSaleClient"): with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: futures = { executor.submit(_throttled_complete_sale, client, r): r for r in records } for future in concurrent.futures.as_completed(futures): order_id, result = future.result() if result == "ok": logger.info(f"[concurrent] {order_id} → 成功") else: logger.error(f"[concurrent] {order_id} → {result}") 注意 MAX_WORKERS の値を安易に大きくしないでください。eBay のサーバーは短時間の集中コールを検知するとレートリミット（Error: request limit exceeded）を返します。実務では MAX_WORKERS=3〜5、DELAY_PER_REQ=0.3 秒程度が安全な上限の目安です。 規模がさらに大きくなり、1日 10,000 件を超える処理が必要な場合は、アーキテクチャ自体を見直す必要があります。具体的には、以下の構成を検討してください。 【分散スケジューリング】: AWS SQS や Google Cloud Tasks などのメッセージキューを導入し、CompleteSale 呼び出しをキューに積んで複数のワーカープロセスで消費する構成。1 プロセスがクラッシュしても他のプロセスが処理を継続でき、DLQ（Dead Letter Queue）に失敗レコードが貯まるため再処理が容易です。 【API コール数の申請増加】: eBay Developer Support に連絡し、利用実績を提示することで API コール上限を引き上げる申請が可能です。大規模セラーであれば、1 日あたり 50,000〜150,000 コールへの引き上げが承認されることがあります。 【Bulk Fulfillment Feeds API への移行検討】: 非常に大量の発送処理（1日 5,000 件超）が継続的に必要な場合、Trading API の CompleteSale ではなく、eBay の Fulfillment API や Order Management API（REST）への移行も検討に値します。REST API では一括操作のエンドポイントが提供されており、API コール効率が大幅に向上します。ただし、現時点（2026年）では Trading API の方が機能的に成熟しているため、移行前に機能差分を必ず確認してください。 まとめ 本記事では、出荷後の最重要タスクである「CompleteSale による発送済みマークと追跡番号の一括登録」を実装しました。 ベースライン: zeep を使った CompleteSale の最小実装で、ItemID + TransactionID のペアと正規のキャリアコードを指定し、Shipped=True で発送を通知する基本パターンを習得しました。 深いポイント: ItemID・TransactionID の正しい取得方法、eBay 規定のキャリアコード必須要件、重複呼び出し時の冪等性確保（処理済み OrderID の記録）という3つの実務の壁と、その具体的な解決策を学びました。 スケーリング: 1日 200 件超の処理では concurrent.futures によるスレッドセーフな並列化と、メッセージキューを使った分散処理アーキテクチャへの移行パスを理解しました。 CompleteSale を自動化することで、人手によるセラーハブの手動操作が不要になり、バイヤーへの発送通知が即時化されます。これは INR 紛争の予防だけでなく、セラーパフォーマンス指標（Defect Rate の改善、TRS ステータスの維持）にも直結する、EC オートメーションの中でも費用対効果の高い実装の一つです。 次のステップ 発送処理が自動化できたら、次に重要なのはバイヤーとのコミュニケーション自動化です。「商品は届きましたか？」「ご不明な点はありますか？」といったメッセージを手動で送っていませんか？ 次回（#14）は、GetMemberMessages と AddMemberMessageAAQToPartner API を使って、バイヤーからのメッセージを自動取得し、テンプレートに基づいた返信を自動送信する仕組みを実装します。人手を介さない完全自動レスポンスシステムの構築にチャレンジしましょう！ 次の記事はこちら

eBayにおける収集用コインのコンディション要件の新規導入 (New Coin Condition Requirements Coming to eBay) eBay では、バイヤー（購入者）の信頼感向上、出品の一貫性確保、およびマーケットプレイス全体のパフォーマンス改善に向けた重要なステップとして、コイン（Coins）カテゴリにおける標準化された「コンプライアンス/コンディション要件（Condition Requirements）」を導入いたします。 本アップデートは、貴社のシステム統合（インテグレーション）および貴社プラットフォーム経由でコインを出品するすべてのセラー（販売者）に直接影響を与えます。 対象カテゴリ (Impacted Categories) 2026年5月6日 より、以下のリーフカテゴリにおいてこれらの変更が適用されます。 253 - Coins: US 256 - Coins: World 3377 - Coins: Canada 4733 - Coins: Ancient 18466 - Coins: Medieval 変更内容 (What’s changing?) 2026年5月初旬 より、API ユーザーは上記のカテゴリで出品を作成または修正（Revise）する際、コンディション要件情報の提供が求められます。フェーズ 2（下記スケジュール参照）以降、この情報を含まない出品や修正リクエストはブロックされるか、非表示となるか、あるいは公開に失敗します。 これらの変更は段階的に適用されます。詳細は以下の「主要スケジュール」セクションを参照してください。 対象カテゴリにおいて、セラーはコインの状態（コンディション）を以下のいずれかに設定する必要があります。 鑑定済み (Graded) Grading company (鑑定会社): 必須 Grade (文字 + 数値グレード): 必須 Certification number (証明番号): 任意 未鑑定 (Ungraded / Raw) 標準化されたコンディション区分（例: Uncirculated、About Circulated、Extra Fine、Fine、Below Fine など）から選択する必要があります。 API メタデータの更新: 更新されたメタデータ詳細は、Sell Metadata API の getItemConditionPolicies コールのレスポンスにも反映されます。 主要スケジュール (Timelines) 以下のマイルストーンに先立ち、貴社システムの統合ロジックが更新されていることを確認してください。 フェーズ 1(2026年5月初旬) フェーズ 2(2026年6月初旬) フェーズ 3(2026年7月初旬) 上記対象カテゴリのすべての新規および既存の出品に対し、API 警告（Warning）のみ を返却 上記対象カテゴリの 新規出品に対する API 義務化（Mandate） を開始 上記対象カテゴリの 既存出品に対する API 義務化（Mandate） を開始 注意: フェーズ 2 および フェーズ 3 の期間中、必須のコンディションデータが欠落している出品は、ブロック、非表示、または公開失敗となる可能性があります。 コンディションデータの移行戦略 (Condition data migration strategy) 本変更をサポートするため、eBay はフェーズ 1 の期間中に既存のアクティブ出品（Live listings）を自動的に移行（Migrate）します。 現在の商品アスペクト（Item Specifics）と新しいコンディション要件フィールドとの間に明確な1対1のマッピングが存在する出品は、自動的に更新されます。割り当てられた新しいアスペクトやコンディション要件を確認し、不一致がある場合は変更する責任は開発者/セラー側にあります。 出品に移行に必要な十分なデータが含まれていない場合、新しいコンディションフィールドは空のままとなります。 重要な注意事項: フェーズ 1 の時点では、必須のコンディションフィールドが欠落している既存出品が終了（End）されることはありません。 ご対応いただきたい手順 (What we need from you) セラーがスムーズに移行できるよう、以下の対応を完了させてください。 システム統合の更新 新しいコンディション評価（Condition grading）フィールドをサポートし、コインカテゴリにおける必須入力値のバリデーションを実装してください。 出品フローのテスト 鑑定済み（Graded）および未鑑定（Ungraded）の両方の入力パスがサポートされていること、および欠落・無効なデータに対するエラーハンドリングが正常に機能することを確認してください。 移行計画の策定 既存の出品がどのように更新されるか、またはセラーに対してどのように入力促しを行うかを検討・計画してください。 ご不明な点がございましたら、eBay デベロッパーポータル (developer.ebay.com) のデベロッパーサポートまでお問い合わせください。

前回の記事はこちら 【連載#12】eBay Trading API：注文管理の核心 —— GetOrdersで注文ロストと入金未済を防ぐ自動同期 はじめに 本記事は、全 42 回にわたる「eBay API 実践ガイド」の第 12 回です。 前回（#11）までに、出品から在庫同期、品質監査ツール（QA）の構築といった「商品・在庫軸（Inventory）」のパイプラインが完成しました。ストアに売上が発生し始めると、システムの主役は次のフェーズである 【注文管理（Order Management）】 へと移行します。 注文データの同期遅延や取得漏れは、出荷遅延ペナルティや未入金発送といった致命的な実害に直結します。今回は、分散 DB 特有の遅延対策、ページネーションによる「サイレントな注文ロスト」の防御、通貨情報の厳格な保持などを網羅した、本番環境仕様の注文自動取得エンジンを構築します。 この記事で得られること: CreateTime と ModTime のトレードオフに基づいたフィルタリング設計。 HasMoreOrders を用いたページネーション制御による、注文ロストゼロのループ処理。 同梱発送（Combined Shipping）の多重ネスト構造とマルチ通貨（currencyID）を安全にハンドリングするパース技術。 背景・なぜこれが重要か (Motivation) EC のバックオフィス自動化において、「注文同期システム」の設計ミスはストアのアカウント健全性を一瞬で破壊します。 ページネーション漏れによるサイレントロスト: 注文が急増した時間帯に、API が 1 ページで返せる上限（デフォルト 100 件）を超えた注文データをプログラムが次ページへ追わずに切り捨ててしまうバグ。エラーを吐かないため検知が極めて困難です。 レプリケーション遅延による出荷遅延: eBay 側のデータベース同期タイムラグにより、直前の数秒〜数分間の注文が API レスポンスから漏れ、そのまま永久に同期されないリスク。 未入金商品のフライング発送: バイヤーが注文を確定（Checkout）したものの、決済審査中（Pending 等）であるステータスをシステムが「支払い済み」と誤判定して出荷してしまうリスク。 これらを完全に防ぐためには、API の通信仕様とステータス挙動を深く理解し、厳格なデータハンドシェイクを実装する必要があります。 CreateTime vs ModTime フィルターの選択基準 GetOrders で時間窓フィルタリングを行う際、利用できるアプローチは 2 つあります。目的のバッチ要件に応じて正しく使い分けてください。 CreateTimeFrom / CreateTimeTo (注文作成日時): 特性: 注文が「最初に発生した瞬間」を基準に検索します。 用途: 新規注文の確実な捕捉に向いています。ただし、バイヤーが後から決済を完了した、キャンセルしたなどの「状態変化」を追跡できません。 ModTimeFrom / ModTimeTo (注文更新日時): 特性: 決済完了、キャンセル、発送など、注文データに「何らかの変更が加わった瞬間」を基準に検索します。 用途: 状態変化を検知して出荷指示を出す WMS 連携や自動ステータス同期バッチに最適です。本記事の注文同期エンジンでは、実務で最も多用されるこの ModTime ベースの設計を採用します。 基本的な使い方（ベースライン）：XML 構造の全容 GetOrders のリクエストでは、自身がセラー側（Seller）であることを明示し、適切なルートネームスペースを指定して送信します。 <?xml version="1.0" encoding="utf-8"?> <GetOrdersRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <ModTimeFrom>2026-07-13T00:00:00.000Z</ModTimeFrom> <ModTimeTo>2026-07-13T23:59:59.000Z</ModTimeTo> <OrderRole>Seller</OrderRole> <OrderStatus>Completed</OrderStatus> <Pagination> <EntriesPerPage>100</EntriesPerPage> <PageNumber>1</PageNumber> </Pagination> </GetOrdersRequest> 実務で躓く場面・深いポイント (Core Pitfalls) 1. 「時間窓重複戦略（Overlap Strategy）」の真の自動化 eBay の分散データベースでは、バイヤーの決済完了から API に反映されるまで数秒〜数分のタイムラグ（書き込み遅延）が発生することがあります。 そのため、15 分おきに「前回の終了時刻〜現在の時刻」で完全に区切ってバッチを回すと、境界線上の注文がロストします。 これを防ぐため、「バッチ実行時のインターバル＋5〜10 分前のオーバーラップ時間」を動的に計算し、前方の時間窓を意図的に重複させて取得します。 重複して取得した注文は、後段のシステム（ローカル DB）側で OrderID を主キー（Primary Key）とした UPSERT 処理を行うことで、二重発注を完全に防ぎつつロストをゼロにします。 2. 多値通貨属性（currencyID）のパース漏れとフォールバックの危険性 eBay はグローバルプラットフォームであるため、アメリカ（USD）、イギリス（GBP）、オーストラリア（AUD）など、複数の通貨で注文が発生します。 注文総額を表す <Total> タグは、以下のように属性値として通貨を持っています。 <Total currencyID="USD">29.99</Total> プログラム側で .text だけを抽出して数値化すると、「通貨単位が消失する」ため、財務データが壊れる原因になります。また、取得できなかった際のフォールバックを安易に 'USD' などと固定値で埋めると、他国サイトでの取引データと混ざり重大な計算ミスを引き起こします。パース時には要素の属性（attrib）から currencyID を厳格に抽出し、存在しない場合は None としてハンドリングを分ける必要があります。 3. 同梱発送（Combined Shipping）の多重ネスト バイヤーが同じセラーから複数の異なる商品（ItemID）をカートに入れ、まとめて決済した場合、eBay 側ではそれらが 1 つの <Order> に統合されます。 このとき、XML の階層構造は Order -> TransactionArray -> 複数の Transaction となります。 パース処理の段階で「1 注文＝ 1 商品」と思い込んだ設計をしていると、同梱された 2 商品目以降がシステム上で完全に見落とされる大事故になります。必ず二重のループ構造で安全に走査しなければなりません。 4. OrderStatus=Completed 指定の業務上の根拠 本スクリプトでは <OrderStatus>Completed</OrderStatus> を選択しています。これは、バイヤーが購入手続き（Checkout）を完全に完了させ、注文構成が確定した状態のみを狙い撃ちするためです。 Active（決済手続きの途中）段階の注文は、後からバイヤーによって同梱要請が出されるなどして注文構造そのものが変化するリスクがあるため、出荷指示バッチにおいては Completed に絞り込むのが実務上最も効率的かつ安全なアプローチとなります。 堅牢な実装：自動注文同期パースエンジン（完全版） 「無限リトライを防ぐ Rate Limit 対策」「HasMoreOrders に対応した完全ページネーション」「データ変換例外の完全ディフェンス」「カスタム例外クラスによる堅牢化」を実装した、プロダクション環境仕様の注文同期スクリプトです。Python 3.7+ 互換の型ヒントスタイルで統一しています。 注意 (コード内XMLについて): 下記Pythonコード中のXMLリクエスト文字列は、CMSのHTMLパースによるタグ消失を回避するため、文字列連結構文（'...' '...'）で記述しています。実際の < > 文字は上記のXML構造例を参照してください。コードをローカルで実行する際は、タグが正しく含まれていることをご確認ください。 # ebay_order_syncer.py import requests import xml.etree.ElementTree as ET import time import random from datetime import datetime, timedelta, timezone from typing import List, Dict, Any, Tuple, Optional from tenacity import retry, stop_after_attempt, wait_exponential from config import eBayConfig class EbayApiError(Exception): """プロダクション仕様: eBay APIからのエラーレスポンスを表現するカスタム例外クラス""" pass def _get_text(node: Optional[ET.Element], tag: str, ns: dict) -> str: """静的解析ツール(mypy等)でエラーが出ないよう、Optional型アノテーションで安全に宣言""" if node is None: return "" el = node.find(tag, ns) return el.text if el is not None and el.text is not None else "" @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), reraise=True ) def _execute_api_post(url: str, headers: dict, payload: str) -> str: """ HTTP 429 や瞬断に対するリトライ制限付きの通信実行器。 スロットリングによる無限ループを防ぐため、最大3回で例外を投げる設計。 ※本番環境でHTTP 429が発生した際、eBayが返却する Retry-After レスポンスヘッダーを 動的に読み取って待機時間を決定するロジックを挟むと、よりスマートなスロットリング制御が可能です。 """ res = requests.post(url, headers=headers, data=payload.encode('utf-8'), timeout=30) res.raise_for_status() return res.text def parse_ebay_orders_xml(xml_text: str) -> Tuple[List[Dict[str, Any]], bool]: """XML レスポンスをパースし、注文データリストと次ページ有無を返す""" ns = {'ns': 'urn:ebay:apis:eBLBaseComponents'} root = ET.fromstring(xml_text) ack = _get_text(root, 'ns:Ack', ns) if ack not in ['Success', 'Warning']: errors = root.findall('ns:Errors', ns) msg = "; ".join([_get_text(e, 'ns:LongMessage', ns) for e in errors]) raise EbayApiError(f"GetOrders API Failure: {msg}") orders_list = [] order_array_node = root.find('ns:OrderArray', ns) if order_array_node is not None: for order_node in order_array_node.findall('ns:Order', ns): order_id = _get_text(order_node, 'ns:OrderID', ns) order_status = _get_text(order_node, 'ns:OrderStatus', ns) checkout_node = order_node.find('ns:CheckoutStatus', ns) paid_status = _get_text(checkout_node, 'ns:PaidStatus', ns) buyer_id = _get_text(order_node, 'ns:BuyerUserID', ns) # 深いポイント①: 金額の数値変換エラー対策と通貨ID(currencyID)の厳格な抽出 total_node = order_node.find('ns:Total', ns) if total_node is not None and total_node.text: try: total_amount = float(total_node.text) except (ValueError, TypeError): total_amount = 0.0 currency_id = total_node.attrib.get('currencyID', None) # サイレントなUSD埋めを回避 else: total_amount = 0.0 currency_id = None # 深いポイント②: 二重ループ構造による同梱決済（Combined Shipping）の完全走査 transactions_extracted = [] tx_array_node = order_node.find('ns:TransactionArray', ns) if tx_array_node is not None: for tx_node in tx_array_node.findall('ns:Transaction', ns): item_node = tx_node.find('ns:Item', ns) sku = _get_text(item_node, 'ns:SKU', ns) item_id = _get_text(item_node, 'ns:ItemID', ns) # 安全対策: 数量パース時の数値例外に対する一貫した堅牢な保護 qty_text = _get_text(tx_node, 'ns:QuantityPurchased', ns) try: qty = int(qty_text) except (ValueError, TypeError): qty = 0 tx_id = _get_text(tx_node, 'ns:TransactionID', namespace=ns) transactions_extracted.append({ "transaction_id": tx_id, "item_id": item_id, "sku": sku, "quantity": qty }) orders_list.append({ "order_id": order_id, "order_status": order_status, "paid_status": paid_status, "buyer_id": buyer_id, "total_amount": total_amount, "currency_id": currency_id, "items": transactions_extracted }) # 核心: ページネーション継続判定フラグの抽出 has_more = _get_text(root, 'ns:HasMoreOrders', ns).lower() == 'true' return orders_list, has_more def sync_ebay_orders( config: eBayConfig, token: str, start_dt: datetime, end_dt: datetime ) -> List[Dict[str, Any]]: """タイムウィンドウを指定し、ページネーションを完全に回して全注文を網羅する関数""" headers = { "X-EBAY-API-CALL-NAME": "GetOrders", "X-EBAY-API-SITEID": "0", "X-EBAY-API-COMPATIBILITY-LEVEL": "1323", "X-EBAY-API-IAF-TOKEN": token, "Content-Type": "text/xml" } from_str = start_dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.000Z') to_str = end_dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.000Z') all_orders: List[Dict[str, Any]] = [] page_number = 1 while True: # 核心②: XMLはPython文字列連結で構築 — CMSのHTMLパースによるタグ消失を構造的に回避 xml_payload = ( '<?xml version="1.0" encoding="utf-8"?>' '<GetOrdersRequest xmlns="urn:ebay:apis:eBLBaseComponents">' '<ErrorLanguage>en_US</ErrorLanguage>' '<WarningLevel>High</WarningLevel>' f'<ModTimeFrom>{from_str}</ModTimeFrom>' f'<ModTimeTo>{to_str}</ModTimeTo>' '<OrderRole>Seller</OrderRole>' '<OrderStatus>Completed</OrderStatus>' '<Pagination>' '<EntriesPerPage>100</EntriesPerPage>' f'<PageNumber>{page_number}</PageNumber>' '</Pagination>' '</GetOrdersRequest>' ) print(f"Fetching page {page_number}...") res_text = _execute_api_post(config.trading_api_url, headers, xml_payload) orders, has_more = parse_ebay_orders_xml(res_text) all_orders.extend(orders) if not has_more: break page_number += 1 # デバイス遅延やスパイク（429エラー）を防ぐためジッターを付与したウェイトを入れる time.sleep(0.5 + random.uniform(0, 0.5)) return all_orders # --- バッチ定期実行シミュレーション --- if __name__ == "__main__": from ebay_token_manager import eBayTokenManager config = eBayConfig() manager = eBayTokenManager( config.client_id, config.client_secret, config.refresh_token, config.env ) # 窓重複戦略: 15分間隔のcron実行を想定し、10分の余白を持たせて過去25分間を指定 # 本番環境では「前回バッチの正常終了時刻」をDBに永続化し、そこからN分引いて # 動的にウィンドウを算出するロジックを必ず実装してください。 end_window = datetime.now(timezone.utc) start_window = end_window - timedelta(minutes=25) try: active_token = manager.get_token() pulled_orders = sync_ebay_orders(config, active_token, start_window, end_window) print( f"\\n[Sync Window] " f"({start_window.strftime('%H:%M:%S')} - {end_window.strftime('%H:%M:%S')}) " f"の取得注文数: {len(pulled_orders)}件" ) for order in pulled_orders: is_shippable = ( order["order_status"] == "Completed" and order["paid_status"] == "Paid" ) ship_badge = "WMS出荷指示可能" if is_shippable else "決済未完了/保留" # 通貨が未取得(None)の場合のハンドリングで表示の整合性を保護 currency_display = order['currency_id'] or 'N/A' print( f"[{ship_badge}] 注文ID: {order['order_id']} " f"| 総額: {order['total_amount']} {currency_display}" ) for it in order["items"]: print(f" └─ SKU: {it['sku']} x 数量: {it['quantity']}") except Exception as ex: print(f"注文同期処理で致命的例外が発生しました: {ex}") ⚡ パフォーマンス・スケーリング視点 (深度) 新世代 REST API（Fulfillment API）への移行パス Trading API の GetOrders は実績の多い安定した機能ですが、XML のパースにかかるシステム CPU 負荷や、データ量に比例してページ数（リクエスト回数）が増大する制限があります。 将来的に月間数万件以上のトランザクションをさばくエンタープライズシステムへとスケールさせる場合は、次世代の REST API（Fulfillment API） へのリプレイスを設計する必要があります。 対応する REST API メソッド: GET /order エンドポイント例 (GET): https://api.ebay.com/sell/fulfillment/v1/order?filter=lastmodifieddate:[2026-07-13T00:00:00Z..2026-07-13T23:59:59Z] REST の構造的メリット: JSON 形式の標準採用: 配列構造（lineItems）をネイティブに扱えるため、メモリ効率が向上します。 URLエンコードの必須性: REST で上記の filter パラメータを送信する際は、予約文字であるブラケット（[]）やコロン（:）を %5B や %3A へ適切にパーセントエンコード（URLエンコード）して送信する必要があります。初学者がそのまま生文字でリクエストを投げると HTTP 400 エラーになるため注意してください。 Webhook（Notification API）への発展: ポーリング（Pull型）から、注文発生時のみ駆動するイベント駆動型アーキテクチャ（Push型）への移行が極めてスムーズになります。 まとめ 本記事では、EC システムの基盤となる注文データの安全なインポートについて解説しました。 ベースライン: ModTime フィルターを用いた増量取得と OrderRole=Seller 指定の必須性。 深いポイント: 分散 DB の同期遅延を相殺する「タイムウィンドウ重複戦略」の実装、および同梱決済に対応する二重ループパースロジック。 致命的エラーの防御: HasMoreOrders を用いたページネーションによるサイレント注文ロストの完全撲滅。 スケーリング: 出荷指示のための二段階ステータス監査（Completed & Paid）の定義と、次世代 RESTful Fulfillment API へのロードマップ。 これで注文データがローカルシステムへ安全に引き込まれました。 次のステップ 注文が確定し、入金が確認された後にシステムが行うべき次のアクションは「出荷手配とバイヤーへの追跡番号の通知」です。 次回（#13）は、「CompleteSale API で追跡番号（Tracking Number）を自動回伝し、eBay 上の発送通知を自動化する」 方法について解説します。バイヤーの顧客満足度を高め、未着トラブル（INR）から身を守るための物流自動化ロジックをお楽しみに！ 次の記事はこちら