hermes-chatto-plugin/platform_config.py

"""
Chatto Platform Config


"""

import os

# Put the vendored dependencies for THIS platform on sys.path before importing
# anything from chattolib. Imported relatively as part of the plugin package and
# absolutely when this module is loaded standalone (e.g. by the tests).
try:
    from .vendor_path import setup_vendor_path
except ImportError:  # pragma: no cover - depends on how the module is loaded
    from vendor_path import setup_vendor_path

setup_vendor_path()

import logging
from typing import Any, Dict, Generic, Optional, TypeVar

from gateway.config import PlatformConfig
import utils

# Absolute import — the vendor dir is on sys.path (see above) and chattolib's
# own modules import each other absolutely ("from chattolib.x import y").
# Importing it relatively as well would load a *second* copy of every module
# under a different name, so isinstance() checks across the two would fail.
from chattolib.client import ChattoClient

logger = logging.getLogger(__name__)

# --------------------------------------------------------------------------- #
# Constants
# --------------------------------------------------------------------------- #

class ChattoConstants:
    """
    Chatto Platform Constants
    """

    def __init__(self):
        pass

    PLATFORM_NAME: str = "chatto-platform"
    PLATFORM_LABEL: str = "Chatto"

    INSTALL_HINT = "Requires a Chatto server. See https://docs.chatto.run"

    # NOTE: env var names live on the ConfigFields below (ConfigField.env_name),
    # so there is exactly one source of truth for them.

    MAX_MESSAGE_LENGTH = 10000
    SEEN_CAP = 500

    # WebSocket / realtime protocol
    WS_PATH = "/api/realtime"
    WS_AUTH_TIMEOUT = 20.0
    WS_MAX_MESSAGE_BYTES = 4_000_000
    WS_PING_INTERVAL = 30.0
    WS_PING_FAILURE_THRESHOLD = 3
    WS_RECONNECT_INITIAL_BACKOFF = 1.0
    WS_RECONNECT_MAX_BACKOFF = 30.0

    # HTTP timeout used for outbound URL fetches
    HTTP_TIMEOUT = 30

    # Emoji shortcode mapping (Chatto uses shortcode names, not unicode emoji)

    # Emoji shortcode mapping (Chatto uses shortcode names, not unicode emoji)
    EMOJI_TO_SHORTCODE: Dict[str, str] = {
        "👍": "thumbsup",
        "👎": "thumbsdown",
        "❤️": "heart",
        "❤": "heart",
        "✅": "white_check_mark",
        "❌": "x",
        "👀": "eyes",
        "🎉": "tada",
        "😂": "joy",
        "🚀": "rocket",
        "🔥": "fire",
        "💯": "100",
        "🤔": "thinking",
        "👏": "clap",
        "🙏": "pray",
        "😅": "sweat_smile",
        "😴": "sleeping",
        "⏳": "hourglass",
    }

    # Chunk size for asset uploads (256 KB)
    UPLOAD_CHUNK_SIZE = 256 * 1024


def _get_env_or_extra_str_opt(env_var: str, extra_val: Optional[str | bool], default: Optional[str] = None) -> 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:
        if isinstance(extra_val, str):
            return extra_val.strip()
        elif isinstance(extra_val, bool):
            return str(extra_val)
        else:
            logger.error("extra_val: %s is supposed to be str, but %s was found.", extra_val, str(type(extra_val)))

    if default is not None:
        logger.debug("Chatto: Defaulting to '%s'", default)
        return default.strip() 
    return None

def _get_env_or_extra_str(env_var: str, extra_val: Optional[str | bool], default: Optional[str] = None) -> str:
    """Get a value from environment variable or extra config."""
    my_string = _get_env_or_extra_str_opt(env_var, extra_val, default)
    if my_string:
        return my_string
    return ""


def _get_env_or_extra_truthy(env_var: str, extra_val: Optional[str | bool], 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, str(default)), default)


def _split_str_to_list(mystring: str) -> list[str]:
    """Split a comma-separated string, trimming whitespace and dropping empties."""
    return [part.strip() for part in mystring.split(",") if part.strip()]


def _get_env_or_extra_list(env_var: str, extra_val: Optional[list[str]]) -> 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 [str(c).strip() for c in extra_val if str(c).strip()]
        if isinstance(extra_val, str):
            return _split_str_to_list(extra_val)

    return []


T = TypeVar("T")


