eBay Inventory Mapping API + Inventory API：AI推奨結果をInventory APIに渡して高品質な出品を自動作成する

前回の記事はこちら

【連載#20】eBay Inventory Mapping API + Inventory API：AI推奨結果をInventory APIに渡して高品質な出品を自動作成する

はじめに

本記事は、全42回にわたる「eBay API 実践ガイド」の第20回です。

前回（#19）では、Inventory Mapping APIにサブミットしたタスクをポーリングで監視し、タスク完了後にListingPreviewオブジェクト群（AI推奨のカテゴリ・タイトル・アスペクト・説明文・SKU・mappingReferenceId）を取得しました。これでAIの推奨結果が手元に揃った状態です。

本記事では、その「プレビュー結果を眺めるだけ」で終わらせず、Inventory APIの createOrReplaceInventoryItem へ実際に渡して出品アイテムレコードを自動作成する——Inventory Mapping API三部作の「仕上げ」を実装します。

背景・なぜこれが重要か (Motivation)

「推奨結果をコピペして手動で出品すればいいのでは？」

Inventory Mapping APIを初めて使う開発者が必ず抱く疑問です。数件であれば確かに手動コピーも現実的ですが、数百・数千件のカタログを毎日処理するECオペレーションでは、すぐに壁に当たります。

プログラム化が必要な理由は3つあります。「スケーラビリティ」——Mapping APIは1タスクで最大500件をバッチ処理できますが、その結果を手動で出品しては処理能力を活かせません。「データ一貫性」——コピペによるタイプミスや項目の抜けが出品エラーや検索品質低下に直結します。「トレーサビリティ」——mappingReferenceIdをどのSKUに紐付けたか記録しなければ、AI推薦がビジネスにどれだけ貢献したかを後から検証できません。

この3課題を同時解決するのが、本記事で実装する「AI推奨→Inventory API自動連携パイプライン」です。

基本的な使い方（ベースライン）：ListingPreviewからcreateOrReplaceInventoryItemへ

まず最小限の動作確認コードから始めます。Inventory APIの createOrReplaceInventoryItem はPUT /sell/inventory/v1/inventory_item/{sku} として呼び出します。リクエストボディの主要フィールドを確認しましょう。

• product.title: 商品タイトル（最大80文字）。
• product.description: 商品説明文（基本的なHTMLタグ可）。
• product.aspects: Item Specificsの辞書形式。キー＝アスペクト名（最大40文字）、値＝文字列の配列（各値最大50文字）。
• product.imageUrls: 商品画像URLの配列（HTTPS必須、offer公開前に最低1枚必要）。
• condition: 商品コンディションの列挙値（例: NEW, USED_EXCELLENT など）。
• availability.shipToLocationAvailability.quantity: 利用可能在庫数。

なお、createOrReplaceInventoryItemはPUTリクエストで、新規作成・更新ともにHTTP 204（No Content）を返します。このAPIだけでは出品は完了せず、その後createOrReplaceOffer → publishOfferが必要です（Offer APIシリーズは第21回以降で解説予定）。

以下は最小限の実装例です。

# baseline_inventory_upload.py
import requests
from typing import Any


def _convert_aspects_to_rest(gql_aspects: list[dict]) -> dict[str, list[str]]:
    """
    GraphQL ListingPreviewProductAspect配列をInventory API REST形式の辞書に変換する。

    GraphQL形式（第19回で取得した構造）:
      [{"localizedAspectName": "Brand", "value": [{"localizedValue": "Nike"}]}, ...]
    REST形式（createOrReplaceInventoryItemが期待する構造）:
      {"Brand": ["Nike"], "Color": ["Black"], ...}
    """
    result: dict[str, list[str]] = {}
    for aspect in (gql_aspects or []):
        name = aspect.get("localizedAspectName") or aspect.get("name", "")
        if not name:
            continue
        raw_values = aspect.get("value") or aspect.get("values") or []
        if raw_values and isinstance(raw_values[0], dict):
            values = [v.get("localizedValue", "") for v in raw_values if v.get("localizedValue")]
        else:
            values = [str(v) for v in raw_values if v]
        if values:
            result[name] = values
    return result


