雅安市网站建设_网站建设公司_Sketch_seo优化

你需要的是乐天（Rakuten）商品详情 API 接口的完整调用示例与可直接运行的代码实现，核心覆盖「单商品详情查询、批量商品查询、异常处理、数据解析」四大核心场景，以下是适配新手的极简版 + 企业级稳健版代码，兼顾易用性与生产环境稳定性。

一、 前置准备（必做）

1. 核心凭证获取

• 关注博主，可创建应用并获取Application ID（32 位字符串，所有请求必填）；
• 确认接口版本：推荐使用IchibaItem/Item 20170426（商品详情）、IchibaItem/Search 20220601（批量搜索）。

2. 环境依赖安装

仅需 Python 基础库，执行以下命令安装：

bash

运行

pip install requests # 处理HTTP请求 pip install pandas # 可选，用于批量数据处理

二、 极简版调用示例（快速上手）

适合新手快速验证 API 可用性，实现单商品详情查询的核心功能：

python

运行

import requests import json # -------------------------- 配置项（替换为你的信息） -------------------------- APP_ID = "你的32位Application ID" # 替换为真实ID ITEM_CODE = "shop001:item123456" # 替换为真实商品编码（乐天商品页URL中提取） # -------------------------- 核心调用逻辑 -------------------------- def get_rakuten_item_simple(): """极简版：调用乐天商品详情API并打印结果""" # 1. 构造请求URL和参数 api_url = "https://app.rakuten.co.jp/services/api/IchibaItem/Item/20170426" params = { "applicationId": APP_ID, "itemCode": ITEM_CODE, "format": "json", # 响应格式为JSON "formatVersion": 2 # 推荐版本，结构更清晰 } # 2. 发送请求 try: response = requests.get(api_url, params=params, timeout=15) response.raise_for_status() # 抛出HTTP错误（如404/429） item_data = response.json() # 3. 解析核心字段 print("=== 乐天商品详情（极简版）===") print(f"商品编码：{item_data.get('itemCode', '未知')}") print(f"商品名称：{item_data.get('itemName', '未知')}") print(f"商品价格：{item_data.get('itemPrice', 0)} 日元") print(f"库存状态：{'有货' if item_data.get('availability') == 1 else '无货'}") print(f"店铺名称：{item_data.get('shopName', '未知')}") return item_data except Exception as e: print(f"调用失败：{e}") return None # 执行调用 if __name__ == "__main__": get_rakuten_item_simple()

运行说明

1. 替换APP_ID和ITEM_CODE为真实值；
2. 直接运行代码，即可打印商品核心信息；
3. 若报错401 Unauthorized，检查APP_ID是否有效；若报错404，检查ITEM_CODE是否正确。

三、 企业级稳健版代码（生产环境可用）

包含频率控制、异常重试、缓存、数据清洗、批量查询等生产级特性，适配海外仓 / 选品系统对接：

python

运行

