137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
"""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)
|