staging 镜像同步指南（MySQL 原表 → staging.db）

内网 MySQL xuangubao 库 5 张源表 → 本地 data/staging.db（原样镜像）的使用说明。工具：scripts/sync/sync_staging.py（单文件、零项目内依赖，可整体拷走）。表结构与字段释义见 staging-schema.md；本体库的加工同步（ETL）见 sync.md。方案背景与设计取舍：data-distribution-design.md §5.1。

当前状态（2026-10-01）：镜像、ETL、图谱、TOS 备份全链路已实测跑通（见 §1 更新次序）。 ETL 已切换为读 staging（StagingConn 适配层，取数代码零改动）——staging 现在是 ETL 的唯一取数入口， "同步（搬数据）"与"加工（ETL 本体化）"两段正式解耦，加工重算不再依赖 VPN。

1. 概览：三库关系与更新次序

内网 MySQL (10.0.46.53, 只读)
│  live_tuoshui_news / tuoshui_msgs / 两张 tag 关联表 / plate_rank_infos
│
│  ① scripts/sync/sync_staging.py（增量，秒级）
▼
data/staging.db ──── 原样镜像（表名字段与源一致，agent 结构说明 staging_schema.json）
│                    字段字典：staging-schema.md
│  ② scripts/sync/sync_data.py（ETL 增量，五阶段校验回退；⚠ 需先停本地 Web 服务）
▼
data/xgb_wiki.db ─── 本体库（report/theme/stock/关联/聚合 23 表）
│                    字段字典：../core/data-model.md §1–§6
│  ③ scripts/kg/build_graph.py（三元组抽取，增量按 extraction_log）
▼
data/graph.db ────── 知识图谱（entity/edge/mention/cluster/image 6 表）
│                    字段字典：../core/data-model.md §7
│  ④ scripts/ops/backup_tos.py（三库一致性快照 + .env）
▼
火山 TOS  xgb-wiki/backup/{时间戳}/（滚动保留 6 份，§6）

日常更新次序（每步独立成败、独立进度；哪步失败修好后从该步续跑）：

步	命令	产物	实测耗时	注意
⓪ TOS 预检	`python scripts/ops/tos_preflight.py`	本地与云端基线对齐（多机场景他机备份过时先恢复三库，旧库移入 backups/preflight-*/）	~10s（恢复 ~33s）	`--check` 只报告；TOS 不可达可跳过
① 镜像	`python scripts/sync/sync_staging.py`	staging.db + staging_schema.json	增量 ~1.3s / 全量 ~17s	首次或列漂移用 `--full`
② ETL	`python scripts/sync/sync_data.py`	xgb_wiki.db（含聚合/观点单元/时间窗）	~12s	先停本地 Web（5010 端口）；同步前自动备份到 data/backups/
③ 图谱	`python scripts/kg/build_graph.py`	graph.db（entity/edge/mention）	~20s	`fail=N` 为无可抽内容篇目的正常跳过；主题群重算另跑 `scripts/kg/kg_recluster.py`
④ 备份	`python scripts/ops/backup_tos.py`	TOS backup/{ts}/ + latest.json	~30s	前置校验 staging 新鲜度（24h）；staging 与 ETL 进度独立属正常

镜像只做原样搬运、不含任何解析——这是"同步（搬数据）"与"加工（ETL 做本体化）"分层的稳定性来源：镜像层出错环节极少，加工逻辑迭代时从本地 staging 重算即可，不依赖 VPN。

2. 环境准备

项	要求
Python	conda 环境 `xgb-wiki`（Python 3.13）。已激活时直接 `python`；未激活用 `conda run -n xgb-wiki python …`（注意：`conda run` 不透传 heredoc stdin，内嵌脚本请用 env 的 `python.exe` 全路径）
依赖	pymysql、python-dotenv（环境内已装；工具本身只用标准库 + 这两个）
网络	VPN 在线（MySQL 在内网 10.0.46.53）；`--selftest` 离线可用，无需任何网络
配置	`.env`：`DB_HOST/DB_PORT/DB_DATABASE/DB_USER/DB_PASSWORD`（MySQL）；`TOS_*` 六键（备份通道，§6）。`.env` 不进 git

