Message、Artifact 与异步交接

mkdir l6-artifacts
cd l6-artifacts

"""本地文件产物、内容校验与持久交付收据。"""
import hashlib
import json
import tempfile
from pathlib import Path

def digest(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def create_artifact(root, content):
    # 内容寻址：相同内容得到相同编号，文件变更能被哈希检测。
    artifact_id = digest(content)
    artifact = {"artifact_id": artifact_id, "task_id": "T-17",
                "content": content, "sha256": artifact_id}
    path = Path(root) / (artifact_id + ".json")
    path.write_text(json.dumps(artifact, ensure_ascii=False), encoding="utf-8")
    return path

class Receiver:
    def __init__(self, root):
        self.receipt_path = Path(root) / "receipts.json"
        self.receipts = json.loads(self.receipt_path.read_text(encoding="utf-8")) if self.receipt_path.exists() else {}

    def receive(self, event_id, artifact_path, expected_task):
        artifact = json.loads(Path(artifact_path).read_text(encoding="utf-8"))
        if artifact["task_id"] != expected_task:
            raise ValueError("产物任务不匹配")
        actual = digest(artifact["content"])
        if artifact["sha256"] != actual or artifact["artifact_id"] != actual:
            raise ValueError("产物内容校验失败")
        if event_id in self.receipts:
            if self.receipts[event_id] != actual:
                raise ValueError("同一交付编号携带不同产物")
            return "duplicate", artifact["content"]
        # 收据写盘后，新建接收者仍能识别这次交付。
        self.receipts[event_id] = actual
        self.receipt_path.write_text(json.dumps(self.receipts), encoding="utf-8")
        return "accepted", artifact["content"]

def main():
    # 临时目录放在当前学习目录；退出时清理固定演示的临时文件。
    with tempfile.TemporaryDirectory(prefix="handoff-", dir=".") as root:
        path = create_artifact(root, "O-17可申请退款120元；依据R-1")
        print("进度消息：资料已查完，正在整理")
        status, content = Receiver(root).receive("E-1", path, "T-17")
        print("首次交付：" + status)
        print("产物内容：" + content)
        status, _ = Receiver(root).receive("E-1", path, "T-17")
        print("重新加载后重送：" + status)
        artifact = json.loads(path.read_text(encoding="utf-8"))
        artifact["content"] = "金额被改为999元"
        path.write_text(json.dumps(artifact, ensure_ascii=False), encoding="utf-8")
        try:
            Receiver(root).receive("E-2", path, "T-17")
        except ValueError as error:
            print("篡改产物：" + str(error))

if __name__ == "__main__":
    main()

ls
python -X utf8 artifacts_demo.py

进度消息：资料已查完，正在整理
首次交付：accepted
产物内容：O-17可申请退款120元；依据R-1
重新加载后重送：duplicate
篡改产物：产物内容校验失败

python -X utf8 -m unittest -v test_artifacts_demo.py