slack_utils.py

"""
Slack utilities for message posting and request verification
"""

import hmac
import hashlib
import json
from typing import Optional, List, Dict
import httpx


async def post_slack_message(
        channel_id: str,
        text: str,
        blocks: Optional[List[Dict]] = None,
        thread_ts: Optional[str] = None,
        bot_token: Optional[str] = None
) -> Dict:
    """
    Post a message to a Slack channel

    Args:
        channel_id: Slack channel ID
        text: Text content
        blocks: Optional Slack block Kit blocks
        thread_ts: Optional thread timestamp for threaded messages
        bot_token: Slack bot token (uses env var if not provided)

    Returns:
        Response from Slack API
    """
    import os
    if bot_token is None:
        bot_token = os.getenv("SLACK_BOT_TOKEN")

    payload = {
        "channel": channel_id,
        "text": text
    }

    if blocks:
        payload["blocks"] = blocks

    if thread_ts:
        payload["thread_ts"] = thread_ts

    async with httpx.AsyncClient() as client:
        try:
            print(f"📤 Posting to: {channel_id}")
            print(f"   Text: {text[:50]}..." if len(text) > 50 else f"   Text: {text}")
            print(f"   Token: {bot_token[:20] if bot_token else 'MISSING'}...")

            response = await client.post(
                "https://slack.com/api/chat.postMessage",
                json=payload,
                headers={"Authorization": f"Bearer {bot_token}"},
                timeout=30.0
            )
            result = response.json()

            if not result.get("ok"):
                error = result.get('error', 'Unknown error')
                print(f"⚠️ Slack API error: {error}")
                print(f"   Full response: {result}")
            else:
                print(f"✅ Message posted successfully")

            return result

        except httpx.HTTPError as e:
            print(f"❌ HTTP error posting to Slack: {e}")
            return {"ok": False, "error": str(e)}
        except Exception as e:
            print(f"❌ Error posting to Slack: {e}")
            return {"ok": False, "error": str(e)}


def verify_slack_request(
        body: bytes,
        signature: str,
        timestamp: str,
        signing_secret: str
) -> bool:
    """
    Verify that a request came from Slack

    Args:
        body: Raw request body (bytes)
        signature: x-slack-request-signature header value
        timestamp: x-slack-request-timestamp header value
        signing_secret: Slack app signing secret

    Returns:
        True if signature is valid, False otherwise
    """

    # Check timestamp is within 5 minutes (Slack requirement)
    import time
    current_time = int(time.time())
    request_time = int(timestamp)

    if abs(current_time - request_time) > 300:
        print(f"⚠️ Request timestamp too old: {current_time - request_time} seconds")
        return False

    # Verify signature
    sig_basestring = f"v0:{timestamp}:{body.decode('utf-8')}".encode('utf-8')

    my_signature = (
            "v0=" +
            hmac.new(
                signing_secret.encode('utf-8'),
                sig_basestring,
                hashlib.sha256
            ).hexdigest()
    )

    is_valid = hmac.compare_digest(my_signature, signature)

    if not is_valid:
        print(f"❌ Invalid request signature")

    return is_valid


def parse_slack_payload(payload_json: str) -> Dict:
    """
    Parse Slack interactive payload JSON

    Args:
        payload_json: JSON string from form data

    Returns:
        Parsed payload dictionary
    """
    try:
        return json.loads(payload_json)
    except json.JSONDecodeError as e:
        print(f"❌ Failed to parse Slack payload: {e}")
        return {}


def create_text_block(text: str, block_id: Optional[str] = None) -> Dict:
    """
    Create a simple text block
    """
    return {
        "type": "section",
        "text": {
            "type": "mrkdwn",
            "text": text
        },
        **({"block_id": block_id} if block_id else {})
    }


def create_button_block(
        text: str,
        action_id: str,
        value: str,
        style: str = "primary",
        block_id: Optional[str] = None
) -> Dict:
    """
    Create a button element
    """
    return {
        "type": "button",
        "text": {
            "type": "plain_text",
            "text": text
        },
        "action_id": action_id,
        "value": value,
        "style": style,
        **({"block_id": block_id} if block_id else {})
    }


def create_actions_block(
        elements: List[Dict],
        block_id: Optional[str] = None
) -> Dict:
    """
    Create an actions block with buttons
    """
    return {
        "type": "actions",
        "elements": elements,
        **({"block_id": block_id} if block_id else {})
    }


def create_divider_block() -> Dict:
    """
    Create a divider block
    """
    return {"type": "divider"}


def create_header_block(text: str) -> Dict:
    """
    Create a header block
    """
    return {
        "type": "header",
        "text": {
            "type": "plain_text",
            "text": text
        }
    }


def create_context_block(text: str) -> Dict:
    """
    Create a context block (small text)
    """
    return {
        "type": "context",
        "elements": [
            {
                "type": "mrkdwn",
                "text": text
            }
        ]
    }


# Slack emoji map
EMOJI_MAP = {
    "success": "✅",
    "error": "❌",
    "warning": "⚠️",
    "loading": "⏳",
    "rocket": "🚀",
    "paper": "📄",
    "github": "🔗",
    "database": "📊",
    "clock": "⏰",
    "chart": "📈",
    "code": "💻",
    "gear": "⚙️",
    "thinking": "🤔",
    "tada": "🎉",
    "wave": "👋",
    "down": "⬇️"
}


def get_emoji(name: str) -> str:
    """
    Get emoji by name
    """
    return EMOJI_MAP.get(name.lower(), "•")