抖音解析器 Cloudflare Workers 反代实现全记录

背景

在开发 AstrBot 抖音解析插件时，遇到了一个常见但棘手的问题：服务器 IP 被抖音风控，导致 API 请求返回空数据。本文记录了从简单的重试机制到最终使用 Cloudflare Workers 反代成功解决问题的完整过程。

问题分析

初始问题

插件在解析抖音链接时，dysk.py 返回 None，表现为：

视频可以偶尔解析成功
图片和实况（live photo）完全无法解析
日志显示 API 返回了空响应

问题根源

抖音的风控机制会检测请求来源 IP，当检测到异常请求模式时，会返回空响应（HTTP 200 但 body 为空）。

解决方案演进

阶段一：实现重试机制

思路：解析失败时重新创建 DouyinDownloader 实例并重试。

实现要点：

最多重试 5 次
每次重试间隔 5 秒
重试时强制重新创建实例（获取新的 Cookie）

代码片段：

PYTHON

def _parse_douyin_sync(self, url):
    max_retries = 5
    retry_delay = 5

    for attempt in range(max_retries):
        try:
            current_time = time.time()

            if attempt == 0:
                # 首次尝试：复用实例（5分钟内）
                if self.dy_downloader is None or (current_time - self.dy_downloader_time) > 300:
                    self.dy_downloader = DouyinDownloader()
                    self.dy_downloader_time = current_time
            else:
                # 重试时：强制重新创建实例
                self.dy_downloader = DouyinDownloader()
                self.dy_downloader_time = current_time

            result = self.dy_downloader.get_detail(url)

            if result is not None:
                return (result, self.dy_downloader)

            if attempt < max_retries - 1:
                time.sleep(retry_delay)
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(retry_delay)
            else:
                raise

    return (None, self.dy_downloader)

效果：重试机制能提高成功率，但治标不治本，IP 被风控后依然无法解析。

阶段二：引入 Cloudflare Workers 反代

思路：

API 请求通过 CF Workers 代理，利用 CF 的 IP 池避免风控
视频/图片下载直连 CDN，节省 CF 流量

预期成功率：75-93%

阶段三：实现过程中的坑

坑1：CF Workers 响应体为空

现象：

Response status: 200
Response headers: {'content-length': '0', 'content-type': 'text/plain'}
Response text length: 0

原因：CF Workers 的自动 gzip 压缩导致响应体传输失败。

尝试的解决方案：

❌ 直接返回 fetch() 结果 - 响应体丢失
❌ 使用 response.arrayBuffer() - 返回空 ArrayBuffer
❌ 设置 Content-Length 头 - CF 仍然强制压缩
✅ Base64 编码传输
- 成功绕过压缩问题

坑2：Cookie 未正确传递

现象：

原始响应长度: 20
原始响应内容(hex): 1f8b08000000000000ff03000000000000000000
解压后长度: 0

这是一个空的 gzip 文件，说明抖音服务器返回了空响应。

原因分析：

ttwid API 请求成功（Response text length: 205）
抖音详情 API 请求失败（Response text length: 0）
CF Workers 日志显示：Request Cookie: No Cookie

根本原因：Python 的 requests.Session() 虽然管理 Cookie，但通过 CF Workers 代理时，Cookie 没有被自动发送到 CF Workers。

解决方案：手动构建 Cookie 请求头。

最终解决方案

架构设计

┌─────────────┐      ┌──────────────────┐      ┌─────────────┐
│   Python    │─────▶│  CF Workers      │─────▶│   抖音 API   │
│   Client    │      │  (反代 + Base64)  │      │             │
└─────────────┘      └──────────────────┘      └─────────────┘
      │                                                │
      │                                                │
      └────────────────────────────────────────────────┘
                    直连下载视频/图片

完整代码

1. Cloudflare Workers 反代脚本

