GetCategoriesとGetCategoryFeaturesでカテゴリ構造とItem Specificsルールを自動取得する

前回の記事はこちら

【連載#16】eBay Trading API：GetCategoriesとGetCategoryFeaturesでカテゴリ構造とItem Specificsルールを自動取得する

はじめに

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

前回（#15）は、SetNotificationPreferencesを用いてeBayからのイベント通知を設定し、PythonサーバーでWebhookとして受信する仕組みを構築しました。これにより、注文や支払いの変化をリアルタイムに検知できるようになりました。

しかし実務でよくある落とし穴として、「通知を受け取る前の段階」、すなわち出品（Listing）時にカテゴリ選択を誤り、後から修正が効かない状況に陥ることがあります。間違ったカテゴリに出品すると、検索露出が著しく低下するだけでなく、「そのカテゴリで必須のItem Specifics（商品の詳細属性）」が不明なまま出品してしまい、eBayから出品差し止めのエラーを受けることもあります。

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

「だいたい合ってそうなカテゴリに入れておけばいいじゃないか？」

eBay開発を始めたエンジニアから、この言葉を何度か聞いたことがあります。その気持ちは理解できます。カテゴリIDは単なる整数値であり、「6000」と「6001」のどちらが正解なのかは、eBayのサイトを実際に確認しなければわかりません。

しかし、eBayのカテゴリは単なる「分類ラベル」ではありません。カテゴリには以下の情報が紐付いています。

• 必須Item Specifics: そのカテゴリで出品するために必ず入力しなければならない属性（例: 衣類カテゴリなら「Brand」「Size Type」「Size」など）。欠損するとAPIレベルでエラー、もしくはeBay品質スコアが低下し検索下位に沈みます。
• バリエーション対応可否 (VariationsEnabled): カテゴリによってはバリエーション出品（色・サイズ展開）が禁止されています。連載#6で実装したバリエーションXMLを送っても、このフラグがfalseのカテゴリでは容赦なくエラーが返ります。
• コンディション制約 (ConditionEnabled): 「新品」「中古」などのコンディション指定が必須か、または特定コンディションが禁止されているかどうか。

これらの情報をプログラムから取得するためのAPIがGetCategories（カテゴリツリーの取得）とGetCategoryFeatures（カテゴリ別ルールの取得）です。出品システムを本番稼働させるなら、この二つのAPIを使いこなすことは避けて通れません。

基本的な使い方（ベースライン）：GetCategoriesでカテゴリツリーを取得する

まずはGetCategoriesの最小構成から始めます。このAPIはeBayのサイト（US、JPなど）全体のカテゴリ階層をSOAPレスポンスとして返します。zeepライブラリを使ったPython実装は以下の通りです。

# get_categories_baseline.py
import json
import zeep
from zeep import Client
from zeep.transports import Transport
import requests

WSDL_URL = "https://api.ebay.com/wsapi?callname=GetCategories&siteid=0&version=1265"
APP_TOKEN = "YOUR_OAUTH_TOKEN"  # User tokenまたはApp token

def get_categories_raw(site_id: int = 0, level_limit: int = 2) -> dict:
    """
    GetCategoriesを呼び出してカテゴリツリーを取得する（最小実装）。
    site_id: 0=US, 101=Italy, 189=Switzerland など
    level_limit: 取得する階層の深さ。None指定で全階層（注意：レスポンスが巨大になる）
    """
    transport = Transport(session=requests.Session())
    client = Client(
        "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl",
        transport=transport
    )

    header_data = {
        "RequesterCredentials": {
            "eBayAuthToken": APP_TOKEN
        }
    }
    header = client.get_element("ns0:RequesterCredentials")(**header_data)

    request_body = {
        "CategorySiteID": site_id,
        "DetailLevel": "ReturnAll",
        "LevelLimit": level_limit,
        "ViewAllNodes": True,
    }

    response = client.service.GetCategories(
        _soapheaders={"RequesterCredentials": {"eBayAuthToken": APP_TOKEN}},
        **request_body
    )
    return zeep.helpers.serialize_object(response, target_cls=dict)

if __name__ == "__main__":
    result = get_categories_raw(site_id=0, level_limit=2)
    categories = result.get("CategoryArray", {}).get("Category", [])
    print(f"取得カテゴリ数: {len(categories)}")
    for cat in categories[:5]:
        print(f"  ID={cat['CategoryID']}, Name={cat['CategoryName']}, Leaf={cat.get('LeafCategory', False)}")

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

