姚章浩
4455d02262
scripts/restore_drill.py：把留档 PG 行按原文 \copy FROM 回填进目标库，核对每张表行数与清单、 抽样页面正文/标题逐字段比对，再用临时跟踪表 _drill_restored 精确删除并断言残留为 0。 实测（本地测试库 weknora_full，留档 20261009T041913Z-0b217e8a）：13,539 行全部回填成功、 8 张表行数与清单逐一致、冲突 0、抽样页正文一致（388 字符）、残留 0。 同时记录一个格式事实：留档 JSONL 是 psql \copy text 格式（转义），按行 json.loads 会失败 （knowledges 3830/3830、wiki_pages 218/1982），回填/程序化读取须先按 text 格式反转义。
356 lines
19 KiB
Bash
Executable File
356 lines
19 KiB
Bash
Executable File
#!/usr/bin/env bash
|
||
# 目标 WeKnora 知识库的只读行级快照（KB 维度）。
|
||
#
|
||
# 用途：发布 / 撤权 / 回滚前，把目标库**当前状态**留档，出事后有可回退的现场。
|
||
# 只做 SELECT + \copy，不写目标库任何一行。
|
||
#
|
||
# 用法：
|
||
# bash scripts/target_kb_snapshot.sh # 已绑定的目标库；云内执行（快）
|
||
# EXECUTION=tunnel bash scripts/target_kb_snapshot.sh # 走本机 SSH 隧道（慢，兜底）
|
||
# bash scripts/target_kb_snapshot.sh <kb_id> [<kb_id>] # 指定目标库
|
||
# INCLUDE_CHUNKS=1 bash scripts/target_kb_snapshot.sh # 连 chunks 一起导（大，按需）
|
||
# PULL_CHUNKS=1 bash scripts/target_kb_snapshot.sh # 云内执行时把 chunks 结果也拉回本机
|
||
#
|
||
# 环境变量：
|
||
# EXECUTION cloud（默认，与 DB 同 VPC 的云主机上跑）| tunnel（本机经 SSH 隧道）
|
||
# SNAPSHOT_ROOT 输出根目录（默认 backups/target-kb-snapshots）
|
||
# INCLUDE_CHUNKS 1=连 chunks 一起导（默认 0：秒级完成；向量在 ES 侧，行可从源端重发）
|
||
# PULL_CHUNKS 1=云内执行时把 chunks 结果也拉回本机（默认 0，留在云主机）
|
||
# PULL_ES 1=把 ES 文档导出（gzip JSONL）也拉回本机。默认 0：ES 导出有数百 MB
|
||
# （实测两库 692 MB），家用上行只有 ~0.6 MB/s（约 19 分钟），留在目标主机更实际；
|
||
# 门禁按 location=cloud:… 核清单声明与哈希，同机部署时直接读本地文件
|
||
# ES_BATCH ES scroll 每次取多少条（默认 2000）
|
||
# TUNNEL_HOST/TUNNEL_PORT 隧道模式的目标 PG 地址（默认 172.17.0.1:25432）
|
||
# SSH_CONFIG SSH 配置路径（默认 ~/.ssh/config.weknora-cloud-test）
|
||
# SSH_HOST 云主机别名（默认 weknora-test）
|
||
# APP_CONTAINER 云上提供 DB 凭据的容器（默认 weknora-test-v071-app-1）
|
||
# REMOTE_ROOT 云上快照目录（默认 /opt/weknora-test/kb-snapshots）
|
||
# WITH_ES 1=同时记录目标 ES 的文档数（默认 1）
|
||
# NOTE 附加到清单的备注（例如发布单号）
|
||
#
|
||
# 覆盖范围（PG 行级）：knowledge_bases / knowledges / knowledge_tags /
|
||
# knowledge_tag_relations / wiki_pages / wiki_folders / resources /
|
||
# resource_bindings（+ chunks，可选）。含软删行，不含表结构。
|
||
# 另外：ES 文档按库导出 es_documents.jsonl.gz（含 _id/_source，可按 _id 重建）。
|
||
#
|
||
# 不覆盖：Elasticsearch 文档与向量（见 ES 仓库 kb_backup 与 SLM 策略 kb-hourly/kb-daily）、
|
||
# OSS 对象本体（只在 resources.jsonl 里记路径/大小/哈希做清单）、表结构与索引。
|
||
# 整库级恢复请用阿里云 RDS 快照/PITR。
|
||
set -euo pipefail
|
||
|
||
WORKDIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||
cd "$WORKDIR"
|
||
|
||
EXECUTION="${EXECUTION:-cloud}"
|
||
SNAPSHOT_ROOT="${SNAPSHOT_ROOT:-$WORKDIR/backups/target-kb-snapshots}"
|
||
TUNNEL_HOST="${TUNNEL_HOST:-172.17.0.1}"
|
||
TUNNEL_PORT="${TUNNEL_PORT:-25432}"
|
||
INCLUDE_CHUNKS="${INCLUDE_CHUNKS:-0}"
|
||
PULL_CHUNKS="${PULL_CHUNKS:-0}"
|
||
# ES 文档也留档（只读 scroll 导出）；PULL_ES=0 时只在云主机留一份
|
||
PULL_ES="${PULL_ES:-0}"
|
||
ES_BATCH="${ES_BATCH:-2000}"
|
||
SSH_CONFIG="${SSH_CONFIG:-$HOME/.ssh/config.weknora-cloud-test}"
|
||
SSH_HOST="${SSH_HOST:-weknora-test}"
|
||
APP_CONTAINER="${APP_CONTAINER:-weknora-test-v071-app-1}"
|
||
# 备份目录契约：由 scripts/backup_layout.sh 写到 /etc/knowledge-governance/backup.env。
|
||
# 换主机（测试云主机 → 生产云主机）时只改这一处，脚本本身不动。
|
||
if [ -z "${KB_BACKUP_ROOT:-}" ] && [ -f /etc/knowledge-governance/backup.env ]; then
|
||
# shellcheck disable=SC1091
|
||
. /etc/knowledge-governance/backup.env
|
||
fi
|
||
KB_BACKUP_ROOT="${KB_BACKUP_ROOT:-/srv/kb-backups}"
|
||
REMOTE_ROOT="${REMOTE_ROOT:-$KB_BACKUP_ROOT/target-kb-snapshots}"
|
||
WITH_ES="${WITH_ES:-1}"
|
||
NOTE="${NOTE:-}"
|
||
|
||
KB_IDS=("$@")
|
||
if [ "${#KB_IDS[@]}" -eq 0 ]; then
|
||
mapfile -t KB_IDS < <(docker exec knowledge-governance-p1-postgres-1 psql -U governance -d governance -tA -c \
|
||
"SELECT unnest(ARRAY[internal_kb_id, external_kb_id]) FROM production_kb_bindings WHERE internal_kb_id <> '' OR external_kb_id <> ''" \
|
||
| sed '/^$/d' | sort -u)
|
||
fi
|
||
if [ "${#KB_IDS[@]}" -eq 0 ]; then
|
||
echo "no target knowledge base id given and none bound"; exit 1
|
||
fi
|
||
|
||
for id in "${KB_IDS[@]}"; do
|
||
case "$id" in
|
||
[0-9a-f]*-[0-9a-f]*-[0-9a-f]*-[0-9a-f]*-[0-9a-f]*) ;;
|
||
*) echo "invalid knowledge base id: $id"; exit 1 ;;
|
||
esac
|
||
done
|
||
|
||
STAMP="$(date -u +%Y%m%dT%H%M%SZ)"
|
||
PREFIX="$(printf '%s' "${KB_IDS[0]}" | cut -c1-8)"
|
||
OUT="$SNAPSHOT_ROOT/${STAMP}-${PREFIX}"
|
||
mkdir -p "$OUT"
|
||
|
||
KB_SQL="'$(printf '%s' "${KB_IDS[*]}" | sed "s/ /','/g")'"
|
||
|
||
# ---------- 生成导出脚本（本身也是留档的一部分） ----------
|
||
cat >"$OUT/export.sql" <<SQL
|
||
-- 目标知识库只读行级快照：$STAMP
|
||
-- 生成者：scripts/target_kb_snapshot.sh（EXECUTION=$EXECUTION）
|
||
-- 目标库：${KB_IDS[*]}
|
||
-- 说明：全部为 SELECT + \copy，不修改目标库。
|
||
\pset pager off
|
||
\timing off
|
||
|
||
\copy (SELECT to_jsonb(t) FROM knowledge_bases t WHERE t.id IN ($KB_SQL)) TO 'out/knowledge_bases.jsonl'
|
||
\copy (SELECT to_jsonb(t) FROM knowledges t WHERE t.knowledge_base_id IN ($KB_SQL)) TO 'out/knowledges.jsonl'
|
||
\copy (SELECT to_jsonb(t) FROM knowledge_tags t WHERE t.knowledge_base_id IN ($KB_SQL)) TO 'out/knowledge_tags.jsonl'
|
||
\copy (SELECT to_jsonb(r) FROM knowledge_tag_relations r WHERE r.knowledge_id IN (SELECT id FROM knowledges WHERE knowledge_base_id IN ($KB_SQL))) TO 'out/knowledge_tag_relations.jsonl'
|
||
\copy (SELECT to_jsonb(t) FROM wiki_pages t WHERE t.knowledge_base_id IN ($KB_SQL)) TO 'out/wiki_pages.jsonl'
|
||
\copy (SELECT to_jsonb(t) FROM wiki_folders t WHERE t.knowledge_base_id IN ($KB_SQL)) TO 'out/wiki_folders.jsonl'
|
||
-- 归属集合用 UNION 子查询（云上实测 0.6~0.8s；OR+IN 会退化成慢查询）。
|
||
-- 注意：psql 的 \copy 是单行元命令，且不接受 WITH 开头，整条语句必须写成一行。
|
||
\copy (SELECT to_jsonb(b) FROM resource_bindings b WHERE b.owner_id IN (SELECT id FROM knowledges WHERE knowledge_base_id IN ($KB_SQL) UNION SELECT id FROM chunks WHERE knowledge_base_id IN ($KB_SQL))) TO 'out/resource_bindings.jsonl'
|
||
\copy (SELECT to_jsonb(r) FROM resources r WHERE r.id IN (SELECT b.resource_id FROM resource_bindings b WHERE b.owner_id IN (SELECT id FROM knowledges WHERE knowledge_base_id IN ($KB_SQL) UNION SELECT id FROM chunks WHERE knowledge_base_id IN ($KB_SQL)))) TO 'out/resources.jsonl'
|
||
SQL
|
||
if [ "$INCLUDE_CHUNKS" = "1" ]; then
|
||
cat >>"$OUT/export.sql" <<SQL
|
||
\copy (SELECT to_jsonb(t) FROM chunks t WHERE t.knowledge_base_id IN ($KB_SQL)) TO 'out/chunks.jsonl'
|
||
SQL
|
||
fi
|
||
|
||
REMOTE_DIR=""
|
||
CLOUD_REF=""
|
||
if [ "$EXECUTION" = "cloud" ]; then
|
||
# ---------- 云内执行：与 DB 同 VPC，内网直连，不受家用带宽限制 ----------
|
||
REMOTE_DIR="$REMOTE_ROOT/$STAMP-${PREFIX}"
|
||
CLOUD_REF="${SSH_HOST}:$REMOTE_DIR"
|
||
ssh -F "$SSH_CONFIG" "$SSH_HOST" "mkdir -p '$REMOTE_DIR'"
|
||
scp -q -F "$SSH_CONFIG" "$OUT/export.sql" "$SSH_HOST:$REMOTE_DIR/export.sql"
|
||
scp -q -F "$SSH_CONFIG" "$SCRIPT_DIR/es_kb_export.py" "$SSH_HOST:$REMOTE_DIR/es_kb_export.py"
|
||
ssh -F "$SSH_CONFIG" "$SSH_HOST" \
|
||
APP_CONTAINER="$APP_CONTAINER" REMOTE_DIR="$REMOTE_DIR" INCLUDE_CHUNKS="$INCLUDE_CHUNKS" \
|
||
ES_INDEX="${ES_INDEX:-weknora_test_chunks}" ES_BATCH="$ES_BATCH" KB_LIST="$(IFS=,; echo "${KB_IDS[*]}")" \
|
||
KB_BACKUP_ROOT="$KB_BACKUP_ROOT" \
|
||
'bash -s' <<'REMOTE'
|
||
set -euo pipefail
|
||
env_value() { docker inspect "$APP_CONTAINER" --format '{{range .Config.Env}}{{println .}}{{end}}' | grep -E "^$1=" | head -1 | cut -d= -f2-; }
|
||
DB_HOST_VALUE=$(env_value DB_HOST); DB_PORT_VALUE=$(env_value DB_PORT)
|
||
DB_NAME_VALUE=$(env_value DB_NAME); DB_USER_VALUE=$(env_value DB_USER)
|
||
DB_PASS_VALUE=$(env_value DB_PASSWORD)
|
||
PGURL=$(python3 - "$DB_USER_VALUE" "$DB_PASS_VALUE" "$DB_HOST_VALUE" "$DB_PORT_VALUE" "$DB_NAME_VALUE" <<'PY'
|
||
import sys, urllib.parse as u
|
||
user, password, host, port, name = sys.argv[1:6]
|
||
print(f"postgres://{u.quote(user)}:{u.quote(password)}@{host}:{port}/{name}")
|
||
PY
|
||
)
|
||
docker run --rm --network host -e PGURL="$PGURL" \
|
||
-v "$REMOTE_DIR:/out" -v "$REMOTE_DIR/export.sql:/scripts/export.sql:ro" postgres:16-alpine \
|
||
sh -c 'psql "$PGURL" -q -f /scripts/export.sql' >/dev/null
|
||
rm -f "$REMOTE_DIR/export.sql"
|
||
if [ "$INCLUDE_CHUNKS" = "1" ] && [ -f "$REMOTE_DIR/chunks.jsonl" ]; then
|
||
gzip -f "$REMOTE_DIR/chunks.jsonl"
|
||
fi
|
||
python3 - "$REMOTE_DIR" <<'PY'
|
||
import hashlib, json, pathlib, sys
|
||
out = pathlib.Path(sys.argv[1]); files = []
|
||
for path in sorted(out.glob("*.jsonl*")):
|
||
digest = hashlib.sha256(); size = 0
|
||
with path.open("rb") as handle:
|
||
for block in iter(lambda: handle.read(1 << 20), b""):
|
||
digest.update(block); size += len(block)
|
||
files.append({"file": path.name, "bytes": size, "sha256": digest.hexdigest()})
|
||
(out / "REMOTE_MANIFEST.json").write_text(json.dumps({"host_files": files}, indent=2) + "\n")
|
||
print("remote files:", " ".join(f"{f['file']}={f['bytes']}" for f in files))
|
||
PY
|
||
# ES 文档导出（只读 scroll）：云主机可直连 ES 容器
|
||
if [ -f "$REMOTE_DIR/es_kb_export.py" ]; then
|
||
ES_IP=$(docker inspect weknora-test-v071-elasticsearch-1 --format "{{json .NetworkSettings.Networks}}" \
|
||
| python3 -c "import json,sys; d=json.load(sys.stdin); print([v['IPAddress'] for v in d.values() if v.get('IPAddress')][0])")
|
||
ES_KB_ARGS=""
|
||
for kb in $(printf '%s' "$KB_LIST" | tr ',' ' '); do ES_KB_ARGS="$ES_KB_ARGS --kb $kb"; done
|
||
python3 "$REMOTE_DIR/es_kb_export.py" --endpoint "http://$ES_IP:9200" --index "$ES_INDEX" \
|
||
$ES_KB_ARGS --output "$REMOTE_DIR/es_documents.jsonl.gz" --batch "$ES_BATCH"
|
||
rm -f "$REMOTE_DIR/es_kb_export.py"
|
||
python3 - "$REMOTE_DIR" <<'PY'
|
||
import hashlib, json, pathlib, sys
|
||
out = pathlib.Path(sys.argv[1]); files = []
|
||
for path in sorted(out.glob("*.jsonl*")):
|
||
digest = hashlib.sha256(); size = 0
|
||
with path.open("rb") as handle:
|
||
for block in iter(lambda: handle.read(1 << 20), b""):
|
||
digest.update(block); size += len(block)
|
||
files.append({"file": path.name, "bytes": size, "sha256": digest.hexdigest()})
|
||
(out / "REMOTE_MANIFEST.json").write_text(json.dumps({"host_files": files}, indent=2) + "\n")
|
||
print("remote files:", " ".join(f"{f['file']}={f['bytes']}" for f in files))
|
||
PY
|
||
fi
|
||
exit 0
|
||
REMOTE
|
||
# 小表拉回本机；chunks 默认留在云主机（PULL_CHUNKS=1 才拉）
|
||
scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/*.jsonl" "$OUT/" 2>/dev/null || true
|
||
scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/REMOTE_MANIFEST.json" "$OUT/" 2>/dev/null || true
|
||
if [ "$PULL_CHUNKS" = "1" ]; then
|
||
scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/chunks.jsonl.gz" "$OUT/" 2>/dev/null || true
|
||
fi
|
||
# ES 文档：默认拉回本机；PULL_ES=0 时只拉清单（大文件留在云主机）
|
||
scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/es_documents.jsonl.gz.manifest.json" "$OUT/" 2>/dev/null || true
|
||
if [ "$PULL_ES" = "1" ]; then
|
||
scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/es_documents.jsonl.gz" "$OUT/" 2>/dev/null || true
|
||
fi
|
||
else
|
||
# ---------- 隧道模式（兜底）：本机经 SSH 隧道直连 ----------
|
||
SECRET_FILE="secrets/release-sim/target_database_url"
|
||
[ -f "$SECRET_FILE" ] || { echo "missing $SECRET_FILE"; exit 1; }
|
||
PGURL="$(python3 - "$SECRET_FILE" "$TUNNEL_HOST" "$TUNNEL_PORT" <<'PY'
|
||
import re, sys, pathlib
|
||
url = pathlib.Path(sys.argv[1]).read_text().strip()
|
||
print(re.sub(r'@[^/@]+:\d+/', '@%s:%s/' % (sys.argv[2], sys.argv[3]), url))
|
||
PY
|
||
)"
|
||
docker run --rm --network host -e PGURL="$PGURL" \
|
||
-v "$OUT:/out" -v "$OUT/export.sql:/scripts/export.sql:ro" postgres:16 \
|
||
sh -c 'psql "$PGURL" -q -f /scripts/export.sql' >/dev/null
|
||
ES_KB_ARGS=""
|
||
for kb in "${KB_IDS[@]}"; do ES_KB_ARGS="$ES_KB_ARGS --kb $kb"; done
|
||
python3 "$SCRIPT_DIR/es_kb_export.py" --endpoint "http://${TUNNEL_HOST}:29200" --index "${ES_INDEX:-weknora_test_chunks}" \
|
||
$ES_KB_ARGS --output "$OUT/es_documents.jsonl.gz" --batch "$ES_BATCH"
|
||
fi
|
||
|
||
# ---------- ES 文档数（只读，失败不阻塞） ----------
|
||
ES_SUMMARY=""
|
||
if [ "$WITH_ES" = "1" ]; then
|
||
for id in "${KB_IDS[@]}"; do
|
||
count="$(curl -s --max-time 15 -H 'Content-Type: application/json' \
|
||
-X POST "http://${TUNNEL_HOST}:29200/weknora_test_chunks/_count" \
|
||
-d "{\"query\":{\"term\":{\"knowledge_base_id.keyword\":\"$id\"}}}" \
|
||
| python3 -c 'import json,sys; print(json.load(sys.stdin).get("count","?"))' 2>/dev/null || echo "?")"
|
||
ES_SUMMARY="${ES_SUMMARY}${id}=${count} "
|
||
done
|
||
fi
|
||
|
||
# ---------- 清单 + 说明 ----------
|
||
python3 - "$OUT" "$STAMP" "$EXECUTION" "$INCLUDE_CHUNKS" "$ES_SUMMARY" "$NOTE" "$CLOUD_REF" "${KB_IDS[@]}" <<'PY'
|
||
import hashlib, json, pathlib, sys
|
||
out = pathlib.Path(sys.argv[1]); stamp, execution = sys.argv[2], sys.argv[3]
|
||
include_chunks = sys.argv[4] == "1"; es_summary = sys.argv[5].strip(); note = sys.argv[6]
|
||
remote_ref = sys.argv[7]; kb_ids = sys.argv[8:]
|
||
files = []
|
||
for path in sorted(out.glob("*.jsonl")):
|
||
digest = hashlib.sha256()
|
||
with path.open("rb") as handle:
|
||
for block in iter(lambda: handle.read(1 << 20), b""):
|
||
digest.update(block)
|
||
files.append({"file": path.name, "rows": sum(1 for _ in path.open("rb")),
|
||
"bytes": path.stat().st_size, "sha256": digest.hexdigest(), "location": "local"})
|
||
for path in sorted(out.glob("chunks.jsonl.gz")):
|
||
digest = hashlib.sha256()
|
||
with path.open("rb") as handle:
|
||
for block in iter(lambda: handle.read(1 << 20), b""):
|
||
digest.update(block)
|
||
files.append({"file": path.name, "bytes": path.stat().st_size,
|
||
"sha256": digest.hexdigest(), "location": "local"})
|
||
# ES 文档留档（本地有就按本地哈希；只在云主机则在 remote 段补）
|
||
es_sidecar = out / "es_documents.jsonl.gz.manifest.json"
|
||
es_counts = {}
|
||
if es_sidecar.exists():
|
||
es_sidecar_data = json.loads(es_sidecar.read_text())
|
||
es_counts = {str(k): v for k, v in es_sidecar_data.get("knowledge_base_document_counts", {}).items()}
|
||
for path in sorted(out.glob("es_documents.jsonl.gz")):
|
||
digest = hashlib.sha256()
|
||
with path.open("rb") as handle:
|
||
for block in iter(lambda: handle.read(1 << 20), b""):
|
||
digest.update(block)
|
||
files.append({"file": path.name, "bytes": path.stat().st_size,
|
||
"sha256": digest.hexdigest(), "location": "local"})
|
||
remote_manifest = out / "REMOTE_MANIFEST.json"
|
||
if remote_manifest.exists():
|
||
for item in json.loads(remote_manifest.read_text()).get("host_files", []):
|
||
if not any(existing["file"] == item["file"] for existing in files):
|
||
item["location"] = f"cloud:{remote_ref}"
|
||
files.append(item)
|
||
manifest = {
|
||
"kind": "weknora-target-kb-snapshot",
|
||
"created_at": stamp,
|
||
"execution": execution,
|
||
"knowledge_base_ids": kb_ids,
|
||
"profile": "kb-rows" + ("" if include_chunks else " (chunks excluded: 向量在 ES 侧，行可从源端重发)"),
|
||
"covers": ["knowledge_bases", "knowledges", "knowledge_tags", "knowledge_tag_relations",
|
||
"wiki_pages", "wiki_folders", "resource_bindings", "resources",
|
||
"elasticsearch_documents"]
|
||
+ (["chunks"] if include_chunks else []),
|
||
"does_not_cover": ["OSS object bodies (listed by reference in resources.jsonl; versioning protects overwrite/delete)",
|
||
"Elasticsearch index settings/mappings (use ES repository kb_backup snapshots)",
|
||
"table structure / indexes", "rows outside the listed knowledge bases",
|
||
"whole-database recovery (use Aliyun RDS snapshot / PITR)"],
|
||
"es_document_counts": es_counts or (es_summary or None),
|
||
"es_documents_file": "es_documents.jsonl.gz" if (es_counts or (out / "es_documents.jsonl.gz.manifest.json").exists()) else None,
|
||
"remote_directory": remote_ref or None,
|
||
"note": note or None,
|
||
"files": files,
|
||
}
|
||
(out / "MANIFEST.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n")
|
||
print("snapshot:", out)
|
||
print("files:", " ".join(f"{item['file']}={item.get('rows', '-')}" for item in files))
|
||
print("local bytes:", sum(item["bytes"] for item in files if item.get("location") == "local"))
|
||
print("es docs:", es_summary or "-")
|
||
PY
|
||
|
||
cat >"$OUT/README.md" <<'MD'
|
||
# 目标知识库快照（只读行级）
|
||
|
||
## 它是什么
|
||
发布 / 撤权 / 回滚**之前**的目标库现场，两层一起留：
|
||
|
||
1. **PG 行**：KB 维度的行（含软删行），每张表一个 JSONL（`to_jsonb(row)` 一行一条）；
|
||
2. **ES 文档**：`es_documents.jsonl.gz`（一行一条 `{"_index","_id","_source"}`，含向量字段）
|
||
+ `es_documents.jsonl.gz.manifest.json`（每库文档数 / 字节 / sha256）。
|
||
|
||
附 `MANIFEST.json`（行数 / 字节 / sha256 / 覆盖范围 / 位置）与 `export.sql`（导出用的原始查询）。
|
||
生成过程只有 SELECT / `\copy` / `_count` / `_search`，不修改目标库与索引。
|
||
|
||
> 若 `MANIFEST.json` 里某文件的 `location` 是 `cloud:<主机>:<路径>`，说明字节留在目标主机上
|
||
> （家用带宽拉不动），本地只有清单与哈希；同机执行时按该路径直接读。
|
||
>
|
||
> JSONL 是 psql `\copy` **text 格式**（反斜杠/控制字符被转义）：回填用对称的 `\copy FROM`
|
||
> 即可原文还原（`scripts/restore_drill.py` 走的就是这条路，并核对行数与抽样内容）；
|
||
> 用程序按行 `json.loads` 会失败，需先反转义。
|
||
>
|
||
> `es_documents.jsonl.gz` 的 sha256 每次导出都不同（scroll 顺序非确定）；判断"两份留档是不是
|
||
> 同一份现场"请看 `es_document_counts` 与逐条的 `_id` 集合，而不是文件哈希。
|
||
|
||
## 它不是什么
|
||
- 不含 Elasticsearch 文档与向量（见 ES 仓库 `kb_backup` 与 SLM 策略 `kb-hourly`/`kb-daily`）、
|
||
不含 OSS 对象本体（`resources.jsonl` 里只有路径 / 大小 / 哈希清单）、不含表结构与索引。
|
||
- 不是整库级灾备。**整库级恢复请用阿里云 RDS 的快照 / 按时间点恢复（PITR）**：
|
||
恢复到临时实例后，再按 KB 导出需要的行，这样连索引和约束一起回到某个时间点。
|
||
|
||
## 怎么恢复（行级）
|
||
先确认目标库当前状态已被另一份快照覆盖（避免二次覆盖），然后按外键顺序灌回：
|
||
|
||
```bash
|
||
URL=<目标库连接串>
|
||
OUT=<本快照目录>
|
||
docker run --rm --network host -e PGURL="$URL" -v "$OUT:/snap:ro" postgres:16 \
|
||
psql "$PGURL" -v ON_ERROR_STOP=1 <<'SQL'
|
||
BEGIN;
|
||
DO $$
|
||
DECLARE t text;
|
||
BEGIN
|
||
FOREACH t IN ARRAY ARRAY[
|
||
'knowledge_bases','knowledges','chunks','knowledge_tags','knowledge_tag_relations',
|
||
'resources','resource_bindings','wiki_folders','wiki_pages'] LOOP
|
||
EXECUTE format('CREATE TEMP TABLE stage_%s (raw jsonb)', t);
|
||
EXECUTE format($f$\copy stage_%s(raw) FROM '/snap/%s.jsonl'$f$, t, t);
|
||
EXECUTE format('INSERT INTO %I SELECT (jsonb_populate_record(NULL::%I, raw)).* FROM stage_%s
|
||
ON CONFLICT (id) DO NOTHING', t, t, t);
|
||
EXECUTE format('DROP TABLE stage_%s', t);
|
||
END LOOP;
|
||
END $$;
|
||
COMMIT;
|
||
SQL
|
||
```
|
||
|
||
注意：
|
||
- 只恢复了**行**；缺行时不会删、也不会改（`ON CONFLICT DO NOTHING`）。
|
||
- 与目标库既有唯一索引冲突的行会被跳过，恢复后请核对 `MANIFEST.json` 的行数。
|
||
- 向量在 Elasticsearch 侧，行恢复后必须重建索引 / 重放 ES 文档，否则检索不到。
|
||
MD
|
||
|
||
echo "done: $OUT" |