""" 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(), "•")