把隔离运营起来 — 堆积、熔断、重投

目标

创建守卫 qgate.py：把期望写成文件，每次运行按行过滤投放文件，分别堆进 sqlite 的事实表和隔离表。让错得太多的运行整体中断，把修好的行重新放回去而不产生重复计算，找出存龄大的隔离，并把运行状态和数据状态分开汇报。

为什么重要

不能用的行不丢弃，而是留在隔离里，这一步通常都会做。问题在后面。隔离会在每次运行中重新堆积，半年后打开看，里面有 4 万行，而在此期间汇总一直少着这些数据在往外发。管道每天都成功了——成功是指没有报错就结束了，并不是该放进去的都放进去了。 一行出错和整个文件出错也是不同的事件。上游把列顺序换了再发来时，几乎每一行都会错位，这时如果按行隔离，隔离表里会进去一万行，汇总则是空的。这样的文件，不放进一行、直接中断运行更好。划分的依据是比例，而这个比例不是整数，而是要从平时的失败比例来定。 还需要把修好的行重新放回去的闭环。重新投入是由人手动来跑的，一定会运行多次，这时很容易在事实表里放进两次，或者再次关闭已经关闭的隔离，使“解决了 3 条”被报告两次。 评分器不会相信你写出来的文字。它会在临时目录里摆好评分器生成的投放文件和期望，真正运行你的守卫，然后直接打开生成的 sqlite 文件，把事实、隔离和运行记录与评分器亲自数出来的值相比对。阈值、店铺名和金额每次运行都会变化。

步骤

1. 创建并运行 /root/quarantine/gen_drops.py，在 /root/quarantine/drops 下生成四个投放文件，并生成 /root/quarantine/expectations.json。
2. 在 /root/quarantine/qgate.py 中实现 check，只按期望进行计数。
3. 加上 run，把事实、隔离和运行记录堆进 sqlite。
4. 加上 run --max-fail-ratio，让错得太多的运行整体中断。
5. 加上 requeue，把修好的行重新放回去，并且放两次数字也不变。
6. 加上 aging，找出存龄大的隔离。
7. 加上 status，把运行状态和数据状态分开汇报。
8. 依次运行自己的四个投放文件，生成 /root/quarantine/pipeline.db，并写出 /root/quarantine/status.json 和 /root/quarantine/quarantine_report.md。

参考

