淘宝商品视频接口实战：从合规调用到视频解析（2026 独家版）

import requests
import time
import hashlib
import json
import re
import os
from urllib.parse import urlparse, parse_qs
from moviepy.editor import VideoFileClip

# 核心配置（替换为自己的凭证）
APP_KEY = "你的AppKey"
APP_SECRET = "你的AppSecret"
ACCESS_TOKEN = "你的AccessToken"
API_URL = "https://eco.taobao.com/router/rest"
DOWNLOAD_DIR = "taobao_videos"  # 视频保存目录

class TaobaoVideoApi:
    def __init__(self, app_key, app_secret, access_token):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        # 创建保存目录
        if not os.path.exists(DOWNLOAD_DIR):
            os.makedirs(DOWNLOAD_DIR)

    def generate_sign(self, params):
        """生成接口签名（严格按阿里规范，避免签名错误）"""
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        sign_str = self.app_secret
        for key, value in sorted_params:
            if value:
                sign_str += f"{key}{value}"
        sign_str += self.app_secret
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()

    def parse_video_url(self, raw_url):
        """
        解析防盗链视频地址（网上教程未覆盖的核心逻辑）
        将带签名的原始地址转为可直接下载的链接
        """
        try:
            # 提取核心播放地址，去除临时签名参数
            parsed_url = urlparse(raw_url)
            query_params = parse_qs(parsed_url.query)
            # 保留视频核心参数，剔除过期签名
            keep_params = ["v", "vid", "type"]
            new_query = "&".join([f"{k}={v[0]}" for k, v in query_params.items() if k in keep_params])
            # 重构播放地址
            base_url = f"{parsed_url.scheme}://{parsed_url.netloc}{parsed_url.path}"
            valid_url = f"{base_url}?{new_query}"
            return valid_url
        except Exception as e:
            raise Exception(f"视频地址解析失败：{str(e)}")

    def convert_to_mp4(self, flv_path):
        """将FLV格式视频转为MP4（解决格式兼容性问题）"""
        try:
            mp4_path = flv_path.replace(".flv", ".mp4")
            clip = VideoFileClip(flv_path)
            clip.write_videofile(mp4_path, codec="libx264")
            clip.close()
            os.remove(flv_path)  # 删除原FLV文件
            return mp4_path
        except Exception as e:
            raise Exception(f"格式转换失败：{str(e)}")

    def get_video(self, item_id):
        """
        获取商品视频并下载
        :param item_id: 淘宝商品ID
        :return: 视频保存路径/错误信息
        """
        # 组装接口参数
        params = {
            "method": "taobao.item.video.get",
            "app_key": self.app_key,
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "access_token": self.access_token,
            "item_id": item_id
        }
        # 生成签名
        params["sign"] = self.generate_sign(params)

        try:
            # 调用接口
            response = requests.post(
                API_URL,
                data=params,
                timeout=30,
                headers={"Content-Type": "application/x-www-form-urlencoded;charset=utf-8"}
            )
            response.raise_for_status()
            result = response.json()

            # 处理接口错误
            if "error_response" in result:
                return {"success": False, "msg": f"接口错误：{result['error_response']['msg']}（错误码：{result['error_response']['code']}）"}

            # 提取视频信息
            video_info = result.get("item_video_get_response", {}).get("video", {})
            if not video_info:
                return {"success": False, "msg": "该商品无视频"}

            video_url = video_info.get("play_url")
            video_duration = float(video_info.get("duration", 0))
            video_resolution = video_info.get("resolution", "unknown")

            # 过滤无效视频（时长<3秒）
            if video_duration < 3:
                return {"success": False, "msg": "视频时长过短（<3秒），跳过下载"}

            # 解析并下载视频
            valid_url = self.parse_video_url(video_url)
            video_filename = f"{item_id}_{int(time.time())}.flv"
            video_path = os.path.join(DOWNLOAD_DIR, video_filename)

            # 下载视频
            video_resp = requests.get(valid_url, timeout=60, stream=True)
            with open(video_path, "wb") as f:
                for chunk in video_resp.iter_content(chunk_size=1024*1024):
                    if chunk:
                        f.write(chunk)

            # 转换为MP4格式
            mp4_path = self.convert_to_mp4(video_path)

            return {
                "success": True,
                "msg": "视频下载成功",
                "video_path": mp4_path,
                "duration": video_duration,
                "resolution": video_resolution
            }

        except Exception as e:
            return {"success": False, "msg": f"视频获取失败：{str(e)}"}

# 实战调用示例
if __name__ == "__main__":
    # 初始化接口实例
    video_api = TaobaoVideoApi(APP_KEY, APP_SECRET, ACCESS_TOKEN)
    
    # 调用接口：替换为实际商品ID
    item_id = "1234567890"  # 示例商品ID，需替换为有效ID
    result = video_api.get_video(item_id)

    # 打印结果
    if result["success"]:
        print(f"✅ {result['msg']}")
        print(f"📁 视频路径：{result['video_path']}")
        print(f"⏱️ 视频时长：{result['duration']}秒")
        print(f"🖥️ 分辨率：{result['resolution']}")
    else:
        print(f"❌ {result['msg']}")