ForcePilot/backend/package/yuxi/channels/adapters/slack/message_actions.py

from __future__ import annotations

from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any


class ActionCategory(StrEnum):
    REACTIONS = "reactions"
    MESSAGES = "messages"
    PINS = "pins"
    MEMBER_INFO = "memberInfo"
    EMOJI_LIST = "emojiList"
    CALLS = "calls"
    WORKFLOWS = "workflows"
    SLACK_CONNECT = "slackConnect"
    BOOKMARKS = "bookmarks"
    CANVASES = "canvases"
    ASSISTANT = "assistant"
    LISTS = "lists"


ACTION_CATEGORY_DEFAULTS: dict[ActionCategory, bool] = {
    ActionCategory.REACTIONS: True,
    ActionCategory.MESSAGES: True,
    ActionCategory.PINS: True,
    ActionCategory.MEMBER_INFO: True,
    ActionCategory.EMOJI_LIST: True,
    ActionCategory.CALLS: False,
    ActionCategory.WORKFLOWS: False,
    ActionCategory.SLACK_CONNECT: False,
    ActionCategory.BOOKMARKS: False,
    ActionCategory.CANVASES: False,
    ActionCategory.ASSISTANT: False,
    ActionCategory.LISTS: False,
}

ACTION_CATEGORY_ACTIONS: dict[ActionCategory, list[str]] = {
    ActionCategory.REACTIONS: ["react", "remove_reaction", "list_reactions", "remove_own_reactions"],
    ActionCategory.MESSAGES: ["send", "read", "edit", "delete", "upload-file", "download-file"],
    ActionCategory.PINS: ["pin", "unpin", "list-pins"],
    ActionCategory.MEMBER_INFO: ["member-info"],
    ActionCategory.EMOJI_LIST: ["emoji-list"],
    ActionCategory.CALLS: ["calls-add", "calls-end", "calls-update"],
    ActionCategory.WORKFLOWS: ["workflow-step-execute", "workflow-step-failed"],
    ActionCategory.SLACK_CONNECT: ["connect-invite", "connect-accept"],
    ActionCategory.BOOKMARKS: ["bookmark-add", "bookmark-edit", "bookmark-list", "bookmark-remove"],
    ActionCategory.CANVASES: ["canvas-create", "canvas-edit", "canvas-delete"],
    ActionCategory.ASSISTANT: ["assistant-thread-start", "assistant-thread-message"],
    ActionCategory.LISTS: ["list-create", "list-edit", "list-delete", "list-item-add"],
}


@dataclass
class MessageAction:
    action_id: str
    label: str
    value: str = ""
    style: str = ""
    type: str = "button"
    uri: str = ""

    def to_dict(self) -> dict[str, Any]:
        result: dict[str, Any] = {"action_id": self.action_id, "label": self.label, "value": self.value}
        if self.style:
            result["style"] = self.style
        if self.type != "button":
            result["type"] = self.type
        if self.uri:
            result["uri"] = self.uri
        return result


@dataclass
class MessageActionsConfig:
    enabled: bool = True
    max_actions_per_message: int = 5
    actions: list[MessageAction] = field(default_factory=list)
    categories: dict[ActionCategory, bool] = field(default_factory=lambda: dict(ACTION_CATEGORY_DEFAULTS))

    def is_category_enabled(self, category: ActionCategory) -> bool:
        return self.categories.get(category, True)

    def is_action_enabled(self, action_name: str) -> bool:
        for category, actions in ACTION_CATEGORY_ACTIONS.items():
            if action_name in actions:
                return self.is_category_enabled(category)
        return self.enabled

    def enabled_categories(self) -> set[ActionCategory]:
        return {c for c, enabled in self.categories.items() if enabled}

    def disabled_categories(self) -> set[ActionCategory]:
        return {c for c, enabled in self.categories.items() if not enabled}

    @classmethod
    def from_config(cls, config: dict[str, Any] | None) -> MessageActionsConfig:
        if not config:
            return cls()
        actions_cfg = config.get("messageActions", {}) or {}
        actions = []
        for a in actions_cfg.get("actions", []):
            if isinstance(a, dict):
                actions.append(
                    MessageAction(
                        action_id=a.get("action_id", ""),
                        label=a.get("label", ""),
                        value=a.get("value", ""),
                        style=a.get("style", ""),
                        type=a.get("type", "button"),
                        uri=a.get("uri", ""),
                    )
                )
        categories: dict[ActionCategory, bool] = {}
        category_config = actions_cfg.get("categories") or actions_cfg.get("actions") or {}
        if isinstance(category_config, dict):
            for cat in ActionCategory:
                cat_val = category_config.get(cat.value)
                if cat_val is not None:
                    categories[cat] = bool(cat_val)
                else:
                    categories[cat] = ACTION_CATEGORY_DEFAULTS.get(cat, True)
        else:
            categories = dict(ACTION_CATEGORY_DEFAULTS)

        return cls(
            enabled=bool(actions_cfg.get("enabled", True)),
            max_actions_per_message=int(actions_cfg.get("maxActionsPerMessage", 5)),
            actions=actions,
            categories=categories,
        )


def gate_message_actions(
    actions_config: MessageActionsConfig,
    *,
    chat_type: str = "",
) -> bool:
    if not actions_config.enabled:
        return False
    if not actions_config.actions:
        return False
    return True


def gate_action_category(
    actions_config: MessageActionsConfig,
    category: ActionCategory,
) -> bool:
    if not actions_config.enabled:
        return False
    return actions_config.is_category_enabled(category)


def gate_action(
    actions_config: MessageActionsConfig,
    action_name: str,
) -> bool:
    if not actions_config.enabled:
        return False
    return actions_config.is_action_enabled(action_name)


def validate_action_allowed(
    actions_config: MessageActionsConfig,
    action_name: str,
) -> tuple[bool, str]:
    if not actions_config.enabled:
        return False, "message actions disabled"
    if not actions_config.is_action_enabled(action_name):
        return False, f"action category for '{action_name}' is disabled"
    return True, ""


def compile_message_actions(
    actions_config: MessageActionsConfig,
    *,
    override_action_id_prefix: str = "",
) -> list[dict[str, Any]]:
    result = []
    for a in actions_config.actions[: actions_config.max_actions_per_message]:
        d = a.to_dict()
        if override_action_id_prefix:
            d["action_id"] = f"{override_action_id_prefix}_{d['action_id']}"
        result.append(d)
    return result