def create_inventory_item_from_preview(
    preview: dict[str, Any],
    access_token: str,
    condition: str = "NEW",
    quantity: int = 10,
) -> dict:
    """ListingPreviewデータをInventory APIに渡して在庫アイテムを作成する（最小実装）。"""
    sku = preview["sku"]
    images = preview.get("images", [])
    image_urls = [img if isinstance(img, str) else img.get("value", "") for img in images]
    body = {
        "product": {
            "title": preview.get("title", ""),
            "description": preview.get("description", ""),
            "aspects": _convert_aspects_to_rest(preview.get("aspects", [])),
            "imageUrls": image_urls,
        },
        "condition": condition,
        "availability": {
            "shipToLocationAvailability": {"quantity": quantity}
        },
    }
    url = f"https://api.ebay.com/sell/inventory/v1/inventory_item/{sku}"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "Content-Language": "en-US",
    }
    resp = requests.put(url, json=body, headers=headers, timeout=30)
    resp.raise_for_status()
    return {"sku": sku, "http_status": resp.status_code}

実務で躓く場面・深いポイント (Core)

ベースラインコードは動きますが、本番に乗せると必ず遭遇する3つの落とし穴を解説します。

1. GraphQLのaspects形式とRESTのaspects形式——データ構造の違いと安全な変換

第19回で取得したListingPreview.aspectsはGraphQLの「配列形式」です。一方、Inventory API REST側が期待するproduct.aspectsは「辞書（オブジェクト）形式」です。この変換を誤ると全商品でHTTP 400が返り続けます。

GraphQL形式（ListingPreviewProductAspect配列）の例:

# GraphQL レスポンス（第19回で取得済み）
gql_aspects = [
    {"localizedAspectName": "Brand", "value": [{"localizedValue": "Nike"}]},
    {"localizedAspectName": "Color",
     "value": [{"localizedValue": "Black"}, {"localizedValue": "White"}]},
    {"localizedAspectName": "Size",  "value": [{"localizedValue": "M"}]},
]

REST API形式（createOrReplaceInventoryItemが受け付ける形式）の例:

# Inventory API REST リクエストボディ内の product.aspects
rest_aspects = {
    "Brand": ["Nike"],
    "Color": ["Black", "White"],
    "Size": ["M"],
}

変換自体は単純ですが、見落としやすい制限が2点あります。アスペクト名の最大長は40文字、値の最大長は1つにつき50文字です。Mapping APIの推奨値がこの制限を超えることは稀ですが、超えた場合はInventory APIがerrorId: 25044または25045を返します。また、localizedValueが空文字のエントリをそのまま送るとバリデーションエラーになるため、変換時に除外処理を入れておくことが重要です。

2. mappingReferenceIdのトレーサビリティ管理を怠ると後でAI推薦効果を検証できなくなる

eBay公式ドキュメントは「Inventory Mapping APIの推奨結果を使って出品を作成・更新する場合、mappingReferenceIdをその出品に含めること」と明記しています。これはトレーサビリティとAI推薦精度の効果測定のためです。

本記事執筆時点（Inventory API v1.18.5）において、createOrReplaceInventoryItemのリクエストボディにmappingReferenceId専用のフィールドは確認できませんでした。そのため、SKUとmappingReferenceIdの対応を内部DB（SQLite等）に永続化する設計を推奨します。憶測で存在しないフィールド名を指定すると、APIが400エラーを返す原因になるため注意してください。

# sku_mapping_store.py — mappingReferenceId を SQLite で管理する
import sqlite3
from datetime import datetime


