京东的 item_video 接口是用于获取商品相关视频资源的专业接口，能够获取商品主视频、细节展示视频、使用教程视频等多种类型的视频资源。这些视频内容对于商品展示、用户体验提升、竞品分析等场景具有重要价值，是电商内容分析中不可或缺的一部分。
一、接口核心特性分析
1. 接口功能与定位
2. 认证机制
3. 核心参数与响应结构
请求参数
响应核心字段
二、Python 脚本实现
import requests
import time
import json
import logging
import re
from typing import Dict, Optional, List
from requests.exceptions import RequestException
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
class JDItemVideoAPI:
def __init__(self, appkey: str, appsecret: str):
"""
初始化京东商品视频API客户端
:param appkey: 京东开放平台appkey
:param appsecret: 京东开放平台appsecret
"""
self.appkey = appkey
self.appsecret = appsecret
self.base_url = "https://api.jd.com"
self.access_token = None
self.token_expires_at = 0 # token过期时间戳
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36"
})
def _get_access_token(self) -> Optional[str]:
"""获取访问令牌"""
# 检查token是否有效
if self.access_token and self.token_expires_at > time.time() + 60:
return self.access_token
logging.info("获取新的access_token")
url = f"{self.base_url}/oauth2/token"
params = {
"grant_type": "client_credentials",
"appkey": self.appkey,
"appsecret": self.appsecret
}
try:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
result = response.json()
if "access_token" in result:
self.access_token = result["access_token"]
self.token_expires_at = time.time() + result.get("expires_in", 86400) # 默认为24小时
return self.access_token
else:
logging.error(f"获取access_token失败: {result.get('error_description', '未知错误')}")
return None
except RequestException as e:
logging.error(f"获取access_token请求异常: {str(e)}")
return None
def get_item_videos(self,
sku_id: str,
video_type: Optional[str] = None,
need_all: bool = False) -> Optional[Dict]:
"""
获取商品视频
:param sku_id: 商品SKU ID
:param video_type: 视频类型筛选
:param need_all: 是否返回所有视频
:return: 视频数据
"""
# 验证参数
valid_types = ["main", "detail", "scene", "compare"]
if video_type and video_type not in valid_types:
logging.error(f"无效的视频类型: {video_type}，支持: {valid_types}")
return None
# 获取有效的access_token
if not self._get_access_token():
return None
url = f"{self.base_url}/item/video"
# 构建请求参数
params = {
"sku_id": sku_id,
"access_token": self.access_token,
"need_all": "true" if need_all else "false"
}
# 添加视频类型筛选
if video_type:
params["video_type"] = video_type
try:
response = self.session.get(url, params=params, timeout=15)
response.raise_for_status()
result = response.json()
# 检查响应状态
if result.get("code") == 200:
# 格式化视频数据
return self._format_video_data(result.get("data", {}))
else:
logging.error(f"获取商品视频失败: {result.get('message', '未知错误')} (错误码: {result.get('code')})")
return None
except RequestException as e:
logging.error(f"获取商品视频请求异常: {str(e)}")
return None
except json.JSONDecodeError:
logging.error(f"商品视频响应解析失败: {response.text[:200]}...")
return None
def _format_video_data(self, video_data: Dict) -> Dict:
"""格式化视频数据"""
# 基础商品信息
item_info = {
"sku_id": video_data.get("sku_id"),
"item_id": video_data.get("item_id"),
"title": video_data.get("title"),
"total_videos": int(video_data.get("total_count", 0))
}
# 格式化视频列表
videos = []
for video in video_data.get("videos", []):
# 处理视频URL（可能有多种清晰度）
video_urls = {}
if video.get("url_list"):
for url_info in video.get("url_list"):
quality = url_info.get("quality", "unknown")
video_urls[quality] = url_info.get("url")
# 提取视频时长（转换为秒）
duration_seconds = self._parse_duration(video.get("duration", ""))
videos.append({
"video_id": video.get("video_id"),
"title": video.get("title"),
"description": video.get("description"),
"type": video.get("type"),
"type_name": self._get_video_type_name(video.get("type")),
"urls": video_urls,
"cover_url": video.get("cover_url"),
"duration": {
"original": video.get("duration"),
"seconds": duration_seconds
},
"resolution": video.get("resolution"),
"size": video.get("size"), # 视频大小，单位字节
"format": video.get("format"),
"play_count": int(video.get("play_count", 0)),
"upload_time": video.get("upload_time")
})
# 按视频类型分组
videos_by_type = {}
for video in videos:
type_name = video["type_name"]
if type_name not in videos_by_type:
videos_by_type[type_name] = []
videos_by_type[type_name].append(video)
return {
"item_info": item_info,
"videos": videos,
"videos_by_type": videos_by_type,
"raw_data": video_data # 保留原始数据
}
def _parse_duration(self, duration_str: str) -> int:
"""解析视频时长字符串为秒数"""
if not duration_str:
return 0
# 处理格式如 "00:01:23" 或 "01:23"
try:
parts = list(map(int, duration_str.split(':')))
if len(parts) == 3: # 时:分:秒
return parts[0] * 3600 + parts[1] * 60 + parts[2]
elif len(parts) == 2: # 分:秒
return parts[0] * 60 + parts[1]
elif len(parts) == 1: # 秒
return parts[0]
except:
logging.warning(f"无法解析时长: {duration_str}")
return 0
def _get_video_type_name(self, type_code: str) -> str:
"""将视频类型代码转换为名称"""
type_map = {
"main": "主视频",
"detail": "细节视频",
"scene": "场景视频",
"compare": "对比视频",
"other": "其他视频"
}
return type_map.get(type_code, type_code or "未知类型")
def download_video_cover(self, video_info: Dict, save_dir: str = "./covers/") -> bool:
"""
下载视频封面图
:param video_info: 视频信息字典
:param save_dir: 保存目录
:return: 是否下载成功
"""
import os
from urllib.parse import urlparse
if not video_info.get("cover_url"):
logging.warning("没有封面图URL")
return False
# 创建保存目录
os.makedirs(save_dir, exist_ok=True)
# 生成文件名
url_path = urlparse(video_info["cover_url"]).path
ext = os.path.splitext(url_path)[1] or ".jpg"
filename = f"{video_info['video_id']}{ext}"
save_path = os.path.join(save_dir, filename)
try:
response = self.session.get(video_info["cover_url"], timeout=15, stream=True)
response.raise_for_status()
with open(save_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=1024):
if chunk:
f.write(chunk)
logging.info(f"封面图已保存至: {save_path}")
return True
except Exception as e:
logging.error(f"下载封面图失败: {str(e)}")
return False
# 示例调用
if __name__ == "__main__":
# 替换为实际的appkey和appsecret（从京东开放平台获取）
APPKEY = "your_appkey"
APPSECRET = "your_appsecret"
# 替换为目标商品SKU ID
SKU_ID = "100012345678"
# 初始化API客户端
api = JDItemVideoAPI(APPKEY, APPSECRET)
# 获取商品视频
video_result = api.get_item_videos(
sku_id=SKU_ID,
# video_type="main", # 可选，指定视频类型
need_all=True # 获取所有视频
)
if video_result:
print(f"=== 商品视频信息 (SKU: {SKU_ID}) ===")
print(f"商品标题: {video_result['item_info']['title']}")
print(f"视频总数: {video_result['item_info']['total_videos']}")
print(f"视频类型分布: {', '.join([f'{k}: {len(v)}个' for k, v in video_result['videos_by_type'].items()])}\n")
# 打印所有视频信息
for i, video in enumerate(video_result["videos"], 1):
print(f"{i}. {video['type_name']}: {video['title']}")
print(f" 视频ID: {video['video_id']}")
print(f" 时长: {video['duration']['original']} ({video['duration']['seconds']}秒)")
print(f" 分辨率: {video['resolution'] or '未知'}")
print(f" 播放次数: {video['play_count']}")
print(f" 上传时间: {video['upload_time'] or '未知'}")
print(f" 可用清晰度: {', '.join(video['urls'].keys())}")
print(f" 封面图: {'有' if video['cover_url'] else '无'}")
# 下载第一个视频的封面图
if i == 1:
api.download_video_cover(video)
print("-" * 100)