ここからは、ベースラインのコードを実運用に乗せる際に必ずぶつかる落とし穴を3つ解説します。

1. カテゴリツリーの巨大さとキャッシュ戦略

LevelLimitを指定せずにGetCategoriesを呼ぶと、USサイトでは20,000件を超えるカテゴリが返ってきます。このAPIを商品1件出品するたびに都度呼び出すのは、パフォーマンス的にも費用的にも論外です。

eBayのカテゴリツリーは頻繁に変わるものではありませんが、四半期ごとに新カテゴリが追加・廃止されることがあります。推奨される実装パターンは次の通りです。

まず初回起動時または週次バッチでGetCategoriesを呼び出し、全カテゴリをローカルのJSONファイルまたはSQLiteに保存します。アプリケーション起動時はそのキャッシュをメモリに読み込み、カテゴリ検索はすべてローカルで処理します。eBayはGetCategoriesのレスポンスにCategoryVersionという整数値を含めており、前回取得時のバージョンと比較することで「差分更新が必要かどうか」を事前判定できます。

# category_cache.py
import json
import os
from datetime import datetime

CACHE_FILE = "ebay_categories_us.json"

def load_or_refresh_categories(fetcher_func, force_refresh: bool = False) -> list[dict]:
    """
    ローカルキャッシュがあればそれを返し、なければAPIを呼んで保存する。
    force_refresh=Trueで強制的にAPIから再取得する。
    """
    if not force_refresh and os.path.exists(CACHE_FILE):
        with open(CACHE_FILE, "r", encoding="utf-8") as f:
            data = json.load(f)
        print(f"キャッシュから読み込み: {len(data['categories'])}件 (更新日時: {data['fetched_at']})")
        return data["categories"]

    print("APIからカテゴリ取得中...")
    raw = fetcher_func()
    categories = raw.get("CategoryArray", {}).get("Category", [])
    cache_data = {
        "fetched_at": datetime.utcnow().isoformat(),
        "category_version": raw.get("CategoryVersion"),
        "categories": [
            {
                "id": c["CategoryID"],
                "name": c["CategoryName"],
                "parent_id": c.get("CategoryParentID"),
                "level": c["CategoryLevel"],
                "is_leaf": c.get("LeafCategory", False),
                "virtual": c.get("Virtual", False),
            }
            for c in categories
        ]
    }
    with open(CACHE_FILE, "w", encoding="utf-8") as f:
        json.dump(cache_data, f, ensure_ascii=False, indent=2)
    print(f"保存完了: {len(categories)}件")
    return cache_data["categories"]

2. GetCategoryFeaturesでVariationsEnabledを判定する

連載#6で実装したバリエーション出品（Multi-SKU）を特定カテゴリで行う前に、そのカテゴリがバリエーションをサポートしているか確認しなければなりません。これを怠ると、精巧に組み上げたVariations XMLがError 21916616（Variations not supported for this category）で弾かれます。

GetCategoryFeaturesのリクエストでFeatureIDに「VariationsEnabled」を指定することで、その情報だけをピンポイントで取得できます。

# check_variations_enabled.py
import zeep
import requests
from zeep.transports import Transport

def check_variations_enabled(category_id: str, token: str) -> bool:
    """
    指定カテゴリがバリエーション出品をサポートしているか確認する。
    Returns: True=バリエーション対応, False=非対応
    """
    transport = Transport(session=requests.Session())
    client = zeep.Client(
        "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl",
        transport=transport
    )

    response = client.service.GetCategoryFeatures(
        _soapheaders={"RequesterCredentials": {"eBayAuthToken": token}},
        CategoryID=category_id,
        FeatureID=["VariationsEnabled"],
        DetailLevel="ReturnAll",
        ViewDataOfAllDescendents=False,
    )
    result = zeep.helpers.serialize_object(response, target_cls=dict)

    # SiteDefaultsはサイト全体のデフォルト値
    # Category[0].VariationsEnabledはカテゴリ固有の値（上書き）
    site_default = result.get("SiteDefaults", {}).get("VariationsEnabled", False)
    categories = result.get("Category", []) or []
    if categories:
        cat_val = categories[0].get("VariationsEnabled")
        if cat_val is not None:
            return bool(cat_val)
    return bool(site_default)

