当前位置：首页 > news >正文

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

news 2026/7/28 22:03:58

用 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 在修改状态判定代码时自动运行。