staging 同步最小工具实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking. 状态 v1.4（2026-09-29 更新）：TOS 切换至上海桶 arkclaw-tos-2101988067-cn-shanghai（cn-shanghai，前缀 xgb-wiki/staging），新凭证已验证；接入用官方 Python SDK（包名 tos，原生端点），全链路验证通过，.env 端点已切换为原生域名——详见文末"P1c 衔接"。此前 v1.2：conda 环境 xgb-wiki（Python 3.13.15，对齐 CI）已建好并装入全部依赖，计划内命令统一以 conda run -n xgb-wiki python … 显式运行。v1.1 的"无 venv、用系统 Python"修正作废。

Goal: 交付一个单文件、零项目内依赖的最小工具，把内网 MySQL 的 5 张源表原样镜像到本地 data/staging.db（表名字段与源完全一致），并配套含字段释义的人类/agent 双受众结构说明——拿到库即可直接上手开发。

Architecture: 单文件脚本 scripts/sync/sync_staging.py，运行时读 SHOW FULL COLUMNS 动态建表（杜绝两边结构漂移）；大表按主键 id 记同步进度增量拉取并同步删除标记，小表每轮整表刷新；每次同步结束导出 data/staging_schema.json（agent 机读结构说明）。离线自检 --selftest 用内置模拟 MySQL 驱动临时 staging，全程不连网。

Tech Stack: conda 环境 xgb-wiki（Python 3.13.15，与 CI 的 3.13 对齐）；仅标准库 + pymysql + python-dotenv（分片阶段另用 pyarrow 25 / boto3，均已装入该环境并验证），SQLite3。

Spec: docs/data-sync/data-distribution-design.md §5.1（v1.3）

Global Constraints

运行环境：统一在 conda 环境 xgb-wiki 中运行——计划命令写作 conda run -n xgb-wiki python …；已激活环境（conda activate xgb-wiki）时直接 python 即可。工具 docstring 与文档示例中简洁的 python 写法按此约定理解；
不 import 项目内模块（src.* 一律不用）——工具可整体拷走独立运行；
staging 表名 = MySQL 源表名，字段名 = 源字段名，全部由 SHOW FULL COLUMNS 动态生成，代码中不得硬编码业务字段清单（REQUIRED_COLUMNS 仅作环境自检下限，允许源表多列）；
配置复用现有 .env 键：DB_HOST/DB_PORT/DB_DATABASE/DB_USER/DB_PASSWORD（默认值同 src/config.py：10.0.46.53:3306/xuangubao/read_user）；
退出码对齐现有 sync_data.py：0 成功，2 环境检查失败（未开始同步），3 数据校验失败；
面向用户的输出与错误信息用中文；
测试不引入 pytest：工具自带 --selftest（断言列表逐步扩展），run_tests.py 末尾以 subprocess 方式接入（先例：wk:push dry-run 节）；
commit 信息 conventional 风格（feat(sync): ...），过项目 githooks 校验。

文件清单

文件	职责
`scripts/sync/sync_staging.py`	全部逻辑：配置、类型映射、动态建表、增量/整表同步、进度与日志、schema.json 导出、--selftest
`docs/data-sync/staging-schema.md`	人类读的结构说明（表关系、逐字段释义、tag 字典、陷阱清单、查询示例）
`data/staging_schema.json`	agent 机读结构说明（工具每次同步自动导出，不进 git）
`scripts/ops/run_tests.py`	末尾新增一节，subprocess 调 `--selftest`
`.gitignore`	增 `data/staging.db`、`data/staging.db-*`、`data/staging_schema.json`

Task 1: 骨架 + 类型映射 + 动态建表

Files: - Create: scripts/sync/sync_staging.py - 无独立测试文件（selftest 即测试，本 task 建立雏形）

Interfaces: - Produces（后续 task 依赖的确切签名）: - MYSQL_TYPE_TO_SQLITE(mysql_type: str) -> str - fetch_table_schema(cursor, table: str) -> list[tuple[str, str, str]]（返回 (字段名, MySQL类型, Key标志) 列表，列序即源表列序） - build_ddl(table: str, schema: list[tuple[str, str, str]]) -> str - ensure_tables(my_conn, staging: sqlite3.Connection) -> dict[str, list[tuple]]（返回 {表名: schema}；表缺失则建，已存在但列集不一致则抛 SchemaDriftError） - TYPE_MAP: dict[str, str] - 常量 INCREMENTAL_TABLES = ("live_tuoshui_news", "tuoshui_msgs")、FULL_REFRESH_TABLES = ("live_tuoshui_tag_related", "tuoshui_tag_related", "plate_rank_infos")

[ ] Step 1: 写工具骨架与失败的自检

创建 scripts/sync/sync_staging.py，内容为完整骨架：docstring、常量、类型映射、三个 schema 函数、SchemaDriftError、--selftest 入口（本步断言只建表相关的 4 条，函数体可先 raise NotImplementedError）：

