#!/usr/bin/env python3 """留档回填演练：把发布前留档（PG 行级 JSONL）真的插进一个目标库，核对行数与内容，再清理。 为什么需要它：哈希校验只能证明"文件没坏"，不能证明"能恢复"。这个演练完整走一遍 "psql 按原文回填 → 行数与清单一致 → 抽样逐字段比对 → 精确删除插入的行（残留必须为 0）"。 格式注意：留档 JSONL 是 psql `\\copy` **text 格式**输出（反斜杠与控制字符被转义）， 所以回填必须用对称的 `\\copy FROM`（由 psql 反转义）；按行 `json.loads` 会失败， 程序化读取需先反转义（见 unescape_text_format）。 用法： python3 restore_drill.py --snapshot-dir <快照目录> --target-url <目标库URL> [--keep] [--slug <页面slug>] 目标库选择：演练用本地测试库（如 WKNFull 的 weknora_full），不要用生产库。 演练会在目标库创建并用完即删一张跟踪表 `_drill_restored`（记录插入了哪些行，保证清理精确、无残留）。 """ import argparse import json import pathlib import subprocess import sys # 依赖顺序：主表 → 子表；resources 在 resource_bindings 之前 TABLES = [ "knowledge_bases", "knowledges", "knowledge_tags", "wiki_folders", "wiki_pages", "resources", "resource_bindings", "knowledge_tag_relations", ] TRACKING_TABLE = "_drill_restored" def psql(image, snapshot_dir, url, sql): command = [ "docker", "run", "--rm", "-i", "--network", "host", "-e", f"PGURL={url}", "-v", f"{snapshot_dir}:/snap:ro", image, "sh", "-c", 'psql "$PGURL" -v ON_ERROR_STOP=1 -f -', ] process = subprocess.run(command, input=sql, text=True, capture_output=True) if process.returncode != 0: sys.stderr.write(process.stdout[-4000:] + "\n" + process.stderr[-4000:]) raise SystemExit(f"回填失败（psql 退出码 {process.returncode}）") return process.stdout def build_insert(table): """回填一张表：staging(jsonb 原文) → 目标表（冲突跳过）→ 记录插入了哪些行。""" return f"""WITH ins AS ( INSERT INTO {table} SELECT (jsonb_populate_record(NULL::{table}, s.row)).* FROM stage_{table} s ON CONFLICT DO NOTHING RETURNING to_jsonb({table}) AS row_json ), rec AS ( INSERT INTO {TRACKING_TABLE} (table_name, row) SELECT '{table}', ins.row_json FROM ins RETURNING 1 ) SELECT 'drill:{table}' AS marker, (SELECT count(*) FROM stage_{table}) AS staged, (SELECT count(*) FROM rec) AS inserted;""" def main(): parser = argparse.ArgumentParser() parser.add_argument("--snapshot-dir", required=True) parser.add_argument("--target-url", required=True) parser.add_argument("--pg-image", default="postgres:17-alpine") parser.add_argument("--slug", default="", help="抽样比对的 wiki 页面 slug（默认自动挑一个）") parser.add_argument("--keep", action="store_true", help="保留插入的行（默认核对后精确删除）") arguments = parser.parse_args() snapshot = pathlib.Path(arguments.snapshot_dir).resolve() manifest = json.loads((snapshot / "MANIFEST.json").read_text()) expected = {item["file"].removesuffix(".jsonl"): item.get("rows") for item in manifest["files"] if item["file"].endswith(".jsonl")} print(f"留档: {snapshot.name} | created_at={manifest['created_at']} " f"| 库: {', '.join(manifest['knowledge_base_ids'])}") slug = arguments.slug if not slug: with (snapshot / "wiki_pages.jsonl").open("rb") as handle: for raw in handle: try: row = json.loads(unescape_text_format(raw)) except Exception: continue if row.get("content"): slug = row["slug"] break print(f"抽样页面: {slug or '（留档里没有带正文的页面）'}") statements = [ "\\set ON_ERROR_STOP on", f"DROP TABLE IF EXISTS {TRACKING_TABLE};", f"CREATE TABLE {TRACKING_TABLE} (table_name text NOT NULL, row jsonb NOT NULL);", "BEGIN;", ] for table in TABLES: if not (snapshot / f"{table}.jsonl").exists(): continue statements += [ f"CREATE TEMP TABLE stage_{table} (row jsonb);", f"\\copy stage_{table} (row) FROM '/snap/{table}.jsonl'", build_insert(table), ] if slug: statements.append(f"""SELECT 'drill:sample' AS marker, ((s.r).title = t.title) AS title_equal, ((s.r).content = t.content) AS content_equal, length((s.r).content) AS chars FROM (SELECT jsonb_populate_record(NULL::wiki_pages, row) AS r FROM stage_wiki_pages WHERE (jsonb_populate_record(NULL::wiki_pages, row)).slug = '{slug}') s JOIN wiki_pages t ON t.id = (s.r).id;""") statements.append("COMMIT;") if not arguments.keep: statements.append("BEGIN;") for table in reversed(TABLES): if not (snapshot / f"{table}.jsonl").exists(): continue statements.append( f"DELETE FROM {table} t USING {TRACKING_TABLE} d " f"WHERE d.table_name = '{table}' AND to_jsonb(t) = d.row;") statements.append( f"SELECT 'drill:cleanup:{table}' AS marker, count(*) AS left " f"FROM {table} t JOIN {TRACKING_TABLE} d ON d.table_name = '{table}' AND to_jsonb(t) = d.row;") statements.append("COMMIT;") statements.append(f"DROP TABLE {TRACKING_TABLE};") output = psql(arguments.pg_image, snapshot, arguments.target_url, "\n".join(statements) + "\n") staged, inserted, left, sample = {}, {}, {}, {} for line in output.splitlines(): parts = [part.strip() for part in line.split("|")] if not parts or not parts[0].startswith("drill:"): continue marker = parts[0].split(":", 1)[1] if marker.startswith("cleanup:"): if len(parts) != 2: raise SystemExit(f"清理结果列数异常（{len(parts)}）: {line}") left[marker.split(":", 1)[1]] = int(parts[1]) elif marker == "sample": if len(parts) != 4: raise SystemExit(f"抽样结果列数异常（{len(parts)}）: {line}") sample = {"title_equal": parts[1] == "t", "content_equal": parts[2] == "t", "chars": parts[3]} else: if len(parts) != 3: raise SystemExit(f"回填结果列数异常（{len(parts)}）: {line}") staged[marker] = int(parts[1]) inserted[marker] = int(parts[2]) print(f"\n{'表':<26}{'清单行数':>10}{'回填尝试':>10}{'已插入':>8}{'冲突跳过':>10}{'清理后残留':>12}") bad = [] for table in TABLES: if table not in staged: continue want, got, added = expected.get(table), staged[table], inserted.get(table, 0) residue = left.get(table) if want != got: bad.append(f"{table} 行数与清单不符（清单 {want} / 回填 {got}）") if arguments.keep: residue_text = "（保留）" elif residue is None: residue_text = "未验证" bad.append(f"{table} 清理结果未取到，无法证明无残留") else: residue_text = str(residue) if residue: bad.append(f"{table} 清理后仍有 {residue} 行") print(f"{table:<26}{str(want):>10}{got:>10}{added:>8}{got - added:>10}{residue_text:>12}") if not sample: bad.append("抽样页面没有完成比对（结果未取到）") else: print(f"\n抽样比对: 标题{'一致' if sample['title_equal'] else '不一致'} / " f"正文{'一致' if sample['content_equal'] else '不一致'}（{sample['chars']} 字符）") if not sample["content_equal"] or not sample["title_equal"]: bad.append("抽样页面的正文/标题与留档不一致") if bad: raise SystemExit("回填演练未通过: " + "; ".join(bad)) print(f"结论: {sum(staged.values())} 行按原文回填成功（与清单一致、抽样逐字段相同）；" + ("插入的行保留在目标库中。" if arguments.keep else "核对后已精确删除，残留为 0。")) def unescape_text_format(raw: bytes) -> bytes: """psql text 格式反转义（\\\\→\\，\\n→换行，\\t→制表符…）。""" out = bytearray() index = 0 while index < len(raw): char = raw[index] if char == 0x5C and index + 1 < len(raw): following = raw[index + 1] out.append({0x62: 0x08, 0x66: 0x0C, 0x6E: 0x0A, 0x72: 0x0D, 0x74: 0x09, 0x76: 0x0B}.get(following, following)) index += 2 else: out.append(char) index += 1 return bytes(out) if __name__ == "__main__": main()