""" Chatto Platform Config """ from dataclasses import dataclass from email.policy import default import os from typing import Any, Optional from gateway.config import PlatformConfig import utils from vendor.chattolib.client import ChattoClient class ChattoConstants: """ Chatto Platform Constants """ PLATFORM_ID: str = "CHATTO" PLATFORM_NAME: str = "chatto-platform" PLATFORM_LABEL: str = "Chatto" def _get_env_or_extra_str(env_var: str, extra_val: Optional[str]) -> Optional[str]: """Get a value from environment variable or extra config.""" env_value = os.getenv(env_var) if env_value is not None: return env_value.strip() if extra_val is not None: return extra_val.strip() return None def _get_env_or_extra_truthy(env_var: str, extra_val: Optional[str], default: bool = False) -> bool: """Get a boolean value from environment variable or extra config.""" return utils.is_truthy_value(_get_env_or_extra_str(env_var, extra_val), default) def _split_str_to_list(mystring: str) -> list: return list(c.strip() for c in mystring.split(",")) def _get_env_or_extra_list(env_var: str, extra_val: Optional[set]) -> list[str]: """Get a list of values from environment variable or extra config.""" env_value = os.getenv(env_var) if env_value is not None: return _split_str_to_list(env_value) if extra_val is not None: if isinstance(extra_val, list): return list( c.strip() for c in extra_val ) if isinstance(extra_val, str): return _split_str_to_list(extra_val) return [] @dataclass class ChattoConfiguration: """ Chatto Platform Config Some constants and getting from environment/config.yaml """ _token: Optional[str] = None def __init__(self, pconfig: PlatformConfig): """PlatformConfig from Hermes provides our own configuration within the "extra" object. But here, we allow overriding via environment variables again. We take our Configuration from this typed Class, because it is easier access than using extra.get["base_url"]. """ self._base_url = _get_env_or_extra_str("CHATTO_BASE_URL", pconfig.extra.get("base_url")) or ChattoClient.DEFAULT_BASE_URL self._token = _get_env_or_extra_str("CHATTO_TOKEN", pconfig.extra.get("token")) self._login = _get_env_or_extra_str("CHATTO_LOGIN", pconfig.extra.get("login")) self._password = _get_env_or_extra_str("CHATTO_PASSWORD", pconfig.extra.get("password")) self._channels_str = _get_env_or_extra_str("CHATTO_CHANNELS", pconfig.extra.get("channels")) self._channels_list = _get_env_or_extra_list("CHATTO_CHANNELS", pconfig.extra.get("channels")) self._home_channel = _get_env_or_extra_str("CHATTO_HOME_CHANNEL", pconfig.extra.get("home_channel")) self._require_mention = _get_env_or_extra_truthy("CHATTO_REQUIRE_MENTION", pconfig.extra.get("require_mention")) # free_response_channels: room IDs where the bot responds without being mentioned via "@botname" self._free_response_channels_list = _get_env_or_extra_list("CHATTO_FREE_RESPONSE_CHANNELS", pconfig.extra.get("free_response_channels")) # Auto-thread: by default, Chatto creates a thread for replies to room # messages (not DMs, not already in a thread). This keeps conversations # organized in the room. Can be disabled via extra.auto_thread=false. self._auto_thread = _get_env_or_extra_truthy("CHATTO_AUTO_THREAD", pconfig.extra.get("auto_thread")) def _add_env_to_seed(self, seed: dict, our_key: str) -> dict env_value = os.getenv("CHATTO_" + our_key.upper()) if env_value: seed[our_key.lower()] = env_value return seed def env_only_to_seed_extra(self) -> Optional[dict]: """Seed PlatformConfig.extra from env vars. Returns a dict compatible with the PlatformConfig merge hook (or None when no env-provided values are present). Called by the platform registry during load_gateway_config(). Return None when the platform isn't minimally configured — the caller then skips auto-enabling. Return a dict to seed extras. The special 'home_channel' key is extracted and becomes a proper HomeChannel dataclass on the PlatformConfig; every other key is merged into PlatformConfig.extra. """ seed = {} seed["base_url"] = (os.getenv("CHATTO_BASE_URL") or ChattoClient.DEFAULT_BASE_URL).strip() seed = self._add_env_to_seed(seed, "TOKEN") seed = self._add_env_to_seed(seed, "LOGIN") seed = self._add_env_to_seed(seed, "PASSWORD") seed = self._add_env_to_seed(seed, "CHANNELS") seed = self._add_env_to_seed(seed, "HOME_CHANNEL") seed = self._add_env_to_seed(seed, "REQUIRE_MENTION") seed = self._add_env_to_seed(seed, "FREE_RESPONSE_CHANNELS") seed = self._add_env_to_seed(seed, "AUTO_THREAD") return seed def _config_key_not_set(self, config_key: str) -> ValueError: envname_upper: str = "CHATTO_" + config_key.upper() keyname_lower: str = "chatto.extra." + config_key.lower() valueErrorString: str = "Environment variable " + envname_upper + " or config.yaml YAML key " + keyname_lower + " is not set." return ValueError(valueErrorString) def _ensure_configuration(self) -> None: """Ensure that the Chatto configuration is valid.""" if not self._base_url: raise self._config_key_not_set("base_url") if not self._login: raise self._config_key_not_set("login") if not self._password: raise self._config_key_not_set("password") return None