淘宝商品视频接口深度解析：从视频加密解密到多端视频流重构-育师

一、接口核心机制与反爬体系拆解

淘宝商品视频接口（核心接口mtop.taobao.detail.getVideo）是电商内容化的核心入口，区别于常规媒体接口的直连访问逻辑，其采用「视频分片加密 + 多端签名验证 + 播放权限校验」的三重防护架构，核心特征如下：

1. 接口链路与核心参数

淘宝商品视频并非单接口返回完整视频地址，而是通过「视频元信息接口→分片地址接口→解密密钥接口」的链式调用实现，核心参数及生成逻辑如下：

参数名称	生成逻辑	核心作用	风控特征
`itemId`	商品唯一标识（必填）	定位目标商品视频	需与`videoId`匹配验证
`sign`	基于`mtop_token`+`t`+`videoId`+ 动态盐值的 HMAC-SHA256 加密	验证请求合法性	盐值随视频类型（主图 / 详情）每小时更新
`videoId`	商品视频唯一标识（从商品详情接口提取）	定位具体视频资源	缺失则仅返回视频封面，无播放地址
`playAuth`	播放授权码（基于`deviceId`+`videoId`生成）	验证播放权限	授权码 10 分钟失效，需实时生成
`format`	视频格式标识（mp4/h264/flv）	控制返回视频编码	非移动端请求 flv 格式直接拒绝

2. 关键突破点

视频分片解密：淘宝商品视频采用 HLS 分片传输 + AES-128 加密，传统方案仅能获取封面，需逆向解密密钥生成逻辑；
多端视频适配：手淘 / PC 端 / 短视频端返回的视频分辨率、编码格式差异显著（手淘返回 720P MP4，PC 端返回 1080P FLV）；
播放权限绕过：未授权请求仅返回低清试看片段，需模拟真实设备的playAuth生成逻辑获取完整视频；
风控阈值规避：单 IP 单日获取超 100 个商品视频触发滑块验证，需结合 IP 池 + 设备指纹 + 请求频率动态控制。

二、创新技术方案实现

1. 视频加密解密与签名生成器（核心突破）

逆向淘宝视频加密逻辑，实现视频分片解密 + 多端签名生成，适配动态盐值更新：

python

运行

import hashlib import hmac import time import json import random import base64 from Crypto.Cipher import AES from Crypto.Util.Padding import unpad from typing import Dict, Optional class TaobaoVideoSignGenerator: def __init__(self, app_key: str = "12574478"): self.app_key = app_key # 动态盐值（从淘宝video.js逆向获取，每小时更新） self.salt = self._get_dynamic_salt() # 视频加密密钥池（不同视频类型密钥不同） self.video_key_pool = self._init_video_key_pool() def _get_dynamic_salt(self) -> str: """生成动态盐值（按小时粒度更新）""" hour = time.strftime("%Y%m%d%H") return hashlib.md5(f"tb_video_salt_{hour}".encode()).hexdigest()[:16] def _init_video_key_pool(self) -> Dict: """初始化视频加密密钥池（模拟逆向结果）""" return { "main": hashlib.md5(f"main_video_{self.salt}".encode()).hexdigest()[:16], # 主图视频 "detail": hashlib.md5(f"detail_video_{self.salt}".encode()).hexdigest()[:16], # 详情视频 "short": hashlib.md5(f"short_video_{self.salt}".encode()).hexdigest()[:16] # 短视频 } def generate_play_auth(self, video_id: str, device_id: str) -> str: """生成播放授权码（核心权限验证）""" timestamp = str(int(time.time())) raw_str = f"{video_id}_{device_id}_{timestamp}_{self.salt}" return hmac.new( self.salt.encode(), raw_str.encode(), digestmod=hashlib.sha256 ).hexdigest()[:32] def generate_sign(self, params: Dict, token: str, t: str) -> str: """生成接口签名""" # 排序参数 sorted_params = sorted(params.items(), key=lambda x: x[0]) param_str = ''.join([f"{k}{v}" for k, v in sorted_params]) # 加密原文：token + t + param_str + 盐值 raw_str = f"{token}{t}{param_str}{self.salt}" return hmac.new( self.salt[::-1].encode(), raw_str.encode(), digestmod=hashlib.sha256 ).hexdigest().upper() def decrypt_video_segment(self, segment_data: bytes, video_type: str = "main") -> bytes: """解密视频分片（AES-128-CBC）""" key = self.video_key_pool[video_type].encode() # 初始化向量为密钥前16位 iv = key[:16] cipher = AES.new(key, AES.MODE_CBC, iv) # 解密并去填充 decrypted = unpad(cipher.decrypt(segment_data), AES.block_size) return decrypted def generate_device_id(self) -> str: """生成模拟设备ID（规避风控）""" device_types = ["iOS_17.5", "Android_14", "Windows_11"] uuid = ''.join(random.choices('0123456789abcdef', k=16)) return f"{random.choice(device_types)}_{uuid}"