• 运行契约：python3 /root/quarantine/qgate.py <명령> ...（占位符为命令）。结果以一个 JSON 对象输出到标准输出。成功时退出码为 0，文件不存在时为 3，用法错误时为 2，运行被中断时为 5。
• 投放文件和重新投入文件的表头是 order_id,shop,qty,amount,status。重新投入文件放在哪里都可以——第 5 步的答案文件会生成在 /tmp/fixes.csv。
• 期望文件是 {"version": 1, "rules": [...]}，每条规则都有 name、column、type。type 有五种——not_null、regex（pattern）、range（min、max，包含边界）、enum（values）、unique（在该文件内只有第一次出现的值通过）。
• 规则的适用规则：取值时先去掉前后空白。空值除了 not_null 以外，全部算失败。正则必须与整个值匹配。唯一性从文件前面的行开始看。
• check 的响应是 {"drop": 이름, "rows": 정수, "passed": 정수, "failed": 정수, "fail_ratio": 소수, "by_rule": {규칙이름: 정수}}（占位符依次为名称、整数、整数、整数、小数、规则名和整数）。failed 是违反一条以上规则的行数，by_rule 是每条规则被违反的行数，所以合计可能不同。by_rule 中期望的所有规则名即使是 0 也要放进去。fail_ratio 在小数点后第六位四舍五入。
• sqlite 表有三张。runs(run_id, drop_name, status, rows, passed, failed, loaded, quarantined)、facts(order_id, shop, qty, amount_cents, run_id)、quarantine(q_id, order_id, rule, detail, raw, first_run_id, resolved_run_id)。run_id 和 q_id 是自增的，order_id 是事实表的主键。amount_cents 是把金额换成整数分的值。
• run 的响应：在 check 的键之上加上 run_id、status、loaded、quarantined、max_fail_ratio。status 是 loaded 或 broken。
• 隔离即使同一行的同一原因再次到来，也只打开一次（如果已经有处于打开状态的相同 order_id 和 rule，就不再新放）。一行违反两条规则，隔离就是两行。
• --max-fail-ratio 默认值是 1.0，所以不会中断。fail_ratio 超过阈值时（相等则通过）就中断运行。被中断的运行在 runs 中记为 broken，事实和隔离一行都不放进去，退出码为 5。
• requeue 的响应是 {"run_id": 정수, "status": "requeued", "rows": 정수, "fixed": 정수, "still_failing": 정수, "resolved": 정수, "facts": 정수}（占位符均为整数）。重新投入也是一次运行，所以会在 runs 里以 requeued 留下一行。通过的行以更新的方式放入事实表，并且只关闭该 order_id 仍处于打开状态的隔离。facts 是放入之后事实表的条数。
• aging 的响应是 {"max_age_runs": 정수, "latest_run_id": 정수, "open": 정수, "aged": 정수, "by_rule": {규칙이름: 정수}, "oldest": {"q_id": 정수, "order_id": 문자열, "rule": 문자열, "age": 정수} 또는 null, "alert": 참거짓}（占位符依次为整数、整数、整数、整数、规则名和整数、整数、字符串、字符串、整数、表示“或”的词、布尔值）。存龄是 latest_run_id - first_run_id，大于或等于 max_age_runs 就是存龄大。by_rule 中只放有存龄大的隔离的规则，oldest 是处于打开状态的隔离中 first_run_id 最小的那一条（相同则取 q_id 较小的）。
• status 的响应是 {"runs": {"total": 정수, "loaded": 정수, "broken": 정수, "requeued": 정수, "last_status": 문자열}, "data": {"facts": 정수, "open": 정수, "resolved": 정수, "aged": 정수}, "runs_ok": 참거짓, "data_ok": 참거짓}（占位符依次为整数、整数、整数、整数、字符串、整数、整数、整数、整数、布尔值、布尔值）。runs_ok 表示有运行记录，并且最后一次运行不是 broken；data_ok 表示处于打开状态的隔离为 0，并且存龄大的隔离为 0。--max-age-runs 默认值是 3。
• 第 8 步的报告写成 ## 이번 실행（韩文，意为“这次运行”）、## 격리에 무엇이 쌓였나（韩文，意为“隔离里堆了什么”）、## 재투입과 이중 계산（韩文，意为“重新投入与重复计算”）、## 임계치를 무엇으로 정했나（韩文，意为“阈值是按什么定的”）四节，并用数字写出事实条数和处于打开状态的隔离条数。
• 官方文档：sqlite3 模块 · UPSERT · re.fullmatch · csv
• 想直接看数据库，可以使用 sqlite3 /root/quarantine/pipeline.db 'select * from runs'。
• 常见错误：每次运行都重新放入隔离，导致同一行不断堆积；把重新投入用插入放进去，导致事实翻倍；再次关闭已经关闭的隔离；把阈值定成整数而不是平时的值。

把投放文件和期望放在文件里

创建并运行 /root/quarantine/gen_drops.py，在 /root/quarantine/drops 下生成四个投放文件，并生成 /root/quarantine/expectations.json。期望要用到全部五种 type，并且其中一个投放文件必须有超过一半的行违反期望。

如果把期望放在文件里而不是代码里，就能拿着这个文件和上游谈，而违反的规则的名字直接就是隔离原因。投放文件里要混入这些行：状态值不在列表内的行、金额为负的行、数量超出范围的行、店铺名为空的行、单据号不符合规格的行、与前面的行单据号相同的行。其中三个文件违反的行要少于一半，一个要超过一半。

