用 Python、JSON 与幂等键实现可验证的多平台发布状态机
用不可变父源身份修复每日多平台覆盖误判
本文给出一套可复制的 Python 3.11 实现：从规范源稿提取不可变身份，用严格匹配计算平台覆盖矩阵，并以 unittest 证明昨天的发布、缺失哈希的旧记录和同源重复写入会被正确区分。
问题：发布成功不能只看日期或平台最近一次状态
每日多平台任务最容易出现的误判，是把墙钟日期、点击时间或平台最近一次成功当成今天源稿已经覆盖。可靠系统必须区分 content identity、daily edition identity、parent_source_identity 与 platform coverage key。前两者描述内容和日版，父源身份绑定规范源稿，覆盖键则把父源字段压成可比较的稳定值。
第一步：从 JSON 提取不可变父源身份
from hashlib import sha256
from json import dumpsREQUIRED = ("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 Falseidentity = 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 unittestFIELDS = ("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 在修改状态判定代码时自动运行。
可复现实施步骤
-
提取规范源稿身份：读取并验证 slug、report_date、source_hash、content_hash 与 idempotency_key。
-
绑定平台衍生稿：让每个平台衍生稿保留完整且不可变的 parent_source_identity。
-
验证每日覆盖矩阵：用父源精确匹配和 published_verified 状态计算覆盖并测试去重边界。
常见问题
为什么不能只用 report_date 判断今天是否完成？
同一天可能出现正文修订，昨天的记录也可能被错误标注到今天。至少要同时比较父源 slug、source_hash、content_hash 与 idempotency_key。
submitted_unverified 为什么不能自动重试？
点击后平台可能已经收稿，只是公开回读尚未完成。自动重试会制造重复内容，因此应等待审核窗口后只读对账，并把提交意图、内容哈希和最早核验时间写入状态记录。
旧记录缺少哈希时怎样处理？
缺哈希意味着无法证明与当前父源相同，应标记为不完整或待人工迁移，不能用标题相似度替代精确身份；迁移时也要保留旧记录原值和修正依据。
平台覆盖键能否替代全部父源字段？
覆盖键适合索引和快速比较，但审计记录仍应保留组成它的父源字段，便于解释冲突、核对哈希生成规则和迁移旧数据，不能只保存一个不可解释的摘要。
为什么测试中不用真实线上哈希？
单元测试验证的是判定逻辑，不需要生产数据。固定占位值更可复现，也避免把运行记录或敏感上下文写进示例；集成测试再使用临时目录和匿名回读验证真实接口边界。
发布状态机最需要记录哪些错误信息？
记录错误类别、阶段、是否发生外部写入、内容哈希、进程清理结果和可重试性即可；认证材料、Cookie、Token 与完整平台响应不应进入日志或测试夹具。