if __name__ == "__main__":
    TOKEN = "YOUR_TOKEN"
    test_cats = [
        ("11450", "Clothing, Shoes & Accessories"),
        ("99", "Everything Else"),
        ("619", "Computers/Tablets & Networking"),
    ]
    for cat_id, name in test_cats:
        enabled = check_variations_enabled(cat_id, TOKEN)
        print(f"カテゴリ {cat_id} ({name}): VariationsEnabled={enabled}")

3. 必須Item Specificsと推奨Item Specificsの見分け方

GetCategoryFeaturesのFeatureIDに「ItemSpecifics」を指定すると、そのカテゴリで定義されているItem Specificsの一覧と、それぞれが必須（Required）か推奨（Recommended）かオプション（Optional）かが返ってきます。

ここで注意が必要なのは「SelectionMode」と「MinValues」の組み合わせです。

• SelectionMode=FreeText かつ MinValues=0 → 任意入力（Optional）
• SelectionMode=SelectionOnly かつ MinValues=1 → リストから選択必須（Required）
• SelectionMode=SelectionOrFreeText かつ MinValues=1 → リストか自由入力で必須（Required）

この三つを判定して「必須フィールドが全部埋まっているか」を出品前にチェックするロジックを組むことで、APIエラーになる前に問題を検知できます。

頻出エラーコード早見表

エラーコード	メッセージ	対処法
Error 21916585	Invalid category ID specified	カテゴリIDが存在しない、または廃止済み。キャッシュを最新化して再確認する。
Error 21916616	Variations are not supported for this category	バリエーション非対応カテゴリにVariationsブロックを送信した。GetCategoryFeaturesでVariationsEnabled=Trueを事前確認する。
Error 21916587	The feature VariationPictures is not supported	VariationPictures非対応カテゴリで画像の軸指定を行った。
Error 878	Category ID is required	CategoryIDフィールドが空またはnull。LeafCategoryのIDのみ指定可能。

堅牢な実装：カテゴリツリーJSON保存と必須フィールド自動チェックスクリプト

ここまでの知識をまとめ、以下の二つの責務を持つ本番品質のスクリプトを実装します。

(1) CategoryCacheManager: GetCategoriesを呼び出し、全カテゴリをJSONに永続化・読み込みする管理クラス

(2) CategoryFeatureChecker: GetCategoryFeaturesを呼び出し、必須Item Specifics・VariationsEnabled・ConditionEnabledを検査し、出品前の可否判定レポートを返す関数

# ebay_category_tools.py
from __future__ import annotations
import json
import os
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional
import zeep
import zeep.helpers
import requests
from zeep.transports import Transport

logger = logging.getLogger(__name__)


@dataclass
class CategoryInfo:
    """単一カテゴリの基本情報。"""
    id: str
    name: str
    parent_id: Optional[str]
    level: int
    is_leaf: bool
    virtual: bool = False


@dataclass
class FeatureCheckResult:
    """GetCategoryFeaturesの解析結果。"""
    category_id: str
    variations_enabled: bool
    condition_enabled: bool
    required_item_specifics: list[str] = field(default_factory=list)
    recommended_item_specifics: list[str] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)

    def is_listing_safe(self, provided_specifics: set[str]) -> tuple[bool, list[str]]:
        """
        出品時に提供したItem Specificsキー一覧を受け取り、
        必須フィールドが揃っているか検証する。
        Returns: (ok: bool, missing_fields: list[str])
        """
        missing = [s for s in self.required_item_specifics if s not in provided_specifics]
        return (len(missing) == 0), missing