// cloudflare_worker.js
// Cloudflare Workers 反代脚本 - 代理抖音 API
export default {
  async fetch(request) {
    // 处理 OPTIONS 预检请求
    if (request.method === "OPTIONS") {
      return new Response(null, {
        status: 200,
        headers: {
          "Access-Control-Allow-Origin": "*",
          "Access-Control-Allow-Methods": "GET, POST, PUT, DELETE, OPTIONS",
          "Access-Control-Allow-Headers": "*",
          "Access-Control-Max-Age": "86400",
        },
      });
    }

    const url = new URL(request.url);

    // 支持多个目标域名
    const targetHosts = {
      "douyin": "www.douyin.com",
      "ttwid": "ttwid.bytedance.com"
    };

    // 从路径中提取目标类型
    // 例如: /douyin/aweme/v1/web/aweme/detail/ 或 /ttwid/ttwid/union/register/
    const pathMatch = url.pathname.match(/^\/(douyin|ttwid)(\/.*)/);
    if (!pathMatch) {
      return new Response("Invalid path. Use /douyin/* or /ttwid/*", { status: 400 });
    }

    const targetType = pathMatch[1];
    const targetPath = pathMatch[2];
    const targetHost = targetHosts[targetType];

    // 构建目标请求 URL
    const targetUrl = `https://${targetHost}${targetPath}${url.search}`;

    try {
      // 复制请求头
      const headers = new Headers(request.headers);
      headers.set("Host", targetHost);

      // 调试：打印请求中的 Cookie（生产环境应注释掉）
      // console.log('Request Cookie:', headers.get('Cookie') || 'No Cookie');

      // 删除 CF 相关头
      headers.delete("cf-connecting-ip");
      headers.delete("cf-ipcountry");
      headers.delete("cf-ray");
      headers.delete("cf-visitor");

      // 发起请求
      const response = await fetch(targetUrl, {
        method: request.method,
        headers: headers,
        body: request.body,
        redirect: 'follow'
      });

      // 调试：打印响应状态（生产环境应注释掉）
      // console.log('Response status:', response.status);
      // console.log('Response Content-Type:', response.headers.get('Content-Type'));

      // 完全读取响应
      const responseText = await response.text();

      // 调试：打印响应长度（生产环境应注释掉）
      // console.log('Response text length:', responseText.length);

      // Base64 编码以绕过 CF 自动压缩
      const base64Data = btoa(unescape(encodeURIComponent(responseText)));

      // 返回 Base64 编码的数据
      return new Response(JSON.stringify({
        data: base64Data,
        encoding: 'base64'
      }), {
        status: response.status,
        headers: {
          'Content-Type': 'application/json; charset=utf-8',
          'Access-Control-Allow-Origin': '*',
        },
      });
    } catch (error) {
      return new Response(
        JSON.stringify({
          error: "代理请求失败",
          message: error.message,
        }),
        {
          status: 500,
          headers: {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
          },
        }
      );
    }
  },
};

2. Python 端关键代码（dysk.py 片段）

PYTHON

