hermes-chatto-plugin/platform_config.py

"""
Chatto Platform Config


"""

import sys
import os
from pathlib import Path

# 1. Den absoluten Pfad zum 'vendor'-Ordner in diesem Plugin ermitteln
current_dir = Path(__file__).parent
vendor_dir = current_dir / "vendor"

# 2. Den vendor-Ordner an den Anfang des Suchpfads (sys.path) setzen
if str(vendor_dir) not in sys.path:
    sys.path.insert(0, str(vendor_dir))

from dataclasses import dataclass
import logging
import os
from typing import Any, Dict, Optional

from gateway.config import PlatformConfig
import utils

from .vendor.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"

    EXTRA_ENV_MAPPING = {
        "base_url": "CHATTO_BASE_URL",
        "home_channel": "CHATTO_HOME_CHANNEL",
        "require_mention": "CHATTO_REQUIRE_MENTION",
        "free_response_channels": "CHATTO_FREE_RESPONSE_CHANNELS",
        "auto_thread": "CHATTO_AUTO_THREAD",
        "allow_all_users": "CHATTO_ALLOW_ALL_USERS",
        "allowed_users": "CHATTO_ALLOWED_USERS",
    }

    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], 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, str(extra_val), "False"), default)


def _split_str_to_list(mystring: str) -> list:
    logger.info("mystring: %s", mystring)
    return list(c for c in mystring.split(","))


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:
        #logger.info("extra_val is1: " + str(type(extra_val)))
        #logger.info("extra_val is2: " + str(extra_val))
        if isinstance(extra_val, list):
            return list(
                c for c in extra_val
            )
        if isinstance(extra_val, str):
            return _split_str_to_list(extra_val)

    return []

from typing import TypeVar, Generic, Any

# 1. Define a generic Type Variable
T = TypeVar('T')

# 2. Inherit from Generic[T]
class ConfigField(Generic[T]):
    """Repräsentiert ein einzelnes Konfigurationsfeld mit IDE-Support."""
# 3. Type 'value' as T (or T | None since it starts as None)
    value: T
    field_name: str = ""
    env_name: str = ""

    # 4. Hint that the init argument should match type T
    # We add `| Any` as a fallback because complex types like `list[str]` 
    # can sometimes confuse older type checkers when used with `type[T]`
    def __init__(self, type_: type[T] | Any):
        self._type = type_

    # 5. Ensure the IDE knows only type T can be assigned
    def __set__(self, instance: Any, value: T) -> None:
        self.value = value
        self._type = type(value)
        logger.info("Chatto: configuration field '%s' set to '%s'", self.field_name, str(value))

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

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


@dataclass
class ChattoConfiguration:
    """
    Chatto Platform Config

    Some constants and getting from environment/config.yaml
    """
    base_url = ConfigField(str)
    token = ConfigField(str | None)
    login = ConfigField(str)
    password = ConfigField(str)
    channels = ConfigField(str)
    channels_list = ConfigField(list[str])
    home_channel = ConfigField(str)
    allowed_users = ConfigField(list[str])
    require_mention = ConfigField(bool)
    free_response_channels_list = ConfigField(list[str])
    auto_thread = ConfigField(bool)
    allow_all_users = ConfigField(bool)
    reactions = ConfigField(bool)

    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.value = _get_env_or_extra_str(self.base_url.env_name,
            pconfig.extra.get(self.base_url.field_name), ChattoClient.DEFAULT_BASE_URL)

        self.token.value = _get_env_or_extra_str_opt(self.token.env_name, pconfig.extra.get(self.token.field_name))
        self.login.value = _get_env_or_extra_str(self.login.env_name, pconfig.extra.get(self.login.field_name))
        self.password.value = _get_env_or_extra_str(self.password.env_name, pconfig.extra.get(self.password.field_name))
        
        self.channels.value = str(_get_env_or_extra_list(self.channels.env_name, pconfig.extra.get(self.channels.field_name)))
        self.channels_list.value = _get_env_or_extra_list(self.channels.env_name, pconfig.extra.get(self.channels.field_name))

        self.home_channel.value = _get_env_or_extra_str(self.home_channel.env_name, pconfig.extra.get(self.home_channel.field_name))

        self.allowed_users.value = _get_env_or_extra_list(self.allowed_users.env_name, pconfig.extra.get(self.allowed_users.field_name))
        logger.info("self.allowed_users.value: %s", self.allowed_users.value)

        self.require_mention.value = _get_env_or_extra_truthy(self.require_mention.env_name, pconfig.extra.get(self.require_mention.field_name))
        logger.info("self.require_mention.value: %s", self.require_mention.value)

        # free_response_channels: room IDs where the bot responds without being mentioned via "@botname"
        self.free_response_channels_list.value = _get_env_or_extra_list(self.free_response_channels_list.env_name,
                                                                   pconfig.extra.get(self.free_response_channels_list.field_name))

        # 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.value = _get_env_or_extra_truthy(self.auto_thread.env_name, pconfig.extra.get(self.auto_thread.field_name))

        self.allow_all_users.value = _get_env_or_extra_truthy(self.allow_all_users.env_name, pconfig.extra.get(self.allow_all_users.field_name))
        self.reactions.value = _get_env_or_extra_truthy(self.reactions.env_name, pconfig.extra.get(self.reactions.field_name))

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