"""MySQL 原表 → staging.db 镜像同步工具（最小独立版）。

把内网 MySQL 的 5 张源表原样搬运到本地 SQLite（data/staging.db）：
- 表名、字段名与源表完全一致（运行时 SHOW FULL COLUMNS 动态建表）
- 大表（live_tuoshui_news / tuoshui_msgs）按主键 id 记同步进度增量拉取，
  并把源端 is_deleted 标记同步过来
- 小表（两张 tag 关联表 + plate_rank_infos）每轮整表刷新
- 每次同步结束导出 data/staging_schema.json（agent 机读结构说明）

用法：
    python scripts/sync/sync_staging.py            # 增量同步（默认）
    conda run -n xgb-wiki python scripts/sync/sync_staging.py --full     # 全量重建
    conda run -n xgb-wiki python scripts/sync/sync_staging.py --check    # 仅环境检查，不动数据
    conda run -n xgb-wiki python scripts/sync/sync_staging.py --selftest # 离线自检（不连网）

退出码：0 成功；2 环境检查失败；3 数据校验失败。
结构说明（人类读）：docs/data-sync/staging-schema.md；agent 读：data/staging_schema.json。
单文件无项目内依赖，可整体拷走独立使用。
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sqlite3
import sys
import tempfile
import time
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path

STAGING_PATH = Path(__file__).resolve().parents[2] / "data" / "staging.db"
SCHEMA_JSON_PATH = STAGING_PATH.parent / "staging_schema.json"

# 增量组：大表，按主键 id 记同步进度
INCREMENTAL_TABLES = ("live_tuoshui_news", "tuoshui_msgs")
# 整表组：小表，每轮全量刷新（涨到十万行级再改增量，见 plan 注记）
FULL_REFRESH_TABLES = ("live_tuoshui_tag_related", "tuoshui_tag_related", "plate_rank_infos")
ALL_TABLES = INCREMENTAL_TABLES + FULL_REFRESH_TABLES

# 环境自检下限（源自 src/pipeline/sync.py REQUIRED_MYSQL_COLUMNS，独立工具复制一份；
# 仅用于 --check 报错提示，镜像本身是 SHOW COLUMNS 全字段，源表多列照搬）
REQUIRED_COLUMNS = {
    "live_tuoshui_news": ["id", "msg_id", "title", "summary", "content", "stocks",
                          "plates", "msg_display_time", "is_deleted"],
    "live_tuoshui_tag_related": ["live_msg_id", "tag_id"],
    "tuoshui_msgs": ["id", "msg_id", "title", "summary", "content", "related_stocks",
                     "plates", "msg_created_at", "is_premium"],
    "tuoshui_tag_related": ["msg_id", "tag_id"],
    "plate_rank_infos": ["plate_id", "plate_name", "classification"],
}

TYPE_MAP = {
    "bigint": "INTEGER", "int": "INTEGER", "integer": "INTEGER",
    "smallint": "INTEGER", "tinyint": "INTEGER", "mediumint": "INTEGER",
    "decimal": "REAL", "float": "REAL", "double": "REAL",
    # 其余（varchar/char/text/longtext/datetime/date/timestamp/json…）一律 TEXT
}


class SchemaDriftError(Exception):
    """staging 已有表与源表列集不一致——需要 --full 重建。"""


def log(msg: str) -> None:
    print(f"[staging {datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)


def mysql_type_to_sqlite(mysql_type: str) -> str:
    base = re.sub(r"\(.*\)", "", mysql_type.strip().lower()).replace(" unsigned", "")
    return TYPE_MAP.get(base, "TEXT")

同文件继续追加 schema 三函数与 selftest 雏形（Step 3 实现函数体，此处先放签名与 raise NotImplementedError）：

def fetch_table_schema(cursor, table: str) -> list[tuple[str, str, str]]:
    """SHOW FULL COLUMNS → [(字段名, MySQL类型, Key标志)]，列序即源表列序。"""
    raise NotImplementedError


def build_ddl(table: str, schema: list[tuple[str, str, str]]) -> str:
    """按 schema 生成 CREATE TABLE；Key='PRI' 的列组成主键（无则不设）。"""
    raise NotImplementedError


def ensure_tables(my_conn, staging: sqlite3.Connection) -> dict:
    """5 张镜像表 + staging_meta + staging_log 全部就绪。

    表缺失→建；已存在但列集不一致→SchemaDriftError（提示 --full）。
    返回 {表名: schema}。
    """
    raise NotImplementedError

以及 selftest 骨架（断言先行的 4 条本 task 相关 + 后续 task 的占位调用点）：

def _selftest() -> int:
    results: list[tuple[str, bool, str]] = []

    def check(name: str, cond: bool, detail: str = ""):
        results.append((name, bool(cond), detail))

    tmpdir = tempfile.mkdtemp(prefix="staging_selftest_")
    db_path = Path(tmpdir) / "staging.db"

    fake = FakeMySQL()
    staging = sqlite3.connect(db_path)

    schemas = ensure_tables(fake, staging)
    check("建表后 5 张镜像表齐全", all(t in schemas for t in ALL_TABLES))
    ddl_checked = build_ddl("live_tuoshui_news", schemas["live_tuoshui_news"])
    check("DDL 含源字段名与表级主键",
          ', PRIMARY KEY ("id")' in ddl_checked and '"msg_display_time" TEXT' in ddl_checked)
    check("类型映射 datetime→TEXT", mysql_type_to_sqlite("datetime") == "TEXT")
    check("类型映射 bigint unsigned→INTEGER", mysql_type_to_sqlite("bigint unsigned") == "INTEGER")

    n_fail = sum(1 for _, ok, _ in results if not ok)
    for name, ok, detail in results:
        mark = "✓" if ok else "✗"
        print(f"  {mark} {name}" + (f"  [{detail}]" if detail and not ok else ""))
    print(f"===== selftest: {len(results) - n_fail} 通过 / {n_fail} 失败 =====")
    return 1 if n_fail else 0

main() 与 argparse 本步一并给出（--full/--check/--selftest 分派，非 selftest 分支本 task 先打印"Task 2+ 实现"并返回 0）。

[ ] Step 2: 跑 selftest 确认失败

cd /e/Quant_projs/4.\ xuangubao/xgb_wiki && conda run -n xgb-wiki python scripts/sync/sync_staging.py --selftest

预期：NameError/NotImplementedError（FakeMySQL 未定义 / 函数未实现），退出码非 0。

[ ] Step 3: 实现 FakeMySQL 与三个函数

同文件追加（FakeMySQL 为 --selftest 专用模拟层，覆盖 5 表小样本数据与 SHOW FULL COLUMNS；本 task 只需 schema 与建表路径，数据读取在 Task 2/3 扩展断言时已可用）：

class _FakeCursor:
    def __init__(self, owner):
        self._o = owner

    def execute(self, sql, params=None):
        self._rows, self._pos = self._o.run(" ".join(sql.split()), params or []), 0

    def fetchmany(self, n):
        out, self._pos = self._rows[self._pos:self._pos + n], self._pos + n
        return out

    def fetchall(self):
        out, self._pos = self._rows[self._pos:], len(self._rows)
        return out

    def __enter__(self): return self
    def __exit__(self, *a): return False


class FakeMySQL:
    """selftest 专用模拟 MySQL：样本数据 5 表齐全，列序与 _SCHEMAS 一致。"""

    _SCHEMAS = {
        "live_tuoshui_news": [
            ("id", "bigint", "PRI"), ("msg_id", "bigint", ""), ("title", "varchar(255)", ""),
            ("summary", "text", ""), ("content", "longtext", ""), ("stocks", "varchar(512)", ""),
            ("plates", "varchar(512)", ""), ("msg_display_time", "datetime", ""),
            ("is_deleted", "tinyint", "")],
        "tuoshui_msgs": [
            ("id", "bigint", "PRI"), ("msg_id", "bigint", ""), ("title", "varchar(255)", ""),
            ("summary", "text", ""), ("content", "longtext", ""), ("related_stocks", "varchar(512)", ""),
            ("plates", "varchar(512)", ""), ("msg_created_at", "datetime", ""),
            ("is_premium", "tinyint", "")],
        "live_tuoshui_tag_related": [("live_msg_id", "bigint", "PRI"), ("tag_id", "int", "PRI")],
        "tuoshui_tag_related": [("msg_id", "bigint", "PRI"), ("tag_id", "int", "PRI")],
        "plate_rank_infos": [("plate_id", "int", "PRI"), ("plate_name", "varchar(128)", ""),
                             ("classification", "varchar(64)", "")],
    }

    def __init__(self):
        self._deleted_live_ids: set[int] = set()
        # 样本数据：列序与 _SCHEMAS 严格一致
        self.data = {
            "live_tuoshui_news": [
                (1, 101, "标题一", "摘要一", "<p>正文一</p>", "600001.SH", "人工智能", "2026-09-20 09:30:00", 0),
                (2, 102, "标题二", "摘要二", "<p>正文二</p>", "", "", "2026-09-21 10:00:00", 0),
                (3, 103, "标题三（后删）", "摘要三", "<p>正文三</p>", "600002.SH", "", "2026-09-22 11:00:00", 1),
            ],
            "tuoshui_msgs": [
                (1, 201, "研报一", "摘要", "<p>内容</p>", "000001.SZ", "白酒", "2026-09-20 18:00:00", 1),
                (2, 202, "研报二", "摘要", "<p>内容</p>", "", "", "2026-09-21 18:00:00", 0),
            ],
            "live_tuoshui_tag_related": [(101, 11), (101, 12), (102, 13)],
            "tuoshui_tag_related": [(201, 21), (202, 22)],
            "plate_rank_infos": [(11, "人工智能", "科技"), (21, "白酒", "消费")],
        }

    def cursor(self): return _FakeCursor(self)

    def close(self): pass

    def run(self, sql, params):  # Task 2/3 扩展此分派
        if sql.startswith("SHOW FULL COLUMNS FROM"):
            table = sql.split()[-1].strip("`")
            return [r for r in self._SCHEMAS[table]]
        if sql == "SELECT 1":
            return [(1,)]
        raise KeyError(f"FakeMySQL 未实现 SQL: {sql[:80]}")

注意：真实 SHOW FULL COLUMNS 返回多列，fetch_table_schema 只取第 1（Field）、2（Type）、5（Key）列；Fake 返回 3 元组即可（真实连接在 fetch_table_schema 内做 (r[0], r[1], r[4]) 取列——为兼容两者，Fake 的 run 在 SHOW 分支直接返回 5 元组 (Field, Type, "", Key, "") 形式更贴近真实；实现时以真实列位为准，Fake 构造对应占位）。

实现三个函数：

def fetch_table_schema(cursor, table: str) -> list[tuple[str, str, str]]:
    cursor.execute(f"SHOW FULL COLUMNS FROM `{table}`")
    return [(r[0], r[1], r[4]) for r in cursor.fetchall()]


def build_ddl(table: str, schema: list[tuple[str, str, str]]) -> str:
    cols = ", ".join(f'"{name}" {mysql_type_to_sqlite(t)}' for name, t, _ in schema)
    pk = ", ".join(f'"{name}"' for name, _, k in schema if k == "PRI")
    if pk:
        cols += f", PRIMARY KEY ({pk})"
    return f'CREATE TABLE IF NOT EXISTS "{table}" ({cols})'


def ensure_tables(my_conn, staging: sqlite3.Connection) -> dict:
    schemas = {}
    with my_conn.cursor() as cur:
        for table in ALL_TABLES:
            schemas[table] = fetch_table_schema(cur, table)
            staging.execute(build_ddl(table, schemas[table]))
            have = {r[1] for r in staging.execute(f'PRAGMA table_info("{table}")').fetchall()}
            want = {name for name, _, _ in schemas[table]}
            if have != want:
                raise SchemaDriftError(
                    f"staging 表 {table} 列集与源表不一致（源多: {sorted(want-have)}，"
                    f"本地多: {sorted(have-want)}）。请运行 --full 重建。")
    staging.execute("""CREATE TABLE IF NOT EXISTS staging_meta(
        key TEXT PRIMARY KEY, value TEXT)""")
    staging.execute("""CREATE TABLE IF NOT EXISTS staging_log(
        id INTEGER PRIMARY KEY AUTOINCREMENT, started_at TEXT, finished_at TEXT,
        mode TEXT, status TEXT, duration_s REAL, counts TEXT, errors TEXT)""")
    staging.commit()
    return schemas

[ ] Step 4: 跑 selftest 确认通过

conda run -n xgb-wiki conda run -n xgb-wiki python scripts/sync/sync_staging.py --selftest

预期：4 条断言全部 ✓，退出码 0。

[ ] Step 5: Commit

git add scripts/sync/sync_staging.py
git commit -m "feat(sync): staging 工具骨架——动态建表与类型映射（表名字段与源一致）"

Task 2: 整表组刷新 + 进度/日志表

Files: - Modify: scripts/sync/sync_staging.py

Interfaces: - Consumes: ensure_tables、FakeMySQL（Task 1） - Produces: - adapt_row(row: tuple) -> tuple（datetime→ISO 字符串、date→ISO、Decimal→str、bytes→str，其余原样） - refresh_full_table(my_conn, staging, table, batch=2000) -> int（清空重建该表，返回行数） - write_log(staging, started_at, mode, status, duration_s, counts, errors) -> None - meta_get(staging, key, default=None) -> str | None / meta_set(staging, key, value) -> None

[ ] Step 1: selftest 先加 4 条断言（跑应失败）

在 _selftest() 建表断言之后追加：

    n_plates = refresh_full_table(fake, staging, "plate_rank_infos")
    n_ltr = refresh_full_table(fake, staging, "live_tuoshui_tag_related")
    check("整表刷新 plate_rank_infos 2 行", n_plates == 2)
    check("整表刷新 tag 关联表 3 行", n_ltr == 3)
    check("staging 行数与源一致",
          staging.execute("SELECT COUNT(*) FROM plate_rank_infos").fetchone()[0] == 2)
    write_log(staging, "2026-09-29T10:00:00", "full", "success", 0.5,
              {"plate_rank_infos": 2}, "")
    check("staging_log 记录 1 条",
          staging.execute("SELECT COUNT(*) FROM staging_log").fetchone()[0] == 1)

跑 --selftest：预期 NameError: refresh_full_table。

[ ] Step 2: 实现三个函数

def adapt_row(row: tuple) -> tuple:
    return tuple(
        v.isoformat(sep=" ") if isinstance(v, (datetime, date))
        else str(v) if isinstance(v, Decimal)
        else v.decode("utf-8", "replace") if isinstance(v, bytes)
        else v
        for v in row)


def refresh_full_table(my_conn, staging: sqlite3.Connection, table: str, batch: int = 2000) -> int:
    staging.execute(f'DELETE FROM "{table}"')
    total = 0
    with my_conn.cursor() as cur:
        cur.execute(f'SELECT * FROM `{table}`')
        while rows := cur.fetchmany(batch):
            staging.executemany(
                f'INSERT OR REPLACE INTO "{table}" VALUES ({",".join("?" * len(rows[0]))})',
                [adapt_row(r) for r in rows])
            total += len(rows)
    staging.commit()
    return total


def write_log(staging, started_at, mode, status, duration_s, counts, errors):
    staging.execute(
        "INSERT INTO staging_log(started_at, finished_at, mode, status, duration_s, counts, errors)"
        " VALUES (?,?,?,?,?,?,?)",
        (started_at, datetime.now().isoformat(sep=" ", timespec="seconds"), mode,
         status, duration_s, json.dumps(counts, ensure_ascii=False), errors or ""))
    staging.commit()


def meta_get(staging, key, default=None):
    row = staging.execute("SELECT value FROM staging_meta WHERE key=?", (key,)).fetchone()
    return default if row is None else row[0]


def meta_set(staging, key, value):
    staging.execute("INSERT OR REPLACE INTO staging_meta(key, value) VALUES (?,?)",
                    (key, str(value)))
    staging.commit()

同步给 FakeMySQL.run 补整表读取分派（在 SHOW 分支后）：

        for t in ("live_tuoshui_tag_related", "tuoshui_tag_related", "plate_rank_infos"):
            if sql == f"SELECT * FROM `{t}`":
                return list(self.data[t])

[ ] Step 3: 跑 selftest 通过后 commit

conda run -n xgb-wiki conda run -n xgb-wiki python scripts/sync/sync_staging.py --selftest
git add scripts/sync/sync_staging.py
git commit -m "feat(sync): staging 整表组刷新与进度/日志表"

Task 3: 增量组同步（进度推进 + 删除标记）

Files: - Modify: scripts/sync/sync_staging.py

Interfaces: - Consumes: adapt_row、meta_get/meta_set（Task 2） - Produces: - sync_incremental_table(my_conn, staging, table, deleted_id_sql=None, batch=500) -> dict 返回 {"new": 本轮新增行数, "deleted": 本轮打删除标记行数, "wm_before": int, "wm_after": int}； deleted_id_sql 仅 live_tuoshui_news 传入（SELECT id FROM \live_tuoshui_news` WHERE is_deleted=1`），其余表传 None。

[ ] Step 1: selftest 加增量场景断言（跑应失败）

    # 增量：首轮全量 3 行，进度标记=3；删除标记同步 1 行
    r1 = sync_incremental_table(fake, staging, "live_tuoshui_news",
                                deleted_id_sql="SELECT id FROM `live_tuoshui_news` WHERE is_deleted=1")
    check("增量首轮拉全量 3 行", r1["new"] == 3 and r1["wm_after"] == 3)
    check("删除标记同步 1 行", r1["deleted"] == 1 and staging.execute(
        "SELECT is_deleted FROM live_tuoshui_news WHERE id=3").fetchone()[0] == 1)
    # 增量：无新数据时幂等（0 新增，进度标记不变）
    r2 = sync_incremental_table(fake, staging, "live_tuoshui_news")
    check("无新数据幂等 0 新增", r2["new"] == 0 and r2["wm_after"] == 3)
    # 增量：源端新增 2 行后只拉 2 行
    fake.data["live_tuoshui_news"] += [
        (4, 104, "标题四", "s", "c", "", "", "2026-09-23 09:00:00", 0),
        (5, 105, "标题五", "s", "c", "", "", "2026-09-24 09:00:00", 0)]
    r3 = sync_incremental_table(fake, staging, "live_tuoshui_news")
    check("源端新增后只拉 2 行且进度标记推进", r3["new"] == 2 and r3["wm_after"] == 5)

预期失败：NameError: sync_incremental_table。

[ ] Step 2: 实现

def sync_incremental_table(my_conn, staging: sqlite3.Connection, table: str,
                           deleted_id_sql: str | None = None, batch: int = 500) -> dict:
    wm_before = int(meta_get(staging, f"wm_{table}", "0"))
    ncols = len(staging.execute(f'PRAGMA table_info("{table}")').fetchall())
    new, wm_after = 0, wm_before
    with my_conn.cursor() as cur:
        cur.execute(f"SELECT * FROM `{table}` WHERE id > %s ORDER BY id", (wm_before,))
        while rows := cur.fetchmany(batch):
            staging.executemany(
                f'INSERT OR REPLACE INTO "{table}" VALUES ({",".join("?" * ncols)})',
                [adapt_row(r) for r in rows])
            new += len(rows)
            wm_after = max(wm_after, rows[-1][0])
    deleted = 0
    if deleted_id_sql:
        with my_conn.cursor() as cur:
            cur.execute(deleted_id_sql)
            ids = [r[0] for r in cur.fetchall()]
        for i in range(0, len(ids), 500):
            chunk = ids[i:i + 500]
            q = ",".join("?" * len(chunk))
            deleted += staging.execute(
                f'UPDATE "{table}" SET is_deleted=1 WHERE id IN ({q}) AND is_deleted=0',
                chunk).rowcount
    meta_set(staging, f"wm_{table}", wm_after)
    staging.commit()
    return {"new": new, "deleted": deleted, "wm_before": wm_before, "wm_after": wm_after}

FakeMySQL 增量分派（run 内，SHOW/整表分支之后）：

        import re as _re
        m = _re.match(r"SELECT \* FROM `(\w+)` WHERE id > (\d+) ORDER BY id", sql)
        if m:
            t, wm = m.group(1), int(m.group(2))
            return [r for r in self.data[t] if r[0] > wm]
        if sql == "SELECT id FROM `live_tuoshui_news` WHERE is_deleted=1":
            return [(i,) for i, row in enumerate(self.data["live_tuoshui_news"])
                    if row[8] == 1]  # 第 9 列 is_deleted

[ ] Step 3: selftest 通过后 commit

conda run -n xgb-wiki conda run -n xgb-wiki python scripts/sync/sync_staging.py --selftest
git add scripts/sync/sync_staging.py
git commit -m "feat(sync): staging 增量组——主键进度推进与删除标记同步"

Task 4: 编排（--full / 默认增量）、--check、schema.json 导出

Files: - Modify: scripts/sync/sync_staging.py

Interfaces: - Consumes: 前三个 task 全部产物 - Produces: - connect_mysql() -> pymysql.Connection（读 .env / 默认值，connect_timeout=30） - check_environment(my_conn) -> list[str]（返回错误清单：必需列缺失、SHOW 权限；空清单=通过） - export_schema_json(staging, path=SCHEMA_JSON_PATH) -> dict - 常量 FIELD_GLOSSARY: dict[str, dict[str, str]]（表→字段→一句释义，来源 docs/core/data-model.md §5 实测口径，独立工具复制一份；缺失字段导出 desc: ""，不阻塞同步） - 常量 TAG_DICTIONARIES: dict（live/digest 两套栏目 tag_id→名称字典，来源 src/config.py） - run_sync(mode: str) -> int（编排：连接→检查→ensure_tables→按 mode 分派增量/整表/全量→导出 json→写日志；异常按退出码约定返回 2/3）

[ ] Step 1: selftest 加导出断言（跑应失败）

    info = export_schema_json(staging, db_path.parent / "staging_schema.json")
    check("schema.json 含 5 表与行数",
          info["tables"]["plate_rank_infos"]["rows"] == 2 and len(info["tables"]) == 5)
    check("schema.json 含同步进度",
          info["progress"]["wm_live_tuoshui_news"] == "5")
    check("schema.json 落盘且可解析",
          json.loads((db_path.parent / "staging_schema.json").read_text("utf-8"))["generated_at"])
    check("schema.json 含字段释义",
          info["tables"]["live_tuoshui_news"]["columns"][0]["desc"])
    check("schema.json 含栏目 tag 字典",
          info["tag_dictionaries"]["digest"]["4"] == "每日强股")

预期失败：NameError: export_schema_json。

[ ] Step 2: 实现 FIELD_GLOSSARY / TAG_DICTIONARIES 与编排层

字段释义与栏目字典（独立工具内置一份，来源注明，不 import src）：

# 字段释义：来源 docs/core/data-model.md §5（2026-09 实测口径）。源表演进新增的字段
# 若未登记于此，导出时 desc 为空串——同步不受影响，发现后补一行即可。
FIELD_GLOSSARY = {
    "live_tuoshui_news": {
        "id": "主键；增量同步进度基准（staging_meta.wm_live_tuoshui_news）",
        "msg_id": "消息 ID——标签表经它关联（live_tuoshui_tag_related.live_msg_id），注意不是主键 id",
        "title": "研报标题；live 侧标题常带'券商名：'前缀，指被转述方",
        "summary": "摘要",
        "content": "正文 HTML",
        "stocks": "编辑打标个股，分隔符字符串 '代码,名称;…'（源库无外键，需解析；打标率约 20%）",
        "plates": "题材，分隔符字符串 '板块ID,名称;…'（板块 ID 逻辑关联 plate_rank_infos.plate_id）",
        "msg_display_time": "展示时间（业务时间字段）",
        "is_deleted": "1=源端已删除；业务查询必须过滤 is_deleted=0",
    },
    "tuoshui_msgs": {
        "id": "主键；增量同步进度基准（staging_meta.wm_tuoshui_msgs）",
        "msg_id": "消息 ID——tuoshui_tag_related 经它关联，不是主键 id",
        "title": "研报标题",
        "summary": "摘要",
        "content": "正文 HTML；tag_id=4（每日强股）正文含 'N、个股：X 大涨题材：Y' 结构段",
        "related_stocks": "编辑打标个股（此表列名是 related_stocks，不是 stocks），格式同 live 侧 stocks",
        "plates": "题材，格式同 live 侧",
        "msg_created_at": "创建时间（业务时间字段）",
        "is_premium": "APP 端付费标记 0/1",
    },
    "live_tuoshui_tag_related": {
        "live_msg_id": "= live_tuoshui_news.msg_id（不是主键 id）",
        "tag_id": "栏目 ID，释义见 tag_dictionaries.live；缺省/未知归'未分类'",
    },
    "tuoshui_tag_related": {
        "msg_id": "= tuoshui_msgs.msg_id",
        "tag_id": "栏目 ID，释义见 tag_dictionaries.digest",
    },
    "plate_rank_infos": {
        "plate_id": "板块/题材 ID（主数据主键，即本体库 theme.plate_id）",
        "plate_name": "题材名称",
        "classification": "分类：概念 / 行业 / 风格",
    },
}

# 栏目 tag 字典：来源 src/config.py LIVE_TAG_COLUMNS / DIGEST_TAG_COLUMNS
TAG_DICTIONARIES = {
    "live": {"43": "热点", "44": "深度", "42": "消费", "41": "医药", "36": "行业精选",
             "1": "精选", "11": "业绩前瞻", "46": "政策", "40": "未分类"},
    "digest": {"1": "脱水研报", "2": "评级慧选", "3": "直击一线", "4": "每日强股",
               "7": "每日谈", "99": "周回顾"},
}

export_schema_json 实现如下（字段合并释义、顶层带栏目字典）：

def export_schema_json(staging: sqlite3.Connection, path=SCHEMA_JSON_PATH) -> dict:
    tables = {}
    for (table,) in staging.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'staging_%'"):
        glossary = FIELD_GLOSSARY.get(table, {})
        cols = [{"name": r[1], "type": r[2], "desc": glossary.get(r[1], "")} for r in
                staging.execute(f'PRAGMA table_info("{table}")')]
        tables[table] = {"columns": cols,
                         "rows": staging.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]}
    progress = {k: v for k, v in staging.execute(
        "SELECT key, value FROM staging_meta WHERE key LIKE 'wm_%'")}
    info = {"generated_at": datetime.now().isoformat(sep=" ", timespec="seconds"),
            "description": "MySQL 原表镜像。表名/字段与源一致；结构说明见 docs/data-sync/staging-schema.md",
            "tables": tables, "progress": progress, "tag_dictionaries": TAG_DICTIONARIES}
    path.write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding="utf-8")
    return info

def load_db_config() -> dict:
    try:
        from dotenv import load_dotenv
        load_dotenv(Path(__file__).resolve().parents[2] / ".env")
    except ImportError:
        pass
    return {"host": os.getenv("DB_HOST", "10.0.46.53"),
            "port": int(os.getenv("DB_PORT", "3306")),
            "database": os.getenv("DB_DATABASE", "xuangubao"),
            "user": os.getenv("DB_USER", "read_user"),
            "password": os.getenv("DB_PASSWORD", ""),
            "charset": "utf8mb4", "connect_timeout": 30}


def connect_mysql():
    import pymysql
    return pymysql.connect(**load_db_config())


def check_environment(my_conn) -> list[str]:
    errors = []
    try:
        with my_conn.cursor() as cur:
            have: dict[str, set] = {}
            for table in ALL_TABLES:
                cur.execute(f"SHOW FULL COLUMNS FROM `{table}`")
                have[table] = {r[0] for r in cur.fetchall()}
            for table, need in REQUIRED_COLUMNS.items():
                missing = [c for c in need if c not in have.get(table, set())]
                if missing:
                    errors.append(f"源表 {table} 缺列：{', '.join(missing)}")
    except Exception as e:
        errors.append(f"源表结构检查失败：{e}")
    return errors


def export_schema_json(staging: sqlite3.Connection, path=SCHEMA_JSON_PATH) -> dict:
    tables = {}
    for (table,) in staging.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'staging_%'"):
        cols = [{"name": r[1], "type": r[2]} for r in
                staging.execute(f'PRAGMA table_info("{table}")')]
        tables[table] = {"columns": cols,
                         "rows": staging.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]}
    progress = {k: v for k, v in staging.execute(
        "SELECT key, value FROM staging_meta WHERE key LIKE 'wm_%'")}
    info = {"generated_at": datetime.now().isoformat(sep=" ", timespec="seconds"),
            "description": "MySQL 原表镜像。表名/字段与源一致；结构说明见 docs/data-sync/staging-schema.md",
            "tables": tables, "progress": progress}
    path.write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding="utf-8")
    return info


def run_sync(mode: str) -> int:
    started = time.time()
    started_at = datetime.now().isoformat(sep=" ", timespec="seconds")
    try:
        my = connect_mysql()
    except Exception as e:
        log(f"✗ MySQL 连接失败：{e}")
        return 2
    errors = check_environment(my)
    if errors:
        for e in errors:
            log(f"✗ {e}")
        my.close()
        return 2
    staging = sqlite3.connect(STAGING_PATH)
    counts: dict[str, dict] = {}
    status, err = "success", ""
    try:
        if mode == "full":
            staging.close()
            STAGING_PATH.unlink(missing_ok=True)
            staging = sqlite3.connect(STAGING_PATH)
        ensure_tables(my, staging)
        for t in INCREMENTAL_TABLES:
            del_sql = ("SELECT id FROM `live_tuoshui_news` WHERE is_deleted=1"
                       if t == "live_tuoshui_news" else None)
            counts[t] = sync_incremental_table(my, staging, t, deleted_id_sql=del_sql)
        for t in FULL_REFRESH_TABLES:
            counts[t] = {"rows": refresh_full_table(my, staging, t)}
        export_schema_json(staging)
    except SchemaDriftError as e:
        status, err = "failed", str(e)
        log(f"✗ {e}")
        return_code = 3
    except Exception as e:
        status, err = "failed", f"{type(e).__name__}: {e}"
        log(f"✗ 同步失败：{err}")
        return_code = 3
    else:
        return_code = 0
    finally:
        if status == "success" or err:
            write_log(staging, started_at, mode, status, round(time.time() - started, 1),
                      counts, err)
        staging.close()
        my.close()
    return return_code

实现注意：run_sync 须先声明 staging 为可重绑定变量；full 分支在删除 db 文件前必须先 staging.close()（Windows 下活动连接会锁文件）。

main() 补全分派：--check 只跑连接+check_environment（打印结果返回 0/2）；--selftest 调 _selftest()；默认与 --full 调 run_sync("incremental"|"full")。

[ ] Step 3: selftest 通过（真实同步留待 VPN 在线时人工验证）后 commit

conda run -n xgb-wiki conda run -n xgb-wiki python scripts/sync/sync_staging.py --selftest
git add scripts/sync/sync_staging.py
git commit -m "feat(sync): staging 编排——--full/--check 与 staging_schema.json 导出"

Task 5: run_tests.py 接入 + 结构说明文档 + .gitignore

Files: - Modify: scripts/ops/run_tests.py（末尾新增一节，db.close() 之前） - Create: docs/data-sync/staging-schema.md - Modify: .gitignore

Interfaces: - Consumes: sync_staging.py --selftest（subprocess 退出码 0）

[ ] Step 1: run_tests.py 加一节（先跑应失败）

在 db.close() 之前（wk: 节之后）插入，风格对齐现有 wk:push dry-run 先例：

    print("== N. staging 同步工具自检 ==")
    import subprocess as _sp
    r = _sp.run([sys.executable, "scripts/sync/sync_staging.py", "--selftest"],
                capture_output=True, text=True,
                cwd=str(Path(__file__).resolve().parents[2]))
    check("staging:--selftest 退出码0", r.returncode == 0, r.stderr[-200:])
    check("staging:--selftest 全部通过", "失败" not in r.stdout and "selftest:" in r.stdout)

节编号 N 取当前文件实际节数 +1；先跑 conda run -n xgb-wiki python scripts/ops/run_tests.py 确认该节两条 FAIL（工具尚未在 CI 顺序中跑通或路径问题）。

[ ] Step 2: 写 docs/data-sync/staging-schema.md（表结构及字段释义）

# staging.db 结构说明（MySQL 原表镜像·含字段释义）

> 本文档给人类读者；agent 请读 `data/staging_schema.json`（每次同步自动刷新，含字段释义、
> 实时行数、同步进度与栏目 tag 字典）。
> 生成工具：`scripts/sync/sync_staging.py`（单文件，可独立拷走）。用法见其 docstring。
> 字段释义与源库结构以 docs/core/data-model.md §5 的实测口径为准（2026-09 复核）。

## 它是什么

内网 MySQL `xuangubao` 库 5 张源表的本地 SQLite 镜像，供加工层（ETL）读取，
也是"数据同步"与"数据加工"两段解耦的中间层：镜像只做原样搬运，不含任何解析。
真丢了不要紧——`--full` 一条命令从源库全量重建。

## 表关系速览

    live_tuoshui_news.msg_id ──┬── live_tuoshui_tag_related.live_msg_id   （取栏目）
    tuoshui_msgs.msg_id ───────┬── tuoshui_tag_related.msg_id             （取栏目）
    *.plates 内"板块ID" ───────┴── plate_rank_infos.plate_id              （字符串内 ID 逻辑关联，非外键）

## 表结构与字段释义

### live_tuoshui_news — 盘中快讯（live 研报）

单篇完整版、标题即观点，盘中滚动、每天多篇。

| 字段 | 类型 | 释义 |
|---|---|---|
| id | INTEGER PK | 主键；增量同步进度基准 |
| msg_id | INTEGER | 消息 ID，**标签表经它关联**（不是主键 id） |
| title | TEXT | 研报标题；常带"券商名："前缀（被转述方） |
| summary | TEXT | 摘要 |
| content | TEXT | 正文 HTML |
| stocks | TEXT | 编辑打标个股：`"代码,名称;…"` 分隔符字符串（无外键，需解析；打标率约 20%） |
| plates | TEXT | 题材：`"板块ID,名称;…"`；板块 ID 对应 plate_rank_infos.plate_id |
| msg_display_time | TEXT | 展示时间（业务时间字段） |
| is_deleted | INTEGER | 1=源端已删除；**业务查询必须过滤 is_deleted=0** |

### tuoshui_msgs — 脱水研报（digest 每日汇总）

按栏目每日汇总，每栏目约每日 1 篇。

| 字段 | 类型 | 释义 |
|---|---|---|
| id | INTEGER PK | 主键；增量同步进度基准 |
| msg_id | INTEGER | 消息 ID，**标签表经它关联**（不是主键 id） |
| title | TEXT | 研报标题 |
| summary | TEXT | 摘要 |
| content | TEXT | 正文 HTML；tag_id=4（每日强股）正文含"N、个股：X 大涨题材：Y"结构段 |
| related_stocks | TEXT | 编辑打标个股（**此表列名是 related_stocks**，格式同上） |
| plates | TEXT | 题材，格式同 live 侧 |
| msg_created_at | TEXT | 创建时间（业务时间字段） |
| is_premium | INTEGER | APP 端付费标记 0/1 |

### live_tuoshui_tag_related — 快讯×栏目关联

| 字段 | 类型 | 释义 |
|---|---|---|
| live_msg_id | INTEGER PK | = live_tuoshui_news.**msg_id**（不是主键 id） |
| tag_id | INTEGER PK | 栏目 ID，见下方 live 字典；缺省/未知归"未分类" |

### tuoshui_tag_related — 研报×栏目关联

| 字段 | 类型 | 释义 |
|---|---|---|
| msg_id | INTEGER PK | = tuoshui_msgs.msg_id |
| tag_id | INTEGER PK | 栏目 ID，见下方 digest 字典 |

### plate_rank_infos — 板块（题材）主数据

| 字段 | 类型 | 释义 |
|---|---|---|
| plate_id | INTEGER PK | 板块/题材 ID（即本体库 theme.plate_id） |
| plate_name | TEXT | 题材名称 |
| classification | TEXT | 概念 / 行业 / 风格 |

## 栏目 tag 字典

**digest 侧**（tuoshui_tag_related.tag_id）：`1` 脱水研报（主刊）/ `2` 评级慧选 / `3` 直击一线 / `4` 每日强股 / `7` 每日谈 / `99` 周回顾（每周 1 篇）

**live 侧**（live_tuoshui_tag_related.tag_id）：`43` 热点（量最大）/ `44` 深度 / `42` 消费 / `41` 医药 / `36` 行业精选 / `1` 精选 / `11` 业绩前瞻 / `46` 政策 / `40` 未分类

## 新手陷阱（拿库直接开查前必读）

1. 标签关联走 **msg_id**，不是主键 id——最容易写错的 JOIN；
2. `stocks` / `plates` 是分隔符字符串，源库无外键，个股/题材关系需解析字符串；
3. 镜像**含已删除行**（is_deleted=1），业务口径一律加 `is_deleted=0`；
4. 一篇研报可挂多个栏目 tag（关联表多行），别假设一对一。

## 常用查询示例

    -- 看数据同步到哪了
    SELECT * FROM staging_meta;
    -- 最新 10 条快讯
    SELECT id, title, msg_display_time FROM live_tuoshui_news
    WHERE is_deleted=0 ORDER BY id DESC LIMIT 10;
    -- 某日"脱水研报"主刊（tag_id=1）
    SELECT m.id, m.title FROM tuoshui_msgs m
    JOIN tuoshui_tag_related t ON t.msg_id = m.msg_id
    WHERE t.tag_id = 1 AND m.msg_created_at >= '2026-09-24'
      AND m.msg_created_at < '2026-09-25';

## 更新方式

    python scripts/sync/sync_staging.py            # 增量（日常）
    conda run -n xgb-wiki python scripts/sync/sync_staging.py --full     # 全量重建（首次/结构变化/修复）
    conda run -n xgb-wiki python scripts/sync/sync_staging.py --check    # 仅环境检查
    conda run -n xgb-wiki python scripts/sync/sync_staging.py --selftest # 离线自检（不连网）

字段清单不在此罗列全集——源表演进后新增字段以 `staging_schema.json` 为准（自动跟随）。

[ ] Step 3: .gitignore 增项

在 data/xgb_wiki.db-shm 一组附近追加：

# staging 镜像（MySQL 原表同步产物，本地中间层，不进 git）
data/staging.db
data/staging.db-shm
data/staging.db-wal
data/staging_schema.json

[ ] Step 4: 全量验证 + commit

conda run -n xgb-wiki python scripts/ops/run_tests.py          # 新节两条通过，存量全部通过
git add scripts/ops/run_tests.py docs/data-sync/staging-schema.md .gitignore
git commit -m "feat(sync): staging 工具接入测试套件并交付结构说明"

完成后的首次真实验证（需 VPN 在线，人工执行，不在本计划 task 内）

conda run -n xgb-wiki conda run -n xgb-wiki python scripts/sync/sync_staging.py --check     # 环境检查
conda run -n xgb-wiki conda run -n xgb-wiki python scripts/sync/sync_staging.py --full      # 首次全量镜像
conda run -n xgb-wiki python scripts/sync/sync_staging.py             # 再跑一次增量，应 0 新增（幂等）
conda run -n xgb-wiki python -c "import json;print(json.load(open('data/staging_schema.json',encoding='utf-8'))['progress'])"

对照验收标准 §8.1（design v1.3）：五表行数与源一致（is_deleted=0 口径核对大表）、selftest 通过、schema.json 与库内结构一致。

P1c 衔接（分片版本化，本计划完成后另出实施计划）

本工具交付后，P1c（design §5.7）将以子命令形式扩展进 sync_staging.py，P1a 阶段无需预写任何代码，但以下事实已就绪，届时直接使用：

TOS 接入：官方 Python SDK（包名 tos，2026-09-29 已验证）：xgb_wiki/.env 六键就绪——TOS_ACCESS_KEY / TOS_SECRET_KEY / TOS_ENDPOINT（原生端点 tos-cn-shanghai.volces.com，官方 SDK 用原生域名而非 S3 兼容端点）/ TOS_REGION / TOS_BUCKET（arkclaw-tos-2101988067-cn-shanghai）/ TOS_PREFIX（xgb-wiki/staging）。TosClientV2(ak, sk, endpoint, region) 全链路（does_bucket_exist / list_objects_type2 / put_object / get_object / delete_object / does_object_exist）实测通过；
SDK 实施要点：list_objects_type2 返回 ListObjectType2Output 对象，不能直接迭代——清单在 .contents 属性，翻页看 .is_truncated；上传/下载文件有现成的 put_object_from_file / download_file，P1c 分片传输直接用；S3 兼容协议（boto3）留作应急备选，其坑位：S3 兼容端点仅支持 virtual-hosted style（path style 一律 403 / InvalidPathAccess）；
依赖：pyarrow 25.0.1 与 tos SDK（含 boto3 备选）已装入 xgb-wiki conda 环境（分片导出/合并/上传直接可用），无需新增安装；
扩展子命令（design §5.7）：--export-shards（导出新增/变更分片 + manifest 并上传）、--pull（对比远端 manifest 只拉缺失分片）、--assemble（分片 + deleted_ids 合并回 staging.db）；
argparse 预留：本计划 Task 4 的 main() 无需为子命令做任何预埋，P1c 直接追加参数分支即可（保持每期改动最小）。

Self-Review 记录

覆盖 spec §5.1：动态建表 ✓ 增量/整表分组 ✓ 删除标记 ✓ 进度/日志 ✓ schema.json ✓ selftest ✓ 人类文档 ✓ 字段释义（FIELD_GLOSSARY + staging-schema.md 字段表 + 栏目 tag 字典，2026-09-29 应用户要求补充）✓；
占位符扫描：run_tests.py 节编号 N 需实施时取实际值——意图明确的指引而非 TBD；
类型/签名一致性：sync_incremental_table 返回 dict 结构在 Task 3 定义、Task 4 直接消费；adapt_row Task 2 定义、Task 3 消费 ✓；
释义来源可信度：字段含义取自 docs/core/data-model.md §5（2026-09 实测口径，且该节已纠正过上游 Wiki 文档的过时列名），tag 字典取自 src/config.py 实际字典——均为运行中代码/实测佐证，非猜测。