前回の記事はこちら 【連載#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）から身を守るための物流自動化ロジックをお楽しみに！ 次の記事はこちら

eBay ファッションカテゴリにおけるサイズ表記標準化 (Size Standardization for eBay Fashion Listings) eBay では、アパレル＆フットウェア（Apparel & Footwear）出品における「サイズ表記の標準化（Size Standardization）」を導入いたします。これは、出品クオリティの向上、バイヤー（購入者）の信頼感醸成、そしてマーケットプレイス全体のパフォーマンス改善に向けた取り組みの一環です。 本アップデートは、API、File Exchange、およびサードパーティ製ツール経由で作成されたすべての出品に適用されます。 変更内容 (What’s changing?) 2026年6月より: eBay は、確信度の高いサイズ値（例: 「Small」から「S」への自動変換）を含む既存の出品に対して自動正常化（Auto-normalization）を開始します。また、確信度の低い値や無効な入力値（例: 「説明を参照（See description）」や「N/A」など）に対しては警告フラグを付与します。 2026年7月より: 非標準のサイズ値、欠落したサイズ値、および/またはコンディション（状態）値を含むすべての新規および既存の出品は、サイト上での表示がブロックされるか、保留（Hold）状態となります。 システムの統合（Integration）への影響: 長期的な技術的対応は必要となりますが、初期ローンチ時点において即座のシステム修正が必須というわけではありません。 eBay は確信度の高い値の正常化を自動的に処理します。 これにより、ご自身のシステムと eBay 上で表示される値が異なる場合があります（例: システム上は「Small」、eBay上は「S」）。 確信度の高い非標準値が再送信された場合でも、eBay が引き続き自動的に正常化処理を行います。 現時点で必須のアクションはありませんが、整合性を確保するため、時間をかけて貴社システムのサイズスキーマを eBay の標準値へ合わせ込むことを推奨いたします。 導入の重要性 (Why this matters) サイズデータを標準化することにより、以下のメリットが得られます： 絞り込み検索（Filter）および検索結果における出品の可視性（露出）の向上 より一貫した出品パフォーマンスの実現 バイヤーの購買意欲の向上および返品率の削減 主要スケジュール (Key dates to know) 時期 フェーズ 主な内容 6月開始 正常化の開始(Normalization begins) ・確信度の高い値に対する自動正常化の開始 ・確信度の低い値に対して警告メッセージを表示 ・新規出品におけるカスタムサイズ値の入力機能を削除 7月開始 完全強制の開始(Full enforcement begins) ・確信度の高い値への自動正常化は継続 ・確信度の低い値や無効な値を含む出品に対する完全適用（ブロックまたは保留措置） 今後の対応手順 (Next steps) 現在プラットフォーム経由でサイズ値がどのように渡されているかを確認してください。 貴社のサイズフィールドを eBay の標準化された値に整合させる対応を、今後の開発ロードマップに組み込んでください。 完全適用の実施に向けて、今後も主要なアップデートやサポートリソースを継続的に共有してまいります。ご不明な点がございましたら、担当のパートナー窓口またはデベロッパーサポートまでお問い合わせください。