def init_mapping_store(db_path: str = "ebay_mapping.db") -> None:
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS sku_mapping_log (
            id               INTEGER PRIMARY KEY AUTOINCREMENT,
            sku              TEXT NOT NULL,
            mapping_ref_id   TEXT NOT NULL,
            category_id      TEXT,
            created_at       TEXT NOT NULL,
            UNIQUE(sku, mapping_ref_id)
        )
    """)
    conn.commit()
    conn.close()


def save_mapping_reference(
    sku: str, mapping_ref_id: str,
    category_id: str | None = None,
    db_path: str = "ebay_mapping.db",
) -> None:
    """SKUとmappingReferenceIdの対応をDBに保存する。"""
    conn = sqlite3.connect(db_path)
    conn.execute(
        "INSERT OR IGNORE INTO sku_mapping_log"
        " (sku, mapping_ref_id, category_id, created_at)"
        " VALUES (?, ?, ?, ?)",
        (sku, mapping_ref_id, category_id, datetime.utcnow().isoformat()),
    )
    conn.commit()
    conn.close()

このテーブルを持つことで、後日「AI推薦カテゴリで出品したSKUとそうでないSKUの売上CVRを比較する」といった分析が可能になります。mappingReferenceIdはListingPreview1件ごとに発行される個別IDなので、SKUと1対1で紐付けて管理するのが正しい使い方です。

3. COMPLETED_WITH_ERRORの商品を誤って自動出品してしまうリスク

第19回で解説した通り、タスク結果にはCOMPLETED（全成功）とCOMPLETED_WITH_ERROR（一部エラーあり）の2ステータスがあります。COMPLETED_WITH_ERRORのタスクには成功した商品と失敗した商品が混在しています。

よくある実装ミスは、タスクステータスを確認せずlistingPreviewsが空でなければ全件をInventory APIに渡してしまうパターンです。エラーとなった商品のListingPreviewは推奨データが不完全な状態（titleが空・aspectsが0件など）になっている場合があり、品質の低い出品が自動生成されるリスクがあります。

安全策は各ListingPreviewに対して「最低品質チェック」を実施してからAPIを呼ぶことです。チェック項目: title が非空か、aspects が1件以上あるか、imageUrls が1件以上あるか、mappingReferenceId が存在するか。不備があればスキップして警告ログを残し、人手確認に回します。

# ──────────────────────────────────────────────────
# エラーコード早見表 (createOrReplaceInventoryItem)
# ──────────────────────────────────────────────────

# HTTP 400 | errorId: 25002
# 内容: SKU値が無効（使用禁止文字を含む、または長すぎる）
# 対処: SKUは英数字・ハイフン・アンダースコアのみ。最大50文字に収める。

# HTTP 400 | errorId: 25013
# 内容: conditionの値が対象カテゴリで無効
# 対処: Metadata API getItemConditionPoliciesで
#        カテゴリ別の有効なcondition一覧を事前取得して照合する。

# HTTP 400 | errorId: 25044 / 25045
# 内容: アスペクト名が40文字超 / アスペクト値が50文字超
# 対処: _convert_aspects内でname[:40]・value[:50]に切り詰めるか、
#        超過したアスペクトをスキップしてログに残す。

# HTTP 400 | errorId: 25709
# 内容: カテゴリで必須のItem Specificが欠落している
# 対処: Taxonomy API getItemAspectsForCategoryで必須アスペクトを確認し、
#        Mapping API推奨結果に含まれない場合はデフォルト値で補完する。

# HTTP 401 | errorId: 1001
# 内容: アクセストークンが無効または期限切れ
# 対処: OAuthリフレッシュロジックを実装し、401時に自動リフレッシュ&リトライする。

堅牢な実装：E2Eパイプライン（ポーリング結果取得→変換→Inventory API登録→トレーサビリティ保存）

ここまでの知識を統合した、本番運用に耐えるE2Eパイプラインを示します。型アノテーション・docstring・例外処理・入力バリデーションをすべて含みます。

# inventory_mapping_pipeline.py
# Inventory Mapping API 三部作（#18-#20）の仕上げ実装
import logging
import requests
from dataclasses import dataclass, field
from typing import Any

from sku_mapping_store import init_mapping_store, save_mapping_reference

logger = logging.getLogger(__name__)


@dataclass
class ListingPreviewValidationError(Exception):
    """ListingPreviewの品質チェック失敗を表す例外。"""
    sku: str
    reasons: list[str] = field(default_factory=list)

    def __str__(self) -> str:
        return f"SKU={self.sku} バリデーション失敗: {self.reasons}"


class InventoryMappingPipeline:
    """
    Inventory Mapping APIの推奨結果をInventory APIに投入するE2Eパイプライン。

    Attributes:
        access_token: eBay OAuthアクセストークン。
        default_condition: conditionのデフォルト値。SKUごとに上書き可能。
        default_quantity: 初回在庫数のデフォルト値。
        db_path: mappingReferenceId管理用SQLiteのパス。
    """

    BASE_URL = "https://api.ebay.com/sell/inventory/v1"

    def __init__(
        self,
        access_token: str,
        default_condition: str = "NEW",
        default_quantity: int = 10,
        db_path: str = "ebay_mapping.db",
    ) -> None:
        self.access_token = access_token
        self.default_condition = default_condition
        self.default_quantity = default_quantity
        self.db_path = db_path
        init_mapping_store(db_path)

    # ----------------------------------------------------------------
    # パブリックメソッド
    # ----------------------------------------------------------------

    def process_listing_previews(
        self,
        previews: list[dict[str, Any]],
        condition_map: dict[str, str] | None = None,
        quantity_map: dict[str, int] | None = None,
    ) -> dict[str, list[str]]:
        """
        ListingPreviewのリストを受け取り、一括でInventory APIに登録する。

        Args:
            previews: 第19回ポーリング結果のListingPreview配列。
            condition_map: SKU→conditionの上書き辞書。
            quantity_map: SKU→在庫数の上書き辞書。

        Returns:
            {"succeeded": [sku,...], "failed": [sku,...], "skipped": [sku,...]}
        """
        results: dict[str, list[str]] = {"succeeded": [], "failed": [], "skipped": []}
        for preview in previews:
            sku = preview.get("sku", "UNKNOWN_SKU")
            try:
                self._validate_preview(preview)
                condition = (condition_map or {}).get(sku, self.default_condition)
                quantity = (quantity_map or {}).get(sku, self.default_quantity)
                self._upload_inventory_item(preview, condition, quantity)
                self._save_traceability(preview)
                results["succeeded"].append(sku)
                logger.info("Inventory item created: sku=%s", sku)
            except ListingPreviewValidationError as e:
                logger.warning("品質チェック不合格のためスキップ: %s", e)
                results["skipped"].append(sku)
            except requests.HTTPError as e:
                logger.error("HTTP error sku=%s: %s", sku, e.response.text)
                results["failed"].append(sku)
            except Exception as e:
                logger.exception("予期しないエラー sku=%s: %s", sku, e)
                results["failed"].append(sku)
        return results

    # ----------------------------------------------------------------
    # プライベートメソッド
    # ----------------------------------------------------------------

    def _validate_preview(self, preview: dict[str, Any]) -> None:
        """COMPLETED_WITH_ERROR商品の混入を防ぐ最低品質チェック。"""
        sku = preview.get("sku", "UNKNOWN_SKU")
        reasons: list[str] = []
        if not preview.get("title", "").strip():
            reasons.append("title が空")
        if not preview.get("aspects"):
            reasons.append("aspects が空（Item Specificsなし）")
        if not preview.get("images"):
            reasons.append("images が空（画像URLなし）")
        if not preview.get("mappingReferenceId"):
            reasons.append("mappingReferenceId が欠落")
        if reasons:
            raise ListingPreviewValidationError(sku=sku, reasons=reasons)

    def _convert_aspects(self, gql_aspects: list[dict]) -> dict[str, list[str]]:
        """GraphQL aspects配列 → REST辞書形式。名前40文字・値50文字上限を強制。"""
        result: dict[str, list[str]] = {}
        for aspect in (gql_aspects or []):
            name = (aspect.get("localizedAspectName") or aspect.get("name", "")).strip()
            if not name:
                continue
            if len(name) > 40:
                logger.warning("アスペクト名が40文字超のためスキップ: %r", name)
                continue
            raw_values = aspect.get("value") or aspect.get("values") or []
            if raw_values and isinstance(raw_values[0], dict):
                values = [
                    v.get("localizedValue", "")[:50]
                    for v in raw_values if v.get("localizedValue")
                ]
            else:
                values = [str(v)[:50] for v in raw_values if v]
            if values:
                result[name] = values
        return result

    def _upload_inventory_item(
        self, preview: dict[str, Any], condition: str, quantity: int
    ) -> None:
        """createOrReplaceInventoryItemを呼び出す。"""
        sku = preview["sku"]
        images = preview.get("images", [])
        image_urls = [
            img if isinstance(img, str) else img.get("value", "")
            for img in images if img
        ]
        body = {
            "product": {
                "title": preview.get("title", ""),
                "description": preview.get("description", ""),
                "aspects": self._convert_aspects(preview.get("aspects", [])),
                "imageUrls": image_urls,
            },
            "condition": condition,
            "availability": {
                "shipToLocationAvailability": {"quantity": quantity}
            },
        }
        url = f"{self.BASE_URL}/inventory_item/{sku}"
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json",
            "Content-Language": "en-US",
        }
        resp = requests.put(url, json=body, headers=headers, timeout=30)
        resp.raise_for_status()

    def _save_traceability(self, preview: dict[str, Any]) -> None:
        """SKUとmappingReferenceIdの対応をDBに永続化する。"""
        cat = preview.get("category") or {}
        category_id = cat.get("categoryId") or cat.get("id") if isinstance(cat, dict) else None
        save_mapping_reference(
            sku=preview.get("sku", ""),
            mapping_ref_id=preview.get("mappingReferenceId", ""),
            category_id=category_id,
            db_path=self.db_path,
        )


# ====================================================
# 使用例
# ====================================================
if __name__ == "__main__":
    import json
    # 第19回ポーリングスクリプトが出力したJSONを読み込む
    with open("listing_previews.json") as f:
        previews = json.load(f)

    pipeline = InventoryMappingPipeline(
        access_token="v^1.1#i^1#...",  # OAuthトークン
        default_condition="NEW",
        default_quantity=5,
    )
    result = pipeline.process_listing_previews(
        previews=previews,
        condition_map={"SKU-12345": "USED_EXCELLENT"},  # SKUごとの上書き
    )
    print(f"成功: {len(result['succeeded'])}件")
    print(f"スキップ: {len(result['skipped'])}件")
    print(f"失敗: {len(result['failed'])}件")

このパイプラインは3つの責務を明確に分離しています。_validate_preview（品質チェック）、_convert_aspects（データ変換）、_save_traceability（トレーサビリティ）。各メソッドをPytestで個別にユニットテストできるため、CI組み込みも容易です。

パフォーマンス・スケーリング視点 (深度)

数百〜数千件の商品を定期バッチで処理する場合、1件ずつcreateOrReplaceInventoryItemを呼ぶアーキテクチャではすぐに限界が来ます。eBayが提供する一括投入エンドポイントと再処理キューを組み合わせて対処します。

bulkCreateOrReplaceInventoryItemで最大25件を一括投入する

Inventory APIには POST /sell/inventory/v1/bulk_create_or_replace_inventory_item という一括エンドポイントがあり、1リクエストで最大25件を処理できます。単発PUTと比較してAPI呼び出し回数を最大1/25に削減でき、スループットが大幅に向上します。

# bulk_upload.py — 25件単位のバッチ投入
import requests
from itertools import islice

BULK_URL = "https://api.ebay.com/sell/inventory/v1/bulk_create_or_replace_inventory_item"
CHUNK_SIZE = 25


def _chunk(lst: list, size: int):
    it = iter(lst)
    while chunk := list(islice(it, size)):
        yield chunk


def bulk_create_inventory_items(
    inventory_bodies: list[dict],
    access_token: str,
) -> list[dict]:
    """
    skuキーを含む各SKU分のリクエストボディを25件ずつ一括投入する。

    Returns:
        各チャンクのAPIレスポンスのリスト。
        responses[i]["responses"][j]["statusCode"] で個別の成否を確認すること。
    """
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "Content-Language": "en-US",
    }
    all_responses = []
    for chunk in _chunk(inventory_bodies, CHUNK_SIZE):
        resp = requests.post(
            BULK_URL, json={"requests": chunk},
            headers=headers, timeout=60,
        )
        resp.raise_for_status()
        all_responses.append(resp.json())
    return all_responses

失敗SKUの再処理キュー設計

バッチ処理中の部分失敗に対して、「その場でリトライ」よりも「失敗キューに積んで後で処理する」アーキテクチャが堅牢です。以下はJSONLファイルベースの軽量なキュー実装例です。

# retry_queue.py — 失敗SKUをJSONLファイルで管理する
import json
from pathlib import Path
from datetime import datetime

RETRY_QUEUE = Path("retry_queue.jsonl")


def enqueue_failed(sku: str, error_msg: str, body: dict) -> None:
    """失敗したSKUを再処理キューに追記する。"""
    entry = {
        "sku": sku,
        "error": error_msg,
        "body": body,
        "queued_at": datetime.utcnow().isoformat(),
        "retry_count": 0,
    }
    with RETRY_QUEUE.open("a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")


def drain_retry_queue(access_token: str, max_retries: int = 3) -> None:
    """キューの失敗SKUを順次再処理し、成功分はキューから削除する。"""
    if not RETRY_QUEUE.exists():
        return
    remaining = []
    for line in RETRY_QUEUE.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        entry = json.loads(line)
        if entry["retry_count"] >= max_retries:
            # 上限超過 → 人手確認リストに転記（実装省略）
            continue
        # ここで createOrReplaceInventoryItem を再呼び出し（実装省略）
        # 失敗した場合のみ remaining に積む
        entry["retry_count"] += 1
        remaining.append(entry)
    RETRY_QUEUE.write_text(
        "\n".join(json.dumps(e, ensure_ascii=False) for e in remaining),
        encoding="utf-8",
    )

JSONLファイルは行ごとに1件のJSONを格納する形式で、追記が容易かつ部分読み込みが可能なため、ジョブキューの簡易実装に適しています。スループット要件が高まった際はこのキューをAWS SQSやRedisに差し替えるだけで、パイプラインの主要ロジックを変えずにスケールアップできます。

並列処理の観点では、concurrent.futures.ThreadPoolExecutorを使って複数SKUのAPI呼び出しを同時実行することも有効です。ただし、eBay APIには1セラーあたりのレートリミットがあるため、max_workers=5程度から始めて429（Too Many Requests）が出ないか確認しながら調整してください。

まとめ

本記事では、Inventory Mapping APIが出力したAI推奨結果（ListingPreview）をInventory APIのcreateOrReplaceInventoryItemに連携するE2Eパイプラインを実装しました。

• ベースライン: ListingPreviewのGraphQL aspects配列（[{localizedAspectName, value: [...]}]）をInventory API REST形式の辞書（{"Brand": ["Nike"]}）に変換する_convert_aspects関数と、title・description・imageUrlsのシンプルなフィールドマッピングを確立しました。
• 深いポイント: ①mappingReferenceIdはREST API専用フィールドがないため内部SQLiteでSKUと1対1に永続化するトレーサビリティ設計を採用。②COMPLETED_WITH_ERRORの商品が品質不足のまま出品されるリスクを_validate_previewの事前チェックで排除。③conditionはInventory Mapping APIが推奨しないため外部から注入する設計が必須。
• スケーリング: bulkCreateOrReplaceInventoryItemで25件単位の一括投入を実現し、部分失敗はJSONLキューで管理して後回し再処理することで、大量カタログにも対応できるアーキテクチャを構築しました。

これでInventory Mapping API三部作（第18-20回）が完結しました。第18回のタスク生成・第19回のポーリングと結果取得・第20回の実出品連携——この3ステップを繋ぐことで、商品メタデータを入力するだけでAIがカテゴリ・アスペクトを決定し、Inventory APIへ自動登録される完全自動化パイプラインが完成しました。

次のステップ

次回（#21）からは新しいAPIカテゴリ「Message API」に入ります。

「Message API入門：getConversationsとgetConversationでバイヤーとのメッセージを取得する」をテーマに、ECオペレーションで欠かせないカスタマーコミュニケーション自動化の第一歩を解説します。お楽しみに！

次の記事はこちら