Auto-update config.json with new scalar fields from config.example.json on startup
On each startup, if config.json exists alongside config.example.json, any scalar fields (strings, ints, booleans) present in the example but absent in the running config are added automatically and the file is rewritten atomically. Existing values and dict/list sections are never touched. Added fields are logged at INFO level. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,11 +1,15 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
except ModuleNotFoundError: # pragma: no cover - optional dependency fallback
|
||||
@@ -88,6 +92,35 @@ def _read_json_file(path: Path) -> dict[str, Any]:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def _write_json_file(path: Path, data: dict[str, Any]) -> None:
|
||||
serialized = json.dumps(data, ensure_ascii=False, indent=2)
|
||||
with tempfile.NamedTemporaryFile(
|
||||
"w",
|
||||
encoding="utf-8",
|
||||
delete=False,
|
||||
dir=path.parent,
|
||||
prefix=path.name,
|
||||
suffix=".tmp",
|
||||
) as handle:
|
||||
handle.write(serialized)
|
||||
temp_path = Path(handle.name)
|
||||
temp_path.replace(path)
|
||||
|
||||
|
||||
def _merge_example_defaults(target: dict[str, Any], source: dict[str, Any], path: str = "") -> list[str]:
|
||||
"""Add scalar keys present in source but absent in target; recurse into existing dicts."""
|
||||
added: list[str] = []
|
||||
for key, value in source.items():
|
||||
full_path = f"{path}.{key}" if path else key
|
||||
if key not in target:
|
||||
if not isinstance(value, (dict, list)):
|
||||
target[key] = value
|
||||
added.append(full_path)
|
||||
elif isinstance(target[key], dict) and isinstance(value, dict):
|
||||
added.extend(_merge_example_defaults(target[key], value, full_path))
|
||||
return added
|
||||
|
||||
|
||||
def _parse_int(value: Any, *, field_name: str) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
@@ -350,6 +383,16 @@ def load_app_config(root: Path) -> AppConfig:
|
||||
raw: dict[str, Any] = {}
|
||||
if candidate.exists():
|
||||
raw = _read_json_file(candidate)
|
||||
example_path = root / "config.example.json"
|
||||
if example_path.exists():
|
||||
try:
|
||||
example_raw = _read_json_file(example_path)
|
||||
added = _merge_example_defaults(raw, example_raw)
|
||||
if added:
|
||||
_write_json_file(candidate, raw)
|
||||
log.info("Auto-updated %s with new fields: %s", candidate.name, ", ".join(added))
|
||||
except Exception as exc:
|
||||
log.warning("Could not auto-update config from example: %s", exc)
|
||||
|
||||
env_api_id = os.environ.get("TG_API_ID")
|
||||
env_api_hash = os.environ.get("TG_API_HASH")
|
||||
|
||||
Reference in New Issue
Block a user