用 Python、JSON 与幂等键实现可验证的多平台发布状态机

2026年7月29日
| 没有评论

用不可变父源身份修复每日多平台覆盖误判

本文给出一套可复制的 Python 3.11 实现：从规范源稿提取不可变身份，用严格匹配计算平台覆盖矩阵，并以 unittest 证明昨天的发布、缺失哈希的旧记录和同源重复写入会被正确区分。

问题：发布成功不能只看日期或平台最近一次状态

每日多平台任务最容易出现的误判，是把墙钟日期、点击时间或平台最近一次成功当成今天源稿已经覆盖。可靠系统必须区分 content identity、daily edition identity、parent_source_identity 与 platform coverage key。前两者描述内容和日版，父源身份绑定规范源稿，覆盖键则把父源字段压成可比较的稳定值。

第一步：从 JSON 提取不可变父源身份

from hashlib import sha256
from json import dumps

REQUIRED = ("source_function", "report_date", "source_hash", "content_hash", "idempotency_key")

def canonical_source_identity(source: dict) -> dict:
    origin = source.get("origin") if isinstance(source.get("origin"), dict) else {}
    values = {
        "source_slug": str(source.get("slug") or ""),
        "source_function": str(origin.get("source_function") or ""),
        "source_report_date": str(origin.get("report_date") or ""),
        "source_hash": str(origin.get("source_hash") or ""),
        "source_content_hash": str(source.get("content_hash") or origin.get("content_hash") or ""),
        "source_idempotency_key": str(source.get("idempotency_key") or origin.get("idempotency_key") or ""),
    }
    if not values["source_slug"] or not all(values.get({"report_date": "source_report_date", "content_hash": "source_content_hash", "idempotency_key": "source_idempotency_key"}.get(name, name), "") for name in REQUIRED):
        raise ValueError("canonical source identity is incomplete")
    material = dumps(values, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    values["platform_coverage_key"] = "sha256:" + sha256(material.encode("utf-8")).hexdigest()
    return values

这里不使用当前时间补字段。report_date 来自源稿版本，content_hash 绑定正文，idempotency_key 绑定业务意图。源稿日期相同但正文哈希不同，或者正文相同但日版不同，都不能被视为同一次覆盖。具体哈希在示例中只由输入计算，不声称对应任何线上数据。

第二步：严格匹配父源，再计算平台覆盖矩阵

IDENTITY_FIELDS = (
    "source_slug",
    "source_function",
    "source_report_date",
    "source_hash",
    "source_content_hash",
    "source_idempotency_key",
    "platform_coverage_key",
)

def same_parent_identity(left: dict, right: dict) -> bool:
    return all(str(left.get(field) or "") == str(right.get(field) or "") for field in IDENTITY_FIELDS)

def record_is_complete(record: dict, expected: dict) -> bool:
    if record.get("status") != "published_verified":
        return False
    identity = record.get("parent_source_identity")
    return isinstance(identity, dict) and same_parent_identity(identity, expected)

def coverage_matrix(platforms: list[str], records: list[dict], expected: dict) -> dict:
    matrix = {}
    for platform in platforms:
        complete = any(
            record.get("platform") == platform and record_is_complete(record, expected)
            for record in records
        )
        matrix[platform] = {"complete": complete}
    return matrix

覆盖矩阵只接受 published_verified 与父源精确匹配同时成立。submitted_unverified 只能进入只读对账队列，不能自动重发；login_required 和 verification_required 属于外部阻塞，也不能伪装成准备成功。这样，业务状态和网络写入状态被明确分开。

第三步：用 unittest 锁住三个最容易回归的边界

import unittest

FIELDS = ("source_slug", "source_function", "source_report_date", "source_hash", "source_content_hash", "source_idempotency_key", "platform_coverage_key")

def same_identity(left: dict, right: dict) -> bool:
    return all(bool(left.get(field)) and left.get(field) == right.get(field) for field in FIELDS)

def complete(record: dict, expected: dict) -> bool:
    return record.get("status") == "published_verified" and same_identity(record.get("parent_source_identity") or {}, expected)

class CoverageTests(unittest.TestCase):
    def setUp(self):
        self.today = {
            "source_slug": "daily-example",
            "source_function": "daily",
            "source_report_date": "2026-07-28",
            "source_hash": "source-example",
            "source_content_hash": "content-example",
            "source_idempotency_key": "daily:2026-07-28:article",
            "platform_coverage_key": "sha256:coverage-example",
        }

    def test_yesterday_publication_does_not_complete_today(self):
        yesterday = dict(self.today, source_report_date="2026-07-27")
        record = {"status": "published_verified", "parent_source_identity": yesterday}
        self.assertFalse(complete(record, self.today))

    def test_legacy_record_without_hash_is_not_complete(self):
        legacy = dict(self.today)
        legacy.pop("source_content_hash")
        record = {"status": "published_verified", "parent_source_identity": legacy}
        self.assertFalse(complete(record, self.today))

    def test_exact_same_source_skips_duplicate_write(self):
        record = {"status": "published_verified", "parent_source_identity": dict(self.today)}
        should_write = not complete(record, self.today)
        self.assertFalse(should_write)

if __name__ == "__main__":
    unittest.main()

失败处理与可观测性

  • 准备阶段只允许一次主生成路由和一次后备，避免三个子任务逐条放大超时。
  • 点击发布后立即记录 submission intent；结果不明时禁止自动重放，只允许只读对账。
  • 日志只保存错误类别、字符数与哈希，不写 Cookie、Token、密码或完整认证响应。
  • 每个平台记录 external_write_count、公开 URL、内容哈希与回读结果，便于复盘状态机。

复现方法

把第三段代码保存为 test_coverage.py，在 Python 3.11 下执行 python -m unittest test_coverage.py。三个测试应全部通过。工程中还应增加空文本、批内近重复、登录跳转、公开回读 404 与提交后等待窗口等测试，并让 CI 在修改状态判定代码时自动运行。

可复现实施步骤

  1. 提取规范源稿身份：读取并验证 slug、report_date、source_hash、content_hash 与 idempotency_key。

  2. 绑定平台衍生稿：让每个平台衍生稿保留完整且不可变的 parent_source_identity。

  3. 验证每日覆盖矩阵：用父源精确匹配和 published_verified 状态计算覆盖并测试去重边界。

常见问题

为什么不能只用 report_date 判断今天是否完成？

同一天可能出现正文修订，昨天的记录也可能被错误标注到今天。至少要同时比较父源 slug、source_hash、content_hash 与 idempotency_key。

submitted_unverified 为什么不能自动重试？

点击后平台可能已经收稿，只是公开回读尚未完成。自动重试会制造重复内容，因此应等待审核窗口后只读对账，并把提交意图、内容哈希和最早核验时间写入状态记录。

旧记录缺少哈希时怎样处理？

缺哈希意味着无法证明与当前父源相同，应标记为不完整或待人工迁移，不能用标题相似度替代精确身份；迁移时也要保留旧记录原值和修正依据。

平台覆盖键能否替代全部父源字段？

覆盖键适合索引和快速比较，但审计记录仍应保留组成它的父源字段，便于解释冲突、核对哈希生成规则和迁移旧数据，不能只保存一个不可解释的摘要。

为什么测试中不用真实线上哈希？

单元测试验证的是判定逻辑，不需要生产数据。固定占位值更可复现，也避免把运行记录或敏感上下文写进示例；集成测试再使用临时目录和匿名回读验证真实接口边界。

发布状态机最需要记录哪些错误信息？

记录错误类别、阶段、是否发生外部写入、内容哈希、进程清理结果和可重试性即可；认证材料、Cookie、Token 与完整平台响应不应进入日志或测试夹具。

文章摘自：https://www.cnblogs.com/yuanxi2000/p/22015861/python-json-idempotent-publishing-state-machine-20260728

flask, python, pytorch, tornado