处理了三条，宇宙节庆报告却错了

目标

观测太空庆典取消作业的批准、审计和当前行，制作出客户说明和机器判定基于同一依据的报告。把只读采集和保留已有文件的发布连接起来。

为什么重要

笔数相同、哈希通过、报告文字很长，都不是正确对象和当前状态的证据。请先学习 Python dict、list、异常、文件，SQL 事务，以及前面的批准版本、分块恢复课程。预计 130 分钟，所以请在到期前用“＋时间”延长。必须在最长 180 分钟内完成，会话结束后文件会消失。需要的代码请另外保存。

环境与通用契约

产出物是 /root/evidence/worker.py。镜像里有 PostgreSQL 16、psycopg 3.2.3、Python 3。无需额外安装、网络和权限，用 postgres 用户写入 /root。

评分器会在本地 labdb 的独有临时 schema 中创建下面的表和虚构数据，并且只清理自己建立的 schema。使用传入的 con 的 search_path 和 DSN，不修改 public 或生产数据库。表和列的名称是下面固定的契约，SQL 数据值通过参数传递。不要把客户、ID、观测时间、schema 写死。

CREATE TABLE orders(id integer PRIMARY KEY,tenant text NOT NULL,
 qty integer NOT NULL CHECK(qty BETWEEN 1 AND 1000),
 state text NOT NULL CHECK(state IN ('pending','paid','cancelled')),
 revision integer NOT NULL CHECK(revision>=0));
CREATE TABLE changes(change_id text PRIMARY KEY,tenant text NOT NULL,targets jsonb NOT NULL);
CREATE TABLE audit(change_id text REFERENCES changes(change_id),id integer NOT NULL,
 previous_revision integer NOT NULL,new_revision integer NOT NULL,qty integer NOT NULL,
 PRIMARY KEY(change_id,id));

change_id 和 tenant 是 str 本身，ASCII 英文、数字、下划线、连字符 1–64 个字符。id 是类型恰好为 int 的 1–2147483647，qty 是类型恰好为 int 的 1–1000，批准 revision 是类型恰好为 int 的 0–2147483646。当前和审计的 revision、previous_revision、new_revision 是类型恰好为 int 的 0–2147483647。bool 不允许当作整数。

targets 是 1–64 个、只有 id、revision、qty 三个键的 dict 本身所组成的 list 本身。audit 是键为 id、previous_revision、new_revision、qty 的 dict 本身组成的列表，current 是键为 id、tenant、qty、state、revision 的 dict 本身组成的列表，各为 0–64 个。每个列表内部禁止重复 ID。current 的 state 是 pending、paid、cancelled 之一，并且禁止批准之外的 ID。audit 中批准之外的 ID 不是格式错误，而是要分析的异常。列表在规范化时生成按 ID 排序的深拷贝，不修改输入。

借来的 con 是 autocommit=True、Read Committed，调用开始时是 IDLE。capture 不关闭它，成功和失败之后恢复原隔离级别、只读设置、锁和语句限制，并回到 IDLE。不要把它包在其他外部事务里。只有 publish 拥有连接。有 fault 时，以明确的字符串为参数调用，不隐藏原始错误。

证据与分类格式

evidence 是只有 format、change_id、tenant、targets、audit、current、snapshot、observed_at 八个键的 dict 本身。format 是不是 bool 的 int 1。snapshot 是 PostgreSQL pg_current_snapshot() 的文本，不超过 4096 个字符，形式是十进制数:十进制数:可选的用逗号分隔的十进制数列表。observed_at 是 UTC ISO 字符串，以 Z 结尾，与 Python datetime.isoformat() 以 UTC 输出的形式相同。例：2026-09-13T01:02:03.123456Z。它是把数据库的 transaction_timestamp() 转换而来，并不表示权威认证时间。

capture 的 targets 用 plan 规范化。audit 里只放该 change_id 的，current 里只放与批准 ID 对应的当前行。即使当前行变成了其他客户的，要看到这个差异，也必须观测该 ID。不要因为原审计损坏就去修改或删除它。不要混入其他变更的审计和批准之外的订单。

