短视频文案提取 API 实战：一条链接拿到口播全文，附轮询与批量方案

做内容的人都有过这种经历：刷到一条爆款短视频，想把口播文案扒下来拆解结构。手动做法是打开、暂停、往回拖、逐句打字，一条 3 分钟的视频要花二十分钟。

这类工作已经有现成接口：给它一条短视频链接，返回口播全文、作者、视频描述和时长。本文按 /doc/28 讲清楚三件事——参数怎么传、异步任务怎么等、批量怎么不触发限流，并给出两段可直接用的 Python。

一、这个接口适合干什么

需要提前说清的四条边界，避免需求评审时理解错：

1. 只给文本，不给时间轴。 返回的是 resultText 整段文案，没有分句时间码，做字幕文件还得自己对齐。
2. 是语音转写，不是画面理解。 视频里没有口播、只有背景音乐，结果可能为空。
3. 不处理私密内容。 需要登录、需要密码、仅好友可见的视频取不到。
4. 版权自负。 文档写得很直接：需确认拥有该内容的权利，使用不拥有的视频可能违反版权法。

二、接口说明

2.1 请求参数

关键设计就在 taskId 这一个字段上：同一个接口，靠它有没有值来区分"提交"和"查询"。 所以不需要记两个 URL。

2.2 返回字段

2.3 三处文档内部不一致，先说在前面

这几条我核对时发现的，直接按示例硬编会在线上出问题：

1. language 的取值写法不一致。 示例值写 zh-CN，可选值枚举写的是 zh。建议按枚举传 zh，首次接入用两种各试一次，确认哪种被接受。
2. creditCost 示例值 0.5 与当前计费对不上。 页面计费区块写的是 10 点/次（约 0.1 元），返回示例里的 creditCost: 0.5 应是旧示例。成本核算以计费区块和控制台账单为准，别拿返回字段反推单价。
3. status 只给了 completed 一个值。 文档没列全枚举（如处理中、失败）。代码里的判断逻辑应该是"等于 completed 且 resultText 非空才算成功"，其余值一律按未完成处理，别用 != "failed" 这种反推写法。

三、调用流程：提交 + 查询两段式

文档把任务类型标成 sync，但从参数设计看，它实际是"提交拿 taskId，再带 taskId 查结果"的两段式。按这个模型写代码最稳：

① POST content=<视频链接>
        ↓
   data.taskId + data.status
        ↓
② POST content=<原链接> & taskId=<上一步的 ID>
        ↓
   status=completed ? 取 resultText : 等几秒再查

一个容易被忽略的细节：查询时把原始 content 一起带上。 文档把 content 标为必填，虽然查询阶段理论上只需要 taskId，但带上原链接不会出错，也能兼容某些后端实现的非空校验。

四、Python 实战：提交与轮询

依赖只有 requests，用表单提交（和文档标的 Content-Type 一致）。

import os
import time

import requests

ENDPOINT = "https://xiaoapis.com/api/video/text"
API_KEY = os.environ["XIAOAPIS_API_KEY"]

# 文档：付费用户 10 次/秒；免费用户 3 秒 1 次
POLL_INTERVAL = 3.0
TOTAL_TIMEOUT = 600.0


class ApiError(RuntimeError):
    pass


def _call(payload: dict) -> dict:
    """统一请求入口：表单提交，按 code 判成败。"""
    body = {
   "key": API_KEY, **payload}
    response = requests.post(ENDPOINT, data=body, timeout=(10, 120))

    # 401/402/429 等会返回非 200 的 HTTP 状态，先落到统一错误分支
    try:
        result = response.json()
    except ValueError:
        raise ApiError(f"HTTP {response.status_code}，响应非 JSON：{response.text[:200]}")

    code = result.get("code")
    if code != 200:
        raise ApiError(f"[{code}] {result.get('msg')} (HTTP {response.status_code})")
    return result


def extract_text(
    video_url: str,
    *,
    language: str | None = "zh",
    poll_interval: float = POLL_INTERVAL,
    total_timeout: float = TOTAL_TIMEOUT,
) -> dict:
    """提交任务并轮询到完成，返回 data 字典。"""
    if not video_url.startswith(("http://", "https://")):
        raise ValueError("content 必须是 http/https 链接；分享口令请先转成链接")

    payload = {
   "content": video_url}
    if language:
        payload["language"] = language

    first = _call(payload)
    data = first.get("data") or {
   }
    task_id = data.get("taskId")
    if not task_id:
        raise ApiError(f"提交未返回 taskId：{first}")

    deadline = time.monotonic() + total_timeout
    while True:
        if data.get("status") == "completed" and data.get("resultText"):
            return data

        if time.monotonic() > deadline:
            raise TimeoutError(f"taskId={task_id} 超过 {total_timeout}s 未完成")

        time.sleep(poll_interval)
        # 查询时同时带上 content，避免后端校验 content 非空
        data = _call({
   "content": video_url, "taskId": task_id}).get("data") or {
   }


if __name__ == "__main__":
    result = extract_text("https://v.douyin.com/xxxxxxx/")
    print(f"作者：{result.get('author')}")
    print(f"描述：{result.get('videoDesc')}")
    print(f"文案：{result.get('resultText')}")

三个设计点值得说明：

