适用场景：什么时候要补发布 API 幂等

个人 Flask 博客做到自动发布后，最容易被忽略的不是“能不能发出去”，而是网络抖动、脚本重试、人工补发和搜索提交之间的边界。Stepnex 这类中小型独立站在百度网盟审核冲刺期，需要稳定新增全栈技术内容，同时避免重复文章、半截正文、同一 slug 多次写入、中文英文对应关系断掉。发布 API 如果只做 Token 校验和 db.session.add()，短期看能跑，长期会把内容库变成难复盘的状态：第一次请求超时但数据库已写入，第二次重试又生成新 slug；中文发布成功、英文失败；百度提交拿到旧 URL；管理员后来不知道哪一次请求是真正生效的。

幂等不是把所有错误吞掉，而是让同一篇文章在同一请求指纹下重复执行时得到同一个结果。它适合自动化发文、批量补稿、后台草稿转发布、搜索平台提交前的 URL 确认等流程。本文延续站内已有的 Flask 博客自动发布接口如何做得更稳：Token、重试和发布后校验，把重点放到“如何避免重复写入、如何记录审计日志、如何验证回滚路径”。

技术取舍：先约束入口，再记录结果

最小可用方案分三层。第一层是请求入口约束：slug、language、translation_key、content_hash 或 idempotency_key 必须稳定，缺一项就返回明确 400。第二层是数据库约束：文章表至少要保证 slug 唯一，审计表保证 idempotency_key 唯一；不要只依赖 Python 代码判断，因为并发请求可能同时通过内存检查。第三层是发布结果记录：每次请求都写入请求摘要、操作者、状态码、文章 ID、错误信息和外部提交结果，方便后续从日志追溯。

配置上不建议一上来引入复杂消息队列。小站可以先用 SQLite 或 MySQL 的唯一约束配合事务完成，搜索提交放在文章写入成功之后执行，并允许单独补跑。这样即使百度提交失败，也不会把文章发布回滚掉；反过来，文章写入失败时也不会提交一个不存在的 URL。

落地前还要先定一个发布前检查口径：标题、摘要、分类、标签、slug、正文长度、站内链接、状态值都应在本地脚本里验证，只有验证通过才调用线上 API。这样幂等接口负责“同一请求不要写乱”，质量脚本负责“低质量请求不要进入发布链路”，两者边界清楚，排查时也不会互相甩锅。

步骤一：定义请求指纹和唯一约束

发布脚本应在生成 JSON 时写入稳定字段。中文和英文文章使用不同 slug，但共享同一个 translation_key。如果自动化框架支持请求头，可以额外传 Idempotency-Key；如果没有，就用 language + slug + sha256(content_md) 生成。

import hashlib

def article_fingerprint(payload: dict) -> str:
    raw = "|".join([
        payload.get("language", ""),
        payload.get("slug", ""),
        payload.get("translation_key", ""),
        hashlib.sha256(payload.get("content_md", "").encode("utf-8")).hexdigest(),
    ])
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

数据库侧要有硬约束。SQLite 可以先建唯一索引，MySQL/PostgreSQL 迁移时保留同样语义：

create unique index if not exists uq_article_slug on article(slug);

create table if not exists publish_audit (
    id integer primary key autoincrement,
    idempotency_key text not null unique,
    article_slug text not null,
    language text not null,
    payload_hash text not null,
    status text not null,
    article_id integer,
    error_message text,
    created_at text not null
);

验证方式是先在测试库执行两次同一 payload。第一次应插入文章并写 success 审计记录；第二次不应新增文章，而是返回已有文章 ID 和 URL。可以用命令确认数量：

sqlite3 database.db "select slug,count(*) from article group by slug having count(*)>1;"
sqlite3 database.db "select idempotency_key,status,article_id from publish_audit order by id desc limit 5;"

步骤二：把 API 返回分成三类

发布接口要避免所有错误都返回 500。建议分成三类：参数错误返回 400，鉴权失败返回 401 或 403，幂等命中返回 200 并带 idempotent_replay=true，真实冲突返回 409。真实冲突指的是同一个 slug 已存在，但内容哈希、语言或 translation_key 不一致。此时不要覆盖旧文章，也不要自动改 slug，因为自动改 slug 会让后续搜索提交和中英文对应关系变得不可预测。

伪代码如下：

@app.post("/api/articles/publish")
def publish_article():
    payload = request.get_json(silent=False)
    key = request.headers.get("Idempotency-Key") or article_fingerprint(payload)
    payload_hash = hash_payload(payload)

    audit = PublishAudit.query.filter_by(idempotency_key=key).first()
    if audit and audit.status == "success":
        article = Article.query.get(audit.article_id)
        return jsonify({"ok": True, "slug": article.slug, "idempotent_replay": True})

    existing = Article.query.filter_by(slug=payload["slug"]).first()
    if existing and existing.content_hash != payload_hash:
        return jsonify({"ok": False, "error": "slug_conflict"}), 409

    article = create_article_from_payload(payload)
    db.session.add(PublishAudit(...))
    db.session.commit()
    return jsonify({"ok": True, "slug": article.slug, "id": article.id})

这里的避坑点是不要在事务外先查后写再提交。真正落库时仍然可能遇到唯一约束异常，所以要捕获 IntegrityError，回查审计表或文章表，再决定返回幂等结果还是 409。日志也要区分 replay、conflict、created、failed，否则事后只能看到一串 200 或 500。

