前回の記事はこちら 【連載#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 を使って、バイヤーからのメッセージを自動取得し、テンプレートに基づいた返信を自動送信する仕組みを実装します。人手を介さない完全自動レスポンスシステムの構築にチャレンジしましょう！

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

前回の記事はこちら 【連載#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 させる際の実務上の注意点を掘り下げます。お楽しみに！ 次の記事はこちら

【重要】Compliance APIの提供終了（Decommission）に関するお知らせ eBay開発者の皆様、 日頃よりeBay Developers Programをご利用いただきありがとうございます。 本お知らせは、当社の記録により、提供終了（Decommission）が予定されている Compliance API を呼び出しているアプリケーションをお持ちの開発者様へご案内しております。 対象となるAPIと今後のスケジュール Compliance APIは主に、セラーが出品をeBayのポリシーに確実に準拠させるための支援を目的として提供されていました。以下のCompliance APIのメソッドはすでに非推奨（Deprecated）となっており、以下の期日をもって完全に廃止されます。 getListingViolationsSummary getListingViolations 重要な日程 (Key Dates) 完全廃止日 (Decommission Date): 2026年5月31日 上記の日付以降、これらのエンドポイントへのアクセスは完全に遮断され、呼び出しは機能しなくなります。 代替手段について (Alternatives) 上記の非推奨APIはまもなく利用できなくなりますが、現時点において、これらのAPIに代わる代替のAPIの提供は予定されておりません。 しかしながら、セラーは引き続き Seller Hub（セラーハブ） を通じて、ご自身の「出品コンプライアンスレポート（Listing compliance report）」にアクセスし、違反状況を確認・管理することが可能です。APIを利用したプログラムによる自動取得機能は廃止となりますが、運用上の確認はSeller Hubをご活用いただくようお願いいたします。 今後ともeBay Developers Programをよろしくお願い申し上げます。 eBay Developers Program / eBay Japan APIサポートチーム