• 用 data= 而不是 json=。 文档 Header 写的是 application/x-www-form-urlencoded，用表单提交最贴合；若你的环境强制 JSON，接口一般也兼容，但别在同一个项目里两种混用。
• 轮询间隔 3 秒，别设 0.5 秒。 免费档本身就是 3 秒一次的限制，付费档虽然是 10 QPS，也没必要用轮询把额度打满。
• 超时给到 10 分钟。 12 小时的长音频不可能几秒出结果，超时设太短会把正常任务判死。

五、批量：把 10 QPS 用满而不触发限流

单条调通之后，真实需求往往是"一次几百条"。这里的问题不是接口能力，是别把并发打爆。

import time
from concurrent.futures import ThreadPoolExecutor, as_completed


class RateLimiter:
    """简单的令牌桶：允许 qps 次/秒，跨线程共享。"""

    def __init__(self, qps: float = 8.0) -> None:
        self.interval = 1.0 / qps
        self.next_at = time.monotonic()

    def wait(self) -> None:
        now = time.monotonic()
        if now < self.next_at:
            time.sleep(self.next_at - now)
        self.next_at = max(self.next_at, time.monotonic()) + self.interval


def batch_extract(urls: list[str], qps: float = 8.0, workers: int = 4) -> dict:
    """qps 留 20% 余量，避免瞬时抖动触发 429。"""
    limiter = RateLimiter(qps)
    results: dict[str, dict] = {
   }
    failures: dict[str, str] = {
   }

    def task(url: str) -> dict:
        limiter.wait()
        return extract_text(url, poll_interval=3.0)

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {
   pool.submit(task, u): u for u in urls}
        for future in as_completed(futures):
            url = futures[future]
            try:
                results[url] = future.result()
            except Exception as exc:
                failures[url] = str(exc)

    print(f"成功 {len(results)} 条，失败 {len(failures)} 条")
    return {
   "ok": results, "failed": failures}

两个要点：

• 限流放在"提交动作"上，不要放在每条任务的整个生命周期上。 一个任务的轮询要跑一两分钟，如果整段都占着令牌，吞吐会被严重低估。
• 失败要分类。 参数错误（400）重试一百次也没用，429 和 500 才值得退避重试。文档明确"只计 200 和 500"，意味着 500 是计费的，重试前先确认业务能不能接受这笔消耗。

六、拿到的文案还要处理一下

resultText 是"能用的原文"，不是"能直接发的稿"。真正投产前建议补四步：

1. 断句与分段。 转写结果常常一句到底，按中文标点切分、按语义合并短句，可读性会好很多。
2. 去口水词。 "嗯""那个""就是说"这类词用正则清掉，但注意别误删引语中的语气词。
3. 专有名词校正。 品牌名、人名、英文缩写最容易错（比如把 SpaceX 写成 Spacex）。维护一张替换表，比每次人工校对便宜。
4. 保留元数据。 author、videoDesc、duration 一起入库。做选题分析时，"谁说的、什么时候、多长"往往比正文更有信息量。

七、六个容易踩的坑

1. 链接要编码。 带中文、表情、多个查询参数的分享链接直接拼进表单会报错，先 quote 或交给 requests 的 data= 处理。
2. 短链先展开。 部分平台返回 302 跳转，接口拿不到最终地址就会失败。提交前先跟随一次跳转，取 response.url。
3. 2GB / 12 小时是硬上限。 长直播回放请先切片或只提音频轨，纯音频（mp3、m4a）识别更快、体积更小。
4. 纯音频链接同样可用。 content 不只收视频，音频直链也支持——做播客转写时不用先合成视频。
5. 别用重试兜参数错误。 400 是参数问题（链接错、格式不支持），403/审核类是内容问题，这两类重试都无意义；只有 429 与 500 值得退避。
6. 版权与授权要留痕。 文档要求确认拥有内容权利。批量抓取第三方内容做二创，建议在业务侧保留授权记录，别把风险留给接口。

八、常见问题

Q1：能直接传"复制口令"那种文本吗？

文档写的是"短视频分享链接"。口令需要先解析成 URL 再提交。如果业务上必须支持口令，先在自己服务里做一层口令→链接的转换。

Q2：返回的 duration 是什么单位？

文档标为 string，示例值是 0.5，没有明确单位说明。做时长统计前建议用几条已知时长的视频实测确认，别直接当秒用。

Q3：为什么 status 一直是处理中？

两种可能：视频较长（12 小时上限意味着处理时间会明显拉长），或者音频里没有可识别的人声。建议把轮询超时设到 10 分钟以上，并对"超时"和"空文案"分开告警。

Q4：一次能提多条吗？

一次一条。批量靠并发 + 限流，参考第五节；付费档 10 QPS，留 20% 余量比较稳。

Q5：失败了会不会扣点？

计费口径是"只计 200 和 500"。审核不通过、参数错误、任务取消不计费。但 500 是计费的，所以重试策略要克制。

Q6：能拿到字幕时间轴吗？

不能。这个接口只返回整段文本。需要时间码的话，得走字幕类接口或自己做强制对齐。

结语

这个接口的能力很单一：一条链接进，一段文案出。 真正决定它好不好用的，是三件工程细节——按"提交 + 查询"两段式写、把限流放在提交动作上、对返回的文案做二次清洗。

接入顺序建议：先用一条 1 分钟内的公开视频跑通提交与查询，确认 status 与 resultText 的行为；再传一条长视频看轮询节奏；最后接批量和清洗逻辑。别一上来就灌 500 条，否则出问题时你分不清是链接失效、限流，还是任务超时。