import logging import os from slack_sdk import WebClient from slack_sdk.errors import SlackApiError def format_story(story: dict) -> list: title_text = ( f"<{story['source_url']}|{story['title']}>" if story["source_url"] else story["title"] ) return [ { "type": "section", "text": { "type": "mrkdwn", "text": f"*{title_text}*", }, }, { "type": "context", "elements": [ { "type": "plain_text", "text": f"Categories: {', '.join(story['categories'])}" if story["categories"] else "No categories", } ], }, { "type": "section", "text": { "type": "mrkdwn", "text": story["summary"], }, }, ] def format_slack_blocks(grouped_stories: dict[str, list[dict]]) -> list[dict]: """Format grouped stories into Slack block format.""" blocks = [] # Header block blocks.append( {"type": "header", "text": {"type": "plain_text", "text": "🚀 Tech Updates"}} ) # Add stories for each group for group_name, stories in grouped_stories.items(): # Group section header section_title = ( "*Other Stories*" if group_name == "Other" else f"*{group_name}*" ) blocks.append( {"type": "section", "text": {"type": "mrkdwn", "text": section_title}} ) for story in stories: blocks.extend(format_story(story)) # Add divider after each group (except the last one) blocks.append({"type": "divider"}) return blocks def send_message(channel: str, blocks: list) -> None: client = WebClient(token=os.environ["SLACK_BOT_TOKEN"]) try: response = client.chat_postMessage( channel=channel, text="Tech updates", blocks=blocks, unfurl_links=False, unfurl_media=False, ) response.validate() logging.info(f"Message sent successfully to channel {channel}") except SlackApiError as e: logging.error(f"Error sending message: {e.response['error']}")