class CategoryCacheManager:
    """
    GetCategoriesのレスポンスをJSONファイルにキャッシュし、
    ローカル検索・ID解決を提供するマネージャー。
    """

    def __init__(self, token: str, cache_path: str = "ebay_categories.json",
                 site_id: int = 0, cache_ttl_days: int = 7):
        if not token or not token.strip():
            raise ValueError("eBay OAuth tokenは必須です")
        self.token = token
        self.cache_path = cache_path
        self.site_id = site_id
        self.cache_ttl = timedelta(days=cache_ttl_days)
        self._categories: dict[str, CategoryInfo] = {}
        self._client: Optional[zeep.Client] = None

    def _get_client(self) -> zeep.Client:
        if self._client is None:
            transport = Transport(session=requests.Session())
            self._client = zeep.Client(
                "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl",
                transport=transport
            )
        return self._client

    def _is_cache_fresh(self) -> bool:
        """キャッシュファイルが存在し、TTL内であればTrueを返す。"""
        if not os.path.exists(self.cache_path):
            return False
        try:
            with open(self.cache_path, "r", encoding="utf-8") as f:
                data = json.load(f)
            fetched_at = datetime.fromisoformat(data["fetched_at"])
            return datetime.utcnow() - fetched_at < self.cache_ttl
        except (KeyError, ValueError, json.JSONDecodeError):
            return False

    def _fetch_from_api(self) -> list[dict]:
        """GetCategories APIを呼び出して全カテゴリを取得する。"""
        client = self._get_client()
        logger.info("GetCategories APIを呼び出し中 (site_id=%d)...", self.site_id)
        try:
            response = client.service.GetCategories(
                _soapheaders={"RequesterCredentials": {"eBayAuthToken": self.token}},
                CategorySiteID=self.site_id,
                DetailLevel="ReturnAll",
                ViewAllNodes=True,
            )
        except zeep.exceptions.Fault as e:
            raise RuntimeError(f"GetCategories SOAPエラー: {e.message}") from e

        result = zeep.helpers.serialize_object(response, target_cls=dict)
        categories = result.get("CategoryArray", {}).get("Category", []) or []
        version = result.get("CategoryVersion")
        logger.info("取得完了: %d件 (CategoryVersion=%s)", len(categories), version)
        return categories, version

    def load(self, force_refresh: bool = False) -> None:
        """
        カテゴリデータをロードする。
        キャッシュが新鮮であればファイルから、そうでなければAPIから取得する。
        """
        if not force_refresh and self._is_cache_fresh():
            logger.info("キャッシュから読み込み中: %s", self.cache_path)
            with open(self.cache_path, "r", encoding="utf-8") as f:
                data = json.load(f)
            raw_cats = data["categories"]
        else:
            raw_cats, version = self._fetch_from_api()
            cache_data = {
                "fetched_at": datetime.utcnow().isoformat(),
                "category_version": version,
                "site_id": self.site_id,
                "categories": [
                    {
                        "id": str(c["CategoryID"]),
                        "name": c["CategoryName"],
                        "parent_id": str(c.get("CategoryParentID", "")),
                        "level": int(c.get("CategoryLevel", 1)),
                        "is_leaf": bool(c.get("LeafCategory", False)),
                        "virtual": bool(c.get("Virtual", False)),
                    }
                    for c in raw_cats
                ],
            }
            with open(self.cache_path, "w", encoding="utf-8") as f:
                json.dump(cache_data, f, ensure_ascii=False, indent=2)
            logger.info("キャッシュ保存完了: %s", self.cache_path)
            raw_cats = cache_data["categories"]

        self._categories = {
            c["id"]: CategoryInfo(**c)
            for c in raw_cats
        }
        logger.info("メモリにロード完了: %d件のカテゴリ", len(self._categories))

    def find_by_id(self, category_id: str) -> Optional[CategoryInfo]:
        """IDでカテゴリを検索する。"""
        return self._categories.get(str(category_id))

    def search_by_name(self, keyword: str, leaf_only: bool = True) -> list[CategoryInfo]:
        """カテゴリ名（部分一致）でカテゴリを検索する。"""
        kw = keyword.lower()
        results = [
            cat for cat in self._categories.values()
            if kw in cat.name.lower() and (not leaf_only or cat.is_leaf)
        ]
        return sorted(results, key=lambda c: c.name)

    def get_ancestors(self, category_id: str) -> list[CategoryInfo]:
        """指定カテゴリの祖先カテゴリを根から順に返す。"""
        path = []
        current = self.find_by_id(category_id)
        visited = set()
        while current and current.id not in visited:
            path.append(current)
            visited.add(current.id)
            if current.parent_id and current.parent_id != current.id:
                current = self.find_by_id(current.parent_id)
            else:
                break
        return list(reversed(path))