analyze 的所有 ID list 都是升序。

approved：原批准 ID 的全部。
committed：是批准 ID，并且审计的 previous_revision=批准 revision、new_revision=批准 revision+1、qty=批准 qty 全部相符的 ID。
invalid_audit：批准之外的 ID，或者上述值中有任何一个不同的审计 ID。不放入 committed。
remaining：从 approved 中去掉 committed 的 ID。并不意味着可以重新执行。
matching：committed 中，当前行的客户、qty、cancelled、revision=批准 revision+1 全部相符的 ID。
missing：committed 中当前行不存在的 ID。
drifted：committed 中有行但不是 matching 的 ID。
decision：只要有 invalid_audit、missing、drifted 中任何一个，就是 hold；不是的话，只要有 remaining，就是 incomplete，其余是 complete。这是观测时点的判定。

客户说明与证据包契约

describe 只根据 normalize 和 analyze 的结果，按下面的顺序返回纯文本。每行末尾是 LF，最后一行之后也有一个 LF。方括号的位置填入实际的值。ID 之间以不带空格的逗号分隔，空列表用表示“无”的韩文词。

변경: [change_id] / 고객: [tenant]
관측: [observed_at] / 스냅샷: [snapshot]
판정: [hold=보류, incomplete=미완료, complete=관측 시점 완료]
승인 ID: [approved]
확정 ID: [committed]
미확정 ID: [remaining]
현재 일치 ID: [matching]
후속 차이 ID: [drifted]
누락 ID: [missing]
감사 불일치 ID: [invalid_audit]
범위: 일관된 과거 관측이며 현재 상태·출처 인증을 보장하지 않습니다.

包有 payload、sha256 两个键。payload 由 format=1、scope=historical-snapshot-not-current-state-or-authenticity、规范化的 evidence、分析 report、纯文本 summary 五个键组成。sha256 是对 payload 的 canonical 字节取的小写 64 位十六进制数。校验器会从 evidence 重新计算 payload，不只比较哈希，还比较材料、分类、说明和范围的全部。格式版本的 True 不认作 1。用不同的观测时间一致地重新生成的包也可能通过，这就是这种方式在真伪认证上的局限。

文件输出契约

输出 path 是 str 本身的绝对路径，是用户拥有的已有目录里的普通文件位置。允许文件不存在，但符号链接、目录、相对路径是 ValueError。不自动创建父目录。评分会把自己建立的临时输出文件夹作为参数传入，所以输出路径也不要写死。学员自己测试时，请使用像 /root/evidence/report.json 这样自己的文件夹。

把完整的字节写入与目标同一文件夹的唯一临时文件，经过 flush、fsync、关闭之后再替换。普通异常时清理临时文件，保留已有的或新的完整版本。它不是连强制终止、断电、其他进程替换父路径这类攻击都能应对的文件系统安全边界。只在可信的、自己拥有的文件夹里使用。

步骤

