327 lines
11 KiB
Python
327 lines
11 KiB
Python
# -*- 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())
|