hermes-chatto-plugin/platform_config.py

"""
Chatto Platform Config


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

from gateway.config import PlatformConfig
import utils

from vendor.chattolib.client import ChattoClient

from .adapter import ChattoAdapter, hermes_standalone_sender_fn

logger = logging.getLogger(__name__)

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

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

    def __init__(self):
        pass

    PLATFORM_ID: str = "chatto"
    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_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], 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:
        return extra_val.strip()

    if default is not None:
        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], 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, "False"), 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 []

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)

    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(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_str(self.allowed_users.env_name, pconfig.extra.get(self.allowed_users.field_name))

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

        # 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))


def hermes_validate_config_fn(config: PlatformConfig) -> bool:
    """"
    Function name should be the same as register argument name with "hermes_" prefix, so we
    know that it is needed for plugin register(). Do not change signature.
    - config
    Check whether Chatto Plugin is configured."""

    chatto_config = ChattoConfiguration(pconfig=config)

    if chatto_config.base_url.value and (chatto_config.token.value or (chatto_config.login.value and chatto_config.password.value)):
        return True
    return False


def hermes_check_fn() -> bool:
    """Check if Chatto is configured and dependencies are available.
    Add real logic?! Or just .. there are no dependencies.. always return true. Really. Docs suck."""
    try:
        from vendor.chattolib.client import ChattoClient
        return True
    except ImportError:
        return False
    return True


# ---------------------------------------------------------------------------
# is_connected probe
# ---------------------------------------------------------------------------

def hermes_is_connected(config: PlatformConfig) -> bool:
    """Check whether Chatto Plugin is connected. But where to? To the Hermes Agent? To Chatto Server?
    The Hermes Agent plugin docs suck and it seems there are many functions to do the same."""
    return bool(hermes_validate_config_fn(config) and config.enabled)



# ---------------------------------------------------------------------------
# YAML → env config bridge
# ---------------------------------------------------------------------------

@DeprecationWarning
def hermes_apply_yaml_config_fn(yaml_dict: dict, platform_dict: dict) -> Optional[dict]:
    """Translate config.yaml chatto.extra keys into CHATTO_* env vars.
    I don't actually get why Hermes wants us to modify OS environment variables.
    Bad behavior in my book.
    
    Also .. I don't think we need this"""

    if not isinstance(platform_dict, dict):
        platform_dict = {}
    extra = platform_dict.get("extra", {}) or {}
    if not isinstance(extra, dict):
        extra = {}

    for yaml_key, env_key in ChattoConstants.EXTRA_ENV_MAPPING.items():
        val = extra.get(yaml_key)
        if val is not None and not os.getenv(env_key):
            if isinstance(val, bool):
                env_val = str(val).lower()
            elif isinstance(val, list):
                env_val = ",".join(str(v) for v in val)
            else:
                env_val = str(val)
            os.environ[env_key] = env_val

    channels = extra.get(ChattoConfiguration.channels.field_name)
    if isinstance(channels, list) and not os.getenv(ChattoConfiguration.channels.env_name):
        os.environ[ChattoConfiguration.channels.env_name] = ",".join(str(c) for c in channels)

    allowed = extra.get(ChattoConfiguration.allowed_users.field_name)
    if isinstance(allowed, list) and not os.getenv(ChattoConfiguration.allowed_users.env_name):
        os.environ[ChattoConfiguration.allowed_users.env_name] = ",".join(str(u) for u in allowed)

    if ChattoConfiguration.allow_all_users.field_name in extra and not os.getenv(ChattoConfiguration.allow_all_users.env_name):
        os.environ[ChattoConfiguration.allow_all_users.env_name] = str(extra[ChattoConfiguration.allow_all_users.field_name]).lower()

    return None