前回の記事はこちら eBay Trading API：ReviseInventoryStatusで複数商品の在庫・価格を高速同期する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第8回です。 前回（#7）は ReviseFixedPriceItem を用いた「カタログ情報（タイトル等）の部分更新」を実装しましたが、今回は全く別の要件にフォーカスします。それは 「在庫と価格のリアルタイム同期」 です。 この記事で得られること: なぜ在庫同期に ReviseFixedPriceItem を使ってはいけないのかというアーキテクチャの理解。 ReviseInventoryStatus を用いた、軽量かつ一括（Bulk）での在庫・価格更新メソッドの習得。 チャンク分割（4件制限）と並列処理（マルチスレッド）を組み合わせた、本番稼働に耐えうる高速同期エンジンの実装。 背景・なぜこれが重要か (Motivation) 自社ECや他モールと併売している越境ECセラーにとって、「在庫がズレて欠品キャンセルが発生する」ことはアカウントの致命傷（Defect Rate悪化）に直結します。そのため、在庫は最短数分〜十数分間隔で eBay に同期し続ける必要があります。 ここで、前回の ReviseFixedPriceItem を使って数千件の在庫を同期しようとすると、以下の壁に激突します。 APIが重すぎる: 検索インデックスの再構築が走るため、レスポンスに数秒かかります。 Bulk処理ができない: 1リクエストにつき1商品しか更新できません。 Rate Limitの枯渇: 1日あたりの Call Limit をあっという間に使い果たします。 この課題を解決するために eBay が用意した専用の軽量 API が ReviseInventoryStatus です。この API は「在庫数（Quantity）」と「価格（StartPrice）」の更新のみに特化しており、検索インデックスの深い更新をバイパスするため、圧倒的な速度で動作します。 基本的な使い方（ベースライン）：一括更新XMLの全体像 ReviseInventoryStatus は、1回の API コールで 最大4件まで の商品を同時に更新できます。 第6回（バリエーション出品）で解説した InventoryTrackingMethod=SKU の設定を行っていれば、eBay の ItemID を知らなくても、自社の SKU だけで直接更新をかけることが可能です。 <?xml version="1.0" encoding="utf-8"?> <ReviseInventoryStatusRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <InventoryStatus> <SKU>TSHIRT-RED-S</SKU> <Quantity>15</Quantity> <StartPrice currencyID="USD">18.99</StartPrice> </InventoryStatus> <InventoryStatus> <ItemID>112233445566</ItemID> <Quantity>0</Quantity> </InventoryStatus> </ReviseInventoryStatusRequest> 実務で躓く場面・深いポイント (Core) この API は強力ですが、独自の制約があるため実装時にいくつかの罠が存在します。 1. 厳格な「最大4件」ルール 1つのリクエストに <InventoryStatus> ブロックを5つ以上入れると、API は即座にエラーを返します。プログラム側で、更新対象のリストを必ず4件ずつのチャンク（Chunk）に分割してリクエストを投げるロジックを組む必要があります。 2. レスポンス解析の罠（トップレベルの Ack） eBay API に慣れてくると、「リクエストで配列を送ったのだから、結果も配列ごとに Ack（成功/失敗）が入っているはずだ」と思い込みがちですが、この API のレスポンスはトップレベルに1つの <Ack> があり、エラーは <Errors> に集約される構造になっています。個別のノードで成否を判定しようとするとバグの温床になります。 3. バリエーション商品の ItemID 指定の罠 InventoryTrackingMethod=SKU を設定していないバリエーション商品の在庫を更新する場合、<ItemID> だけを送ると「どのバリエーションの在庫か分からない」とエラーになります。この場合は <ItemID> と <SKU> を両方指定しなければなりません。（※だからこそ、第6回で「絶対に SKU ベースでのトラッキングを有効にせよ」と強調しました。） 堅牢な実装：チャンク分割と高速同期エンジン 実務で数千件の在庫を同期するための「バッチプロセッサ」を実装します。入力されたリストを自動で4件ずつに分割（Chunking）し、エラーハンドリングを厳格に行った動的ジェネレータです。 # sync_inventory.py import requests import xml.etree.ElementTree as ET import html import time import random from typing import List, Dict, Any from config import eBayConfig from ebay_token_manager import eBayTokenManager def _get_text(node: 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 "" def api_call_with_backoff(fn, max_retries=3): """HTTP 429エラー(Too Many Requests)対策の指数バックオフ""" 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) print(f"[WARNING] Rate limited. Retrying in {wait:.2f}s...") time.sleep(wait) else: raise raise Exception("Max retries exceeded after HTTP errors") def revise_inventory_batch(config: eBayConfig, token: str, status_list: List[Dict[str, Any]]) -> Dict[str, Any]: """ 最大4件の在庫/価格を一括更新するコア関数 """ if len(status_list) > 4: raise ValueError("ReviseInventoryStatus accepts a maximum of 4 items per request.") if not status_list: return {"success": [], "errors": []} status_tags = "" for item in status_list: # 【深いポイント①】: 必須キーの欠如に対する防御 if 'sku' not in item and 'item_id' not in item: raise ValueError(f"Each item must have 'sku' or 'item_id': {item}") sku_tag = f"<SKU>{html.escape(str(item['sku']))}</SKU>" if 'sku' in item else "" item_id_tag = f"<ItemID>{html.escape(str(item['item_id']))}</ItemID>" if 'item_id' in item else "" qty_tag = "" if 'quantity' in item: qty_tag = f"<Quantity>{int(item['quantity'])}</Quantity>" price_tag = "" if 'price' in item: price_tag = f'<StartPrice currencyID="USD">{float(item["price"])}</StartPrice>' status_tags += f""" <InventoryStatus> {sku_tag} {item_id_tag} {qty_tag} {price_tag} </InventoryStatus> """ xml_payload = f"""<?xml version="1.0" encoding="utf-8"?> <ReviseInventoryStatusRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> {status_tags} </ReviseInventoryStatusRequest> """ headers = { "X-EBAY-API-CALL-NAME": "ReviseInventoryStatus", "X-EBAY-API-SITEID": "0", "X-EBAY-API-COMPATIBILITY-LEVEL": "1323", "X-EBAY-API-IAF-TOKEN": token, "Content-Type": "text/xml" } 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) results = {"success": [], "errors": []} # 【深いポイント②】: 正しいレスポンス解析ロジック（トップレベルのAckを確認） 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"ReviseInventoryStatus failed: {errors}") # 成功した場合、各 InventoryStatus ノードから更新された ItemID / SKU を取得 for node in root.findall('ns:InventoryStatus', namespace): sku = _get_text(node, 'ns:SKU', namespace) item_id = _get_text(node, 'ns:ItemID', namespace) identifier = sku if sku else item_id results["success"].append(identifier) return results def run_bulk_inventory_sync(config: eBayConfig, manager: eBayTokenManager, all_updates: List[Dict[str, Any]]): """ 大量の更新リストを4件ずつのチャンクに分割して処理する（直列版） """ chunk_size = 4 total_success = 0 total_errors = 0 chunks = [all_updates[i:i + chunk_size] for i in range(0, len(all_updates), chunk_size)] for idx, chunk in enumerate(chunks): token = manager.get_token() try: res = revise_inventory_batch(config, token, chunk) total_success += len(res["success"]) if res["errors"]: total_errors += len(res["errors"]) print(f"[ERROR] Chunk {idx+1} Errors: {res['errors']}") else: print(f"[SUCCESS] Chunk {idx+1} Processed successfully.") except Exception as e: print(f"[FATAL ERROR] Fatal Error on Chunk {idx+1}: {e}") total_errors += len(chunk) time.sleep(0.2) # 直列時の簡易Rate Limit対策 print("\n--- Sync Complete ---") print(f"Successfully Updated: {total_success} | Errors: {total_errors}") パフォーマンス・スケーリング視点 (深度) さらなる高速化：並列処理（Threading）の導入 数万件規模の SKU を管理するセラーの場合、4件ずつ直列で処理していても時間がかかりすぎます。 I/O 待ちがボトルネックとなるため、チャンク分割した後に concurrent.futures.ThreadPoolExecutor を用いて 複数のチャンクを並列で送信する のが究極のスケーリング手法です。 【注意】: 連載第1回で eBayTokenManager に threading.Lock を組み込んだのはこのためです。マルチスレッド環境では、各スレッド内 で get_token() を評価することで、トークン更新時の競合を防ぐことができます。 import concurrent.futures import threading def run_concurrent_inventory_sync(config: eBayConfig, manager: eBayTokenManager, all_updates: List[Dict[str, Any]]): chunk_size = 4 chunks = [all_updates[i:i + chunk_size] for i in range(0, len(all_updates), chunk_size)] # 【深いポイント③】: スレッド内でのトークン評価とRate Limit制御 # Semaphoreを用いて同時実行数を制限し、急激なスパイク(429エラー)を防ぐ semaphore = threading.Semaphore(5) # 内部関数として定義し、直列版関数との混同を避ける def _sync_chunk_concurrent(chunk_data): with semaphore: time.sleep(0.1) # スレッド起動時の最小間隔を確保 # 必ずスレッド内部で get_token を呼ぶこと（メインスレッドで評価するとタイミングがズレます） token = manager.get_token() return revise_inventory_batch(config, token, chunk_data) with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: futures = [executor.submit(_sync_chunk_concurrent, chunk) for chunk in chunks] for future in concurrent.futures.as_completed(futures): try: res = future.result() # 成功ログ・エラーログの記録など except Exception as e: print(f"[ERROR] Thread execution failed: {e}") # --- 使い方の例 --- if __name__ == "__main__": config = eBayConfig() manager = eBayTokenManager(config.client_id, config.client_secret, config.refresh_token, config.env) sync_data = [ {"sku": "TS-RED-S", "quantity": 10, "price": 19.99}, # 【注記】: OutOfStockControlが有効な場合は「検索非表示」になるだけですが、 # 無効の場合は quantity=0 を送ると出品が終了（End）します。第6回の設定を必ず確認してください。 {"sku": "TS-RED-M", "quantity": 0}, {"sku": "TS-BLU-S", "quantity": 5}, {"sku": "TS-BLU-M", "quantity": 2, "price": 18.99}, {"sku": "ACC-WATCH-01", "quantity": 15} # 5件目（自動的に次のチャンクへ） ] run_concurrent_inventory_sync(config, manager, sync_data) この実装により、API の Call Limit を節約しつつ、自社システムの在庫変動を「数分以内」に eBay 側の全 SKU へ反映させる強靭な同期パイプラインが完成します。 まとめ 本記事では、eBay 運用における心臓部とも言える「在庫と価格の高速同期」について解説しました。 ベースライン: ReviseFixedPriceItem ではなく、軽量な ReviseInventoryStatus を使う意義と、最大4件の制約。 深いポイント: トップレベルの Ack 判定による正しいレスポンス解析。必須項目（SKU/ItemID）の欠落防止。 スケーリング: マルチスレッド環境下におけるスレッドセーフなトークン取得と、セマフォを用いた Rate Limit 制御アーキテクチャ。 これで、商品の「出品」から「日々のメンテナンス（カタログ更新・在庫同期）」までを自動化する基盤が完全に整いました。 次のステップ これまでは「こちらから eBay にデータを送る（Push）」処理を実装してきました。しかし、「今、eBay 上に何が出品されているのか？」という現状をプログラムから把握できなければ、同期システムは成り立ちません。 次回（#9）は、「GetSellerList / GetMyeBaySellingで出品中の一覧・販売状況を自動取得する」 方法について解説します。取得した一覧データをローカルDBにマッピングする実務的な手法をお楽しみに！ 次の記事はこちら

