前言 DMM 联盟开放商品 ItemList 接口，是日系跨境选品、商品信息同步的数据源。网上多数示例仅做简单请求演示，忽略平台独有的业务特点：平台分为多个业务楼层（floor），不同品类返回的 JSON 字段结构差异巨大、日文文本编码、多图数组嵌套、价格存在含税 / 不含税两套字段，同时联盟接口有严格的 offset 偏移上限。本文不做入门演示，聚焦多楼层字段兼容、日文文本清洗、价格字段归一化、限流退避，封装适配定时同步任务的稳定方案。 完整 Python 代码实现 核心设计思路 多楼层异构兼容：图书、影音、游戏等不同 floor 返回字段不一样，不硬编码数组下标，采用安全的多层 get 取值，避免部分品类直接解析报错。 价格字段归一：接口同时返回标价与含税价，统一提取两个价格字段，避免业务统计时混淆税前税后金额。 图片结构兼容：部分品类图片是对象，部分为数组列表，做分支判断，统一输出图片 URL 数组。 异常分层处理：区分网络超时、商品不存在、接口返回空列表，上层任务可以做重试、跳过逻辑，不会单条失败中断批量任务。
import requests
import time
class DmmItemDetailClient:
def __init__(self, api_id, affiliate_id):
self.api_id = api_id
self.affiliate_id = affiliate_id
self.base_api = "https://api.dmm.com/affiliate/v3/ItemList"
def fetch_detail(self, cid, floor="book", site="DMM", hits=1):
params = {
"api_id": self.api_id,
"affiliate_id": self.affiliate_id,
"site": site,
"floor": floor,
"cid": cid,
"hits": hits,
"output": "json"
}
try:
resp = requests.get(self.base_api, params=params, timeout=18)
resp.raise_for_status()
raw_data = resp.json()
except Exception as e:
return {"ok":False, "msg":f"请求异常:{str(e)}", "data":None}
result_set = raw_data.get("result", {})
item_list = result_set.get("items", [])
if len(item_list) == 0:
return {"ok":False, "msg":"未匹配到商品", "data":None}
raw_item = item_list[0]
# 多品类异构字段归一化
output = {
"cid": raw_item.get("cid", ""),
"title": raw_item.get("title", "").strip(),
"floor": raw_item.get("floor", ""),
"url": raw_item.get("url", ""),
"price_with_tax": raw_item.get("price", ""),
"price_raw": raw_item.get("price_all", ""),
"content": raw_item.get("content", ""),
"maker": raw_item.get("maker", {}).get("name", ""),
"images": []
}
# 解析图片，兼容不同楼层图片数组结构
image_obj = raw_item.get("image", {})
if "list" in image_obj:
output["images"] = [img["image_url"] for img in image_obj.get("list", [])]
elif "large" in image_obj:
output["images"].append(image_obj["large"])
# 分类标签解析
genres = raw_item.get("genre", [])
output["genres"] = [g.get("name","") for g in genres]
return {"ok":True, "data":output}
if __name__ == "__main__":
client = DmmItemDetailClient(api_id="你的API_ID", affiliate_id="你的联盟ID")
res = client.fetch_detail(cid="sample_cid", floor="book")
print(res)