class DouyinDownloader:
    def __init__(self, enable_cf_proxy=False, cf_proxy_url=""):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": USERAGENT,
            "Referer": "https://www.douyin.com/",
        })
        self.ab = ABogus(USERAGENT)
        self.extractor = Extractor()
        self.enable_cf_proxy = enable_cf_proxy
        self.cf_proxy_url = cf_proxy_url.rstrip("/") if cf_proxy_url else ""

        print("正在初始化 (获取 ttwid/msToken)...")
        self._init_tokens()

    def _init_tokens(self):
        base_str = string.digits + string.ascii_letters
        ms_token = "".join(random.choice(base_str) for _ in range(156))
        self.session.cookies.set("msToken", ms_token, domain=".douyin.com")
        data = {"region": "cn", "aid": 1768, "needFid": False, "service": "www.ixigua.com",
                "migrate_info": {"ticket": "", "source": "node"}, "cbUrlProtocol": "https", "union": True}

        # 使用 CF 代理或直连
        if self.enable_cf_proxy and self.cf_proxy_url:
            url = f"{self.cf_proxy_url}/ttwid/ttwid/union/register/"
            resp = self.session.post(url, json=data, timeout=30)
            if resp.status_code == 200:
                # 从响应中提取 ttwid cookie 并设置到 session
                if 'set-cookie' in resp.headers or 'Set-Cookie' in resp.headers:
                    cookie_header = resp.headers.get('set-cookie') or resp.headers.get('Set-Cookie')
                    # 调试：打印 Set-Cookie（生产环境应注释掉）
                    # print(f"收到 Set-Cookie: {cookie_header[:100] if cookie_header else 'None'}")

                    if cookie_header and 'ttwid=' in cookie_header:
                        ttwid_match = re.search(r'ttwid=([^;]+)', cookie_header)
                        if ttwid_match:
                            ttwid_value = ttwid_match.group(1)
                            self.session.cookies.set("ttwid", ttwid_value, domain=".douyin.com")
                            # 调试：打印设置的 cookie（生产环境应注释掉）
                            # print(f"设置 ttwid cookie: {ttwid_value[:50]}...")
        else:
            url = "https://ttwid.bytedance.com/ttwid/union/register/"
            resp = self.session.post(url, json=data, timeout=30)

        if resp.status_code != 200:
            raise Exception(f"初始化 ttwid 失败: HTTP {resp.status_code}")

    def get_detail(self, url_input):
        url = url_input.strip()
        aweme_id = self._resolve_short_url(url)

        if not aweme_id:
            return None

        print(f"解析到 ID: {aweme_id}")

        params = {
            "device_platform": "webapp",
            "aid": "6383",
            "channel": "channel_pc_web",
            "aweme_id": aweme_id,
            "update_version_code": "170400",
            "pc_client_type": "1",
            "version_code": "190500",
            "version_name": "19.5.0",
            "cookie_enabled": "true",
            "platform": "PC",
            "downlink": "10",
            "msToken": self.session.cookies.get("msToken")
        }

        params["a_bogus"] = self.ab.get_value(params)

        try:
            # 使用 CF 代理或直连
            if self.enable_cf_proxy and self.cf_proxy_url:
                api = f"{self.cf_proxy_url}/douyin/aweme/v1/web/aweme/detail/"
            else:
                api = "https://www.douyin.com/aweme/v1/web/aweme/detail/"

            self.session.headers.update({"User-Agent": USERAGENT})

            # 如果使用 CF 代理，手动添加 Cookie 到请求头
            if self.enable_cf_proxy and self.cf_proxy_url:
                # 获取所有 cookies 并构建 Cookie 头
                cookie_str = "; ".join([f"{k}={v}" for k, v in self.session.cookies.items()])
                # 调试：打印发送的 Cookie（生产环境应注释掉）
                # print(f"发送 Cookie: {cookie_str[:100]}...")

                headers_with_cookie = {"Cookie": cookie_str}
                resp = self.session.get(api, params=params, timeout=30, headers=headers_with_cookie)
            else:
                resp = self.session.get(api, params=params, timeout=30)

            # 调试：打印请求信息（生产环境应注释掉）
            # print(f"API 请求: {api}")
            # print(f"响应状态码: {resp.status_code}")
            # print(f"响应头: {dict(resp.headers)}")

            if resp.status_code == 200:
                try:
                    resp_json = resp.json()

                    # 如果使用 CF 代理，响应会被 Base64 编码
                    if self.enable_cf_proxy and self.cf_proxy_url and isinstance(resp_json, dict) and 'encoding' in resp_json:
                        if resp_json.get('encoding') == 'base64':
                            import base64
                            decoded_text = base64.b64decode(resp_json['data']).decode('utf-8')
                            data = json.loads(decoded_text)
                            # 调试：打印解码成功信息（生产环境应注释掉）
                            # print(f"Base64 解码成功，JSON keys: {list(data.keys())}")
                        else:
                            data = resp_json
                    else:
                        data = resp_json

                    if data.get("aweme_detail"):
                        return self.extractor.extract_data(data["aweme_detail"])
                    else:
                        print(f"未获取到 aweme_detail")
                except Exception as e:
                    print(f"JSON 解析失败: {e}")
                    # 调试：打印响应内容（生产环境应注释掉）
                    # print(f"响应内容前500字符: {resp.text[:500]}")
            else:
                print(f"API 请求失败: {resp.status_code}")
                # 调试：打印响应内容（生产环境应注释掉）
                # print(f"响应内容: {resp.text[:500]}")
        except Exception as e:
            print(f"请求异常: {e}")
        return None

3. 插件配置文件（confschema.json）

JSON

{
  "enable_cf_proxy": {
    "description": "是否启用 Cloudflare 代理",
    "type": "bool",
    "hint": "启用后将通过 CF Workers 代理请求抖音 API，可以有效避免 IP 被风控",
    "default": false
  },
  "cf_proxy_url": {
    "description": "Cloudflare Workers 代理地址",
    "type": "string",
    "hint": "填写你部署的 CF Workers 地址，例如: https://your-worker.workers.dev",
    "default": ""
  }
}

技术要点总结

1. CF Workers 自动压缩问题

问题：CF Workers 会自动对响应进行 gzip 压缩，即使设置了 Content-Length 头也无法禁用。

解决方案：使用 Base64 编码传输数据。

原理：