前回の記事はこちら eBay Trading API：ReviseFixedPriceItemで出品済みの価格・在庫・タイトルを安全に更新する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第7回です。 これまでの連載で、単一商品（#5）およびバリエーション商品（#6）の新規出品プロセスが完成しました。しかし、EC サイトの運用において「出品して終わり」ということはあり得ません。為替変動による価格調整、SEO 改善のためのタイトル変更など、出品後データの更新（Revision） は日常業務の大半を占めます。 この記事で得られること: ReviseFixedPriceItem を用いた部分更新（Partial Update）のアーキテクチャ理解と空タグの危険性。 変更不可フィールドと DeletedField の正しい使い方。 CSV を用いた一括更新バッチと、HTTP 429 (Too Many Requests) を防ぐ Exponential Backoff（指数バックオフ） の実装。 背景・なぜこれが重要か (Motivation) eBay の Trading API において、既存の出品を更新するメソッドは主に ReviseFixedPriceItem です。この API は非常に強力で、タイトルから画像、Item Specifics に至るまで、出品時に設定したほぼすべての項目を後から書き換えることができます。 しかし、その強力さゆえに、リクエストの組み方を一つ間違えると「更新したかったのは価格だけなのに、商品説明が全部消えてしまった」という大事故を引き起こします。「何を送り、何を送らないべきか」。この 部分更新（Delta Update） の思想を理解することが、安全な運用システムの絶対条件となります。 基本的な使い方（ベースライン）：部分更新の原則 ReviseFixedPriceItem の最大の特徴は、「XML に含めたタグだけが更新され、省略したタグは現状維持される」 という点です。 例えば、ItemID: 112233445566 の商品の「タイトル」と「価格」だけを変更したい場合、ベースラインの XML は以下のようになります。 <?xml version="1.0" encoding="utf-8"?> <ReviseFixedPriceItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <Item> <ItemID>112233445566</ItemID> <Title>New SEO Optimized Title for Vintage Camera</Title> <StartPrice currencyID="USD">289.99</StartPrice> </Item> </ReviseFixedPriceItemRequest> 【部分更新の処理フローイメージ】 送信するXML eBay側の処理 ───────────────── ───────────────────────── <Title>新しい値</Title> → Titleを「新しい値」に上書き (反映) <StartPrice>289</StartPrice> → Priceを289に上書き (反映) （Descriptionタグなし） → Descriptionは現状維持 (安全) <Description></Description> → Descriptionを空に上書き (危険) 実務で躓く場面・深いポイント (Core) ここからは、エンジニアがよくハマる「更新 API の罠」を解説します。 1. 空タグによる意図せぬデータ消失 (Accidental Wipe) 最も多いバグがこれです。プログラム側で「今回は価格だけ更新するから、タイトルは空文字でいいや」と考えて <Title></Title> を送信すると、eBay は「タイトルを空にしろ」と解釈します。 後述する CSV 一括更新スクリプトでは、「CSV の空欄」をプログラム側でフィルタリングし、XML タグ自体を生成しない（現状維持にする） という厳格な制御を行っています。 2. 変更不可フィールドの存在 Revise API でも変更できない（または条件付きでしか変更できない）フィールドがあります。これを変更しようとすると Error 21916635 などで弾かれます。 フィールド 制約内容 PrimaryCategory 出品から14日経過後、または販売実績がある場合は変更不可。 ListingType 固定（Fixed Price から Auction への変更などは不可）。 VariationSpecificsSet バリエーションの「軸の名前（例: Color）」の変更は不可（値の追加は可）。 3. 項目の明示的な削除 (DeletedField) 空タグが使えない項目において、既存データを完全に削除するためには eBay 特有の <DeletedField> タグを使用します。 <ReviseFixedPriceItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <DeletedField>Item.SubTitle</DeletedField> <DeletedField>Item.SecondaryCategory</DeletedField> <DeletedField>Item.ShippingDetails.ShippingServiceOptions</DeletedField> # 配送オプション全体が削除されます。FreeShipping設定も消えるため注意！ <Item><ItemID>112233445566</ItemID></Item> </ReviseFixedPriceItemRequest> 4. バリエーション商品（Multi-SKU）に関する罠 本記事の実装は 単一商品（Single-SKU） が対象です。前回（#6）で解説したバリエーション商品に対して、本記事のようにトップレベルの <Quantity> を送信しても無視されます。バリエーションの在庫を更新する場合は、特定の <Variation> ブロック内で <SKU> と <Quantity> を指定する必要があります。 5. UUID は Revise では冪等性を保証しない 第5回で紹介した <UUID> による冪等性（Idempotency）保証は、実は Add 系の呼び出し（新規出品）に限定された機能 です。ReviseFixedPriceItem に UUID を送っても無害ですが、二重更新を防ぐ効果はありません。そのため、バッチ処理側で適切なリトライ制御を行う必要があります。 堅牢な実装：動的 XML ビルダーによる安全な更新関数 「意図せぬデータ消失の防止」「NoneType クラッシュ対策」「処理スキップの明示」をすべて組み込んだ堅牢な更新関数を実装します。今回はバッチ処理等でも使い回せるよう、Rate Limit 対策のバックオフ関数も同じファイルに定義します。 # revise_item.py import requests import xml.etree.ElementTree as ET import html import time import random from dataclasses import dataclass, field from enum import Enum from typing import List from config import eBayConfig from ebay_token_manager import eBayTokenManager class ReviseStatus(Enum): SUCCESS = "success" SKIPPED = "skipped" @dataclass class ReviseResult: item_id: str status: ReviseStatus warnings: List[str] = field(default_factory=list) def _get_text(node: 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 "" def api_call_with_backoff(fn, max_retries=3): """HTTP 429エラー時に待機時間を倍増させながらリトライするヘルパー""" 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: # Too Many Requests wait = (2 ** attempt) + random.uniform(0, 1) print(f"[警告] Rate limited. Retrying in {wait:.2f}s...") time.sleep(wait) else: raise raise Exception("Max retries exceeded after HTTP errors") def revise_single_item(config: eBayConfig, token: str, item_id: str, updates: dict) -> ReviseResult: """ 指定された項目のみを安全に部分更新する関数。 """ if not item_id: raise ValueError("ItemID is required for revision.") xml_elements = [] if updates.get('title'): xml_elements.append(f"<Title>{html.escape(updates['title'])}</Title>") if updates.get('price') is not None: try: price_val = float(updates['price']) xml_elements.append(f'<StartPrice currencyID="USD">{price_val}</StartPrice>') except ValueError: raise ValueError("Price must be numeric.") if updates.get('quantity') is not None: try: qty_val = int(updates['quantity']) xml_elements.append(f"<Quantity>{qty_val}</Quantity>") except ValueError: raise ValueError("Quantity must be an integer.") # 【ポイント】 更新項目がない場合はAPIコールをスキップし、状態を明確に返す if not xml_elements: return ReviseResult(item_id=item_id, status=ReviseStatus.SKIPPED) inner_xml = "\n".join(xml_elements) headers = { "X-EBAY-API-CALL-NAME": "ReviseFixedPriceItem", "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"?> <ReviseFixedPriceItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <Item> <ItemID>{html.escape(str(item_id))}</ItemID> {inner_xml} </Item> </ReviseFixedPriceItemRequest> """ def execute_request(): 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_request) # レスポンス解析 namespace = {'ns': 'urn:ebay:apis:eBLBaseComponents'} root = ET.fromstring(response.text) ack = _get_text(root, 'ns:Ack', namespace) warnings_list = [] if ack == 'Warning': warnings_list = [ _get_text(e, 'ns:LongMessage', namespace) for e in root.findall('ns:Errors', namespace) if _get_text(e, 'ns:SeverityCode', namespace) == 'Warning' ] if ack not in ['Success', 'Warning']: errors = [ { "code": _get_text(e, 'ns:ErrorCode', namespace), "message": _get_text(e, 'ns:LongMessage', namespace) } for e in root.findall('ns:Errors', namespace) if _get_text(e, 'ns:SeverityCode', namespace) == 'Error' ] raise Exception(f"Revision Failed for {item_id}: {errors}") return ReviseResult(item_id=item_id, status=ReviseStatus.SUCCESS, warnings=warnings_list) 実践：CSVを用いた一括更新バッチ処理とバックオフ制御 実務では「年末商戦に向けて、CSVで数百件の商品のタイトルと価格を一気に変更したい」といったバッチ処理が求められます。 用意するCSV (bulk_updates.csv) の例: item_id,title,price,quantity 112233445566,Holiday Special Vintage Camera,250.00, 998877665544,Limited Edition Lens 50mm,,5 776655443322,,, ※ 3行目の 776655443322,,, は更新項目がすべて空欄のため、プログラム内で SKIPPED 扱いとなります（実際の運用では事前に除外して構いません）。 Exponential Backoff（指数バックオフ）による Rate Limit 対策 Trading API の Rate Limit は「1日あたりの呼び出し回数上限（Daily Call Limit）」と「瞬間的な過負荷による 429 Too Many Requests」の両方が存在します。429 エラーが出た際に即座にリトライするとブロックされるため、前述の指数バックオフ機構が必須です。 # bulk_revise_from_csv.py import csv from config import eBayConfig from ebay_token_manager import eBayTokenManager from revise_item import revise_single_item, ReviseStatus def run_bulk_update(csv_file_path: str): config = eBayConfig() manager = eBayTokenManager(config.client_id, config.client_secret, config.refresh_token, config.env) stats = {"success": 0, "skipped": 0, "error": 0} with open(csv_file_path, mode='r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row in reader: item_id = row.get('item_id', '').strip() if not item_id: continue # CSVの空欄をフィルタリングし、意図せぬデータ消失(空タグ送信)を防ぐ updates = {k: v for k, v in row.items() if k != 'item_id' and v.strip() != ''} try: token = manager.get_token() result = revise_single_item(config, token, item_id, updates) if result.status == ReviseStatus.SKIPPED: print(f"[SKIPPED]: {item_id} (No updates)") stats["skipped"] += 1 else: warn_text = f" (Warnings: {len(result.warnings)})" if result.warnings else "" print(f"[SUCCESS]: {item_id}{warn_text}") stats["success"] += 1 except Exception as e: print(f"[ERROR] updating {item_id}: {e}") stats["error"] += 1 print("\n--- Batch Update Complete ---") print(f"Success: {stats['success']} | Skipped: {stats['skipped']} | Errors: {stats['error']}") if __name__ == "__main__": run_bulk_update("bulk_updates.csv") パフォーマンス・スケーリング視点 (深度) 【高頻度な「在庫数・価格」同期における注意と使い分け】 上記で作成した CSV 一括更新バッチは、「タイトル、説明、Item Specifics などのカタログ情報を一括改修する（例: 季節ごとのSEO対策）」 という目的においては最適です。 しかし、「自社倉庫の在庫が1個売れたから、eBay の在庫数もすぐ減らしたい」「15分おきに価格と在庫だけを他モールと同期させたい」といった高頻度のトランザクション同期に、この ReviseFixedPriceItem をループで回すのはシステム崩壊の原因になります。Revise は eBay 内部で商品カタログ全体を再インデックスするため負荷が大きいです。 ReviseFixedPriceItem: カタログ情報（タイトル等）の変更。日次・週次ベースのバッチ処理や夜間の分散実行。 ReviseInventoryStatus: 「価格」と「在庫数」だけを更新する場合。次回解説するこちらの軽量 API を使用します。 まとめ 本記事では、出品済み商品のデータを安全にメンテナンスするための ReviseFixedPriceItem の使い方を解説しました。 ベースライン: 変更差分だけを XML に含める「部分更新（Delta Update）」の原則。 深いポイント: 空欄によるデータ消失の防止、DeletedField の使い方、そして安全な XML 解析とステータス管理。 スケーリング: HTTP 429 を防ぐ指数バックオフ制御と、目的（カタログ改修 vs 高頻度在庫同期）に応じた API の使い分け。 次のステップ 今回、パフォーマンスの章で触れた「軽量な在庫同期 API」の正体について掘り下げます。 次回（#8）は、「ReviseInventoryStatusで複数商品の在庫・価格を高速同期する」 です。前回のバリエーション出品で仕込んだ InventoryTrackingMethod=SKU がここで真価を発揮します。多店舗展開の在庫連動を組むエンジニア必見の内容です！お楽しみに。 次の記事はこちら

