精品长文一键写作：把批量写作任务排成队列的几种做法
# 精品长文一键写作：把批量写作任务排成队列的几种做法
> 接口：精品长文一键写作（apiCode=3206）· 接入点：3206-1 提交写作任务 / 3206-2 查询任务列表 / 3206-3 查询文章详情
> 计费：计次收费（提交 300 厘/次、两个查询接入点 0 厘/次）· 请求方式：POST/GET · 返回格式：JSON
> 适用人群：内容团队、有批量产出需求的开发者 · 阅读时间：约 8 分钟 · **最后实测核对：2026-09-15**
## 核心要点
- 接口并发量是 2 次/秒，批量提交必须自己做节流，否则超出部分会被拦。
- 没有批量提交参数，也没有回调推送。批量场景的正确姿势是「循环提交 + 集中轮询」，提交和取稿两段解耦。
- 轮询用 `3206-2` 扫列表比逐个调 `3206-3` 便宜得多，一页能拿 20 条状态。
## 批量场景的三个约束
批量写作和单篇调试的调用方式不一样。精品长文一键写作在批量场景下有四个需要先明确的条件：
| 约束 | 具体值 | 对设计的影响 |
|------|------|------|
| 并发上限 | 2 次/秒 | 提交阶段要做节流，并行的消费者数量受限 |
| 无批量参数 | `3206-1` 每次只接受一个 `topic` | 批量 = 循环提交，N 篇就是 N 次调用 |
| 无回调推送 | 文档没有回调地址参数 | 取稿只能主动轮询，需要自己维护任务状态 |
第三个约束决定了整体架构：**提交和取稿必须分成两段**。提交段只管把任务推进去并记录 `task_id`，取稿段独立运行、按自己的节奏扫描。
## 架构：两段式队列
```
提交段 取稿段
┌──────────────┐ ┌──────────────────┐
│ 待写主题队列 │ │ 未完成任务集合 │
│ (业务库/Redis)│ │ (writing_task 表) │
└──────┬───────┘ └────────┬─────────┘
│ 限速 2 次/秒 │ 每 20~30s 一轮
▼ ▼
3206-1 提交 ──► task_id 入库 ──► 3206-2 扫列表（分页）
│
task_status=success 的才取稿
▼
3206-3 取正文 ──► 回填落库
```
两段之间只通过 `task_id` 耦合。提交段挂了不影响已经提交的任务继续生成，取稿段挂了重启后从库里捞 `task_status != success` 的记录接着跑。
## 提交段：按 2 次/秒节流
最省事的节流是令牌桶。用一个长度 2、每 1 秒回填的桶，就能把提交速率压在并发上限之内：
```python
import time
import threading
import requests
class RateLimiter:
"""令牌桶：capacity 个令牌，每 1/capacity 秒回填一个。"""
def __init__(self, capacity=2):
self.capacity = capacity
self.tokens = capacity
self.interval = 1.0 / capacity
self.last = time.monotonic()
self.lock = threading.Lock()
def acquire(self):
while True:
with self.lock:
now = time.monotonic()
self.tokens = min(self.capacity,
self.tokens + (now - self.last) / self.interval)
self.last = now
if self.tokens >= 1:
self.tokens -= 1
return
time.sleep(0.02)
limiter = RateLimiter(capacity=2)
def submit_one(topic, appkey):
limiter.acquire()
r = requests.post("https://route.showapi.com/3206-1",
params={"appKey": appkey},
data={"topic": topic, "lang": "zh"},
timeout=60)
if r.status_code != 200:
return {"ok": False, "status": r.status_code, "raw": r.text[:200]}
body = r.json()["showapi_res_body"]
return {"ok": body["ret_code"] == 0, "task_id": body.get("task_id"),
"ret_code": body.get("ret_code"), "remark": body.get("remark")}
```
实测数据可以帮你估时间：2026-09-15 单次提交耗时 1.31 秒到 4.11 秒。按 2 次/秒的令牌桶，100 篇主题大约需要 1 到 3 分钟才能全部推进去。这个时间是提交前的一次性成本，不用等出稿。
## 取稿段：用 3206-2 扫列表
不要为每篇内容都开一个轮询线程去调 `3206-3`。`3206-2` 一次返回一页 20 条任务的状态，扫一遍列表再对已完成的任务取稿，请求数少一个数量级。
```python
import requests
import time
def scan_and_fetch(appkey, limit_pages=50):
"""扫全量任务列表，对 success 的任务取正文并回填。"""
base = "https://route.showapi.com"
pending_ids = set(load_unfinished_task_ids()) # 本地库中未完成的 task_id
if not pending_ids:
return 0
fetched = 0
page = 1
pages = 1
while page <= pages and page <= limit_pages:
r = requests.post(f"{base}/3206-2", params={"appKey": appkey},
data={"page": page}, timeout=60).json()
body = r["showapi_res_body"]
pages = body.get("allPages") or 1 # 空列表时 allPages 为 0，兜底成 1
for item in body.get("contentlist", []):
tid = item["task_id"]
if tid not in pending_ids:
continue
if item["task_status"] != "success":
update_task(tid, task_status=item["task_status"])
continue
# 出稿了，取正文
d = requests.post(f"{base}/3206-3", params={"appKey": appkey},
data={"task_id": tid, "need_md": "2"}, timeout=60).json()
if d.get("task_status") == "success":
update_task(tid, task_status="success", title=d.get("title"),
content_md=d.get("content"),
keywords=d.get("keywords"), outline=d.get("outline"))
pending_ids.discard(tid)
fetched += 1
page += 1
return fetched
```
两个实测细节会直接影响这段代码：
- **空列表时 `allPages` 返回 0**，如果用 `while page <= allPages` 做循环条件会直接跳过，所以要 `or 1` 兜底。
- **`3206-3` 在生成中返回 HTTP 450**，`need_md=2` 时响应体是 JSON（`ret_code=-1`），`need_md=1` 时是纯文本。上面的代码只在列表里看到 `success` 才去调详情，正常路径下不会碰到 450；但如果有时间差，仍要在取稿函数里对非 200 做分支，别把 450 当成失败。
## 失败与重试
提交阶段的失败要区分两类：
| 情况 | 表现 | 处理 |
|------|------|------|
| 业务失败 | HTTP 200，`ret_code=-1` | 记 `remark`，可原样重试一次；重复失败说明 `topic` 需要调整 |
| 网络/网关错误 | 非 200，或超时 | 指数退避重试，退避基数从 2 秒起，最多 3 次 |
| 超出并发 | 被限流拦截 | 加大令牌桶间隔，不要立刻重试 |
`3206-1` 没有幂等参数。每次调用都会新建任务并计入一次计费，所以重试前要先确认上一次是真的没提交成功，最稳的做法是先调一次 `3206-2` 看最近的任务列表里有没有相同 `topic` 的记录。
任务层面一直停在 `writing` 的，用 `3206-4` 删除失败任务，然后重新提交：
```python
def drop_task(task_id, appkey):
r = requests.post("https://route.showapi.com/3206-4",
params={"appKey": appkey},
data={"task_id": task_id}, timeout=60)
return r.json()
```
## 计费与调用方式
批量场景的成本基本只跟提交次数成正比。按 2026-09-15 的计费口径：提交 300 厘/次，轮询用的 `3206-2`、`3206-3` 都是 0 厘/次，配图 `3206-5` 是 260 厘/次。100 篇带配图的长文，费用集中在 100 次提交加 100 次配图上，轮询本身不产生费用，所以轮询频率可以按业务需要设，不用为了省钱把间隔拉到很大。
调用成功才计费，失败不扣费。档位与费率明细见 https://www.showapi.com/apiGateway/view/3206 的「产品价格」标签。
## FAQ
**Q1：并发 2 次/秒是全局的还是按账号算的？**
该值出现在专用资源包规格表的「并发量」一栏，属于接口的调用上限。批量提交按这个值节流即可，不要靠试错去摸边界。
**Q2：可以开多个线程同时提交吗？**
可以，但令牌桶要放在共享层，让所有线程共用一个桶。每个线程各建一个桶的话总速率会乘上线程数，很快超限。
**Q3：一页 20 条，任务多了要翻很多页吗？**
`3206-2` 的 `maxResult` 是 20，`allPages` 给出总页数。批量场景建议加一个 `ct` 时间范围的快速跳过，只翻最近几页，因为该接入点本身只覆盖近 30 天的任务。
**Q4：怎么知道哪些任务要重试？**
在本地库里维护 `task_status`，扫列表时同步更新。长期停在 `writing` 且超过业务容忍时间的记录标为待处理，用 `3206-4` 清掉后重提。
**Q5：批量产出的内容会重复吗？**
同一批任务如果 `topic` 差异太小，生成结果会接近。要让批量产出有区分度，需要在 `topic` 里写清各自的角度和范围，必要时用 `reference_list` 指定不同来源。
## 下一步阅读
- 状态与轮询细节：[精品长文一键写作：任务写完了没有？看 task_status 三种状态](https://www.showapi.com/guides/longform-writing-task-status-3206)
- 链路与表结构：[从一句主题到带配图长文的完整链路](https://www.showapi.com/guides/longform-writing-pipeline-3206)
- 成本口径：[精品长文一键写作计费口径：各接入点费率与资源包怎么算](https://www.showapi.com/guides/longform-writing-billing-3206)
- **本系列共 9 篇**：查看[精品长文一键写作指南总目录](https://www.showapi.com/guides/longform-writing-guides-3206)