更细一点，接口响应体也要稳定。自动化脚本通常只关心 ok、slug、language、url 和 idempotent_replay，不要在成功和重放时返回两套完全不同的结构。错误响应也应包含机器可读的 error_code，例如 bad_payload、unauthorized、slug_conflict、storage_failed。这样 PowerShell、Python、GitHub Actions 或 Codex 自动化都能用同一段判断逻辑处理结果。日志字段建议包含 request_id、idempotency_key 前 12 位、slug、language、status_code、elapsed_ms，不要只写自然语言。线上排查时，先用 grep slug /var/log/blog/publish.log 找到请求，再回查数据库审计表，最后再看 Nginx access log 的 HTTP 状态码，三者能对上才算闭环。

步骤三：发布后提交搜索平台要可补偿

文章发布成功后再提交百度，不要把百度提交和数据库事务绑死。更稳的流程是：文章写入成功，得到公开 URL；审计表记录 published；执行百度提交；提交成功则记录 baidu_submitted，失败则记录 baidu_pending 和错误摘要。后续 baidu_union_sprint.py --submit-limit 10 或单 URL 提交脚本可以补偿。

def after_publish(article_url: str, audit_id: int) -> None:
    try:
        result = submit_baidu_url(article_url)
    except Exception as exc:
        mark_audit(audit_id, "baidu_pending", str(exc)[:500])
    else:
        mark_audit(audit_id, "baidu_submitted", json.dumps(result, ensure_ascii=False))

验证时不要只看脚本退出码，还要看线上 URL、sitemap 和提交日志：

curl -I https://stepnex.cn/article/flask-publish-api-idempotency-audit-zh
curl -s https://stepnex.cn/sitemap.xml | grep flask-publish-api-idempotency-audit
python scripts/baidu_union_sprint.py --submit-limit 10

如果百度凭据缺失，应报告缺失项，而不是伪造成功。常见缺失包括 BAIDU_PUSH_URL、BAIDU_SITE、BAIDU_TOKEN。Google sitemap 或 URL inspection 也是同理：没有凭据就记录待配置，不能把“已生成 sitemap”写成“已收录”。

补偿任务还要限制批量规模。比如每天只从审计表取最近 10 条 baidu_pending 或最新 sitemap URL，提交后写回状态和返回体摘要。不要无限循环提交同一个失败 URL，也不要把所有历史 URL 每次都推一遍。验证时可以先 dry-run 打印候选 URL，再正式提交。若遇到 4xx，优先检查凭据、站点域名和 URL 编码；若遇到网络超时或 5xx，再做有限重试。这样既能维持更新节奏，又不会把搜索提交脚本变成新的噪声来源。

步骤四：回滚不是删除，而是有记录地撤回

自动发布出错时，第一反应不应是直接删库。更稳的做法是给文章增加 status=draft 或 status=archived，保留 slug、发布时间和审计记录；如果已经提交过搜索平台，公开页面可以短期保留并修正文案，只有内容确实不该公开时才返回 404 或 410。这样做的好处是日志、sitemap、后台列表和搜索提交记录还能对上。

回滚命令可以先做成后台动作或管理脚本：

python scripts/publish_rollback.py --slug flask-publish-api-idempotency-audit-zh --status draft --reason "validation failed after publish"

脚本至少要写三件事：更新文章状态，写入 publish_audit 的 rollback 事件，刷新 sitemap 或触发缓存失效。验证方式是访问文章 URL、分类页、sitemap 和后台列表，确认状态一致。不要只改数据库字段而忘记缓存和 sitemap；这会造成用户看不到文章，但搜索平台仍然从 sitemap 发现旧 URL。

避坑点：幂等和重复发布不是一回事

第一，幂等命中必须返回原来的结果，而不是重新生成一篇文章。第二，slug 冲突不要自动追加随机后缀；自动化发布最怕“看似成功，实际 URL 漂移”。第三，中文和英文要共享 translation_key，但不能共享同一个 slug。第四，正文修订不能用同一个幂等键伪装成重试；修订应走更新接口或新版本审计。第五，日志里不要保存完整 Token、Cookie 或后台密钥，只保存请求摘要和操作者标识。

还有一个常见误区是把 409 当作失败重试目标。409 往往代表人工介入点：旧文章是否该更新、slug 是否被占用、内容是否重复。发布脚本遇到 409 应停止重试并输出明确错误；网络超时、502、503 才适合有限重试。这样才能避免越重试越乱。

复盘清单：每周看一次发布链路

  • 最近一周是否出现同 slug 多篇文章，命令：select slug,count(*) ... having count(*)>1。
  • publish_audit 是否存在 failed、conflict、baidu_pending，每一条是否有处理结果。
  • 中文和英文文章是否共享 translation_key，并分别出现在正确 URL 路径。
  • 新文章是否进入 sitemap，分类页是否可见，站内相关文章链接是否有效。
  • 发布脚本的重试次数是否只覆盖网络错误和 5xx，没有对 400、401、403、409 盲目重试。
  • 日志里是否能查到操作者、请求指纹、文章 ID、公开 URL 和搜索提交结果。
  • 回滚演练是否能把文章状态、sitemap、缓存和审计记录一起处理。

审核冲刺期的重点是稳，不是堆数量。发布 API 有了幂等、冲突处理和回滚审计后，自动化才能放心承担每天的技术文章更新；如果某一步失败，也能清楚知道失败在生成、发布、公开访问、sitemap 还是百度提交。