前回の記事はこちら eBay Trading API：AddFixedPriceItemでバリエーション商品(Multi-SKU)を構築する極意 はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第6回です。 前回（#5）は、堅牢な単一商品（Single-SKU）の出品システムを構築しました。しかし、アパレル、靴、アクセサリーなどを扱う越境 EC においては、サイズや色ごとに異なる在庫と価格を持つ 「バリエーション商品（Multi-SKU）」 の出品が避けられません。 この記事で得られること: eBay API 屈指の複雑さを誇る <Variations> ブロックの完全な構造理解。 共通スペック（ItemSpecifics）と個別スペック（VariationSpecifics）の厳格な検証ロジック。 今後の在庫管理を劇的に楽にする InventoryTrackingMethod の SKU ベース化と OutOfStockControl の活用。 背景・なぜこれが重要か (Motivation) 「サイズ違いの商品を別々の ItemID で出品してはダメなのか？」という疑問を抱く方も多いですが、ビジネス的には絶対に避けるべきです。 別々に出品すると、eBay の出品手数料（Insertion Fee）が余分にかかるだけでなく、最も重要な 「販売履歴（Sales History）」 が分散してしまいます。バリエーションとして1つの ItemID に統合することで、「このページから累計1,000着売れている」という実績が合算され、Best Match（eBayの検索アルゴリズム）における SEO が飛躍的に向上します。 しかし、その代償としてバリエーション出品の XML ペイロードは非常に深く、ネストされた構造になります。1箇所でも整合性が崩れるとエラーが返るため、アーキテクチャの正確な理解が求められます。 基本的な使い方（ベースライン）：バリエーションXMLの全体像 バリエーション出品では、特に「全体定義（Set）」と「個別定義（Variation）」の二重構造を理解することが重要です。 <Item> <Title>Premium Cotton T-Shirt</Title> <ListingDuration>GTC</ListingDuration> <OutOfStockControl>true</OutOfStockControl> <InventoryTrackingMethod>SKU</InventoryTrackingMethod> <ItemSpecifics> <NameValueList> <Name>Brand</Name> <Value>AwesomeApparel</Value> </NameValueList> </ItemSpecifics> <Variations> <!-- バリエーション全体で使用する軸の定義 --> <VariationSpecificsSet> <NameValueList> <Name>Color</Name> <Value>Red</Value> <Value>Blue</Value> </NameValueList> <NameValueList> <Name>Size</Name> <Value>S</Value> <Value>M</Value> </NameValueList> </VariationSpecificsSet> <!-- 個別の在庫・価格・属性定義（SKU単位） --> <Variation> <SKU>TSHIRT-RED-S</SKU> <StartPrice>19.99</StartPrice> <Quantity>10</Quantity> <VariationSpecifics> <NameValueList><Name>Color</Name><Value>Red</Value></NameValueList> <NameValueList><Name>Size</Name><Value>S</Value></NameValueList> </VariationSpecifics> </Variation> <!-- 他のバリエーションも同様に続く --> </Variations> </Item> 【用語解説】 GTC (Good 'Til Cancelled) とは GTC は手動で終了するまで自動継続される出品期間の設定です。現在の eBay では、バリエーション商品は基本的に GTC で運用し、後述する OutOfStockControl と組み合わせて使います。 実務で躓く場面・深いポイント (Core) 1. InventoryTrackingMethod は絶対に「SKU」にせよ デフォルトでは、eBay の商品は ItemID で管理されます。しかし、バリエーションを持つ場合、後から「特定のバリエーションだけ在庫を更新したい」という際、ItemID ベースだと非常に複雑な XML を組むことになります。出品時に InventoryTrackingMethod を SKU に指定しておくことで、eBay 側の管理キーが SKU に切り替わり、在庫同期が劇的に簡単になります。 2. VariationPictures の制約（画像は1軸でしか切り替えられない） 「赤いSサイズの画像」「青いMサイズの画像」のように、すべての組み合わせに画像を設定したくなりますが、eBayの仕様上それは不可能です。バリエーション画像（<Pictures> ブロック）は、必ず1つの軸（通常は Color）にしか紐付けられません。これを破ると Error 21916587 が発生します。 3. NameとValueの厳密な文字列一致 <VariationSpecificsSet> で定義した「Color」や「Red」という文字列は、その下の <Variation> ブロック、さらには <Pictures> ブロックでも、大文字小文字・スペースに至るまで完全に一致している必要があります。 【注意】 バリエーション数の上限 eBay のバリエーション数には上限があります（一般的に1出品につき最大250 SKU）。これを超えると Error 21916284 となります。 主要なエラーコード早見表 エラーコード 概要 対策 21916587 画像の切り替え軸が不正 VariationPictures は必ず VariationSpecificsSet の1つの軸（Color等）と完全に一致させること。 21916284 バリエーション上限超過 1出品あたりの上限（通常250SKU）以内に分割して出品する。 21916286 組み合わせの不備 VariationSpecificsSet で定義した属性の組み合わせが、下部の Variation で矛盾している。 堅牢な実装：バリエーションXMLの動的ビルダーと事前検証 API エラーを防ぐために、文字列の不一致や定義の欠落を事前に検知しつつ XML を構築する Python 実装例を紹介します。 import html from typing import Dict def _validate_consistency(variation_data: Dict) -> None: """ VariationSpecificsSet と各Variationの整合性を検証する """ defined_aspects = { name: set(values) for name, values in variation_data['aspects'].items() } defined_aspect_names = set(defined_aspects.keys()) for var in variation_data['skus']: var_aspect_names = set(var['specifics'].keys()) # 1. SKUの値がSetに存在するか（順方向チェック） for spec_name, spec_value in var['specifics'].items(): if spec_name not in defined_aspects: raise ValueError(f"SKU '{var['sku']}' の軸 '{spec_name}' は未定義です。") if spec_value not in defined_aspects[spec_name]: raise ValueError(f"SKU '{var['sku']}' の値 '{spec_value}' が定義済リストにありません。") # 2. Setの全軸がSKUに揃っているか（逆方向チェック） # ※eBayはスパースなバリエーション（全組み合わせの一部欠落）を許容しているため、 # 全組み合わせの網羅チェックは行いません。 missing = defined_aspect_names - var_aspect_names if missing: raise ValueError(f"SKU '{var['sku']}' に軸 {missing} の指定がありません。") def build_variations_xml(variation_data: Dict) -> str: """ 辞書データから堅牢な <Variations> ブロックのXML文字列を生成する """ _validate_consistency(variation_data) # 1. 全体定義 (VariationSpecificsSet) の構築 set_tags = "" for spec_name, spec_values in variation_data['aspects'].items(): val_tags = "".join([f"<Value>{html.escape(str(v))}</Value>" for v in spec_values]) set_tags += f""" <NameValueList> <Name>{html.escape(str(spec_name))}</Name> {val_tags} </NameValueList> """ # 2. 個別定義 (Variation) の構築 variation_tags = "" for var in variation_data['skus']: spec_tags = "".join([f"<NameValueList><Name>{html.escape(str(k))}</Name><Value>{html.escape(str(v))}</Value></NameValueList>" for k, v in var['specifics'].items()]) variation_tags += f""" <Variation> <SKU>{html.escape(var['sku'])}</SKU> <StartPrice>{var['price']}</StartPrice> <Quantity>{var['quantity']}</Quantity> <VariationSpecifics>{spec_tags}</VariationSpecifics> </Variation> """ return f""" <Variations> <VariationSpecificsSet>{set_tags}</VariationSpecificsSet> {variation_tags} </Variations> """ パフォーマンス・スケーリング視点 (深度) GTC と OutOfStockControl の魔法 実務でバリエーションを扱う際、在庫が0になった SKU は通常 Listing 自体が終了してしまいます。これを防ぐために、eBay では <OutOfStockControl>true</OutOfStockControl> という強力なフラグが用意されています。 このフラグを有効にしておくと、ある SKU の Quantity を 0 にしても出品自体は終了せず、検索結果では「Out of stock」として非表示になるだけで販売履歴（SEOパワー）を維持し続けます。商品が再入荷した際に Quantity を 1 以上に戻せば、即座に販売が再開されます。大規模セラーにとって、この機能の有効化は必須の戦略です。 まとめ 本記事では、アパレル等の販売に欠かせないバリエーション商品（Multi-SKU）の出品構造を解説しました。 ベースライン: VariationSpecificsSet と Variation の二重構造を理解する。 深いポイント: 軸の不一致による API エラーを、プログラム側で事前に検知する。 スケーリング: InventoryTrackingMethod=SKU と OutOfStockControl を組み合わせ、SEO を維持しつつ在庫管理を効率化する。 次のステップ 単一商品とバリエーション商品の「出品」が完了しました。しかし、EC の実務において出品は「始まり」に過ぎません。 次回（#7）からは、新しいフェーズである 【Trading API - 在庫管理】 に突入します。まずは 「ReviseFixedPriceItemで在庫数と価格を高速に同期する」 方法について解説します！お楽しみに。 次の記事はこちら