class CategoryFeatureChecker:
    """
    GetCategoryFeaturesを呼び出して、
    カテゴリの出品ルールを解析するクラス。
    """

    def __init__(self, token: str):
        if not token or not token.strip():
            raise ValueError("eBay OAuth tokenは必須です")
        self.token = token
        self._client: Optional[zeep.Client] = None

    def _get_client(self) -> zeep.Client:
        if self._client is None:
            transport = Transport(session=requests.Session())
            self._client = zeep.Client(
                "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl",
                transport=transport
            )
        return self._client

    def check(self, category_id: str) -> FeatureCheckResult:
        """
        指定カテゴリの機能フラグとItem Specifics要件を取得・解析する。
        Args:
            category_id: LeafカテゴリのID（文字列）
        Returns:
            FeatureCheckResult
        Raises:
            ValueError: category_idが空の場合
            RuntimeError: APIエラーが発生した場合
        """
        if not category_id or not str(category_id).strip():
            raise ValueError("category_idは必須です")

        client = self._get_client()
        try:
            response = client.service.GetCategoryFeatures(
                _soapheaders={"RequesterCredentials": {"eBayAuthToken": self.token}},
                CategoryID=str(category_id),
                FeatureID=["VariationsEnabled", "ConditionEnabled", "ItemSpecifics"],
                DetailLevel="ReturnAll",
                ViewDataOfAllDescendents=False,
            )
        except zeep.exceptions.Fault as e:
            raise RuntimeError(f"GetCategoryFeatures SOAPエラー (category={category_id}): {e.message}") from e

        result = zeep.helpers.serialize_object(response, target_cls=dict)
        return self._parse_result(category_id, result)

    def _parse_result(self, category_id: str, result: dict) -> FeatureCheckResult:
        """レスポンスdictをFeatureCheckResultに変換する内部メソッド。"""
        site_defaults = result.get("SiteDefaults", {}) or {}
        cats = result.get("Category", []) or []
        cat_data = cats[0] if cats else {}

        def resolve(key: str, default=None):
            """カテゴリ固有値 > SiteDefaults の優先順位で値を解決する。"""
            val = cat_data.get(key)
            if val is not None:
                return val
            return site_defaults.get(key, default)

        variations_enabled = bool(resolve("VariationsEnabled", False))
        condition_enabled = bool(resolve("ConditionEnabled", False))

        # Item Specificsの解析
        required_fields: list[str] = []
        recommended_fields: list[str] = []
        item_specifics_data = resolve("ItemSpecificsEnabled")
        specifics_list = []
        if isinstance(item_specifics_data, dict):
            specifics_list = item_specifics_data.get("ItemSpecific", []) or []
        elif isinstance(item_specifics_data, list):
            specifics_list = item_specifics_data

        for spec in specifics_list:
            name = spec.get("Name", "")
            min_values = int(spec.get("MinValues", 0) or 0)
            if min_values >= 1:
                required_fields.append(name)
            else:
                recommended_fields.append(name)

        return FeatureCheckResult(
            category_id=category_id,
            variations_enabled=variations_enabled,
            condition_enabled=condition_enabled,
            required_item_specifics=required_fields,
            recommended_item_specifics=recommended_fields,
        )


def print_feature_report(result: FeatureCheckResult) -> None:
    """FeatureCheckResultを見やすく標準出力する。"""
    print(f"\n=== カテゴリ {result.category_id} 出品前チェックレポート ===")
    print(f"  VariationsEnabled  : {result.variations_enabled}")
    print(f"  ConditionEnabled   : {result.condition_enabled}")
    if result.required_item_specifics:
        print(f"  必須Item Specifics ({len(result.required_item_specifics)}件):")
        for s in result.required_item_specifics:
            print(f"    - {s}")
    else:
        print("  必須Item Specifics: なし")
    if result.recommended_item_specifics:
        print(f"  推奨Item Specifics ({len(result.recommended_item_specifics)}件):")
        for s in result.recommended_item_specifics[:10]:  # 長い場合は先頭10件
            print(f"    - {s}")


if __name__ == "__main__":
    import sys
    logging.basicConfig(level=logging.INFO)
    TOKEN = os.environ.get("EBAY_TOKEN", "")
    if not TOKEN:
        print("環境変数 EBAY_TOKEN を設定してください")
        sys.exit(1)

    # Step 1: カテゴリツリーをロード（キャッシュ優先）
    cache_mgr = CategoryCacheManager(token=TOKEN, site_id=0)
    cache_mgr.load()

    # Step 2: キーワードでカテゴリ検索
    results = cache_mgr.search_by_name("T-Shirt", leaf_only=True)
    print(f"\n'T-Shirt'検索結果: {len(results)}件")
    for cat in results[:5]:
        ancestors = cache_mgr.get_ancestors(cat.id)
        path = " > ".join(a.name for a in ancestors)
        print(f"  [{cat.id}] {path}")

    # Step 3: 特定カテゴリの出品ルールをチェック
    checker = CategoryFeatureChecker(token=TOKEN)
    target_cat_id = "15687"  # 例: Men's T-Shirts
    feature_result = checker.check(target_cat_id)
    print_feature_report(feature_result)

    # Step 4: 出品前の必須Item Specifics検証
    my_specifics = {"Brand", "Size", "Color"}  # 自社システムが用意した属性
    ok, missing = feature_result.is_listing_safe(my_specifics)
    if ok:
        print("\n必須Item Specificsは全て揃っています。出品可能です。")
    else:
        print(f"\n不足している必須Item Specifics: {missing}")
        print("  出品前にこれらの属性を追加してください。")

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