前回の記事はこちら 【連載#11】eBay Trading API：GetItem の並列一括処理による自製「GetItems」とデータ品質チェックの自動化 はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第11回です。 前回（#10）までに、商品の新規出品（Add）、部分更新（Revise）、出品一覧の取得（Pull）、そして緊急下架（End）という商品管理のライフサイクル（CRUD）が一通り完成しました。 システムが自動で出品を回せるようになった次の高次元フェーズとして、実務で必ず求められるのが「出品データの品質監査（QA：Quality Assurance）」です。 今回は、eBay から商品詳細を安全かつ高速にバルク取得し、タイトルの最適化状態、画像の枚数、迅速にSEOに最も影響を与える ItemSpecifics の充足率を自動判定する「データ品質チェックツール」の構築方法を解説します。 この記事で得られること: 单件取得 API の仕様と、Trading API に存在しない「一括取得」をクライアント側でセキュアに擬似実装するアーキテクチャ。 複雑にネストされた ItemSpecifics（商品属性）の XML を安全にパースする技術。 出品品質（タイトル文字数、画像枚数、属性充足率）を自動監査する QA スクリプトの実装。 開発環境 (Environment) OS: Linux / macOS / Windows Language: Python 3.7+ Libraries: requests 2.31+, tenacity 8.2+ 背景・なぜこれが重要か (Motivation) eBay の検索アルゴリズムである Best Match において、検索順位（SEO）の上位を狙うための 3 大要素は「タイトル」「画像」「Item Specifics（商品スペック）」です。 タイトル: 最大 80 文字の枠をフルに活用し、コンバージョンに繋がるキーワードが網羅されているか。 画像: バイヤーの購買意欲を高めるため、十分な枚数が登録されているか。 Item Specifics: Brand や MPN（型番）、Color などの識別子が正確に埋められているか。これが抜けていると、バイヤーが左側のサイドバーで絞り込み検索（Filter）をした際に、商品が検索結果から完全に消滅します。 大規模にシステム出品を行っていると、マスターデータの不備やプログラムのバグによって「属性が空っぽのまま出品されてしまった低品質な Listing」が大量に量産されるリスクがあります。これを定期的に自動巡回して監査し、品質スコアが低い出品をアラート抽出する仕組みは、売上最大化とストアの健全性維持のための生命線となります。 基本的な使い方（ベースライン）と技術的背景 商品詳細を取得する基本 Call は GetItem です。 <?xml version="1.0" encoding="utf-8"?> <GetItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <!-- 【ポイント】: 詳細を取得したい商品のItemID --> <ItemID>112233445566</ItemID> <!-- 【ポイント】: ItemSpecifics などの詳細スペックを含めるための指定 --> <DetailLevel>ReturnAll</DetailLevel> </GetItemRequest> 【実務における知恵】: Trading API に「GetItems」は存在しない？ 実は、eBay Trading API（XML 形式）には、複数の ItemID を一括で渡して同時に詳細を取得できる GetItems という名前の Call は存在しません。 「じゃあ、20 件の商品をチェックしたい時は API を 20 回ループで叩くしかないのか？」 その通りに愚直に実装すると、ネットワークの往復レイテンシ（RTT）が積み重なり、深刻な N+1 問題（パフォーマンス低下）を引き起こします。本記事では、前回のマルチスレッド下架の知見を応用し、「最大 20 件の ItemID を並列処理で高速バルク取得するラッパー関数（自作 get_items_bulk）」 を構築することで、この通信ボトルネックを突破します。 ※なお、参照系の軽量一括取得としては Shopping API の GetMultipleItems も存在しますが、より深いセラー内部情報や正確なバリエーション情報を監査するため、本記事では Trading API の GetItem をコンカレントに回す設計を採用します。 実務で躓く場面・深いポイント (Core) 1. XML 注入（インジェクション）脆弱性の防御 外部ソース（ローカル DB や外部スクリプトの出力）から渡された ItemID を用いて動的に XML 文字列を生成する際、文字列結合や f-string を不用意に使うと、XML インジェクションの脆弱性を作り込むリスクがあります。 ItemID が不正な構造（例: </ItemID><ItemID>悪意あるタグ...）に書き換えられていた場合、リクエスト全体が破壊されるか、予期せぬ挙動を引き起こします。ItemID は必ず数値型の文字列（通常 8〜13 桁の数字）であるため、API コール直前のゲートウェイ層で厳格なフォーマットチェック（バリデーション）を行うのが防御的プログラミングの鉄則です。 2. OutputSelector による極限の帯域最適化 商品説明（<Description>）文は非常にデータサイズが大きく、HTML タグを含めると数万〜数十万バイトに及びます。旧来の設計では <DetailLevel>ItemReturnAttributes</DetailLevel> を使って軽量化していましたが、実務における究極の最適化手段は <OutputSelector> の活用です。 OutputSelector を使用すると、指定したフィールド以外のデータを eBay 側で完全に削ぎ落としてレスポンスを生成させることができます。 <GetItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ItemID>112233445566</ItemID> <DetailLevel>ReturnAll</DetailLevel> <!-- 【注意】: 必要なフィールドだけをピンポイント指定。Description等は一切返却されない --> <!-- 【メモ】: AckやErrorsなどの共通メタデータは指定しなくても自動返却されます --> <OutputSelector>Title</OutputSelector> <OutputSelector>PictureDetails</OutputSelector> <OutputSelector>ItemSpecifics</OutputSelector> <OutputSelector>SKU</OutputSelector> </GetItemRequest> これにより、通信ペイロードの大部分を削減することができ、マルチスレッド並列処理時のメモリ逼迫とネットワーク I/O 負荷を劇的に抑えることが可能になります。 3. ItemSpecifics の XML パースの罠（多値属性のハンドリング） ItemSpecifics は NameValueList が不特定多数ネストされる可変構造です。 <ItemSpecifics> <NameValueList> <Name>Features</Name> <Value>Wi-Fi Capable</Value> <Value>Waterproof</Value> <!-- 【注意】: 1つのNameに対してValueが複数あるケースも！ --> </NameValueList> </ItemSpecifics> Python の xml.etree.ElementTree でパースする際、find().text を雑に使うと、「Value が複数あるケースで最初の 1 つしか取れない」というバグが多発します。 全ての属性を一旦「辞書型（{Name: [Value1, Value2]}）」に展開してからバリデーションにかけるのが、実稼働を前提とした堅牢な設計です。 堅牢な実装：商品詳細の一括取得と QA（品質監査）スクリプト 「XML 注入防御」「OutputSelector による最適化」「複数エラーメッセージの完全抽出」「スレッドセーフなトークン評価」を網羅した本番仕様のコードです。 # item_qa_tool.py import requests import xml.etree.ElementTree as ET import threading import logging import concurrent.futures import re from typing import List, Dict, Any logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') # 品質スコア計算のための定数定義 TITLE_MIN_LENGTH = 75 # eBay推奨のタイトル最適化基準（最大80文字） IMAGE_IDEAL_COUNT = 5 # 出品露出を高めるための理想的な最低画像枚数（最低3枚＋マージン） def _validate_item_id(item_id: str) -> str: """【深いポイント①】: XMLインジェクションを防ぐ厳格な型チェック""" if not item_id or not re.fullmatch(r'\d{8,13}', str(item_id)): raise ValueError(f"Security Alert: Invalid ItemID format detected: {item_id}") return str(item_id) def _get_text(node: ET.Element, tag: str, ns: dict) -> str: 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 "" def parse_item_specifics(item_node: ET.Element, ns: dict) -> Dict[str, List[str]]: """複雑な ItemSpecifics 構造を安全に辞書型 {Name: [Values]} にフラット展開する""" specifics_dict = {} specifics_root = item_node.find('ns:ItemSpecifics', ns) if specifics_root is None: return specifics_dict for nvl_node in specifics_root.findall('ns:NameValueList', ns): name = _get_text(nvl_node, 'ns:Name', ns).strip() if not name: continue # 1つのNameに対して複数のValueがあるケースをリストとしてすべて網羅 values = [v.text.strip() for v in nvl_node.findall('ns:Value', ns) if v.text] specifics_dict[name] = values return specifics_dict def get_single_item_detail(config: Any, token: str, item_id: str) -> Dict[str, Any]: """単一商品の詳細を取得してパースする（OutputSelectorによる最適化版）""" valid_id = _validate_item_id(item_id) headers = { "X-EBAY-API-CALL-NAME": "GetItem", "X-EBAY-API-SITEID": "0", "X-EBAY-API-COMPATIBILITY-LEVEL": "1323", "X-EBAY-API-IAF-TOKEN": token, "Content-Type": "text/xml" } # 【深いポイント②】: OutputSelectorを多重指定し、レスポンスペイロードを極限まで絞り込む xml_payload = f"""<?xml version="1.0" encoding="utf-8"?> <GetItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <ItemID>{valid_id}</ItemID> <DetailLevel>ReturnAll</DetailLevel> <OutputSelector>Title</OutputSelector> <OutputSelector>PictureDetails</OutputSelector> <OutputSelector>ItemSpecifics</OutputSelector> <OutputSelector>SKU</OutputSelector> </GetItemRequest> """ res = requests.post(config.trading_api_url, headers=headers, data=xml_payload.encode('utf-8'), timeout=20) res.raise_for_status() ns = {'ns': 'urn:ebay:apis:eBLBaseComponents'} root = ET.fromstring(res.text) ack = _get_text(root, 'ns:Ack', ns) if ack not in ['Success', 'Warning']: # 【深いポイント③】: 複数返却される可能性があるエラー原因を漏らさず全結合して例外を投げる errors = root.findall('ns:Errors', ns) messages = [_get_text(e, 'ns:LongMessage', ns) for e in errors] raise Exception(f"API Error for {valid_id}: {'; '.join(messages)}") item_node = root.find('ns:Item', ns) if item_node is None: raise ValueError(f"Item node not found in response for {valid_id}") picture_nodes = item_node.findall('ns:PictureDetails/ns:PictureURL', ns) pictures = [p.text for p in picture_nodes if p.text] return { "item_id": valid_id, "title": _get_text(item_node, 'ns:Title', ns), "sku": _get_text(item_node, 'ns:SKU', ns), "pictures": pictures, "specifics": parse_item_specifics(item_node, ns) } def get_items_bulk(config: Any, manager: Any, item_ids: List[str]) -> List[Dict[str, Any]]: """ 【自作 GetItems ラッパー】 前提条件: 引数に渡される manager.get_token() は、内部で Lock 制御等が行われており マルチスレッド環境下でも競合を起こさない「スレッドセーフ」な実装であること。 """ chunk_size = 20 all_updates = item_ids[:chunk_size] # 1回の最大数を制限 results = [] # 【パフォーマンス考慮】: ネットワークI/Oのボトルネックをスレッドプールで解消 with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: token = manager.get_token() futures = { executor.submit(get_single_item_detail, config, token, iid): iid for iid in all_updates } # 【注意】: as_completed は処理が完了した順に結果を返すため、入力された item_ids の順序とは一致しません。 for future in concurrent.futures.as_completed(futures): item_id = futures[future] try: res = future.result() results.append(res) except Exception as e: logging.error(f"[Error] Batch task failed for {item_id}: {e}") return results def check_data_quality(item_data: Dict[str, Any], required_aspects: List[str]) -> Dict[str, Any]: """取得データから品質チェック（QA）をスコアリングするロジック""" title = item_data.get("title", "") pictures = item_data.get("pictures", []) specifics = item_data.get("specifics", {}) # 1. タイトル文字数チェック title_len = len(title) title_score = 100 if title_len >= TITLE_MIN_LENGTH else (title_len / TITLE_MIN_LENGTH) * 100 # 2. 画像枚数チェック pic_count = len(pictures) pic_score = 100 if pic_count >= IMAGE_IDEAL_COUNT else (pic_count / IMAGE_IDEAL_COUNT) * 100 # 3. Item Specifics 必須項目の充足率 filled_required = [a for a in required_aspects if a in specifics and specifics[a]] aspect_coverage = (len(filled_required) / len(required_aspects)) * 100 if required_aspects else 100 # 総合スコア（平均） total_score = (title_score + pic_score + aspect_coverage) / 3 return { "item_id": item_data["item_id"], "sku": item_data["sku"], "title_length": title_len, "picture_count": pic_count, "missing_aspects": [a for a in required_aspects if a not in filled_required], "quality_score": round(total_score, 1) } パフォーマンス・スケーリング視点 (深度) 次世代 REST API（Inventory API / GraphQL）への戦略的移行パス 本記事で実装した自製 get_items_bulk は、Trading API の通信制約をマルチスレッドと OutputSelector でチューニングしたクライアント側の最適化アプローチです。 今後、ストアの規模が数十万 SKU へと拡大し、より強固なデータ同期基盤を設計する場合、レガシーな Trading API から新世代の REST API (Inventory API) または GraphQL API へのリプレイスを設計する必要があります。 【注意】 移行先選定の致命的な罠：Browse API を選んではいけない 技術ブログ等で「商品詳細の取得（REST版）には Browse API > getItem が使える」という記述を見かけますが、これは大きな間違いです。 Browse API はあくまで「購入用（Buyer-facing）」のパブリック API であり、レスポンスにセラー独自の機密データ（SKU、カスタム内部備考、仕入れ原価マスタとの紐付け情報、正確なマルチバリエーション在庫数など）が一切含まれません。 【正しい移行パスと設計思考】 商品情報マスタの監査: セラー側のデータ詳細を REST 環境で完全取得するには、Inventory API > getInventoryItem を使用するのが正解です。レスポンスは JSON 形式で返却され、複雑な XML の走査が不要になるため、パース処理に要するシステム CPU 負荷を大幅に削減できます。 GraphQL API の採用: 現在の eBay が提供する最もモダンなエンドポイント（GraphQL）では、まさに本記事の OutputSelector の上位互換として、リクエスト側から「取得したいプロパティのスキーマ構造」をコードで指定できます。ネットワークの帯域消費量を最も綺麗に削ぎ落とせるため、エンタープライズ領域のクローラー設計では GraphQL への移行がファイナルゴールとなります。 まとめ 本記事では、eBay の出品クオリティを担保するための商品詳細パースロジックと、並列バルク監査ツールの実装方法を解説しました。 ベースライン: GetItem Call の基本構造と、XML インジェクション脆弱性に対する防御措置。 深いポイント: OutputSelector による極限の通信軽量化。多値属性を考慮した ItemSpecifics の安全な辞書化。 スケーリング: スレッドセーフな並列化における順序保証の注意点。将来的な REST (Inventory API / GraphQL) への正しい設計アプローチ。 これで、出品（CRUD）のライフサイクル管理から、その品質を高めるための自動チェック体制までが完全に整いました。商品データ管理のフェーズはここで一区切りとなります。 次に進むべき視点 本記事では、クライアント側の並列処理により速度を担保しましたが、大量リクエストによる eBay 側の負荷分散や、より高度なイベント駆動アーキテクチャを目指す場合、API を定期実行（ポーリング）する設計そのものからの脱却が必要です。例えば、eBay 上でデータが変更された瞬間にシステム側へ通知を受け取る Notification API（Webhookモデル） の導入などを検討すると、システムのリアルタイム性はさらに次元が変わります。 次のステップ 11 回にわたる連載で、商品マスタの構築、出品、在庫同期、品質監査という「商品軸（Inventory）」のパイプラインは完璧に完成しました。 次回（#12）からは、いよいよ EC システムの最大の華である 【Trading API - 注文管理】 新章に突入します！ まずは 「GetOrders で eBay 上の売上・注文データを自動取得する」 方法について、バイヤーの支払いステータス（Paid）の安全なハンドリングや未発送データの抽出など、実務直結のバックオフィス自動化ロジックを解説します。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#10】eBay Trading API：EndItemで出品を終了する / 複数商品の一括取り下げ（bulk）自動化 はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第10回です。 前回（#9）は、eBay 上のアクティブな出品一覧を Pull してローカル DB と同期する処理を構築しました。今回は、商品のライフサイクルの最終ステージである「出品の取り下げ（早期終了：Early Termination）」を扱います。 2026年現在、eBay は従来の Trading API から新世代の REST API への移行を急速に推し進めています（直近でも 6 月の GetCategoryFeatures の廃止、9 月の UploadSiteHostedPictures の廃止などが控えています）。本記事では、現行の Trading API を用いた堅牢な一括下架システムの実装方法を解説するとともに、将来のシステム刷新を見据えた REST API への移行パスについても明示します。 この記事で得られること: EndItem API を用いた出品終了処理の基本構造と、アカウントを守る理由の選択。 プラットフォームのポリシー変更（ファッションカテゴリのサイズ規則変更など）に伴う緊急下架シナリオへの対応。 途中でクラッシュしても再開できる「断点回復（レジューム機能）」と「ファイルロギング」を備えた、本番環境仕様の一括下架スクリプト。 背景・なぜこれが重要か (Motivation) EC 運用において、商品ページを即座に削除・取り下げなければならない局面は突発的に発生します。 自社倉庫や併売先で商品の破損・紛失が発覚した。 知的財産権保護（VeRO プログラム）の警告を受け、即時下架が必要になった。 さらに直近の具体例として、「プラットフォーム側の急なポリシー変更による被動的下架」 も挙げられます。例えば、ファッションカテゴリにおける「サイズ表記のグローバルロック（世界統一規格化）」のような大変革が起きる際、旧来の非標準サイズで出品されていた大量の Listing は、システム上「修正（Revise）」を受け付けなくなります。このような場合、セラーツールは「対象商品を一括で EndItem（終了）させ、標準規格に準拠したデータで新規に出品し直す」という緊急バッチ処理を走らせる必要があります。 手動で数十～数百件の対応をしていては手遅れになります。自動化された強固な下架パイプラインを構築しておくことは、アカウントの健全性を死守するための強力な防衛策です。 基本的な使い方（ベースライン）：出品XMLの全体像 Trading API の EndItem は 1 リクエストにつき 1 商品しか処理できません。最もシンプルなリクエスト形式は以下の通りです。 <?xml version="1.0" encoding="utf-8"?> <EndItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <ItemID>110022334455</ItemID> <EndingReason>Incorrect</EndingReason> </EndItemRequest> 実務で躓く場面・深いポイント (Core) 1. EndingReason（終了理由）の選択ミスによるアカウント降格の罠 EndItem のリクエストには、取り下げの理由を示す <EndingReason> が必須です。 Incorrect（出品内容の誤り） LostOrBroken（商品の紛失・破損） OtherListingError（その他のエラー） ここで LostOrBroken を不用意に選択してはいけません。 公式ドキュメントには明記されていませんが、eBay の内部アルゴリズムは LostOrBroken による早期終了が頻発するアカウントを「在庫管理能力が著しく低いセラー」と判定します。これが累積すると、Best Match（検索結果）の表示順位が落とされる（SEOペナルティ）実害が発生します。 実務上、社内システムやタイムアウト起因の取り下げであれば、原則として Incorrect または OtherListingError を選択するのが安全です。 2. 特異なエッジケース：アクティブな取引（注文）がある場合の挙動 バイヤーが購入手続きを完了したが、未発送の状態や、支払い保留（ON_HOLD）、あるいは未着トラブル（INR）の保護期間内にある商品（ItemID）を EndItem で終了させた場合、システム的にどうなるでしょうか？ 注文データは消失しない: 出品を終了させても、既存の注文履歴や進行中のトランザクションは消えません。セラーは引き続き GetOrders 等でデータを取得し、発送処理や返金、ディスピュート（紛失・未着手続き）に対応する義務があります。 新たな購入のブロック: あくまで「これ以上の新規購入・入札」を即座に防ぐ処理として機能します。 3. EndItem と「在庫数 0 更新（OutOfStockControl）」の境界線 EndItem: その ItemID は完全に終了し、蓄積された「販売履歴（Sales History）」やウォッチ数は消滅します。廃盤商品や VeRO 警告など、二度と同じページを使わない場合のみ使用します。 在庫数 0 更新: 出品状態（SEOパワー）を維持したまま検索結果から一時的に非表示にします。再入荷の可能性がある場合は必ずこちら（第8回参照）を選択してください。 堅牢な実装：断点回復とログ保存を備えた一括下架エンジン Trading API の EndItem は 1 リクエストにつき 1 商品しか処理できません。数百件をマルチスレッドで高速に処理しつつ、途中でシステムがクラッシュしても「どこまで処理したか」を記憶して再開できる（断点回復）本番仕様のスクリプトを実装します。 # bulk_end_items.py import requests import xml.etree.ElementTree as ET import time import random import threading import logging import json import os import concurrent.futures from typing import List, Dict, Any from config import eBayConfig from ebay_token_manager import eBayTokenManager # 深いポイント①: 監査追跡のため、ログを標準出力ではなくファイルへ永続化 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[ logging.FileHandler("bulk_end_operations.log", encoding="utf-8"), logging.StreamHandler() ] ) PROGRESS_FILE = "bulk_end_progress.json" def _get_text(node: ET.Element, tag: str, ns: dict) -> str: 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 "" def load_progress() -> Dict[str, str]: """断点回復用：過去の成功/失敗ステータスをロード""" if os.path.exists(PROGRESS_FILE): with open(PROGRESS_FILE, 'r', encoding='utf-8') as f: return json.load(f) return {} def save_progress(item_id: str, status: str): """断点回復用：処理結果を即座にファイルへ同期保存 スケーリング注意点: 数千件規模にスケールする場合、スレッドごとに毎回全量JSONファイルを読み書きすると 深刻なI/Oボトルネックになります。大量データを扱う実稼働環境では、 SQLiteやMySQL等の外部DBへの個別レコード書き込み(UPDATE文)に置き換えてください。 """ progress = load_progress() progress[item_id] = status with open(PROGRESS_FILE, 'w', encoding='utf-8') as f: json.dump(progress, f, ensure_ascii=False, indent=2) def api_call_with_backoff(fn, max_retries=3): for attempt in range(max_retries): try: return fn() except requests.exceptions.HTTPError as e: if e.response is not None and e.response.status_code == 429: wait = (2 ** attempt) + random.uniform(0, 1) logging.warning(f"Rate limited (429). Retrying in {wait:.2f}s...") time.sleep(wait) else: raise raise Exception("Max retries exceeded after HTTP errors") def end_single_item(config: eBayConfig, token: str, item_id: str, reason: str = "Incorrect") -> str: headers = { "X-EBAY-API-CALL-NAME": "EndItem", "X-EBAY-API-SITEID": "0", "X-EBAY-API-COMPATIBILITY-LEVEL": "1323", "X-EBAY-API-IAF-TOKEN": token, "Content-Type": "text/xml" } xml_payload = f"""<?xml version="1.0" encoding="utf-8"?> <EndItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <ItemID>{item_id}</ItemID> <EndingReason>{reason}</EndingReason> </EndItemRequest> """ def execute(): res = requests.post(config.trading_api_url, headers=headers, data=xml_payload.encode('utf-8'), timeout=30) res.raise_for_status() return res response = api_call_with_backoff(execute) namespace = {'ns': 'urn:ebay:apis:eBLBaseComponents'} root = ET.fromstring(response.text) ack = _get_text(root, 'ns:Ack', namespace) if ack not in ['Success', 'Warning']: errors = [{"code": _get_text(e, 'ns:ErrorCode', namespace), "msg": _get_text(e, 'ns:LongMessage', namespace)} for e in root.findall('ns:Errors', namespace)] raise Exception(f"EndItem Failed: {errors}") return _get_text(root, 'ns:EndTime', namespace) def run_concurrent_bulk_end(config: eBayConfig, manager: eBayTokenManager, item_list: List[str], reason: str = "Incorrect"): # 防御的プログラミング: メモリ快照の不整合や二重Endを防ぐため、順序を維持したまま重複を除去 item_list = list(dict.fromkeys(item_list)) logging.info(f"[System] Starting concurrent bulk end for {len(item_list)} items...") # 拡張性の注意: 同時実行数(Semaphore)と最小ディレイ(sleep)は、eBayアカウントのTier（登録クラス）により # 割り当てられる日次上限「Call Limits」が異なるため、本番環境ではトラフィック制限に応じて数値を増減させてください。 semaphore = threading.Semaphore(5) progress_map = load_progress() stats_lock = threading.Lock() def _worker_concurrent(item_id: str): # 深いポイント②: 断点回復チェック（既に過去の実行で成功している場合は二重Endを防ぐためスキップ） if progress_map.get(item_id) == "SUCCESS": logging.info(f"[Skipped] (Already Ended in previous run): {item_id}") return with semaphore: time.sleep(0.1) # スパイク電流的な過負荷を防ぐためのインターバル try: token = manager.get_token() end_time = end_single_item(config, token, item_id, reason) logging.info(f"[Success]: {item_id} at {end_time}") with stats_lock: save_progress(item_id, "SUCCESS") except Exception as e: logging.error(f"[Failed] to end {item_id}: {e}") with stats_lock: save_progress(item_id, f"FAILED: {str(e)}") with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: futures = [executor.submit(_worker_concurrent, item_id) for item_id in item_list] concurrent.futures.wait(futures) logging.info("--- Bulk End Process Complete ---") if __name__ == "__main__": config = eBayConfig() manager = eBayTokenManager(config.client_id, config.client_secret, config.refresh_token, config.env) # テスト対象のItemIDリスト items_to_end = ["110022334455", "110022334456"] run_concurrent_bulk_end(config, manager, items_to_end, reason="Incorrect") 構造的リスクの回避：最新 REST API への移行パス 本記事で紹介した Trading API の EndItem（XML 形式）は、eBay が段階的に廃止を進めているレガシーな通信プロトコルです。将来にわたり安定したシステムを運用するためには、最新の REST API（Inventory API） へのリプレイスを視野に入れる必要があります。 新規開発時やシステム刷新時は、以下の REST エンドポイントへの移行設計を行ってください。 対応する REST API メソッド: DASHBOARD / INVENTORY 領域の Inventory API > withdrawOffer REST でのエンドポイント (POST): https://api.ebay.com/sell/inventory/v1/offer/{offerId}/withdraw アーキテクチャの違い: Trading API では「商品（ItemID）」そのものを直接終了させますが、REST API（Inventory API）では、商品情報マスタ（Inventory Item）から市場へ公開している売り枠（Offer）を「取り下げる（withdraw）」という洗練されたリソース設計に変わっています。 重要な前提条件と注意点： 上記の withdrawOffer は、最新の Inventory API 経由で作成された出品（Offer）にのみ適用可能です。本連載の第5回・第6回で解説した Trading API（AddItem 等）で登録済みのレガシー出品には OfferID が存在しないため、直接この REST API を呼び出すことはできません。既存の古い出品を REST 管理へと移行させるには、別途マイグレーション手順（既存出品の Inventory API モデルへのコンバート）が必要です。これについては後続回で詳しく取り上げます。 まとめ 本記事では、商品の販売を安全かつ迅速に終了させるための EndItem API について解説しました。 ベースライン: ItemID と EndingReason を組み合わせた下架処理の標準プロトコル。 深いポイント: アカウントの SEO ペナルティを回避する EndingReason の選択。GTC（長期無期限出品）商品における規約変更時の緊急下架シナリオの想定。 堅牢化実装: 障害時に未処理データから安全に再開できる JSON 進行状況保存（断点回復）と、監査可能なファイルロギング。 移行パス: 将来の完全 REST 化を見据えた Inventory API（withdrawOffer）への設計アプローチと制約。 これで、商品の「出品」「更新」「取得」「下架」という、商品管理サイクル（CRUD）の全 API パズルが完成しました。 次のステップ システムが自動で出品のライフサイクルを回せるようになると、次に必要になるのが「その出品データが本当に eBay の推奨する品質を満たしているか？」の自動監査です。 次回（#11）は、「GetItem / GetItemsで商品詳細を取得してデータ品質チェックツールを作る」 について解説します。単一・複数商品のデータを一括で取得し、タイトルの文字数や画像の枚数、Item Specifics の充足率を自動判定する QA スクリプトの組み方を深掘りします。お楽しみに！ 次の記事はこちら