前回の記事はこちら eBay Trading API：AddFixedPriceItemで単一商品(Single-SKU)を安全に出品する はじめに 本記事は、全42回にわたる「eBay API 実践ガイド」の第5回です。 これまでの連載で、OAuth 認証基盤、Sandbox 環境、配送などのメタデータ、そして画像（EPS）のアップロード機能が整いました。今回はこれらをすべて結合し、ついに eBay に商品を出品（Listing） します。 この記事で得られること: AddFixedPriceItem を用いた固定価格商品（即決）の XML ペイロード構造の理解。 ネットワークエラーによる「二重出品」を防ぐ、UUIDの永続化と冪等性（Idempotency）の確保。 実務で致命傷になる XML Injection の防止と、適切なデータ型バリデーション手法。 背景・なぜこれが重要か (Motivation) eBay に商品を出品する際、現在は Inventory API（REST）というモダンな選択肢もあります。しかし、越境 EC の実務において、依然として Trading API の AddFixedPriceItem が広く使われているのには理由があります。 それは 「圧倒的な即時性と柔軟性」 です。REST API が内部的に非同期処理を多用するのに対し、Trading API はリクエストを送った瞬間に ItemID が発行され、即座にサイトに反映されます。 ただし、AddFixedPriceItem の XML は巨大かつ複雑です。必須項目が1つでも欠ければエラー弾きに遭うため、「どのブロックが何のために必要なのか」をアーキテクチャの視点から理解することが、堅牢な出品システム構築の鍵となります。 基本的な使い方（ベースライン）：出品XMLの全体像 AddFixedPriceItem のリクエストは、大きく以下のブロックに分かれています。 基本情報: タイトル、カテゴリ、価格、コンディション、数量。 ポリシー情報: 返品・支払い・配送のポリシー（現在主流の Business Policies を使用するか、レガシーな個別指定を行うか）、発送までの日数。 画像情報: 前回取得した EPS の URL。 商品詳細設定 (Item Specifics): ブランドやサイズなどの必須スペック情報。 これらを愚直に XML で組むと以下のようになります（一部省略）。 <?xml version="1.0" encoding="utf-8"?> <AddFixedPriceItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <Item> <Title>Sample Product Title</Title> <Description><![CDATA[Detailed item description goes here.]]></Description> <PrimaryCategory> <CategoryID>12345</CategoryID> </PrimaryCategory> <StartPrice currencyID="USD">99.99</StartPrice> <ConditionID>1000</ConditionID> <Country>JP</Country> <Currency>USD</Currency> <DispatchTimeMax>3</DispatchTimeMax> <ListingDuration>GTC</ListingDuration> </Item> </AddFixedPriceItemRequest> 実務で躓く場面・深いポイント (Core) ここからは、実稼働するシステムを組む上で絶対に避けては通れない、実務レベルの落とし穴とその解決策を解説します。 1. 二重出品を防ぐ魔法のキー「UUID」とその「永続化」 出品 API は処理が重いため、eBay 側のサーバー都合やネットワークの瞬断でタイムアウトが発生することがあります。この時、プログラム側で「失敗した」と判定してリトライ（再送）をかけると、実は eBay 側では最初の処理が成功しており、同じ商品が二重に出品されてしまう 悲劇が起こります。 これを防ぐのが <UUID> タグです。しかし、「リクエストの直前で UUID を生成する」コードは絶対に書いてはいけません。 その直後にクラッシュした場合、UUID はメモリから消失し、再送時に新しい UUID が生成されて冪等性が失われるからです。 【ベストプラクティス】: API をコールする前に、必ず UUID（32桁の16進数）を生成し、自社のデータベース（DB）に出品予定データと一緒に 「保存（永続化）」 してください。リトライ時は常にその DB の UUID を読み込んで送信します。 2. XML Injection 防止と数値バリデーションの使い分け Title の値に & や < が含まれている（例: "Canon AE-1 & AE-1P"）と、そのまま f-string で埋め込んだ瞬間に XML が壊れます。プレーンテキストは html.escape() で必ずエスケープし、HTML タグを含む Description は <![CDATA[ ... ]]> で囲む必要があります。 一方で、価格や数量といった数値フィールドに対してエスケープ処理を行うのはアンチパターンです（フォーマットエラーの原因になります）。数値フィールドや各種プロファイル ID は事前に Python 側で厳格な型チェック（バリデーション）を行うのが正解です。 3. Business Policies (ビジネスポリシー) の必須化 多くのアカウントでは現在、支払い・返品・発送の設定をまとめた Business Policies の利用が強制されています。これらが強制されているアカウントで古い形式の <ReturnPolicy> などを送ると、Error 21919187 で弾かれます。代わりに <SellerProfiles> を使用します。 【メモ】: 自分のアカウントが Business Policies 有効か確認するには？ eBay の Web サイト（Seller Hub）から確認するか、GetUser API を叩くことで判定可能です。現在新規作成されたセラーアカウントの多くはデフォルトで有効化されています。 堅牢な実装：UUID と エスケープ処理を組み込んだ出品スクリプト 上記の実務的な課題をすべてクリアした安全な出品関数を実装します。 # add_fixed_price_item.py import requests import xml.etree.ElementTree as ET import uuid import html from config import eBayConfig from ebay_token_manager import eBayTokenManager # 【注意】: idempotency_key (UUID) は必ずAPI呼び出し前にDBへ保存済みのものを渡してください。 # Noneを渡すとクラッシュ時に冪等性が失われます（テスト用途のみ許容）。 def list_single_item(config: eBayConfig, token: str, item_data: dict, eps_urls: list, idempotency_key: str = None) -> str: """ 堅牢な単一商品出品処理。 """ if idempotency_key is None: idempotency_key = uuid.uuid4().hex # 【ポイント①】: 数値フィールドとIDの事前バリデーション（エスケープの代わり） try: price = float(item_data['price']) quantity = int(item_data['quantity']) category_id = int(item_data['category_id']) condition_id = int(item_data['condition_id']) dispatch_time = int(item_data.get('dispatch_time', 3)) shipping_profile_id = int(item_data['shipping_profile_id']) return_profile_id = int(item_data['return_profile_id']) payment_profile_id = int(item_data['payment_profile_id']) except (ValueError, TypeError, KeyError) as e: raise ValueError(f"Invalid or missing data in item_data: {e}") headers = { "X-EBAY-API-CALL-NAME": "AddFixedPriceItem", "X-EBAY-API-SITEID": "0", "X-EBAY-API-COMPATIBILITY-LEVEL": "1323", "X-EBAY-API-IAF-TOKEN": token, "Content-Type": "text/xml" } # 画像ブロックの動的生成 picture_tags = "".join([f"<PictureURL>{html.escape(url)}</PictureURL>" for url in eps_urls]) # 【ポイント②】: テキストフィールドの XML Injection 防止 (html.escape) specifics_tags = "" for name, value in item_data.get("item_specifics", {}).items(): specifics_tags += f""" <NameValueList> <Name>{html.escape(str(name))}</Name> <Value>{html.escape(str(value))}</Value> </NameValueList> """ # XML ペイロードの組み立て xml_payload = f"""<?xml version="1.0" encoding="utf-8"?> <AddFixedPriceItemRequest xmlns="urn:ebay:apis:eBLBaseComponents"> <ErrorLanguage>en_US</ErrorLanguage> <WarningLevel>High</WarningLevel> <Item> <Title>{html.escape(item_data['title'])}</Title> <Description><![CDATA[{item_data['description']}]]></Description> <PrimaryCategory> <CategoryID>{category_id}</CategoryID> </PrimaryCategory> <StartPrice currencyID="USD">{price}</StartPrice> <Quantity>{quantity}</Quantity> <ConditionID>{condition_id}</ConditionID> <Country>JP</Country> <Currency>USD</Currency> <DispatchTimeMax>{dispatch_time}</DispatchTimeMax> <ListingDuration>GTC</ListingDuration> <UUID>{idempotency_key}</UUID> <PictureDetails> {picture_tags} </PictureDetails> <ItemSpecifics> {specifics_tags} </ItemSpecifics> <SellerProfiles> <SellerShippingProfile> <ShippingProfileID>{shipping_profile_id}</ShippingProfileID> </SellerShippingProfile> <SellerReturnProfile> <ReturnProfileID>{return_profile_id}</ReturnProfileID> </SellerReturnProfile> <SellerPaymentProfile> <PaymentProfileID>{payment_profile_id}</PaymentProfileID> </SellerPaymentProfile> </SellerProfiles> </Item> </AddFixedPriceItemRequest> """ # 【ポイント③】: 無限待機を防ぐため timeout を明示。タイムアウト時はDBのUUIDで安全にリトライする response = requests.post( config.trading_api_url, headers=headers, data=xml_payload.encode('utf-8'), timeout=30 ) response.raise_for_status() # レスポンス解析 namespace = {'ns': 'urn:ebay:apis:eBLBaseComponents'} root = ET.fromstring(response.text) ack_node = root.find('ns:Ack', namespace) ack = ack_node.text if ack_node is not None else "" # 【ポイント④】: Warningログの記録と、複数エラーの網羅的キャッチ if ack == 'Warning': warnings = [ e.find('ns:LongMessage', namespace).text if e.find('ns:LongMessage', namespace) is not None else "" for e in root.findall('ns:Errors', namespace) if e.find('ns:SeverityCode', namespace) is not None and e.find('ns:SeverityCode', namespace).text == 'Warning' ] print(f"[WARNING] Listing succeeded with warnings: {warnings}") # 本番ではloggingモジュールを使用 if ack not in ['Success', 'Warning']: error_nodes = root.findall('ns:Errors', namespace) errors = [ { "code": e.find('ns:ErrorCode', namespace).text if e.find('ns:ErrorCode', namespace) is not None else "", "severity": e.find('ns:SeverityCode', namespace).text if e.find('ns:SeverityCode', namespace) is not None else "", "message": e.find('ns:LongMessage', namespace).text if e.find('ns:LongMessage', namespace) is not None else "", } for e in error_nodes if e.find('ns:SeverityCode', namespace) is not None and e.find('ns:SeverityCode', namespace).text == 'Error' ] if errors: raise Exception(f"Listing Failed with {len(errors)} errors: {errors}") # 成功時の ItemID を抽出 item_id_node = root.find('ns:ItemID', namespace) if item_id_node is None: raise Exception("Success returned but ItemID is missing.") return item_id_node.text パフォーマンス・スケーリング視点 (深度) 実務において、毎月数千〜数万件の出品を行う場合、f-string での巨大な XML 構築はメンテナンス性が著しく低下します。 【テンプレートエンジン（Jinja2）の導入】 スケーリングの第一歩として、XML の雛形を外部ファイルに切り離し、Jinja2 を使って Python コードから変数を流し込むアーキテクチャに移行することを強く推奨します。autoescape=True を使えば、XML Injection の心配も無くなります。 <Title>{{ title | e }}</Title> <Description><![CDATA[{{ description | safe }}]]></Description> <UUID>{{ idempotency_key }}</UUID> 【セキュリティ注意】: safe フィルタは CDATA ブロック内でのみ使用してください。autoescape の保護を無効化するため、プレーンテキストのフィールド（Title等）に使用すると XML Injection の脆弱性を生みます。 # Python側の呼び出し例 from jinja2 import Environment, FileSystemLoader env = Environment(loader=FileSystemLoader('templates'), autoescape=True) template = env.get_template('listing.xml.j2') # item_data辞書を展開してXMLを生成 xml_payload = template.render(**item_data, idempotency_key=db_saved_uuid) これにより、「新しいポリシー ID に一斉に変更したい」「説明文の HTML デザインを一新したい」といったビジネス要求に対し、Python のロジックに触れることなく、テンプレートファイルの修正だけで安全に対応できるようになります。 まとめ 本記事では、eBay 開発における第一の到達点である「商品の出品」を実装しました。 ベースライン: AddFixedPriceItem で要求される複雑な XML ペイロードの基本構造。 深いポイント: DB永続化を前提とした UUID と timeout による冪等性の確保。XML Injection 対策と複数エラー/警告のハンドリング。 スケーリング: f-string から Jinja2 テンプレートエンジン移行への具体的なアプローチ。 これであなたは、プログラム経由で eBay に安全かつ堅牢に商品カタログを展開できるようになりました。 次のステップ 単一商品の出品に成功したら、次なる壁はアパレルや靴などで必須となる 「バリエーション出品 (Multi-SKU)」 です。 次回（#6）は、親商品（Parent）と子商品（Child/Variation）を一つの AddFixedPriceItem リクエストにまとめ、サイズや色ごとに異なる在庫と価格を管理する高度な XML の構築方法を解説します。お楽しみに！ 次の記事はこちら