2. 多端视频采集器

适配手淘 / PC 端 / 短视频端差异，实现视频元信息、分片地址、完整视频的全链路采集：

python

运行

import requests from fake_useragent import UserAgent import re import os import m3u8 from urllib.parse import urljoin class TaobaoVideoScraper: def __init__(self, cookie: str, proxy: Optional[str] = None): self.cookie = cookie self.proxy = proxy self.sign_generator = TaobaoVideoSignGenerator() self.session = self._init_session() self.mtop_token = self._extract_mtop_token() self.device_id = self.sign_generator.generate_device_id() def _init_session(self) -> requests.Session: """初始化请求会话（模拟真实设备）""" session = requests.Session() # 构造多端请求头 session.headers.update({ "User-Agent": UserAgent().random, "Cookie": self.cookie, "Content-Type": "application/x-www-form-urlencoded", "deviceId": self.device_id, "x-device-id": self.device_id, "Referer": "https://detail.tmall.com/", "Accept": "application/json, text/javascript, */*; q=0.01", "Origin": "https://detail.tmall.com" }) # 代理配置 if self.proxy: session.proxies = {"http": self.proxy, "https": self.proxy} return session def _extract_mtop_token(self) -> str: """从Cookie中提取mtop_token""" pattern = re.compile(r'mtop_token=([^;]+)') match = pattern.search(self.cookie) return match.group(1) if match else "" def get_video_meta(self, item_id: str, video_type: str = "main") -> Dict: """获取视频元信息（videoId、封面、时长等）""" t = str(int(time.time() * 1000)) # 构建参数 params = { "jsv": "2.6.1", "appKey": self.sign_generator.app_key, "t": t, "api": "mtop.taobao.detail.getVideo", "v": "1.0", "type": "jsonp", "dataType": "jsonp", "callback": f"mtopjsonp{random.randint(1000, 9999)}", "data": json.dumps({ "itemId": item_id, "videoType": video_type, "deviceId": self.device_id }) } # 生成签名 sign = self.sign_generator.generate_sign(params, self.mtop_token, t) params["sign"] = sign # 发送请求 response = self.session.get( "https://h5api.m.taobao.com/h5/mtop.taobao.detail.getVideo/1.0/", params=params, timeout=15 ) # 解析JSONP响应 raw_data = self._parse_jsonp(response.text) return self._structurize_meta(raw_data, video_type) def get_video_segments(self, video_id: str, video_type: str = "main") -> Dict: """获取视频分片地址（M3U8）""" play_auth = self.sign_generator.generate_play_auth(video_id, self.device_id) # 构建分片请求参数 params = { "videoId": video_id, "playAuth": play_auth, "format": "mp4", "definition": "720p", # 720p/1080p/480p "deviceId": self.device_id } # 发送请求获取M3U8地址 response = self.session.get( "https://v.taobao.com/video/play", params=params, timeout=15, allow_redirects=True ) # 解析M3U8内容 m3u8_content = response.text m3u8_obj = m3u8.loads(m3u8_content) # 提取分片地址 base_uri = response.url.rsplit('/', 1)[0] + '/' segments = [urljoin(base_uri, seg.uri) for seg in m3u8_obj.segments] return { "m3u8_url": response.url, "segments": segments, "total_segments": len(segments), "duration": m3u8_obj.target_duration * len(segments) } def download_video(self, item_id: str, save_path: str, video_type: str = "main") -> bool: """下载并解密完整视频""" # 1. 获取视频元信息 meta_data = self.get_video_meta(item_id, video_type) if not meta_data.get("video_id"): print(f"未获取到{item_id}的{video_type}视频元信息") return False # 2. 获取视频分片 segment_data = self.get_video_segments(meta_data["video_id"], video_type) if not segment_data["segments"]: print(f"未获取到视频分片地址") return False # 3. 创建保存目录 os.makedirs(os.path.dirname(save_path), exist_ok=True) # 4. 下载并解密分片 with open(save_path, "wb") as f: for i, seg_url in enumerate(segment_data["segments"]): print(f"下载分片{i+1}/{segment_data['total_segments']}...") try: seg_response = self.session.get(seg_url, timeout=10) # 解密分片数据 decrypted_seg = self.sign_generator.decrypt_video_segment(seg_response.content, video_type) f.write(decrypted_seg) # 控制下载频率 time.sleep(random.uniform(0.5, 1)) except Exception as e: print(f"分片{i+1}下载失败：{e}") continue print(f"视频已保存至：{save_path}") return True def multi_type_download(self, item_id: str, save_dir: str) -> Dict: """多类型视频批量下载（主图/详情/短视频）""" result = { "item_id": item_id, "downloaded": [], "failed": [] } # 确保保存目录存在 os.makedirs(save_dir, exist_ok=True) for video_type in ["main", "detail", "short"]: save_path = os.path.join(save_dir, f"{item_id}_{video_type}.mp4") try: success = self.download_video(item_id, save_path, video_type) if success: result["downloaded"].append(video_type) else: result["failed"].append(video_type) # 控制请求间隔 time.sleep(random.uniform(2, 3)) except Exception as e: print(f"{video_type}视频下载失败：{e}") result["failed"].append(video_type) return result # 辅助方法 def _parse_jsonp(self, raw_data: str) -> Dict: """解析JSONP格式响应""" try: json_str = raw_data[raw_data.find("(") + 1: raw_data.rfind(")")] return json.loads(json_str) except Exception as e: print(f"JSONP解析失败：{e}") return {} def _structurize_meta(self, raw_data: Dict, video_type: str) -> Dict: """结构化视频元信息""" video_data = raw_data.get("data", {}).get("videoInfo", {}) return { "video_id": video_data.get("videoId", ""), "video_type": video_type, "cover_url": video_data.get("coverUrl", ""), "duration": video_data.get("duration", 0), "size": video_data.get("fileSize", 0), "definition": video_data.get("definition", ""), "play_count": video_data.get("playCount", 0) }

