tools: add split_config_secrets for config.json → jsonc + secrets

DEVELOPMENT.md

@@ -267,6 +267,7 @@ PRAGMA synchronous = NORMAL;
 业务参数放在 `config.jsonc`（JSONC：`//` / `/* */` 注释与尾逗号；实现为自研轻量剥离后再 `json.loads`，无额外依赖）。
 敏感凭据放同目录 `secrets.json`（纯 JSON，供 `auth` 原子回写）。二者都必须被 Git 忽略，日志中不得输出敏感值。
 默认路径优先 `config.jsonc`，否则回落 `config.json`；旧单体 `config.json`（凭据内嵌）仍可加载，若存在 `secrets.json` 则凭据以它为准。
+从旧单体迁到拆分布局：`uv run python tools/split_config_secrets.py [--dry-run] [--force] [--keep-src]`（默认写出同目录 `config.jsonc` + `secrets.json`，并把源改名为 `.bak`）。
 
 `config.jsonc` 示例：
 

README.md

@@ -314,6 +314,13 @@ Copy-Item secrets.example.json secrets.json
 默认启动优先读 `config.jsonc`，若不存在则回落 `config.json`（旧单体文件仍可用）。
 `python app.py auth` **只回写** `secrets.json`，不会冲掉 JSONC 注释。
 
