hermes-chatto-plugin/platform_config.py

"""
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