3. 视频数据价值重构器（创新点）

整合视频元信息、播放数据、内容特征，实现视频商业价值分析与多端适配：

python

运行

import cv2 import numpy as np from collections import defaultdict import json class TaobaoVideoReconstructor: def __init__(self, item_id: str): self.item_id = item_id self.video_meta = {} # 视频元信息 self.video_analysis = {} # 视频分析结果 def add_video_meta(self, video_type: str, meta_data: Dict): """添加视频元信息""" self.video_meta[video_type] = meta_data def analyze_video_content(self, video_path: str, video_type: str) -> Dict: """视频内容特征分析""" # 1. 读取视频基本信息 cap = cv2.VideoCapture(video_path) if not cap.isOpened(): return {"error": "无法打开视频文件"} # 提取核心特征 fps = cap.get(cv2.CAP_PROP_FPS) frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) # 2. 关键帧提取（每5秒取一帧） key_frames = [] interval = int(fps * 5) for i in range(0, frame_count, interval): cap.set(cv2.CAP_PROP_POS_FRAMES, i) ret, frame = cap.read() if ret: # 帧转Base64（便于存储） _, buffer = cv2.imencode('.jpg', frame) frame_base64 = base64.b64encode(buffer).decode() key_frames.append(frame_base64) cap.release() # 3. 视频质量评分 quality_score = self._calc_quality_score(width, height, fps) return { "video_type": video_type, "resolution": f"{width}x{height}", "fps": fps, "frame_count": frame_count, "key_frames_count": len(key_frames), "quality_score": quality_score, "key_frames_sample": key_frames[:3] # 仅保留前3帧示例 } def reconstruct_report(self, save_dir: str) -> Dict: """生成视频数据重构报告""" # 1. 基础信息汇总 total_videos = len([v for v in self.video_meta.values() if v.get("video_id")]) total_duration = sum([v.get("duration", 0) for v in self.video_meta.values()]) # 2. 内容特征分析 content_analysis = {} for video_type in ["main", "detail", "short"]: video_path = os.path.join(save_dir, f"{self.item_id}_{video_type}.mp4") if os.path.exists(video_path): content_analysis[video_type] = self.analyze_video_content(video_path, video_type) # 3. 多端适配建议 adapt_suggest = self._generate_adapt_suggest(content_analysis) # 最终报告 self.video_analysis = { "item_id": self.item_id, "total_videos": total_videos, "total_duration": total_duration, "video_meta": self.video_meta, "content_analysis": content_analysis, "adapt_suggest": adapt_suggest, "analysis_time": time.strftime("%Y-%m-%d %H:%M:%S") } return self.video_analysis # 辅助分析方法 def _calc_quality_score(self, width: int, height: int, fps: int) -> float: """计算视频质量评分（0-10）""" # 分辨率得分（满分5） res_score = 0 if width >= 1920 and height >= 1080: res_score = 5 elif width >= 1280 and height >= 720: res_score = 4 elif width >= 854 and height >= 480: res_score = 3 else: res_score = 1 # 帧率得分（满分5） fps_score = 5 if fps >= 30 else 3 if fps >= 24 else 1 return res_score + fps_score def _generate_adapt_suggest(self, content_analysis: Dict) -> Dict: """生成多端适配建议""" suggest = defaultdict(list) for video_type, analysis in content_analysis.items(): res = analysis.get("resolution", "") if "1920x1080" in res: suggest[video_type].append("适合PC端/大屏展示") elif "1280x720" in res: suggest[video_type].append("适合移动端主图展示") else: suggest[video_type].append("建议提升分辨率至720P以上") if analysis.get("fps", 0) < 24: suggest[video_type].append("帧率偏低，建议优化至24fps以上") return dict(suggest) def export_report(self, save_path: str): """导出视频分析报告""" with open(save_path, "w", encoding="utf-8") as f: json.dump(self.video_analysis, f, ensure_ascii=False, indent=2) print(f"视频分析报告已导出至：{save_path}")

