Codex AI 自媒体自动化实战：选题、脚本、分发全链路批量生产

原创

资源大佬 jzit-top

发布于 2026-11-10 18:54:41

480

单条内容的自动化流水线，和批量生产之间，隔着的不是"循环调用几次"，而是批量本身引入的三类工程问题：选题池里的内容互相同质化、脚本生成批任务互相拖累、分发产物在平台约束前才暴露不合格。一条流水线跑通时，这些问题都不存在；一旦从"每天一条"变成"每天五十条"，它们会同时出现，而且都不会以报错的形式暴露。下面按选题、脚本、分发三个环节给出可运行的代码。

选题：批量之前先做去重与评分，否则批量生产的是同质内容

选题来源通常是 RSS、热点榜、竞品评论。直接把这些条目喂给脚本生成，结果是同一热点被反复写，选题池里"OpenAI 发布新模型"出现十七条变体。批量的第一个工程动作不是生成，而是用 embedding 做语义去重，再用多信号打分排序：

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

def dedupe_and_rank(topics: list[dict], embed_fn, threshold=0.88,
                    top_n=50) -> list[dict]:
    texts = [t["title"] + " " + t.get("summary", "") for t in topics]
    vecs = np.array([embed_fn(x) for x in texts])

    # 贪心去重：保留分数更高的，剔除与它高度相似的
    order = sorted(range(len(topics)),
                   key=lambda i: topics[i]["heat"] * topics[i]["freshness"],
                   reverse=True)
    kept, kept_vecs = [], []
    for i in order:
        if kept_vecs:
            sims = cosine_similarity([vecs[i]], kept_vecs)[0]
            if sims.max() >= threshold:
                continue
        kept.append(topics[i]); kept_vecs.append(vecs[i])
        if len(kept) >= top_n:
            break
    return kept

threshold=0.88 是语义相似度的经验阈值，高于它基本可以判定为"同一件事的不同表述"。heat * freshness 作为排序信号，让热点和时效性同时参与，而不是只看阅读量。去重必须在生成之前完成——在脚本阶段去重，等于已经为同质选题付过生成成本。

脚本：批任务要隔离工作区，单点失败不能污染整批

批量生成脚本最容易写错的模式是 for topic in topics: codex_exec(topic)。这个循环有三个问题：串行执行让五十条的耗时线性叠加；所有任务共享同一个工作目录，一个任务的残留文件会污染下一个；任一任务超时或崩溃，整个循环中断，已完成的成果无法落盘。

正确的结构是每个任务独立工作区、并发执行、失败隔离：

import concurrent.futures as cf
import pathlib, subprocess, json, shutil

def gen_one(topic: dict, root: pathlib.Path, timeout=120) -> dict:
    ws = root / "jobs" / topic["id"]
    ws.mkdir(parents=True, exist_ok=True)
    prompt = (f"为选题《{topic['title']}》写一篇 1200 字技术文章，"
              f"角度：{topic['angle']}。写入 draft.md，不要输出解释。")
    try:
        subprocess.run(
            ["codex", "exec", "--sandbox", "workspace-write",
             "-C", str(ws), prompt],
            check=True, timeout=timeout, capture_output=True)
        draft = ws / "draft.md"
        if not draft.exists() or len(draft.read_text()) < 500:
            return {"id": topic["id"], "ok": False, "error": "empty_draft"}
        return {"id": topic["id"], "ok": True, "path": str(draft)}
    except subprocess.TimeoutExpired:
        return {"id": topic["id"], "ok": False, "error": "timeout"}
    except subprocess.CalledProcessError as e:
        return {"id": topic["id"], "ok": False,
                "error": "codex_failed", "detail": e.stderr[-500:]}

def batch_generate(topics, root=pathlib.Path("./media"), workers=4):
    results = []
    with cf.ThreadPoolExecutor(max_workers=workers) as ex:
        futs = {ex.submit(gen_one, t, root): t["id"] for t in topics}
        for f in cf.as_completed(futs):
            results.append(f.result())
    ok = [r for r in results if r["ok"]]
    json.dump(results, open(root / "batch_report.json", "w"),
              ensure_ascii=False, indent=2)
    return {"total": len(results), "ok": len(ok), "failed": len(results) - len(ok)}

-C str(ws) 让每个任务在独立目录里执行，codex exec 的 workspace-write 沙箱边界正好落在该任务的目录内。workers=4 而不是开满 CPU 核数，是因为每个 codex 调用本身已经在内部并行，外层并发过高只会让 API 限流成为主要失败原因。batch_report.json 记录每条任务的成败与失败原因，让"这批五十条里哪几条没出来、为什么"成为可查询的数据，而不是一个需要重跑的模糊印象。

分发：发布包必须有 manifest，平台约束要在打包时校验

批量分发最常见的失效是"内容都生成了，但发布时被平台拒绝"。公众号标题超长、知乎不支持某些代码块格式、X 的正文长度上限被忽略——这些问题如果等到调用发布接口时才发现，批量分发的重试逻辑会变得极其复杂。

正确做法是把每个平台的约束前置到发布包构建阶段，用 manifest 记录每个产物的平台、字段和校验结果：

import json, pathlib, hashlib

PLATFORM_RULES = {
    "wechat": {"title_max": 64, "body_min": 800, "allow_code": True},
    "zhihu":  {"title_max": 100, "body_min": 300, "allow_code": True},
    "x":      {"title_max": 0, "body_max": 1000, "allow_code": True},
}

def build_manifest(draft: str, topic: dict, platforms: list[str]) -> dict:
    title, body = topic["title"], draft
    items, errors = [], []
    for p in platforms:
        rule = PLATFORM_RULES[p]
        if len(title) > rule["title_max"] > 0:
            errors.append({"platform": p, "field": "title",
                           "len": len(title), "max": rule["title_max"]})
        if len(body) < rule.get("body_min", 0):
            errors.append({"platform": p, "field": "body_min",
                           "len": len(body), "min": rule["body_min"]})
        if len(body) > rule.get("body_max", 10**9):
            errors.append({"platform": p, "field": "body_max",
                           "len": len(body), "max": rule["body_max"]})
        if not errors:
            items.append({"platform": p, "title": title,
                          "body_sha": hashlib.sha256(body.encode()).hexdigest()[:12],
                          "len": len(body)})
    return {"topic_id": topic["id"], "items": items,
            "errors": errors, "ready": not errors}

def package_and_check(draft_path: str, topic: dict,
                      platforms=("wechat", "zhihu", "x")):
    body = pathlib.Path(draft_path).read_text(encoding="utf-8")
    m = build_manifest(body, topic, list(platforms))
    pathlib.Path(draft_path).with_suffix(".manifest.json").write_text(
        json.dumps(m, ensure_ascii=False, indent=2))
    return m

body_sha 记录正文哈希，让"这条内容发到三个平台的是不是同一版"可以被核对；errors 在打包阶段就暴露平台约束冲突，而不是等到发布接口返回 400。ready 字段让批量分发层可以直接跳过不合格的包，而不是让一个平台的失败拖慢整批。

收束

批量生产的工程化，落在三个"批量才会暴露"的位置：选题要在生成前做语义去重和信号排序，否则批量只是同质内容的放大；脚本任务要独立工作区、并发执行、失败隔离，否则一条失败拖垮整批、一个残留污染下一个；分发要构建带 manifest 的发布包并在打包时校验平台约束，否则失败延迟到发布接口才暴露、重试逻辑不可控。这三层做对了，"每天五十条"才是一个可运营的产能；做错了，它只是一个更快地生产出更多需要人工返工的内容的循环。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

codec