"""Loading and validation of the bot's JSON configuration.""" from __future__ import annotations import json import os from dataclasses import dataclass from pathlib import Path from typing import Any DEFAULT_CONFIG: dict[str, Any] = { "matrix_homeserver": "https://matrix.example.org", "matrix_user": "sumka-bot", "store_dir": "session_storage", "sumka_image": "sumka:amd", "ai_api_key": "change-me", "sumka_args": [], } SUMKA_CONTAINER_NAME = "matrix-sumka-job" REPORT_POLL_INTERVAL_SECONDS = 2.0 UPLOAD_TIMEOUT_SECONDS = 3600.0 class ConfigError(RuntimeError): """Raised when config.json cannot be used safely.""" @dataclass(frozen=True, slots=True) class AppConfig: matrix_homeserver: str matrix_user: str store_dir: Path sumka_image: str ai_api_key: str sumka_args: tuple[str, ...] runtime_dir: Path @property def work_dir(self) -> Path: return self.runtime_dir / "work" @property def runner_script(self) -> Path: return Path(__file__).resolve().parent / "run_sumka.sh" @property def sumka_container_name(self) -> str: return SUMKA_CONTAINER_NAME @property def report_poll_interval_seconds(self) -> float: return REPORT_POLL_INTERVAL_SECONDS @property def upload_timeout_seconds(self) -> float: return UPLOAD_TIMEOUT_SECONDS def _non_empty_string(data: dict[str, Any], key: str) -> str: value = data.get(key) if not isinstance(value, str) or not value.strip(): raise ConfigError(f"`{key}` must be a non-empty string") return value.strip() def _runtime_path(value: str, runtime_dir: Path) -> Path: path = Path(value).expanduser() if not path.is_absolute(): path = runtime_dir / path return path.resolve() def _parse_config(data: Any, config_path: Path) -> AppConfig: if not isinstance(data, dict): raise ConfigError("the config root must be a JSON object") expected = set(DEFAULT_CONFIG) missing = expected - set(data) unknown = set(data) - expected if missing: raise ConfigError(f"missing config keys: {', '.join(sorted(missing))}") if unknown: raise ConfigError(f"unknown config keys: {', '.join(sorted(unknown))}") args = data.get("sumka_args") if not isinstance(args, list) or any(not isinstance(arg, str) for arg in args): raise ConfigError("`sumka_args` must be an array of strings") api_key = _non_empty_string(data, "ai_api_key") if api_key == DEFAULT_CONFIG["ai_api_key"]: raise ConfigError("replace the placeholder value in `ai_api_key`") runtime_dir = config_path.parent.resolve() store_dir = _runtime_path(_non_empty_string(data, "store_dir"), runtime_dir) runner_script = Path(__file__).resolve().parent / "run_sumka.sh" if not runner_script.is_file(): raise ConfigError(f"runner script does not exist: {runner_script}") if not os.access(runner_script, os.X_OK): raise ConfigError(f"runner script is not executable: {runner_script}") return AppConfig( matrix_homeserver=_non_empty_string(data, "matrix_homeserver"), matrix_user=_non_empty_string(data, "matrix_user"), store_dir=store_dir, sumka_image=_non_empty_string(data, "sumka_image"), ai_api_key=api_key, sumka_args=tuple(args), runtime_dir=runtime_dir, ) def write_default_config(path: str | Path = "config.json") -> Path: """Write an editable default config and return its absolute path.""" config_path = Path(path).resolve() config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text( json.dumps(DEFAULT_CONFIG, ensure_ascii=False, indent=4) + "\n", encoding="utf-8", ) return config_path def load_config(path: str | Path = "config.json") -> AppConfig | None: """Load config, or create a template and return None on the first run.""" config_path = Path(path).resolve() if not config_path.is_file(): write_default_config(config_path) return None try: data = json.loads(config_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise ConfigError(f"cannot read {config_path}: {error}") from error return _parse_config(data, config_path)