淘宝开放平台淘口令解析实战：合规解密、参数清洗、真实 URL 结构化（附 Python 代码）

import requests
import hashlib
import time
import re
from urllib.parse import unquote

class TaoBaoTpwdParseAPI:
    def __init__(self, app_key, app_secret, session_key):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session_key = session_key
        self.api_url = "https://eco.taobao.com/router/rest"
        self.timeout = 15

    def _make_sign(self, params):
        # 淘宝TOP标准MD5签名（防篡改核心）
        sorted_items = sorted(params.items())
        sign_str = self.app_secret + "".join(f"{k}{v}" for k, v in sorted_items) + self.app_secret
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()

    def _clean_tpwd(self, raw_content):
        # 智能清洗：提取核心口令串（过滤文案、表情、符号）
        if not raw_content:
            return ""
        # 匹配￥...￥或$...$格式口令
        pattern = r"[￥$](.*?)[￥$]"
        match = re.search(pattern, raw_content)
        if match:
            return match.group(1).strip()
        # 兜底：过滤特殊字符后返回
        return re.sub(r"[^\w\d]", "", raw_content).strip()

    def parse_tpwd(self, raw_tpwd_content):
        """
        解析淘口令获取真实URL+商品信息
        :param raw_tpwd_content: 原始淘口令文本（含文案/表情）
        :return: 结构化结果（code、msg、url、num_iid、title等）
        """
        # 1. 清洗口令
        clean_pwd = self._clean_tpwd(raw_tpwd_content)
        if not clean_pwd:
            return {"code": -1, "msg": "口令清洗失败，无有效内容"}

        # 2. 构造公共参数
        params = {
            "method": "taobao.tbk.tpwd.parse",
            "app_key": self.app_key,
            "session": self.session_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "password_content": clean_pwd
        }
        params["sign"] = self._make_sign(params)

        try:
            # 3. 发送请求
            resp = requests.post(self.api_url, data=params, timeout=self.timeout)
            result = resp.json()

            # 4. 处理错误响应
            if "error_response" in result:
                err = result["error_response"]
                return {"code": -2, "msg": f"接口错误：{err.get('sub_msg', '未知异常')}"}

            # 5. 解析真实数据
            data = result.get("tbk_tpwd_parse_response", {})
            origin_url = unquote(data.get("origin_url", ""))  # 真实URL解码
            num_iid = data.get("num_iid", "")
            title = data.get("title", "")
            price = data.get("price", "")
            pic_url = data.get("pic_url", "")

            if not origin_url:
                return {"code": -3, "msg": "解析成功，但无真实URL（可能为非商品口令）"}

            # 防风控间隔
            time.sleep(0.8)
            return {
                "code": 200,
                "msg": "解析成功",
                "origin_url": origin_url,
                "num_iid": num_iid,
                "title": title,
                "price": price,
                "pic_url": pic_url
            }

        except Exception as e:
            return {"code": 500, "msg": f"请求异常：{str(e)}"}

# ———— 调用示例 ————
if __name__ == "__main__":
    # 替换为你的开放平台密钥与授权session
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    SESSION_KEY = "your_session_key"

    api = TaoBaoTpwdParseAPI(APP_KEY, APP_SECRET, SESSION_KEY)
    # 原始淘口令（含文案、表情）
    raw_tpwd = "🔥限时秒杀￥abc123￥点击抢低价商品"
    res = api.parse_tpwd(raw_tpwd)
    print(res)