本番稼働のeBay出品システムにおいて、カテゴリデータの管理はそれ単体で一つのサブシステムとして設計する価値があります。

カテゴリDBキャッシュと定期同期タスクの設計

JSONファイルへのキャッシュは開発・小規模運用では十分ですが、複数サーバーにスケールアウトする場合、各サーバーがそれぞれ別のJSONを持つと整合性が取れません。このフェーズに入ったら、カテゴリデータをPostgreSQLやMySQLなどの共有RDBMSに格納する設計に移行します。

テーブル設計のポイントは、category_idをプライマリキー（VARCHAR）、parent_idを外部キー（自己参照）とし、is_leaf・levelをインデックス付きカラムとして持つことです。カテゴリ名のキーワード検索にはFULL TEXT INDEXを活用すると、数万件のカテゴリ名から数ミリ秒でヒットします。

定期同期タスクは、Celery Beat（またはcron）で週に一度GetCategoriesを呼び出し、DBの内容と差分を取って更新・追加・廃止を反映します。廃止カテゴリはDBから削除するのではなく、is_active=Falseフラグを立てる「論理削除」が安全です。既にそのカテゴリで出品している商品への影響を追跡できるからです。

# sync_categories_task.py（Celeryタスクのイメージ）
from celery import shared_task
from ebay_category_tools import CategoryCacheManager
import logging

logger = logging.getLogger(__name__)

@shared_task(name="sync_ebay_categories")
def sync_ebay_categories() -> dict:
    """
    週次で実行するeBayカテゴリ同期タスク。
    キャッシュTTLを0に設定することでAPIから強制再取得する。
    """
    import os
    token = os.environ["EBAY_TOKEN"]
    manager = CategoryCacheManager(
        token=token,
        cache_path="/shared/cache/ebay_categories.json",
        site_id=0,
        cache_ttl_days=0,  # TTL=0 → 常にAPIから取得
    )
    manager.load(force_refresh=True)
    count = len(manager._categories)
    logger.info("カテゴリ同期完了: %d件", count)
    return {"synced_count": count}

GetCategoryFeaturesについても同様に、頻繁に出品するカテゴリのFeatureCheckResultをRedisにシリアライズしてキャッシュすることを推奨します。TTLは7〜30日程度が現実的です。カテゴリのルールはGetCategoriesほど頻繁には変わりませんが、eBayのポリシー変更シーズン（毎年春・秋）前後は要注意です。

まとめ

本記事では、eBay Trading APIのメタデータ系エンドポイントであるGetCategoriesとGetCategoryFeaturesを使い、出品カテゴリの構造とルールをプログラムから取得・活用する方法を解説しました。

• ベースライン: zeepを使ったGetCategories呼び出しと、カテゴリツリーのJSON永続化。LevelLimitとLeafCategoryフィールドの基本的な扱い方。
• 深いポイント: カテゴリツリーの大量データに対するキャッシュ戦略と強制リフレッシュの実装。GetCategoryFeaturesのSiteDefaults/Category優先順位ロジック。必須Item Specificsと推奨Item Specificsの判定と、VariationsEnabled・MaxGranularFitmentCountを使った出品前バリデーション。
• スケーリング: JSONキャッシュから共有RDBMSへの移行設計、Celery Beatによる週次同期タスク、RedisによるGetCategoryFeaturesのLazy-Loadingキャッシュパターン。

これにより、カテゴリIDを勘で入力することなく、プログラムが自動的に正しいLeafカテゴリを特定し、そのカテゴリに必要なItem Specificsが揃っているかを出品前に検証できる、堅牢な出品パイプラインの基盤が整いました。

次のステップ

次回（#17）は、【Trading API - フィードバック管理】フェーズに進み、GetFeedbackで自分・相手のフィードバック履歴を取得し、LeaveFeedbackで条件に基づく自動評価送信を実装します。

取引完了後のフィードバック自動化は、セラーの評価スコアを維持するための重要な運用タスクです。どうぞお楽しみに！

次の記事はこちら