3. 命令速查

python scripts/sync/sync_staging.py            # 增量同步（日常，秒级）
python scripts/sync/sync_staging.py --full     # 全量重建（首次/结构变化/修复）
python scripts/sync/sync_staging.py --check    # 仅环境检查，不动数据
python scripts/sync/sync_staging.py --selftest # 离线自检（不连网，17 项断言）

场景	命令
日常数据更新	默认增量（只拉上次进度之后的新增行）
首次建库 / 换机器初始化	`--full`（约 17 秒，64MB 级）
出门在外想确认能不能同步	`--check`
拿到工具先验证一遍（人或 agent）	`--selftest`
报 `SchemaDriftError`（源表加列了）	`--full` 重建
staging.db 损坏/误删	直接 `--full`，中间产物无需心疼

退出码（对齐 sync_data.py 约定）：0 成功；2 环境检查失败（未开始同步）；3 同步/校验失败。

4. 同步规则

表分组

组	表	方式
增量组	`live_tuoshui_news`、`tuoshui_msgs`	记住已同步最大 id（`staging_meta` 的 `wm_*`），只拉新增；每轮把源端 `is_deleted` 标记刷过来
整表组	两张 tag 关联表、`plate_rank_infos`	每轮整表刷新（行数小、毫秒级）

关键语义

表名、字段名与 MySQL 源表完全一致——建表由 SHOW FULL COLUMNS 动态生成，源表加字段自动跟随；列集不一致时报 SchemaDriftError 提示 --full；
镜像含已删除行（is_deleted=1 原样保留），业务查询一律加 is_deleted=0；
失败不推进同步进度，下次自动从原位置续跑；
每次同步结束自动导出 data/staging_schema.json（表/字段释义/行数/进度/栏目 tag 字典）——agent 使用 staging 前先读它；
同步历史在 staging.db 的 staging_log 表。

5. 验收与对账

幂等性：增量跑完紧接着再跑一次，应显示 +0 新增、进度不变；
源端对账（五表行数逐一比对，大表 is_deleted=0 口径）：

python - <<'EOF'
import sqlite3, sys; sys.path.insert(0, "scripts/sync")
from sync_staging import connect_mysql, ALL_TABLES
my, st = connect_mysql(), sqlite3.connect("data/staging.db")
with my.cursor() as cur:
    for t in ALL_TABLES:
        w = " WHERE is_deleted=0" if t == "live_tuoshui_news" else ""
        a = cur.execute(f"SELECT COUNT(*) FROM `{t}`{w}").fetchone()[0]
        b = st.execute(f'SELECT COUNT(*) FROM "{t}"{w}').fetchone()[0]
        print(("✓" if a == b else "✗"), t, a, b)
my.close(); st.close()
EOF

离线自检：--selftest 覆盖建表/类型映射/增量幂等/删除标记/导出等 17 项断言（已接入 run_tests.py 第 16 节，CI 每次构建自动跑）。

6. TOS 备份与取回

每次同步后备份（脚本 scripts/ops/backup_tos.py）：三个库（xgb_wiki.db / graph.db / staging.db）的 SQLite 在线一致性快照 + .env（全部凭据）+ manifest 上传到火山 TOS 上海桶，滚动保留最近 6 份：

python scripts/ops/backup_tos.py            # 前置校验 staging 同步新鲜度（24h 内 success）
python scripts/ops/backup_tos.py --dry-run  # 预演（快照不上传）
python scripts/ops/backup_tos.py --force    # 跳过新鲜度校验（灾备：VPN 不可用时备份本地现状）
python scripts/ops/backup_tos.py --keep 6   # 滚动保留份数