先只用期望来计数

在 /root/quarantine/qgate.py 中实现 check --drop <파일> --expect <파일>（占位符依次为投放文件、期望文件），以 JSON 输出 drop、rows、passed、failed、fail_ratio、by_rule。

failed 是违反一条以上规则的行数，by_rule 是每条规则被违反的行数，所以合计不同——因为一行可能违反两条规则。by_rule 里，没人违反的规则也要写成 0。值为空时，除了 not_null 以外都算失败。

把事实和隔离分开堆积

加上 run --drop <파일> --db <파일> --expect <파일>（占位符依次为投放文件、数据库文件、期望文件），通过的行以更新的方式放入 facts，违反的行按所违反的每条规则在 quarantine 中打开一条，并把这次运行记入 runs。即使同一行的同一原因再次到来，隔离也只打开一次。

三张表的列名和顺序写在参考一节里。事实表以 order_id 为主键，所以用更新的方式放入。打开隔离之前，先看是否已经有相同 order_id 和 rule、并且仍处于打开状态的行——不看的话，每次运行都会把同一行堆上去。

错得太多的运行整体中断

加上 run --max-fail-ratio R（占位符为比例），当 fail_ratio 超过阈值时中断运行。被中断的运行在 runs 中记为 broken，事实和隔离都一行也不放进去，退出码为 5。

一行出错和整个文件出错是不同的事件。阈值不要用整数，而要从平时的失败比例来定——用 check 测一测四个投放文件的比例再定。中断的时候如果放进一半就停下，下一个人就得判断放到了哪里。

把修好的行重新放回去

加上 requeue --db <파일> --fixes <파일> --expect <파일>（占位符依次为数据库文件、修正文件、期望文件），按期望重新检查修好的行，通过的话就以更新的方式放入事实表，并只关闭该单据仍处于打开状态的隔离。同一次重新投入运行两次，事实条数和已关闭的隔离数都不能增加。

重新投入是由人手动来跑的，所以一定会运行多次。用插入的话事实会翻倍，关闭时不加“处于打开状态”的条件，“解决了 3 条”就会被报告两次。第二次运行的 resolved 必须是 0。单据号本身坏掉的行无法通过重新投入修好——因为一改号码，就变成了另一行。

找出存龄大的隔离

加上 aging --db <파일> [--max-age-runs N]（占位符依次为数据库文件、次数），用 latest_run_id - first_run_id 量出处于打开状态的隔离的存龄，数出大于或等于 max_age_runs 的，并以 open、aged、by_rule、oldest、alert 输出。

只看条数，就看不到不减少的东西。存龄按运行次数量，比按时间量更好处理——管道没运行的话存龄也不增加，这样更符合人的感觉。oldest 是处于打开状态的隔离中 first_run_id 最小的那一条。

把“成功了”和“是对的”分开汇报

加上 status --db <파일> [--max-age-runs N]，把运行状态（runs）和数据状态（data）分开输出，并用 runs_ok、data_ok 这两个值来回答。

运行成功和数据正确是不同的两条轴。混在一栏里，两者就都看不见。runs_ok 表示有运行记录，并且最后一次运行没有被中断；data_ok 表示处于打开状态的隔离和存龄大的隔离都为 0——两者都为真的日子是很少的。

用自己的投放文件跑一周并汇报

把自己的四个投放文件按日期顺序运行，生成 /root/quarantine/pipeline.db，把 status 的答案留在 /root/quarantine/status.json 中，并把 /root/quarantine/quarantine_report.md 写成四节。中断阈值要先测出平时的失败比例再定。

如果按平时的值来定阈值，只有违反超过一半的那一天的文件会被中断。status.json 不要手写，而要把 status 命令的答案原样保存——评分器会直接打开数据库来核对。报告里要用数字写出事实条数和处于打开状态的隔离条数。