京东 item_video 接口深度分析及 Python 实现

京东的 item_video 接口是用于获取商品相关视频资源的专业接口，能够获取商品主视频、细节展示视频、使用教程视频等多种类型的视频资源。这些视频内容对于商品展示、用户体验提升、竞品分析等场景具有重要价值，是电商内容分析中不可或缺的一部分。

一、接口核心特性分析

1. 接口功能与定位

• 核心功能：获取京东商品的相关视频资源，包括视频 URL、时长、分辨率、封面图等信息

• 视频类型：

  • 主视频：商品主展示视频，通常在商品详情页顶部展示
  • 细节视频：展示商品细节特征的视频片段
  • 场景视频：商品使用场景展示视频
  • 对比视频：与同类商品的对比视频（部分商品有）

• 应用场景：

  • 商品内容聚合平台
  • 竞品视频内容分析
  • 多平台商品展示系统
  • 电商内容营销研究

2. 认证机制

京东开放平台采用 appkey + access_token 的认证方式：

• 开发者在京东开放平台注册应用，获取 appkey 和 appsecret
• 使用 appkey 和 appsecret 获取 access_token（有有效期限制）
• 每次接口调用时，在请求参数中携带 access_token 进行身份验证

3. 核心参数与响应结构

请求参数

响应核心字段

• 视频列表：每个视频包含

  • 基础信息：视频 ID、标题、描述、类型
  • 资源信息：视频 URL（可能包含多种清晰度）、封面图 URL
  • 规格信息：时长、分辨率、大小、格式
  • 额外信息：上传时间、播放次数 二、Python 脚本实现

以下是调用京东 item_video 接口的完整 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 = "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)

三、接口调用注意事项

1. 调用限制与规范

• QPS 限制：京东开放平台对视频接口有 QPS 限制，通常为 5-10 次 / 秒
• 视频权限：部分视频可能受版权保护，仅允许在特定场景下使用
• 带宽考虑：视频文件较大，批量获取时需考虑带宽和存储成本
• URL 时效：视频 URL 可能有时效性（通常 24 小时），需及时使用或缓存
• 合规使用：获取的视频资源需遵守京东平台规定，不得用于非法用途

2. 常见错误及解决方案

3. 数据解析要点

• 视频 URL 处理：同一视频可能有多个清晰度版本，需根据需求选择
• 时长转换：将原始时长字符串转换为秒数便于比较和筛选
• 视频类型识别：不同类型视频有不同用途，需正确分类
• 封面图下载：封面图是视频内容的重要预览，建议同步获取
• 异常处理：部分视频可能缺少某些字段（如时长、分辨率），需做容错处理

四、应用场景与扩展建议 典型应用场景

• 商品内容聚合平台：整合多平台商品视频，提供丰富的商品展示
• 竞品视频分析系统：分析竞争对手的商品视频内容策略
• 电商内容质量评估：评估商品视频的完整性和质量
• 视频推荐系统：基于视频内容为用户推荐相关商品 扩展建议
• 实现视频下载功能：支持按清晰度下载视频（需遵守版权规定）
• 视频内容分析：结合 AI 技术提取视频中的商品特征和关键词
• 视频质量评估：自动评估视频分辨率、时长等质量指标
• 多平台视频对比：对比同一商品在不同平台的视频展示策略
• 视频内容检索：构建视频内容索引，支持按内容搜索相关商品视频 通过合理使用京东 item_video 接口，开发者可以获取丰富的商品视频资源，为商品展示、内容分析和用户体验提升提供有力支持。使用时需遵守京东开放平台的相关规定和版权要求，确保视频资源的合法使用。