桶内结构：xgb-wiki/backup/{YYYYMMDD-HHMM}/（三库 + .env + manifest.json），顶层 xgb-wiki/backup/latest.json 指向最新一份。上传后逐文件做远端 etag 与本地 md5 比对；旧备份按时间戳滚动删除。备份失败独立计退出码（2 前置失败 / 3 上传失败）， 不影响已完成的同步。

架构说明：staging.db 本身 16 秒可再生，备份它是为换设备免 VPN 重拉与灾备冗余； .env 随备份上传使新机器"下载即完整复原"——前提是接受"私有桶 + IAM 受限 AK" 保护下的云端密钥存储（TOS AK 泄漏即全部凭据泄漏，凭据轮换后重跑备份覆盖即可）。

取回（新机器复原）：

python - <<'EOF'
from pathlib import Path
import tos, json
ENV = dict(l.split("=", 1) for l in Path(".env").read_text(encoding="utf-8").splitlines()
           if l.strip() and not l.strip().startswith("#") and "=" in l)
c = tos.TosClientV2(ENV["TOS_ACCESS_KEY"].strip(), ENV["TOS_SECRET_KEY"].strip(),
                    ENV["TOS_ENDPOINT"].strip(), ENV["TOS_REGION"].strip())
b = ENV["TOS_BUCKET"].strip()
latest = json.loads(c.get_object(b, "xgb-wiki/backup/latest.json").read())["generated_at"]
ver = None
for o in c.list_objects_type2(b, prefix="xgb-wiki/backup/", max_keys=200).contents:
    parts = o.key.split("/")
    if len(parts) > 3 and parts[2] and parts[2] != "latest.json":
        ver = max(ver or "", parts[2])
for name in ("xgb_wiki.db", "graph.db", "staging.db", ".env", "manifest.json"):
    c.download_file(b, f"xgb-wiki/backup/{ver}/{name}", name if name == ".env" else f"data/{name}")
    print("已下载", name, "←", ver)
EOF

演进：P1b 落地后成品库分发改走 GitHub Releases（Releases 管"分发"，TOS 备份继续管 "全量灾备"，职责并行）；P1c 落地后 staging 以内容寻址分片增量同步（xgb-wiki/staging/ 前缀），staging 整库备份届时退役。

7. 故障排查

现象	处理
退出码 2，提示 MySQL 连接失败	VPN 未连或凭据变更——先 `--check` 定位；确认 `.env` 的 `DB_*`
退出码 2，提示源表缺列	源表结构变化超出必需列下限——核对 `REQUIRED_COLUMNS` 与源库
`SchemaDriftError`	staging 与源表列集不一致（源端加/删列）——`--full` 重建
退出码 3，同步中断	看 `staging_log` 最新记录的 errors 字段；进度未推进，修复后直接重跑
`conda run` 跑内嵌脚本无输出/报错	`conda run` 不透传 stdin 与多行 `-c`——改用 env 的 `python.exe` 全路径
git push 连不上 127.0.0.1:7897	本机 URL 级代理配置指向未启动的代理——`git -c http.https://github.com.proxy= push` 临时直连
想确认数据有多新	查 `staging_meta` 的 `wm_*` 与 `last_synced_at`，或读 `staging_schema.json`

8. 路线图（与另两条通道的关系）

通道	状态	说明
staging 镜像（本页）	✅ 已上线	MySQL → staging.db，实测通过
ETL 加工（sync.md）	运行中，待切换	当前直连 MySQL；下一步改为读 staging，实现"加工离线重算"（P1a 后半）
成品库分发（GitHub Releases）	未实施	`publish_data.py` 自动发布，WeKnora 一条命令取数（P1b，design §5.2–5.5）
staging 分片版本化（TOS）	未实施	内容寻址分片 + manifest，换设备只拉增量（P1c，design §5.7）