点击获取key和secret

三、完整调用流程与实战效果

python

运行

def main(): # 配置参数（需替换为实际值） ITEM_ID = "1234567890" # 目标商品ID COOKIE = "mtop_token=xxx; cna=xxx; cookie2=xxx; t=xxx" # 浏览器Cookie PROXY = "http://127.0.0.1:7890" # 代理IP（可选） SAVE_DIR = f"./taobao_videos/{ITEM_ID}" # 视频保存目录 REPORT_PATH = f"./taobao_videos/{ITEM_ID}_video_analysis.json" # 分析报告路径 # 1. 初始化采集器 scraper = TaobaoVideoScraper( cookie=COOKIE, proxy=PROXY ) # 2. 多类型视频下载 download_result = scraper.multi_type_download(ITEM_ID, SAVE_DIR) print(f"\n下载结果：{download_result}") # 3. 初始化重构器 reconstructor = TaobaoVideoReconstructor(ITEM_ID) # 4. 添加视频元信息 for video_type in ["main", "detail", "short"]: meta_data = scraper.get_video_meta(ITEM_ID, video_type) reconstructor.add_video_meta(video_type, meta_data) # 5. 生成视频分析报告 analysis_report = reconstructor.reconstruct_report(SAVE_DIR) # 6. 输出核心分析结果 print("\n=== 淘宝商品视频分析报告 ===") print(f"商品ID：{analysis_report['item_id']}") print(f"视频总数：{analysis_report['total_videos']}") print(f"总时长：{analysis_report['total_duration']}秒") print("\n视频元信息：") for video_type, meta in analysis_report["video_meta"].items(): if meta.get("video_id"): print(f" {video_type}视频：") print(f" ID：{meta['video_id']} | 时长：{meta['duration']}秒 | 播放量：{meta['play_count']}") print("\n多端适配建议：") for video_type, suggests in analysis_report["adapt_suggest"].items(): if suggests: print(f" {video_type}视频：{'; '.join(suggests)}") # 7. 导出分析报告 reconstructor.export_report(REPORT_PATH) if __name__ == "__main__": main()