class ConfigValue(Generic[T]):
    """The resolved value of a single config field, bound to one configuration
    instance.

    Access the payload via ``.value``.  ``__bool__``/``__eq__`` delegate to it,
    so a forgotten ``.value`` (``if config.allow_all_users:``) still evaluates
    the actual setting instead of the always-truthy wrapper object.
    """

    __slots__ = ("value", "field_name", "env_name")

    def __init__(self, value: T, field_name: str, env_name: str) -> None:
        self.value = value
        self.field_name = field_name
        self.env_name = env_name

    def __bool__(self) -> bool:
        return bool(self.value)

    def __eq__(self, other: Any) -> bool:
        if isinstance(other, ConfigValue):
            return self.value == other.value
        return self.value == other

    def __hash__(self) -> int:
        return hash(self.value)

    def __contains__(self, item: Any) -> bool:
        return item in self.value  # type: ignore[operator]

    def __iter__(self):
        return iter(self.value)  # type: ignore[call-overload]

    def __str__(self) -> str:
        return str(self.value)

    def __repr__(self) -> str:
        return f"{self.field_name}={self.value!r}"


class ConfigField(Generic[T]):
    """Declarative descriptor for one config field.

    Declares *how* a field is read (kind, env var name, default); the resolved
    payload lives per configuration instance in ``instance._values``, never on
    the descriptor itself.  Reading a field on the class (rather than on an
    instance) yields the descriptor, so ``ChattoConfiguration.token.env_name``
    keeps working for the registration hooks.
    """

    field_name: str = ""
    env_name: str = ""

    def __init__(
        self,
        kind: str,
        *,
        default: Any = None,
        config_key: Optional[str] = None,
        doc: str = "",
    ) -> None:
        # kind: "str" | "str_opt" | "bool" | "list"
        self.kind = kind
        self.default = default
        # config_key: the name used in config.yaml's "extra" block and (upper-cased,
        # CHATTO_-prefixed) as the env var, where it differs from the attribute name.
        self._config_key = config_key
        self.__doc__ = doc

    def __set_name__(self, owner: Any, name: str) -> None:
        self.field_name = str(name).lower().replace("chatto_", "", 1)
        self.config_key = self._config_key or self.field_name
        self.env_name = f"CHATTO_{self.config_key.upper()}"

    def __get__(self, instance: Any, owner: Any = None) -> "ConfigValue[T]":
        if instance is None:
            return self  # type: ignore[return-value]
        return instance._values[self.field_name]

    def __set__(self, instance: Any, value: Any) -> None:
        raise AttributeError(
            f"Chatto: config field '{self.field_name}' is read-only; "
            f"set '{self.field_name}.value' if you really need to override it."
        )

    def resolve(self, extra: Dict[str, Any]) -> ConfigValue:
        """Read this field from the environment, then from ``extra``, then the default."""
        raw = extra.get(self.config_key)

        if self.kind == "list":
            value: Any = _get_env_or_extra_list(self.env_name, raw)
        elif self.kind == "bool":
            value = _get_env_or_extra_truthy(self.env_name, raw, bool(self.default))
        elif self.kind == "str_opt":
            value = _get_env_or_extra_str_opt(self.env_name, raw, self.default)
        else:
            value = _get_env_or_extra_str(self.env_name, raw, self.default)

        return ConfigValue(value, self.field_name, self.env_name)


class ChattoConfiguration:
    """Chatto platform configuration.

    Every field is resolved once per instance, in this order:
    environment variable → ``PlatformConfig.extra`` → declared default.
    """

    base_url = ConfigField("str", default=ChattoClient.DEFAULT_BASE_URL)
    token = ConfigField("str_opt")
    login = ConfigField("str")
    password = ConfigField("str")
    channels_list = ConfigField("list", config_key="channels")
    home_channel = ConfigField("str")
    allowed_users = ConfigField("list")
    require_mention = ConfigField("bool", default=False)
    # free_response_channels: room IDs where the bot responds without being
    # mentioned via "@botname" even when require_mention is true.
    free_response_channels_list = ConfigField("list", config_key="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.
    auto_thread = ConfigField("bool", default=True)
    allow_all_users = ConfigField("bool", default=False)
    reactions = ConfigField("bool", default=True)

    def __init__(self, pconfig: PlatformConfig):
        """Resolve every declared ConfigField against env vars and
        ``PlatformConfig.extra`` (which Hermes pre-populates from config.yaml).
        """
        extra: Dict[str, Any] = getattr(pconfig, "extra", None) or {}

        self._values: Dict[str, ConfigValue] = {
            field.field_name: field.resolve(extra) for field in self.fields()
        }

        logger.debug("ChattoConfiguration: %s", self)

    @classmethod
    def fields(cls) -> list[ConfigField]:
        """All declared config fields, in declaration order."""
        return [v for v in vars(cls).values() if isinstance(v, ConfigField)]

    def __str__(self) -> str:
        redacted = {"token", "password"}
        parts = [
            f"{name}={'***' if name in redacted and cv.value else cv.value!r}"
            for name, cv in self._values.items()
        ]
        return "ChattoConfiguration(" + ", ".join(parts) + ")"