import requests import json import time import redis import random from datetime import datetime from requests.exceptions import RequestException from functools import wraps # -------------------------- 全局配置 -------------------------- # 核心凭证 APP_ID = "你的32位Application ID" # API地址 ITEM_API_URL = "https://app.rakuten.co.jp/services/api/IchibaItem/Item/20170426" SEARCH_API_URL = "https://app.rakuten.co.jp/services/api/IchibaItem/Search/20220601" # 频率控制（免费版QPS≤10，预留冗余） REQUEST_INTERVAL = 0.12 # 120ms/次 # Redis缓存配置（可选，无需缓存可注释） REDIS_CLIENT = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True) CACHE_EXPIRE = 3600 # 缓存1小时 # 代理池（可选，规避IP风控） PROXY_POOL = [ # "http://ip1:port", # "http://ip2:port" ] # -------------------------- 工具函数 -------------------------- def get_random_proxy(): """随机获取代理IP""" return random.choice(PROXY_POOL) if PROXY_POOL else None def get_headers(): """构造模拟浏览器的请求头，避免风控""" return { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0 Safari/537.36", "Accept": "application/json, text/plain, */*", "Accept-Language": "ja-JP,ja;q=0.9,en-US;q=0.8", "Referer": "https://webservice.rakuten.co.jp/" } def cache_decorator(expire=CACHE_EXPIRE): """缓存装饰器：优先从Redis读取，减少API调用""" def wrapper(func): @wraps(func) def inner(item_code): cache_key = f"rakuten:item:{item_code}" # 1. 查缓存 cached_data = REDIS_CLIENT.get(cache_key) if cached_data: return json.loads(cached_data) # 2. 缓存未命中，调用函数 result = func(item_code) # 3. 写入缓存 if result: REDIS_CLIENT.setex(cache_key, expire, json.dumps(result)) return result return inner return wrapper def request_with_retry(url, params, max_retry=3): """指数退避重试：处理429限流/503服务不可用""" retry_count = 0 while retry_count < max_retry: try: # 频率控制 time.sleep(REQUEST_INTERVAL) # 构造请求 headers = get_headers() proxy = get_random_proxy() proxies = {"http": proxy, "https": proxy} if proxy else None # 发送请求 response = requests.get( url, params=params, headers=headers, proxies=proxies, timeout=15 ) # 处理限流 if response.status_code == 429: sleep_time = 2 ** retry_count # 指数退避：1s→2s→4s print(f"触发429限流，等待{sleep_time}秒重试...") time.sleep(sleep_time) retry_count += 1 continue # 校验响应 response.raise_for_status() # 打印剩余配额（可选） remaining_quota = response.headers.get("X-API-QUOTA-REMAINING", "未知") print(f"剩余API配额：{remaining_quota}次") return response.json() except RequestException as e: print(f"请求失败（重试{retry_count+1}/{max_retry}）：{e}") retry_count += 1 time.sleep(1) print("多次重试失败，放弃请求") return None # -------------------------- 核心功能：单商品详情查询 -------------------------- @cache_decorator(expire=3600) # 缓存1小时 def get_rakuten_item_detail(item_code): """ 获取单商品详情（含数据清洗） :param item_code: 乐天商品编码（shop001:item123456） :return: 清洗后的商品字典 """ params = { "applicationId": APP_ID, "itemCode": item_code, "format": "json", "formatVersion": 2, # 按需筛选字段，减少数据量 "elements": "itemCode,itemName,itemPrice,availability,weight,size,shopName,itemUrl,reviewAverage,reviewCount" } # 调用API raw_data = request_with_retry(ITEM_API_URL, params) if not raw_data: return None # 数据清洗与标准化 cleaned_data = { "商品编码": raw_data.get("itemCode", ""), "商品名称": raw_data.get("itemName", "").strip(), "售价(日元)": int(raw_data.get("itemPrice", 0)), "库存状态": "有货" if raw_data.get("availability") == 1 else "无货", "重量(g)": int(raw_data.get("weight", 0)), "规格": raw_data.get("size", "无规格").strip(), "店铺名称": raw_data.get("shopName", "未知"), "商品链接": raw_data.get("itemUrl", ""), "评价平均分": round(float(raw_data.get("reviewAverage", 0.0)), 1), "评价总数": int(raw_data.get("reviewCount", 0)), "采集时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } return cleaned_data # -------------------------- 扩展功能：批量商品搜索 -------------------------- def search_rakuten_items(keyword, page=1, hits=20): """ 按关键词批量搜索商品（适合选品场景） :param keyword: 搜索关键词（日文，如"ワイヤレスイヤホン"=无线耳机） :param page: 页码（1-100） :param hits: 每页条数（1-100） :return: 批量商品列表 """ params = { "applicationId": APP_ID, "keyword": keyword, "page": page, "hits": hits, "sort": "-itemPrice", # 价格降序（+itemPrice=升序） "availability": 1, # 仅返回有货商品 "format": "json", "formatVersion": 2 } # 调用API raw_data = request_with_retry(SEARCH_API_URL, params) if not raw_data or "Items" not in raw_data: return [] # 解析批量数据 item_list = [] for item in raw_data["Items"]: item_info = item.get("Item", {}) item_list.append({ "商品编码": item_info.get("itemCode", ""), "商品名称": item_info.get("itemName", "").strip(), "售价(日元)": int(item_info.get("itemPrice", 0)), "评价平均分": round(float(item_info.get("reviewAverage", 0.0)), 1), "评价总数": int(item_info.get("reviewCount", 0)) }) print(f"搜索关键词「{keyword}」：共找到{raw_data.get('count', 0)}个商品，当前页{len(item_list)}个") return item_list # -------------------------- 测试调用 -------------------------- if __name__ == "__main__": # 1. 单商品详情查询 ITEM_CODE = "shop001:item123456" # 替换为真实编码 item_detail = get_rakuten_item_detail(ITEM_CODE) if item_detail: print("\n=== 单商品详情（清洗后）===") for key, value in item_detail.items(): print(f"{key}：{value}") # 2. 批量商品搜索 # KEYWORD = "ワイヤレスイヤホン" # 日文关键词：无线耳机 # batch_items = search_rakuten_items(KEYWORD, page=1, hits=10) # if batch_items: # print("\n=== 批量搜索结果 ===") # for idx, item in enumerate(batch_items, 1): # print(f"\n{idx}. {item['商品名称']} | 价格：{item['售价(日元)']} 日元")

四、 关键代码说明

五、 常见问题与解决方案

总结

乐天商品详情 API 调用的核心要点：

1. 凭证核心：Application ID是所有请求的必填参数，需确保有效；
2. 稳定性：通过「频率控制 + 指数退避重试 + 缓存」避免触发限流 / 风控；
3. 数据处理：解析时做好容错（get()方法）和标准化，适配业务场景；
4. 批量场景：优先用IchibaItem/Search接口批量查询，减少请求次数。