京东商品数据怎么获取？商品详情、价格与评价 API 教程

python3 -m pip install requests
export JUSTONEAPI_TOKEN='your_token'
export JD_ITEM_ID='your_numeric_item_id'

import json
import os
from datetime import datetime, timezone
from pathlib import Path

import requests

BASE_URL = "https://api.justoneapi.com"
TOKEN = os.environ.get("JUSTONEAPI_TOKEN", "")


def call_api(path, **params):
    try:
        response = requests.get(
            BASE_URL + path,
            params={"token": TOKEN, **params},
            timeout=120, allow_redirects=False,
        )
    except requests.RequestException:
        raise RuntimeError("API network error or timeout") from None
    if response.status_code != 200:
        raise RuntimeError(f"API HTTP {response.status_code}")
    try:
        result = response.json()
    except ValueError:
        raise RuntimeError("API response is not JSON") from None
    if not isinstance(result, dict) or result.get("code") != 0:
        code = result.get("code") if isinstance(result, dict) else None
        raise RuntimeError(f"API failed: code={code}; check the docs")
    data = result.get("data")
    if not isinstance(data, dict):
        raise RuntimeError("Unexpected data shape; inspect the docs")
    inner = data.get("base_resp")
    if isinstance(inner, dict) and inner.get("status_code") not in (0, "0"):
        raise RuntimeError("Platform response was unsuccessful")
    return result


def save_json(folder, name, value):
    (folder / name).write_text(
        json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8"
    )


def require_list(data, key):
    value = data.get(key)
    if not isinstance(value, list):
        raise RuntimeError(f"Expected a list at {key}; inspect saved JSON")
    return value


def output_folder(topic):
    if not TOKEN or TOKEN == "your_token":
        raise RuntimeError("Set JUSTONEAPI_TOKEN before running")
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    folder = Path(topic + "-output") / run_id
    folder.mkdir(parents=True, exist_ok=False)
    return folder


ITEM_ID = os.environ.get("JD_ITEM_ID", "")
MAX_REVIEW_PAGES = 2


def main():
    if not ITEM_ID or not ITEM_ID.isdigit():
        raise RuntimeError("Set JD_ITEM_ID to a numeric product ID")
    folder = output_folder("jd")
    detail = call_api("/api/jd/get-item-detail/v4", itemId=ITEM_ID)
    save_json(folder, "product-detail.json", detail)
    price = call_api("/api/jd/get-item-price/v1", itemId=ITEM_ID)
    save_json(folder, "product-price.json", price)
    rows = require_list(price["data"], "data")
    save_json(folder, "price-snapshot.json", {
        "requestedItemId": ITEM_ID,
        "retrievedAt": datetime.now(timezone.utc).isoformat(),
        "rawPriceRecords": rows,
    })

    reviews, seen = [], set()
    for page in range(1, MAX_REVIEW_PAGES + 1):
        result = call_api(
            "/api/jd/get-item-comments/v2", itemId=ITEM_ID, page=str(page)
        )
        save_json(folder, f"reviews-{page}.json", result)
        data = result["data"]
        items = require_list(data, "items")
        if not items:
            break
        added = 0
        for item in items:
            if not isinstance(item, dict):
                raise RuntimeError("Unexpected review item")
            rid = item.get("recordId")
            if rid is None:
                raise RuntimeError("Missing review recordId; inspect saved JSON")
            key = (ITEM_ID, str(rid))
            if key in seen:
                continue
            seen.add(key)
            added += 1
            reviews.append({
                "itemId": ITEM_ID, "recordId": str(rid),
                "text": item.get("text"), "rating": item.get("rating"),
                "createdAt": item.get("createdAt"),
                "variantName": item.get("variantName"),
            })
        total_pages = data.get("totalPages")
        if added == 0 or (
            isinstance(total_pages, int) and page >= total_pages
        ):
            break
    save_json(folder, "reviews.json", reviews)
    print(f"Saved product data and {len(reviews)} reviews to {folder}")


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as error:
        raise SystemExit(str(error)) from None