前回の記事はこちら 【連載#9】eBay Trading API：GetSellerList / GetMyeBaySellingで出品中の全商品を確実にPull・同期する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第9回です。 前回（#8）までは、商品の出品から在庫・価格の高速同期といった「eBay へのデータ送信（Push）」をメインに解説してきました。今回からは、eBay 側の現在のステータスを正しくインポートする「データの取得（Pull）」フェーズに入ります。 この記事で得られること: 出品一覧・販売状況を取得する 2 大 API GetSellerList と GetMyeBaySelling の決定的な違いと使い分け。 GTC（長期出品）商品特有の「時間の罠」を回避するフィルタリング技術。 ネットワークの一時エラーに耐え、単一・バリエーション商品ともに網羅する堅牢な同期スクリプトの実装。 背景・なぜこれが重要か (Motivation) 自社システムやデータベース（DB）を構築して運用していると、必ず 「データの整合性の漂移（Data Drift）」 が発生します。 セラーが eBay の管理画面（Seller Hub）から手動で出品を取り下げたり、eBay 側のポリシー違反で出品が強制削除されたり、バリエーションの一部が売り切れたりした場合、自社 DB 側がその事実を検知できなければ、実在庫との不整合や二重販売の原因になります。 この同期ズレを防ぐためには、定期的に eBay 側から「現在のアクティブな出品一覧」を Pull して、ローカル DB を監査・一括更新（Upsert）するバッチ処理が不可欠です。 eBay Trading API にはそのための武器が 2 つ用意されていますが、特性を理解せずに使うと「全件取れていない」「レスポンスが重すぎてタイムアウトする」といった問題に直面します。 基本的な使い方（ベースライン）：2大APIの徹底比較 まずは、これら 2 つの API の特性をマトリクスで理解しましょう。 機能・特性 GetMyeBaySelling GetSellerList 主な用途 現在のアカウント状況のクイックな同期、日次バッチ 出品データの全件一括ダウンロード、週次・月次のディープ監査 必須フィルタ 不要（ActiveList などのブロック単位で指定） 時間範囲（StartTime または EndTime）の指定が必須 時間指定の制約 なし 1 回のリクエストで指定できる期間幅が最大 120 日間（※過去や未来の日付上限自体はない） データ軽量化 比較的軽量（必要なブロックのみを Include する） GranularityLevel で制御。軽量化なら Coarse を指定 バリエーション情報 <IncludeVariations>true</IncludeVariations> の明示が必要 GranularityLevel を調整することで詳細まで取得可能 【どちらを選ぶべきか？】 GetMyeBaySelling: 「今現在、アクティブな商品の一覧と在庫数をサクッと確認したい」という日次のクイックな在庫同期に最適です。 GetSellerList: 「新規システム導入時に、過去に出品した数万件の全データを一括インポートしたい」という初期同期や、時間ベースでの厳密な監査に必須です。 実務で躓く場面・深いポイント (Core) 1. GetSellerList における「GTC商品の罠」 GetSellerList を使う際、ほとんどのエンジニアが 「出品開始時間（StartTimeFrom / StartTimeTo）」 でフィルタをかけてしまいます。これが最大の罠です。 eBay の固定価格商品は基本的に GTC（自動再出品）です。3 年前に出品開始され、毎月自動更新されているアクティブな商品は、開始時間が 3 年前のままなので、直近 120 日間のフィルターには絶対にヒットしません。 【解決策】: 現在アクティブな GTC 商品を網羅したい場合は、開始時間ではなく 「出品終了時間（EndTimeFrom / EndTimeTo）」 でフィルターをかけます。GTC 商品は内部的に「30 日後に終了する設定」で毎月ロールオーバーしているため、“今から 30 日後までに終了する予定の商品” を検索すれば、現在動いているすべてのアクティブ商品を確実にキャッチできます。 2. ペイロード肥大化と DetailLevel / GranularityLevel の混同 Trading API にはレスポンスの細かさを制御するパラメータがありますが、API によって挙動が異なります。 GetSellerList で全件取得を試みる際、詳細なデータを求めすぎてタイムアウトを起こすケースが後を絶ちません。一覧の同期や監査が目的であれば、レスポンスを軽量化するために <GranularityLevel>Coarse</GranularityLevel> を指定するのが鉄則です。これにより、重い商品説明などのテキストを除外した必要最小限のフィールドのみが返却されます。 堅牢な実装：一括取得・パーススクリプト 実運用に耐えうるよう、以下の堅牢化を施した Python コードを実装します。 古い実行環境にも配慮した Python 3.7+ 互換の型ヒント（typing.Tuple など）。 eBay の一時的なネットワークエラーや API 瞬断対策として、tenacity ライブラリを用いた指数バックオフ付きの自動リトライ。 出品形式（固定価格 / オークション+BIN）に応じた価格フィールド（StartPrice / BuyItNowPrice）の安全なパース処理。 # fetch_listings.py import requests import xml.etree.ElementTree as ET 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 from ebay_token_manager import eBayTokenManager def _get_text(node: Optional[ET.Element], tag: str, ns: dict) -> str: """NoneType クラッシュを防ぐ安全なテキスト抽出ヘルパー""" 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 "" # 【深いポイント①】: 一時的なAPIエラーに備え、tenacityによるリトライ機構を装備 @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def _post_api_request(url: str, headers: dict, payload: str) -> str: response = requests.post(url, headers=headers, data=payload.encode('utf-8'), timeout=30) response.raise_for_status() return response.text def fetch_active_list_myebay(config: eBayConfig, token: str, page_number: int = 1) -> Tuple[List[Dict[str, Any]], int]: """ GetMyeBaySelling を使用した日次クイック同期用関数 """ headers = { "X-EBAY-API-CALL-NAME": "GetMyeBaySelling", "X-EBAY-API-SITEID": "0", "X-EBAY-API-COMPATIBILITY-LEVEL": "1323", "X-EBAY-API-IAF-TOKEN": token, "Content-Type": "text/xml" } # 【深いポイント②】: バリエーション情報を取得するために IncludeVariations を明示 xml_payload = f"""<?xml version="1.0" encoding="utf-8"?> <GetMyeBaySellingRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <ActiveList> <Include>true</Include> <IncludeVariations>true</IncludeVariations> <Pagination> <EntriesPerPage>100</EntriesPerPage> <PageNumber>{page_number}</PageNumber> </Pagination> </ActiveList> </GetMyeBaySellingRequest> """ res_text = _post_api_request(config.trading_api_url, headers, xml_payload) namespace = {'ns': 'urn:ebay:apis:eBLBaseComponents'} root = ET.fromstring(res_text) ack = _get_text(root, 'ns:Ack', namespace) if ack not in ['Success', 'Warning']: errors = [{"code": _get_text(e, 'ns:ErrorCode', namespace), "msg": _get_text(e, 'ns:LongMessage', namespace)} for e in root.findall('ns:Errors', namespace)] raise Exception(f"GetMyeBaySelling Error: {errors}") items_extracted = [] total_pages_node = root.find('ns:ActiveList/ns:PaginationResult/ns:TotalNumberOfPages', namespace) total_pages = int(total_pages_node.text) if total_pages_node is not None and total_pages_node.text else 1 active_list_node = root.find('ns:ActiveList/ns:ItemArray', namespace) if active_list_node is not None: for item_node in active_list_node.findall('ns:Item', namespace): item_id = _get_text(item_node, 'ns:ItemID', namespace) title = _get_text(item_node, 'ns:Title', namespace) variations_node = item_node.find('ns:Variations', namespace) if variations_node is not None: for var_node in variations_node.findall('ns:Variation', namespace): items_extracted.append({ "item_id": item_id, "title": f"{title} ({_get_text(var_node, 'ns:VariationTitle', namespace)})", "sku": _get_text(var_node, 'ns:SKU', namespace), "price": float(_get_text(var_node, 'ns:StartPrice', namespace) or 0.0), "quantity": int(_get_text(var_node, 'ns:Quantity', namespace) or 0), "is_variation": True }) else: # 【深いポイント③】: 出品形式（固定価格 / オークション+BIN）に応じた価格フィールドから安全に取得する price_val = _get_text(item_node, 'ns:BuyItNowPrice', namespace) or _get_text(item_node, 'ns:StartPrice', namespace) or "0.0" items_extracted.append({ "item_id": item_id, "title": title, "sku": _get_text(item_node, 'ns:SKU', namespace), "price": float(price_val), "quantity": int(_get_text(item_node, 'ns:Quantity', namespace) or 0), "is_variation": False }) return items_extracted, total_pages def fetch_active_list_sellerlist(config: eBayConfig, token: str, page_number: int = 1) -> Tuple[List[Dict[str, Any]], int]: """ GetSellerList を使用したディープ監査用関数（GTC商品の罠をEndTimeで回避） """ headers = { "X-EBAY-API-CALL-NAME": "GetSellerList", "X-EBAY-API-SITEID": "0", "X-EBAY-API-COMPATIBILITY-LEVEL": "1323", "X-EBAY-API-IAF-TOKEN": token, "Content-Type": "text/xml" } # 【核心】: GTC商品を漏らさず取るため、現在から30日後までの「EndTime」でフィルタリング now = datetime.now(timezone.utc) end_time_from = now.strftime('%Y-%m-%dT%H:%M:%S.000Z') end_time_to = (now + timedelta(days=30)).strftime('%Y-%m-%dT%H:%M:%S.000Z') # 【軽量化】: 軽量化のために GranularityLevel=Coarse を指定 xml_payload = f"""<?xml version="1.0" encoding="utf-8"?> <GetSellerListRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <EndTimeFrom>{end_time_from}</EndTimeFrom> <EndTimeTo>{end_time_to}</EndTimeTo> <GranularityLevel>Coarse</GranularityLevel> <Pagination> <EntriesPerPage>200</EntriesPerPage> <PageNumber>{page_number}</PageNumber> </Pagination> </GetSellerListRequest> """ res_text = _post_api_request(config.trading_api_url, headers, xml_payload) namespace = {'ns': 'urn:ebay:apis:eBLBaseComponents'} root = ET.fromstring(res_text) ack = _get_text(root, 'ns:Ack', namespace) if ack not in ['Success', 'Warning']: errors = [{"code": _get_text(e, 'ns:ErrorCode', namespace), "msg": _get_text(e, 'ns:LongMessage', namespace)} for e in root.findall('ns:Errors', namespace)] raise Exception(f"GetSellerList Error: {errors}") items_extracted = [] pagination_node = root.find('ns:PaginationResult', namespace) total_pages = int(_get_text(pagination_node, 'ns:TotalNumberOfPages', namespace) or 1) item_array_node = root.find('ns:ItemArray', namespace) if item_array_node is not None: for item_node in item_array_node.findall('ns:Item', namespace): # 【深いポイント④】: GranularityLevel=Coarse では BuyItNowPrice は返却されません。 # 固定価格（Fixed Price）商品では StartPrice が出品価格そのものになります。 price_val = _get_text(item_node, 'ns:StartPrice', namespace) or "0.0" items_extracted.append({ "item_id": _get_text(item_node, 'ns:ItemID', namespace), "title": _get_text(item_node, 'ns:Title', namespace), "sku": _get_text(item_node, 'ns:SKU', namespace), "price": float(price_val), "quantity": int(_get_text(item_node, 'ns:Quantity', namespace) or 0), "is_variation": False }) return items_extracted, total_pages パフォーマンス・スケーリング視点 (深度) 1. API Call Limit (日次呼び出し制限) への厳格な配慮 数万〜数十万 SKU を持つ大規模運用の環境において、最も注意すべきは API Call Limit（日次制限） です。無策のまま全件ループを頻繁に回すと、上限に達してシステム全体が停止します。自社アカウントに割り当てられた上限値を My Account > API Call Limits で必ず確認し、バッチの実行頻度を調整してください。 2. ローカル DB 同期アーキテクチャの最適化 API 制限を回避するため、以下の「2段階ハイブリッド構成」で運用するのがベストプラクティスです。 【ハイブリッド同期の構成パターン】 フェーズ 1（高頻度・軽量 Pull）: 普段の日次（または時間ごと）バッチでは、GetMyeBaySelling を使ってアクティブ一覧の差分や主要項目のみを素早く確認します。 フェーズ 2（週次・ディープ監査）: 週に 1 回、アクセスが少ない夜間帯に GetSellerList を用いて、GTC 商品の更新漏れや、管理画面側で直接削除された「幽霊出品」の突合クレンジングを行います。 まとめ 本記事では、eBay 側から出品中のデータを安全かつ正確に Pull し、ローカル DB と同期するための設計を解説しました。 ベースライン: GetSellerList（監査・一括用）と GetMyeBaySelling（デイリー用）の役割の違い。 深いポイント: GTC 商品に対する EndTime フィルタ適用の重要性。GranularityLevel=Coarse による軽量化と、tenacity を使ったエラーリトライ。 スケーリング: 日次の軽量 Pull と週次の完全 Pull を切り分ける、Call Limit を保護するハイブリッド同期設計。 次のステップ 出品状況が正確に把握できるようになると、次に必要になるのが「売れ残った古い在庫の整理」や「突発的なトラブルによる出品の一斉取り下げ」です。 次回（#10）は、「EndItemで出品を終了する / 複数商品の一括取り下げ（bulk）自動化」 について解説します。手動下架との違いや、大量の商品を安全に一括 End させる際の実務上の注意点を掘り下げます。お楽しみに！ 次の記事はこちら