Python请求淘宝天猫API常见问题解析与解决方案

python
# 正确配置认证信息示例
class TaobaoClient:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://eco.taobao.com/router/rest"
    def generate_sign(self, params):
        # 淘宝MD5签名算法实现
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        sign_str = self.app_secret + ''.join([f"{k}{v}" for k, v in sorted_params]) + self.app_secret
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    def call_api(self, method, params):
        common_params = {
            "app_key": self.app_key,
            "method": method,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5"
        }
        all_params = {**common_params, **params}
        all_params["sign"] = self.generate_sign(all_params)
        
        try:
            response = requests.get(self.api_url, params=all_params, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            print(f"HTTP错误: {e.response.text}")
            return None

python
def validate_params(params):
    # 商品ID校验
    if 'num_iid' in params and not params['num_iid'].isdigit():
        raise ValueError("商品ID必须为纯数字")
    
    # fields参数处理
    if 'fields' in params:
        fields = params['fields'].split(',')
        valid_fields = {'num_iid', 'title', 'price', 'pic_url', 'sku'}  # 示例字段集
        invalid_fields = set(fields) - valid_fields
        if invalid_fields:
            raise ValueError(f"无效字段: {','.join(invalid_fields)}")
        params['fields'] = ','.join(fields)
    
    return params
# 调用示例
params = {
    "num_iid": "123456789012",  # 正确格式
    "fields": "title,price,pic_url",  # 英文逗号分隔
    "page_no": 1
}
validated_params = validate_params(params)

python
import time
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def rate_limited_call(client, method, params):
    try:
        return client.call_api(method, params)
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 429:
            retry_after = int(e.response.headers.get('Retry-After', 5))
            print(f"触发限流，等待{retry_after}秒后重试...")
            time.sleep(retry_after)
            raise  # 重新抛出异常触发重试机制
        raise
# 使用示例
client = TaobaoClient("your_app_key", "your_app_secret")
for i in range(100):
    params = {"num_iid": f"1000000000{i}"}
    try:
        result = rate_limited_call(client, "taobao.item.get", params)
        print(f"成功获取商品{i}数据")
    except Exception as e:
        print(f"调用失败: {str(e)}")

python
def safe_json_parse(response_text):
    try:
        # 去除可能的BOM头
        if response_text.startswith('\ufeff'):
            response_text = response_text[3:]
        return json.loads(response_text)
    except json.JSONDecodeError:
        # 尝试修复常见JSON错误
        if 'callback(' in response_text:  # 处理JSONP情况
            response_text = response_text.split('(', 1)[1].rsplit(')', 1)[0]
            return json.loads(response_text)
        
        # 记录原始响应用于调试
        with open('error_response.log', 'a', encoding='utf-8') as f:
            f.write(response_text)
        raise ValueError("无效的JSON响应，已记录到error_response.log")
# 使用示例
response = requests.get("https://eco.taobao.com/router/rest", params=all_params)
try:
    data = safe_json_parse(response.text)
    if "error_response" in data:
        handle_error(data["error_response"])
    else:
        process_data(data)
except Exception as e:
    print(f"数据处理失败: {str(e)}")

python
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
@app.route('/taobao_proxy')
def taobao_proxy():
    item_id = request.args.get('item_id')
    if not item_id:
        return jsonify({"error": "缺少商品ID参数"}), 400
    # 构造淘宝API请求（实际项目中应使用后端签名）
    params = {
        "method": "taobao.item.get",
        "app_key": "your_app_key",
        "num_iid": item_id,
        "fields": "title,price,pic_url",
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "format": "json",
        "v": "2.0",
        "sign_method": "md5"
    }
    
    try:
        # 实际项目中应在此处生成正确签名
        params["sign"] = "DUMMY_SIGN"  # 示例中省略签名生成
        response = requests.get("https://eco.taobao.com/router/rest", params=params)
        response.raise_for_status()
        return jsonify(response.json())
    except Exception as e:
        return jsonify({"error": str(e)}), 500
if __name__ == '__main__':
    app.run(port=5000)