""" 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." ), )