固定批准列表的含义——实现继承 Exception 的 Conflict 和 plan(change_id,tenant,targets)。校验通用输入契约，并返回只有 change_id、tenant、targets 的新 dict。targets 是按 ID 升序的深拷贝，不修改输入。错误的输入是 ValueError。
采集时不混入不同时点——capture(con,change_id,fault=None) 校验 ID 后，在自己的事务中应用 REPEATABLE READ READ ONLY 以及锁 500ms、语句 2000ms 的上限。依次采集 changes 中对应的批准、该 change_id 的审计、与批准 ID 对应的当前 orders。每次查询之后调用 after-plan、after-audit、after-current 钩子。放入同一次观测的 pg_current_snapshot() 记法和 transaction_timestamp() 的 UTC 字符串，返回下面的 evidence 格式。没有批准，或者保存的批准格式损坏，是 Conflict。
检查证据的格式和重复——normalize(evidence) 校验下面的键、类型、范围、观测记法，并返回把 targets、audit、current 按 ID 排序的深拷贝。重复 ID、批准之外的 current、未知状态是 ValueError。批准之外的 audit 不删除，而是保留下来以便之后分析。
把过去的确认和当前的差异分开——analyze(evidence) 在 normalize 之后，按下面的分类契约，返回 approved、committed、remaining、matching、drifted、missing、invalid_audit 的升序 ID list 以及 decision。不修改输入。不要只数审计笔数，也不要自动恢复当前值。
让客户说明和机器结果基于同一依据——canonical(value) 返回 UTF-8、ensure_ascii=False、sort_keys=True、separators=(逗号,冒号)、allow_nan=False 的 JSON bytes。describe(evidence) 用下面的纯文本格式说明分析结果。bundle(evidence) 把规范化的 evidence、analyze 的 report、describe 的 summary 放入 payload，并把 payload 规范字节的小写 SHA-256 放入 sha256。
连重新计算过哈希的虚假结论也要拒绝——verify_bundle(value) 校验包的结构、format、scope、哈希记法，并确认 payload 的哈希。从 evidence 重新生成 bundle，与已有的包比对整个规范字节，一致则返回分析 report 的新 dict。字节或结论、说明不一致是 Conflict，错误的格式是 ValueError。decode(raw) 把类型恰好为 bytes 的 1–1048576 个字节按 UTF-8 JSON 解析，返回 verify_bundle 的结果，并以 ValueError 拒绝重复键、非有限常量、被截断的 JSON、错误的 UTF-8。
文件替换失败时不要抹掉以前的结果——export(path,evidence,fault=None) 校验输出路径，先把规范的包做成不超过 1MiB 的 bytes。写入与目标同一目录里单独的临时文件，flush、fsync 之后关闭，并调用 after-write。用 os.replace 替换目标之后调用 after-replace，并返回包的 sha256。替换之前出错，保留已有文件；替换之后出错，保留新的完整版本。无论正常还是普通异常，都只清理自己的临时文件，并传递原始错误。
把数据库观测到报告发布连成一体——publish(dsn,change_id,path,fault=None) 在建立连接之前，校验 ID 和输出路径。用 psycopg.connect(dsn,autocommit=True,connect_timeout=2) 打开自己拥有的连接来做 capture，无论成功还是失败都关闭。采集成功之后，在数据库连接已关闭的状态下调用 export，返回 sha256。原样传递采集和发布的钩子和错误，不自动重试，也不修改业务数据库。

参考

直接诊断：python3 -B /opt/lab/fixtures/evidence/check.py 8 /root/evidence/worker.py。把数字换成当前的步骤，就会累积检查到那一步。一次评分有 40 秒的限制。
检查观测期间其他连接的提交、READ ONLY 写入被拒绝、笔数相同但 ID 不同、后续 revision、对虚假结论重新计算哈希、文件替换前后的错误。
不要对 JSON 使用 eval。重复键可以用 object_pairs_hook，NaN 等可以用 parse_constant 来拒绝。UTF-8 解码错误和 JSON 解析错误也属于 ValueError 一类。
只读并不会带来认证或按客户的访问权限。只采集允许本人使用的虚构材料，不要使用真实的个人信息、生产数据库，也不要交给外部客户。
本实验不验证数据库服务器终止、文件进程强制终止、断电的耐久性。验证的是快照的内部一致性、分类契约，以及普通异常时的文件保留。

固定批准列表的含义

实现继承 Exception 的 Conflict 和 plan(change_id,tenant,targets)。校验通用输入契约，并返回只有 change_id、tenant、targets 的新 dict。targets 是按 ID 升序的深拷贝，不修改输入。错误的输入是 ValueError。

即使笔数相同，也未必是同一个 ID 集合。不要把 bool 当作 int 接受。

采集时不混入不同时点