CF 的自动压缩针对的是文本内容
Base64 编码后的数据被包装在 JSON 中
JSON 格式的响应虽然也会被压缩，但不会出现响应体丢失的问题

问题：requests.Session() 的 Cookie 管理在代理场景下失效。

解决方案：

手动从响应头中提取 Set-Cookie
手动构建 Cookie 请求头

关键代码：

PYTHON

# 提取 Cookie
cookie_header = resp.headers.get('set-cookie')
ttwid_match = re.search(r'ttwid=([^;]+)', cookie_header)
self.session.cookies.set("ttwid", ttwid_match.group(1), domain=".douyin.com")

# 发送 Cookie
cookie_str = "; ".join([f"{k}={v}" for k, v in self.session.cookies.items()])
headers_with_cookie = {"Cookie": cookie_str}
resp = self.session.get(api, headers=headers_with_cookie)

性能与成功率

测试结果

场景	不使用 CF 代理	使用 CF 代理
视频解析	60-70%	95%+
图片解析	0-10%	95%+
实况解析	0-10%	95%+

优势

高成功率：利用 CF 的 IP 池，避免单一 IP 被风控
节省流量：只代理 API 请求，下载直连 CDN
免费额度：CF Workers 免费版每天 10 万次请求
全球加速：CF 的边缘节点提供低延迟访问

注意事项

CF Workers 限制：
- 免费版：10 万次请求/天
- CPU 时间限制：10-50ms/请求
- 响应体大小：无限制（但建议 < 10MB）
Base64 编码开销：
- 数据量增加约 33%
- 编码/解码有 CPU 开销
- 对于小数据量（< 100KB）影响可忽略
Cookie 管理：
- 需要手动处理 Cookie 的提取和发送
- 注意 Cookie 的域名和路径设置

部署指南

1. 部署 CF Workers

BASH

# 1. 登录 Cloudflare Dashboard
# 2. 进入 Workers & Pages
# 3. 创建新的 Worker
# 4. 粘贴 cloudflare_worker.js 代码
# 5. 部署并获取 Worker URL

2. 配置插件

在 AstrBot WebUI 中：

找到抖音解析插件
点击"配置"
启用"是否启用 Cloudflare 代理"
填写"Cloudflare Workers 代理地址"
保存并重载插件

3. 测试

发送一个抖音链接到机器人，观察日志输出：

成功：应该能看到解析结果
失败：检查 CF Workers 日志和 Python 日志

故障排查

问题1：响应体为空

症状：Response text length: 0

排查步骤：

检查 CF Workers 日志中的 Request Cookie
确认 Cookie 是否包含 ttwid 和 msToken
检查 Python 端是否正确构建了 Cookie 头

问题2：Base64 解码失败

症状：JSON 解析失败: Expecting value

排查步骤：

检查响应格式是否为 {"data": "...", "encoding": "base64"}
确认 CF Workers 是否正确编码了响应
检查响应是否被截断

问题3：CF Workers 超时

症状：代理请求失败: timeout

原因：抖音 API 响应慢或 CF Workers CPU 时间超限

解决方案：

增加 Python 端的 timeout 设置
优化 CF Workers 代码（减少不必要的操作）
考虑使用 CF Workers 付费版

总结

通过 Cloudflare Workers 反代，成功解决了抖音 API 风控问题。关键技术点：

Base64 编码：绕过 CF 自动压缩
手动 Cookie 管理：确保认证信息正确传递
重试机制：提高容错能力
分离架构：API 走代理，下载走直连

这个方案不仅适用于抖音，也可以推广到其他有风控机制的平台（如小红书、B站等）。

抖音解析器 Cloudflare Workers 反代实现全记录

抖音解析器 Cloudflare Workers 反代实现全记录

背景

问题分析

初始问题

问题根源

解决方案演进

阶段一：实现重试机制

阶段二：引入 Cloudflare Workers 反代

阶段三：实现过程中的坑

坑1：CF Workers 响应体为空

坑2：Cookie 未正确传递

最终解决方案

架构设计

完整代码

1. Cloudflare Workers 反代脚本

2. Python 端关键代码（dysk.py 片段）

3. 插件配置文件（confschema.json）

技术要点总结

1. CF Workers 自动压缩问题

2. Cookie 传递问题

性能与成功率

测试结果

优势

注意事项

部署指南

1. 部署 CF Workers

2. 配置插件

3. 测试

故障排查

问题1：响应体为空

问题2：Base64 解码失败

问题3：CF Workers 超时

总结

参考资料