from typing import Tuple from botbuilder.core import ActivityHandler, TurnContext from botbuilder.schema import ChannelAccount, Mention, Activity, ActivityTypes from config import DefaultConfig from commands.webhook import WebhookCommand from commands.test import TestCommand from modules.database import DatabaseManager import re class WebhookBot(ActivityHandler): """Webhook bot for Microsoft Teams.""" def __init__(self, config: DefaultConfig, database: DatabaseManager) -> None: super().__init__() self.command_prefix = DefaultConfig.COMMAND_PREFIX self.webhook_manager = WebhookCommand(database) self.database = database def is_command(self, text: str) -> tuple[bool, str | None]: """Check if the text is a command.""" if text.startswith(self.command_prefix) and len(text) > len(self.command_prefix): return True, text[len(self.command_prefix):].split(" ")[0] return False, None def has_sub_command(self, text: str) -> tuple[bool, str | None]: """Check if the text has a sub command.""" split_text = text.split(" ") if len(split_text) > 1: return True, split_text[1] return False, None def is_bot_mentioned(self, turn_context: TurnContext) -> bool: """Check if the bot is mentioned in the activity.""" activity = turn_context.activity # Get bot ID from recipient if activity.recipient is None: return False bot_id = activity.recipient.id # Check entities for mentions if activity.entities is None: return False for entity in activity.entities: if entity.type == "mention": try: # Check if entity has mentioned property (ChannelAccount) if hasattr(entity, 'mentioned') and entity.mentioned: mentioned_id = getattr(entity.mentioned, 'id', None) if mentioned_id == bot_id: return True # Check properties dict (alternative format) if hasattr(entity, 'properties') and entity.properties: props = entity.properties if isinstance(props, dict): mentioned = props.get('mentioned', {}) if isinstance(mentioned, dict): if mentioned.get('id') == bot_id: # type: ignore return True except Exception as e: continue return False def is_channel_conversation(self, turn_context: TurnContext) -> bool: """Check if this is a channel conversation.""" activity = turn_context.activity if activity.conversation is None: return False # In Teams, channel conversations have conversation_type == "channel" return getattr(activity.conversation, 'conversation_type', None) == "channel" def remove_mentions_from_text(self, text: str, activity) -> str: """Remove mention text from the message text.""" # Remove all between and return re.sub(r'.*?', '', text) def send_message(self, channel_id: str, message: str) -> None: channel_account = ChannelAccount(id=channel_id) activity = Activity( type=ActivityTypes.message, text=message, channel_id=channel_id, from_property=channel_account ) async def on_command(self, command: str, turn_context: TurnContext) -> None: """Handle the command.""" await turn_context.send_activity(f"Command {command} received.") return async def on_members_added_activity(self, members_added: list[ChannelAccount], turn_context: TurnContext) -> None: """Handle the members added activity.""" if turn_context.activity.recipient is None: return for member in members_added: if member.id != turn_context.activity.recipient.id: await turn_context.send_activity("Hello and welcome!") return async def on_channel_created_activity(self, turn_context: TurnContext) -> None: """Handle the channel created activity.""" if turn_context.activity.recipient is None: return await turn_context.send_activity(f"Channel created: {turn_context.activity.channel_id}") return async def on_message_activity(self, turn_context: TurnContext) -> None: """Handle the message activity - works for both 1:1 and channel posts.""" activity = turn_context.activity # Check if this is a channel conversation is_channel = self.is_channel_conversation(turn_context) print(f"[Bot] Is channel conversation: {is_channel}") if not is_channel: print(f"[Bot] Not a channel conversation, ignoring message") return # Get text from activity - can be None for channel posts with only attachments text = activity.text or "" # Remove mentions from text for processing text = self.remove_mentions_from_text(text, activity) # Strip whitespace and check if it's a command text = text.strip() print(f"[Bot] Processed text (after removing mentions): '{text}'") is_command, command = self.is_command(text) print(f"[Bot] Is command: {is_command}, Command: {command}") if is_command: match command: case "test": # Get the text behind the command /test # Remove the command from the text message = text.replace(f"/test ", "").strip() if message == "": await turn_context.send_activity(f"Please provide a message for the test command.") return await TestCommand(self.database).handle_test(turn_context, message) return case "webhooks": is_sub_command, sub_command = self.has_sub_command(text) if is_sub_command: match sub_command: case "list": await self.webhook_manager.handle_list_webhooks(turn_context) return case _: await turn_context.send_activity(f"Sub command {sub_command} not found.") return else: # Default to list if no sub-command specified await self.webhook_manager.handle_list_webhooks(turn_context) return case _: await turn_context.send_activity(f"Command {command} not found.") return else: print(f"[Bot] Not a command, echoing message") await turn_context.send_activity(f"Message: {text}") return