Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
77d74c501a |
No files matched your search
@@ -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` 示例:
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -622,32 +622,19 @@ class DashboardQueries:
|
||||
continue
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _task_started_sort_key(value: Any) -> tuple[int, int]:
|
||||
"""缺 started_at 的排末尾;有值的按开始时间倒序(含进行中)。"""
|
||||
if isinstance(value, Mapping):
|
||||
raw = value.get("started_at")
|
||||
try:
|
||||
return (0, -int(raw))
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
return (1, 0)
|
||||
|
||||
def _task_records(self, conn: sqlite3.Connection) -> dict[str, Any]:
|
||||
rows = conn.execute(
|
||||
"SELECT k, v FROM kv WHERE k LIKE 'task:last:%'"
|
||||
"SELECT k, v FROM kv WHERE k LIKE 'task:last:%' ORDER BY k"
|
||||
).fetchall()
|
||||
items: list[tuple[str, Any]] = []
|
||||
result: dict[str, Any] = {}
|
||||
for row in rows:
|
||||
name = str(row["k"]).removeprefix("task:last:")
|
||||
try:
|
||||
value = json.loads(str(row["v"]))
|
||||
except (TypeError, json.JSONDecodeError):
|
||||
value = {"raw": row["v"]}
|
||||
items.append((name, value))
|
||||
items.sort(key=lambda pair: self._task_started_sort_key(pair[1]))
|
||||
# 插入序即开始时间倒序;前端 Object.entries 保留该序,勿再排序。
|
||||
return dict(items)
|
||||
result[name] = value
|
||||
return result
|
||||
|
||||
def _task_running(self, conn: sqlite3.Connection) -> dict[str, Any]:
|
||||
rows = conn.execute(
|
||||
|
||||
@@ -6,7 +6,6 @@ import {
|
||||
} from "../format.js";
|
||||
|
||||
function taskRows(tasks) {
|
||||
// Order comes from backend (_task_records started_at DESC); do not re-sort here.
|
||||
const entries = Object.entries(tasks || {});
|
||||
if (!entries.length) return emptyState("暂无任务记录", "运行一次同步、快照或发现后会写入最近结果。");
|
||||
return `<div class="table-wrap">
|
||||
|
||||
@@ -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()
|
||||
@@ -681,48 +681,6 @@ def test_overview_adjacent_previous_with_seed_and_dual_fields():
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_system_tasks_sorted_by_started_at_desc():
|
||||
"""最近任务按 started_at 倒序;缺开始时间的排末尾(含进行中无 finished)。"""
|
||||
import json
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, q = setup(tmp)
|
||||
try:
|
||||
repo = Repository(db)
|
||||
# 故意打乱写入/键名字母序;finished_at 与 started_at 错开,避免误用完成时间
|
||||
samples = (
|
||||
# name, started_at, finished_at
|
||||
("zzz_old", FIXED_NOW - 300, FIXED_NOW - 10), # 开始最早,完成却很晚
|
||||
("aaa_mid", FIXED_NOW - 100, FIXED_NOW - 90),
|
||||
("mmm_new", FIXED_NOW - 10, None), # 进行中:只有 started
|
||||
("no_time", None, FIXED_NOW), # 缺 started,应末尾
|
||||
)
|
||||
for name, started, finished in samples:
|
||||
payload = {
|
||||
"name": name,
|
||||
"ok": True if finished is not None else True,
|
||||
"error": None,
|
||||
"summary": {},
|
||||
}
|
||||
if started is not None:
|
||||
payload["started_at"] = started
|
||||
if finished is not None:
|
||||
payload["finished_at"] = finished
|
||||
repo.kv_set(f"task:last:{name}", json.dumps(payload))
|
||||
for legacy in ("snapshot", "fans", "discovery"):
|
||||
repo.kv_delete(f"task:last:{legacy}")
|
||||
DASHBOARD_CACHE.clear()
|
||||
system = q.system()
|
||||
assert list(system["tasks"].keys()) == [
|
||||
"mmm_new", "aaa_mid", "zzz_old", "no_time",
|
||||
]
|
||||
overview = q.overview()
|
||||
assert list(overview["tasks"].keys()) == list(system["tasks"].keys())
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def main_test():
|
||||
test_overview()
|
||||
test_video_listing_filters_and_sorting()
|
||||
@@ -738,7 +696,6 @@ def main_test():
|
||||
test_fans_candidates_system_and_export()
|
||||
test_system_status_uses_completed_date_and_running_flags()
|
||||
test_system_next_runs_follow_last_completion_not_wall_clock()
|
||||
test_system_tasks_sorted_by_started_at_desc()
|
||||
test_growth_auto_falls_back_to_one_day()
|
||||
test_snapshot_recent_next_run_follows_completion()
|
||||
test_overview_adjacent_previous_with_seed_and_dual_fields()
|
||||
|
||||
@@ -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())
|
||||
Reference in new issue
Block a user