def hermes_setup_fn() -> None:
    """Interactive setup wizard for Chatto. Is called by and only works in Hermes CLI context.
    Function name should be the same as register argument name with "hermes_" prefix, so we
    know that it is needed for plugin register().
    """
    from hermes_cli.setup import (
        prompt,
        prompt_yes_no,
        save_env_value,
        get_env_value,
        print_header,
        print_info,
        print_warning,
        print_success,
    )

    url = prompt(
        "Chatto server URL (e.g. https://chat.example.com) or leave blank for default ChattoHQ on chat.chatto.run:")
    if url:
        save_env_value(ChattoConfiguration.base_url.env_name, url)

    login = prompt("Chatto login (username):")
    if login:
        save_env_value(ChattoConfiguration.login.env_name, login)

    password = prompt("Chatto password:", password=True)
    if password:
        save_env_value(ChattoConfiguration.password.env_name, password)

    channels = prompt("Room IDs to watch (comma-separated, or empty for all):")
    if channels:
        save_env_value(ChattoConfiguration.channels.env_name, channels)

    home = prompt("Home room ID for notifications (or empty):")
    if home:
        save_env_value(ChattoConfiguration.home_channel.env_name, home)

    allow_all = prompt_yes_no("Allow all users to talk? (true/false):")
    if allow_all:
        save_env_value(ChattoConfiguration.allow_all_users.env_name, str(allow_all))
    print_success("\n✓ Chatto configured. Restart the gateway to activate.")


def hermes_env_enablement_fn() -> 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.

    Function name should be the same as register argument name with "hermes_" prefix, so we
    know that it is needed for plugin register().
    """

    def _add_env_to_seed(seed: dict, our_key: str) -> dict:
        env_value = os.getenv(our_key.upper())
        if env_value:
            seed[our_key.lower()] = env_value
        return seed
    
    seed = {}
    seed["base_url"] = (os.getenv(ChattoConfiguration.base_url.env_name) or ChattoClient.DEFAULT_BASE_URL).strip()

    seed = _add_env_to_seed(seed, ChattoConfiguration.token.env_name)
    seed = _add_env_to_seed(seed, ChattoConfiguration.login.env_name)
    seed = _add_env_to_seed(seed, ChattoConfiguration.password.env_name)

    seed = _add_env_to_seed(seed, ChattoConfiguration.channels.env_name)
    seed = _add_env_to_seed(seed, ChattoConfiguration.home_channel.env_name)
    seed = _add_env_to_seed(seed, ChattoConfiguration.require_mention.env_name)
    seed = _add_env_to_seed(seed, ChattoConfiguration.free_response_channels_list.env_name)
    seed = _add_env_to_seed(seed, ChattoConfiguration.auto_thread.env_name)

    return seed


# ---------------------------------------------------------------------------
# Plugin registration entry point
# ---------------------------------------------------------------------------

def hermes_adapter_factory(config: PlatformConfig):
    """Factory wrapper that constructs ChattoAdapter from a PlatformConfig."""
    return ChattoAdapter(config)


def register(ctx) -> None:
    """Plugin entry point — called by the Hermes plugin system."""
    logger.error("Registering Chatto Plugin")
    ctx.register_platform(
        name=ChattoConstants.PLATFORM_NAME,
        label=ChattoConstants.PLATFORM_LABEL,
        adapter_factory=hermes_adapter_factory,
        check_fn=hermes_check_fn,
        validate_config=hermes_validate_config_fn,
        is_connected=hermes_is_connected,
        install_hint=ChattoConstants.INSTALL_HINT,
        env_enablement_fn=hermes_env_enablement_fn,
        setup_fn=hermes_setup_fn,
        apply_yaml_config_fn=hermes_apply_yaml_config_fn,
        cron_deliver_env_var=ChattoConfiguration.home_channel.env_name,
        standalone_sender_fn=hermes_standalone_sender_fn,
        allowed_users_env=ChattoConfiguration.allowed_users.env_name,
        allow_all_env=ChattoConfiguration.allow_all_users.env_name,
        max_message_length=ChattoConstants.MAX_MESSAGE_LENGTH,
        emoji="💬",
        allow_update_command=True,
        pii_safe=False,
        platform_hint=(
            "You are chatting in Chatto (a self-hosted or cloud-hosted team or community chat server). "
            "Markdown is supported. Users _may_ address you by @-mentioning your name. If configured, " \
            "you also react without a @-mention. Direct messages reach you without a mention."
            "Keep responses conversational."
        ),
    )