capture(con,change_id,fault=None) 校验 ID 后，在自己的事务中应用 REPEATABLE READ READ ONLY 以及锁 500ms、语句 2000ms 的上限。依次采集 changes 中对应的批准、该 change_id 的审计、与批准 ID 对应的当前 orders。每次查询之后调用 after-plan、after-audit、after-current 钩子。放入同一次观测的 pg_current_snapshot() 记法和 transaction_timestamp() 的 UTC 字符串，返回下面的 evidence 格式。没有批准，或者保存的批准格式损坏，是 Conflict。

另一个连接会在钩子之间真实提交。如果只用 BEGIN 把多条 SELECT 包起来，在默认隔离级别下时点会混在一起。

检查证据的格式和重复

normalize(evidence) 校验下面的键、类型、范围、观测记法，并返回把 targets、audit、current 按 ID 排序的深拷贝。重复 ID、批准之外的 current、未知状态是 ValueError。批准之外的 audit 不删除，而是保留下来以便之后分析。

请区分格式错误和业务上的不一致。批准之外的审计不是要隐藏的材料，而是要报告的材料。

把过去的确认和当前的差异分开

analyze(evidence) 在 normalize 之后，按下面的分类契约，返回 approved、committed、remaining、matching、drifted、missing、invalid_audit 的升序 ID list 以及 decision。不修改输入。不要只数审计笔数，也不要自动恢复当前值。

审计是过去确认的依据，当前行是查看后续差异的材料。即使状态相同，客户、数量、revision 也可能不同。

让客户说明和机器结果基于同一依据

canonical(value) 返回 UTF-8、ensure_ascii=False、sort_keys=True、separators=(逗号,冒号)、allow_nan=False 的 JSON bytes。describe(evidence) 用下面的纯文本格式说明分析结果。bundle(evidence) 把规范化的 evidence、analyze 的 report、describe 的 summary 放入 payload，并把 payload 规范字节的小写 SHA-256 放入 sha256。

翻译、空白、键顺序一变，哈希也会变。不要把本次格式的规范化规则，说成是其他全部 JSON 标准的保证。

连重新计算过哈希的虚假结论也要拒绝

verify_bundle(value) 校验包的结构、format、scope、哈希记法，并确认 payload 的哈希。从 evidence 重新生成 bundle，与已有的包比对整个规范字节，一致则返回分析 report 的新 dict。字节或结论、说明不一致是 Conflict，错误的格式是 ValueError。decode(raw) 把类型恰好为 bytes 的 1–1048576 个字节按 UTF-8 JSON 解析，返回 verify_bundle 的结果，并以 ValueError 拒绝重复键、非有限常量、被截断的 JSON、错误的 UTF-8。

被篡改的结论的哈希，任何人也都能重新计算。即使内部核对通过，也不等于出处认证或对当前数据库的确认。

文件替换失败时不要抹掉以前的结果

export(path,evidence,fault=None) 校验输出路径，先把规范的包做成不超过 1MiB 的 bytes。写入与目标同一目录里单独的临时文件，flush、fsync 之后关闭，并调用 after-write。用 os.replace 替换目标之后调用 after-replace，并返回包的 sha256。替换之前出错，保留已有文件；替换之后出错，保留新的完整版本。无论正常还是普通异常，都只清理自己的临时文件，并传递原始错误。

如果先用写入模式打开目标，以前的报告就会被截断。请在同一目录里做成完整版本，再一次性替换。

把数据库观测到报告发布连成一体

publish(dsn,change_id,path,fault=None) 在建立连接之前，校验 ID 和输出路径。用 psycopg.connect(dsn,autocommit=True,connect_timeout=2) 打开自己拥有的连接来做 capture，无论成功还是失败都关闭。采集成功之后，在数据库连接已关闭的状态下调用 export，返回 sha256。原样传递采集和发布的钩子和错误，不自动重试，也不修改业务数据库。

等待文件操作期间，不要一直占着快照。即使保存的报告与新观测的结论不同，也不要勉强把其中任何一个调成一致。