+若部署目录仍是旧版单体 `config.json`，可用拆分脚本一次迁出：
+
+```powershell
+uv run python tools/split_config_secrets.py --dry-run
+uv run python tools/split_config_secrets.py  # 写出 config.jsonc + secrets.json，源改名为 config.json.bak
+```
+
 凭据文件双账号格式：
 
 ```json

tests/test_split_config_secrets.py

@@ -0,0 +1,194 @@
+# -*- coding: utf-8 -*-
+"""config.json → jsonc + secrets 拆分脚本测试（离线）:
+uv run python tests/test_split_config_secrets.py"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.stdout.reconfigure(encoding="utf-8")
+
+from reaction_board import jsonc  # noqa: E402
+from reaction_board.config import load_config  # noqa: E402
+from tools.split_config_secrets import (  # noqa: E402
+    split_config_file,
+    split_raw,
+)
+
+
+def _monolith() -> dict:
+    return {
+        "bilibili": {
+            "read": {"mid": 11, "csrf": "read-csrf", "cookie": "SESSDATA=secret-read"},
+            "write": {"mid": 22, "csrf": "write-csrf", "cookie": "SESSDATA=secret-write"},
+        },
+        "telegram": {
+            "enabled": True,
+            "bot_token": "tg-secret-token",
+            "chat_id": 12345,
+        },
+        "web": {
+            "host": "127.0.0.1",
+            "port": 8080,
+            "auth_token": "web-secret-token",
+            "public_base_url": "",
+        },
+        "folders": [{"year": 2026, "media_id": 1, "title": "t"}],
+        "track_fans": [37754047],
+        "up_blacklist_seed": [],
+    }
+
+
+def test_split_raw_extracts_secrets_keeps_business():
+    business, secrets, stats = split_raw(_monolith())
+    assert "bilibili" not in business
+    assert business["telegram"] == {"enabled": True}
+    assert business["web"]["host"] == "127.0.0.1"
+    assert business["web"]["port"] == 8080
+    assert "auth_token" not in business["web"]
+    assert secrets["bilibili"]["read"]["mid"] == 11
+    assert secrets["telegram"] == {"bot_token": "tg-secret-token", "chat_id": 12345}
+    assert secrets["web"] == {"auth_token": "web-secret-token"}
+    assert "bilibili" in stats["moved_paths"]
+    assert "telegram.bot_token" in stats["moved_paths"]
+    assert "web.auth_token" in stats["moved_paths"]
+
+
+def test_split_raw_omits_absent_secret_stubs():
+    raw = {
+        "telegram": {"enabled": False},
+        "web": {"host": "0.0.0.0", "port": 9},
+        "folders": [],
+    }
+    business, secrets, stats = split_raw(raw)
+    assert secrets == {}
+    assert business["telegram"] == {"enabled": False}
+    assert business["web"]["host"] == "0.0.0.0"
+    assert stats["moved_count"] == 0
+
+
+def test_dry_run_writes_nothing():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8")
+        summary = split_config_file(src, dry_run=True)
+        assert summary["dry_run"] is True
+        assert summary["moved_count"] >= 3
+        assert not (root / "config.jsonc").exists()
+        assert not (root / "secrets.json").exists()
+        assert src.exists()
+        # 摘要不得泄露 cookie/token
+        blob = json.dumps(summary, ensure_ascii=False)
+        assert "SESSDATA" not in blob
+        assert "tg-secret-token" not in blob
+        assert "web-secret-token" not in blob
+
+
+def test_refuses_overwrite_without_force():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8")
+        (root / "config.jsonc").write_text("// existing\n{}\n", encoding="utf-8")
+        try:
+            split_config_file(src, keep_src=True)
+        except FileExistsError as exc:
+            assert "config.jsonc" in str(exc)
+            assert "--force" in str(exc)
+        else:
+            raise AssertionError("已有输出时无 --force 必须失败")
+        assert src.exists()
+        assert "existing" in (root / "config.jsonc").read_text(encoding="utf-8")
+
+
+def test_writes_and_renames_src_by_default():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8")
+        summary = split_config_file(src)
+        cfg_path = root / "config.jsonc"
+        sec_path = root / "secrets.json"
+        bak = root / "config.json.bak"
+        assert summary["ok"] is True
+        assert not src.exists()
+        assert bak.exists()
+        assert cfg_path.exists()
+        assert sec_path.exists()
+
+        text = cfg_path.read_text(encoding="utf-8")
+        assert text.startswith("//")
+        assert "secrets.json" in text
+        business = jsonc.load(cfg_path)
+        secrets = json.loads(sec_path.read_text(encoding="utf-8"))
+        assert "bilibili" not in business
+        assert business["telegram"]["enabled"] is True
+        assert "bot_token" not in business["telegram"]
+        assert business["web"]["host"] == "127.0.0.1"
+        assert "auth_token" not in business["web"]
+        assert secrets["bilibili"]["read"]["mid"] == 11
+        assert secrets["telegram"]["chat_id"] == 12345
+        assert secrets["web"]["auth_token"] == "web-secret-token"
+
+        loaded = load_config(cfg_path, env={}, require_auth=False)
+        assert loaded.bilibili.read.mid == 11
+        assert loaded.bilibili.write.mid == 22
+        assert loaded.telegram.enabled is True
+        assert loaded.telegram.bot_token == "tg-secret-token"
+        assert loaded.web.host == "127.0.0.1"
+        assert loaded.web.auth_token == "web-secret-token"
+        assert summary["verify"]["checked"] is True
+        assert summary["verify"]["read_mid"] == 11
+
+
+def test_force_overwrite_and_keep_src():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8")
+        (root / "config.jsonc").write_text("{}\n", encoding="utf-8")
+        (root / "secrets.json").write_text("{}\n", encoding="utf-8")
+        summary = split_config_file(src, force=True, keep_src=True)
+        assert src.exists()
+        assert summary["src_renamed"] is False
+        secrets = json.loads((root / "secrets.json").read_text(encoding="utf-8"))
+        assert secrets["bilibili"]["read"]["mid"] == 11
+
+
+def test_honors_secrets_path_relative():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        data = _monolith()
+        data["secrets_path"] = "creds/my-secrets.json"
+        src.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+        summary = split_config_file(src, keep_src=True)
+        sec = root / "creds" / "my-secrets.json"
+        assert Path(summary["secrets_out"]) == sec
+        assert sec.exists()
+        business = jsonc.load(root / "config.jsonc")
+        assert business.get("secrets_path") == "creds/my-secrets.json"
+
+
+def main():
+    test_split_raw_extracts_secrets_keeps_business()
+    test_split_raw_omits_absent_secret_stubs()
+    test_dry_run_writes_nothing()
+    test_refuses_overwrite_without_force()
+    test_writes_and_renames_src_by_default()
+    test_force_overwrite_and_keep_src()
+    test_honors_secrets_path_relative()
+    print(
+        "OK: split_config_secrets 抽取 bilibili/telegram/web 凭据、"
+        "dry-run、拒绝覆盖、改名与 secrets_path 测试全部通过"
+    )
+
+
+if __name__ == "__main__":
+    main()

tools/split_config_secrets.py

@@ -0,0 +1,326 @@
+# -*- coding: utf-8 -*-
+"""将旧版单体 ``config.json`` 拆成 ``config.jsonc`` + ``secrets.json``。
+
+从源文件抽出凭据段：
+
+- 整段 ``bilibili``（若存在）
+- ``telegram.bot_token`` / ``telegram.chat_id``（``enabled`` 等非密钥留在业务文件）
+- ``web.auth_token``（``host`` / ``port`` / ``track`` 等留在业务文件）
+
+源里没有的密钥键不会写空壳。``telegram`` / ``web`` 抽空后若无剩余键，则不写入
+secrets；业务侧若还有非密钥键则保留对应段。``bilibili`` 整段移出业务文件。
+
+用法::
+
+    # 先看会挪哪些键（不打印 cookie/token 原文）
+    uv run python tools/split_config_secrets.py --dry-run
+
+    # 默认：写 config.jsonc + secrets.json，并把源文件改名为 config.json.bak
+    uv run python tools/split_config_secrets.py /path/to/config.json
+
+    # 已有输出时需显式覆盖；保留源文件不改名
+    uv run python tools/split_config_secrets.py --force --keep-src
+"""
+
+from __future__ import annotations
+
+import argparse
+import copy
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any, Mapping, Sequence
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+from reaction_board import jsonc  # noqa: E402
+from reaction_board.config import load_config, resolve_secrets_path  # noqa: E402
+
+TELEGRAM_SECRET_KEYS = ("bot_token", "chat_id")
+WEB_SECRET_KEYS = ("auth_token",)
+
+CONFIG_HEADER = """\
+// 业务配置（可含注释与尾逗号）。敏感凭据在同目录 secrets.json。
+// `python app.py auth` 只回写 secrets.json，不会冲掉本文件注释。
+// 本文件由 tools/split_config_secrets.py 从旧版单体配置拆出。
+"""
+
+
+def _is_mapping(value: Any) -> bool:
+    return isinstance(value, Mapping) and not isinstance(value, (str, bytes))
+
+
+def split_raw(raw: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
+    """拆分业务配置与凭据。
+
+    返回 ``(business, secrets, stats)``。``stats`` 仅含键路径计数，不含密钥值。
+    """
+    business: dict[str, Any] = copy.deepcopy(dict(raw))
+    secrets: dict[str, Any] = {}
+    moved: list[str] = []
+
+    bilibili = business.pop("bilibili", None)
+    if _is_mapping(bilibili):
+        secrets["bilibili"] = copy.deepcopy(dict(bilibili))
+        moved.append("bilibili")
+    elif bilibili is not None:
+        # 非法类型原样丢掉业务侧，避免残留；不写入 secrets
+        moved.append("bilibili(non-object-dropped)")
+
+    telegram = business.get("telegram")
+    if _is_mapping(telegram):
+        tg_biz = dict(telegram)
+        tg_sec: dict[str, Any] = {}
+        for key in TELEGRAM_SECRET_KEYS:
+            if key in tg_biz:
+                tg_sec[key] = tg_biz.pop(key)
+                moved.append(f"telegram.{key}")
+        if tg_sec:
+            secrets["telegram"] = tg_sec
+        if tg_biz:
+            business["telegram"] = tg_biz
+        else:
+            business.pop("telegram", None)
+
+    web = business.get("web")
+    if _is_mapping(web):
+        web_biz = dict(web)
+        web_sec: dict[str, Any] = {}
+        for key in WEB_SECRET_KEYS:
+            if key in web_biz:
+                web_sec[key] = web_biz.pop(key)
+                moved.append(f"web.{key}")
+        if web_sec:
+            secrets["web"] = web_sec
+        if web_biz:
+            business["web"] = web_biz
+        else:
+            business.pop("web", None)
+
+    stats = {
+        "moved_paths": moved,
+        "moved_count": len(moved),
+        "business_top_keys": sorted(business.keys()),
+        "secrets_top_keys": sorted(secrets.keys()),
+        "has_bilibili": "bilibili" in secrets,
+        "telegram_secret_keys": sorted(secrets.get("telegram", {}).keys())
+        if isinstance(secrets.get("telegram"), dict)
+        else [],
+        "web_secret_keys": sorted(secrets.get("web", {}).keys())
+        if isinstance(secrets.get("web"), dict)
+        else [],
+    }
+    return business, secrets, stats
+
+
+def _atomic_write_json(path: Path, data: Mapping[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_name(path.name + ".tmp")
+    with tmp.open("w", encoding="utf-8") as fh:
+        json.dump(data, fh, ensure_ascii=False, indent=2)
+        fh.write("\n")
+    os.replace(tmp, path)
+
+
+def _atomic_write_jsonc(path: Path, data: Mapping[str, Any], *, header: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    body = json.dumps(data, ensure_ascii=False, indent=2)
+    if not body.endswith("\n"):
+        body += "\n"
+    text = header.rstrip() + "\n" + body
+    tmp = path.with_name(path.name + ".tmp")
+    with tmp.open("w", encoding="utf-8") as fh:
+        fh.write(text)
+    os.replace(tmp, path)
+
+
+def resolve_outputs(
+    src: Path,
+    raw: Mapping[str, Any],
+    *,
+    config_out: Path | None,
+    secrets_out: Path | None,
+) -> tuple[Path, Path]:
+    """解析输出路径；``secrets_path`` 相对源文件目录。"""
+    base = src.resolve().parent
+    cfg_path = config_out if config_out is not None else (base / "config.jsonc")
+    if secrets_out is not None:
+        sec_path = secrets_out
+    else:
+        sec_path = resolve_secrets_path(src, raw)
+    return cfg_path, sec_path
+
+
+def split_config_file(
+    src: Path,
+    *,
+    config_out: Path | None = None,
+    secrets_out: Path | None = None,
+    dry_run: bool = False,
+    force: bool = False,
+    keep_src: bool = False,
+) -> dict[str, Any]:
+    """执行拆分；返回可 JSON 序列化的摘要（不含密钥原文）。"""
+    src = Path(src)
+    if not src.exists():
+        raise FileNotFoundError(f"源配置不存在: {src}")
+
+    try:
+        raw = jsonc.load(src)
+    except ValueError as exc:
+        raise ValueError(f"源配置不是合法 JSON/JSONC: {src}: {exc}") from exc
+    if not isinstance(raw, Mapping):
+        raise ValueError(f"源配置顶层必须是对象: {src}")
+
+    business, secrets, stats = split_raw(raw)
+    cfg_path, sec_path = resolve_outputs(
+        src, raw, config_out=config_out, secrets_out=secrets_out
+    )
+
+    summary: dict[str, Any] = {
+        "ok": True,
+        "dry_run": bool(dry_run),
+        "src": str(src),
+        "config_out": str(cfg_path),
+        "secrets_out": str(sec_path),
+        "keep_src": bool(keep_src),
+        "src_bak": None if keep_src else str(src.with_name(src.name + ".bak")),
+        **stats,
+    }
+
+    if dry_run:
+        return summary
+
+    conflicts = [p for p in (cfg_path, sec_path) if p.exists()]
+    if conflicts and not force:
+        names = ", ".join(str(p) for p in conflicts)
+        raise FileExistsError(
+            f"输出已存在（需 --force 覆盖）: {names}"
+        )
+
+    _atomic_write_jsonc(cfg_path, business, header=CONFIG_HEADER)
+    _atomic_write_json(sec_path, secrets)
+
+    # 可选校验：合并加载后 mid 与源一致（不打印 cookie）
+    verify: dict[str, Any] = {"checked": False}
+    if secrets.get("bilibili") or raw.get("bilibili"):
+        cfg = load_config(cfg_path, env={}, require_auth=False)
+        src_bili = raw.get("bilibili") if _is_mapping(raw.get("bilibili")) else {}
+        read_mid = None
+        write_mid = None
+        if _is_mapping(src_bili.get("read")):
+            read_mid = src_bili["read"].get("mid")
+        if _is_mapping(src_bili.get("write")):
+            write_mid = src_bili["write"].get("mid")
+        if read_mid is not None and int(cfg.bilibili.read.mid) != int(read_mid):
+            raise RuntimeError(
+                f"校验失败: bilibili.read.mid {cfg.bilibili.read.mid} != {read_mid}"
+            )
+        if write_mid is not None and int(cfg.bilibili.write.mid) != int(write_mid):
+            raise RuntimeError(
+                f"校验失败: bilibili.write.mid {cfg.bilibili.write.mid} != {write_mid}"
+            )
+        verify = {
+            "checked": True,
+            "read_mid": int(cfg.bilibili.read.mid),
+            "write_mid": int(cfg.bilibili.write.mid),
+        }
+    summary["verify"] = verify
+
+    if not keep_src:
+        bak = src.with_name(src.name + ".bak")
+        if bak.exists() and not force:
+            raise FileExistsError(
+                f"备份已存在（需 --force 覆盖）: {bak}"
+            )
+        if bak.exists():
+            bak.unlink()
+        os.replace(src, bak)
+        summary["src_bak"] = str(bak)
+        summary["src_renamed"] = True
+    else:
+        summary["src_renamed"] = False
+
+    return summary
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="拆分旧版 config.json 为 config.jsonc + secrets.json",
+    )
+    parser.add_argument(
+        "src_pos",
+        nargs="?",
+        default=None,
+        help="源配置路径（默认仓库根目录 config.json）",
+    )
+    parser.add_argument(
+        "--src",
+        default=None,
+        help="源配置路径（与位置参数二选一；默认 ROOT/config.json）",
+    )
+    parser.add_argument(
+        "--config-out",
+        default=None,
+        help="业务 JSONC 输出路径（默认与源同目录 config.jsonc）",
+    )
+    parser.add_argument(
+        "--secrets-out",
+        default=None,
+        help="凭据 JSON 输出路径（默认同目录 secrets.json，"
+        "或源内 secrets_path）",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="只打印将挪动的键路径，不写文件、不改名",
+    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="允许覆盖已存在的输出（及 .bak）",
+    )
+    parser.add_argument(
+        "--keep-src",
+        action="store_true",
+        help="成功写入后保留源文件；默认改名为 <src>.bak，"
+        "以便 default_config_path 优先读到新的 config.jsonc",
+    )
+    args = parser.parse_args(list(argv) if argv is not None else None)
+
+    if args.src_pos and args.src and args.src_pos != args.src:
+        print(
+            json.dumps(
+                {"ok": False, "error": "位置参数与 --src 冲突"},
+                ensure_ascii=False,
+            ),
+            file=sys.stderr,
+        )
+        return 2
+    src = Path(args.src_pos or args.src or (ROOT / "config.json"))
+    config_out = Path(args.config_out) if args.config_out else None
+    secrets_out = Path(args.secrets_out) if args.secrets_out else None
+
+    try:
+        summary = split_config_file(
+            src,
+            config_out=config_out,
+            secrets_out=secrets_out,
+            dry_run=args.dry_run,
+            force=args.force,
+            keep_src=args.keep_src,
+        )
+        print(json.dumps(summary, ensure_ascii=False, indent=2))
+        return 0
+    except (OSError, ValueError, RuntimeError) as exc:
+        print(
+            json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False),
+            file=sys.stderr,
+        )
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())