diff --git a/packages/api/src/microsoft/teams/api/activities/message/message.py b/packages/api/src/microsoft/teams/api/activities/message/message.py index 01af550c..bd745719 100644 --- a/packages/api/src/microsoft/teams/api/activities/message/message.py +++ b/packages/api/src/microsoft/teams/api/activities/message/message.py @@ -25,6 +25,16 @@ SuggestedActions, TextFormat, ) +from ...models.entity import ( + AIMessageEntity, + Appearance, + CitationAppearance, + CitationEntity, + Claim, + Entity, + Image, + MessageEntity, +) from ..utils import StripMentionsTextOptions, strip_mentions_text @@ -396,3 +406,75 @@ def add_stream_final(self) -> Self: stream_entity = StreamInfoEntity(type="streaminfo", stream_id=self.id, stream_type="final") return self.add_entity(stream_entity) + + def add_ai_generated(self) -> Self: + """Add the 'Generated By AI' label.""" + message_entity = self._ensure_single_root_level_message_entity() + ai_entity = AIMessageEntity(**message_entity.model_dump()) + if ai_entity.additional_type and "AIGeneratedContent" in ai_entity.additional_type: + return self + + if not ai_entity.additional_type: + ai_entity.additional_type = [] + + ai_entity.additional_type.append("AIGeneratedContent") + + self._update_entity(message_entity, ai_entity) + + return self + + def add_feedback(self) -> Self: + """Enable message feedback.""" + if not self.channel_data: + self.channel_data = ChannelData(feedback_loop_enabled=True) + else: + self.channel_data.feedback_loop_enabled = True + return self + + def add_citation(self, position: int, appearance: CitationAppearance) -> Self: + """Add citations.""" + message_entity = self._ensure_single_root_level_message_entity() + citation_entity = CitationEntity(**message_entity.model_dump()) + if citation_entity.citation is None: + citation_entity.citation = [] + + citation_entity.citation.append( + Claim( + position=position, + appearance=Appearance( + abstract=appearance.abstract, + name=appearance.name, + image=Image(name=appearance.icon) if appearance.icon else None, + keywords=appearance.keywords, + text=appearance.text, + url=appearance.url, + usage_info=appearance.usage_info, + ), + ) + ) + + self._update_entity(message_entity, citation_entity) + + return self + + def _ensure_single_root_level_message_entity(self) -> MessageEntity: + """ + Get or create the base message entity. + There should only be one root level message entity. + """ + message_entity = next( + (e for e in (self.entities or []) if e.type == "https://schema.org/Message" and e.at_type == "Message"), + None, + ) + + if not message_entity: + message_entity = MessageEntity() + self.add_entity(message_entity) + + return message_entity + + def _update_entity(self, old_entity: Entity, new_entity: Entity) -> None: + if self.entities is not None: + index = self.entities.index(old_entity) + self.entities.pop(index) + self.entities.insert(index, new_entity) diff --git a/packages/api/src/microsoft/teams/api/models/activity.py b/packages/api/src/microsoft/teams/api/models/activity.py index 86002a66..573cdd64 100644 --- a/packages/api/src/microsoft/teams/api/models/activity.py +++ b/packages/api/src/microsoft/teams/api/models/activity.py @@ -13,16 +13,7 @@ from microsoft.teams.api.models.channel_data.team_info import TeamInfo from microsoft.teams.api.models.channel_id import ChannelID from microsoft.teams.api.models.conversation.conversation_reference import ConversationReference -from microsoft.teams.api.models.entity.ai_message_entity import AIMessageEntity -from microsoft.teams.api.models.entity.citation_entity import ( - Appearance, - CitationAppearance, - CitationEntity, - Claim, - Image, -) from microsoft.teams.api.models.entity.entity import Entity -from microsoft.teams.api.models.entity.message_entity import MessageEntity from microsoft.teams.api.models.meetings.meeting_info import MeetingInfo from .custom_base_model import CustomBaseModel @@ -197,82 +188,10 @@ def add_entities(self, *values: Entity) -> Self: self.entities.extend(values) return self - def add_ai_generated(self) -> Self: - """Add the 'Generated By AI' label.""" - message_entity = self.ensure_single_root_level_message_entity() - ai_entity = AIMessageEntity(**message_entity.model_dump()) - if ai_entity.additional_type and "AIGeneratedContent" in ai_entity.additional_type: - return self - - if not ai_entity.additional_type: - ai_entity.additional_type = [] - - ai_entity.additional_type.append("AIGeneratedContent") - - self._update_entity(message_entity, ai_entity) - - return self - - def add_feedback(self) -> Self: - """Enable message feedback.""" - if not self.channel_data: - self.channel_data = ChannelData(feedback_loop_enabled=True) - else: - self.channel_data.feedback_loop_enabled = True - return self - - def add_citation(self, position: int, appearance: CitationAppearance) -> Self: - """Add citations.""" - message_entity = self.ensure_single_root_level_message_entity() - citation_entity = CitationEntity(**message_entity.model_dump()) - if citation_entity.citation is None: - citation_entity.citation = [] - - citation_entity.citation.append( - Claim( - position=position, - appearance=Appearance( - abstract=appearance.abstract, - name=appearance.name, - image=Image(name=appearance.icon) if appearance.icon else None, - keywords=appearance.keywords, - text=appearance.text, - url=appearance.url, - usage_info=appearance.usage_info, - ), - ) - ) - - self._update_entity(message_entity, citation_entity) - - return self - def is_streaming(self) -> bool: """Check if this is a streaming activity.""" return bool(self.entities and any(e.type == "streaminfo" for e in self.entities or [])) - def ensure_single_root_level_message_entity(self) -> MessageEntity: - """ - Get or create the base message entity. - There should only be one root level message entity. - """ - message_entity = next( - (e for e in (self.entities or []) if e.type == "https://schema.org/Message" and e.at_type == "Message"), - None, - ) - - if not message_entity: - message_entity = MessageEntity() - self.add_entity(message_entity) - - return message_entity - - def _update_entity(self, old_entity: Entity, new_entity: Entity) -> None: - if self.entities is not None: - index = self.entities.index(old_entity) - self.entities.pop(index) - self.entities.insert(index, new_entity) - class Activity(_ActivityBase): """Output model for received activities with required fields and read-only properties.""" diff --git a/packages/api/tests/unit/test_activity.py b/packages/api/tests/unit/test_activity.py index 6ec68263..d44f7424 100644 --- a/packages/api/tests/unit/test_activity.py +++ b/packages/api/tests/unit/test_activity.py @@ -8,6 +8,7 @@ from typing import cast import pytest +from microsoft.teams.api.activities.message import MessageActivityInput from microsoft.teams.api.models import ( Account, ActivityInputBase, @@ -57,6 +58,18 @@ def test_activity(user: Account, bot: Account, chat: ConversationAccount) -> Con return activity +@pytest.fixture +def message_activity(user: Account, bot: Account, chat: ConversationAccount) -> MessageActivityInput: + """Create a message activity for testing AI-specific features.""" + activity = MessageActivityInput( + id="1", + from_=user, + conversation=chat, + recipient=bot, + ) + return activity + + @pytest.mark.unit class TestActivity: """Unit tests for Activity class.""" @@ -127,34 +140,34 @@ def test_should_have_channel_data_accessors( assert activity.meeting and activity.meeting.id == "meeting-id" assert activity.notification and activity.notification.alert is True - def test_should_add_ai_label(self, test_activity: ConcreteTestActivity) -> None: - activity = test_activity.add_ai_generated() + def test_should_add_ai_label(self, message_activity: MessageActivityInput) -> None: + activity = message_activity.add_ai_generated() - assert activity.type == "test" + assert activity.type == "message" assert activity.entities and len(activity.entities) == 1 message_entity = cast(MessageEntity, activity.entities[0]) assert message_entity.additional_type and message_entity.additional_type[0] == "AIGeneratedContent" - def test_should_add_feedback_label(self, test_activity: ConcreteTestActivity) -> None: - activity = test_activity.add_feedback() + def test_should_add_feedback_label(self, message_activity: MessageActivityInput) -> None: + activity = message_activity.add_feedback() - assert activity.type == "test" + assert activity.type == "message" assert activity.channel_data and activity.channel_data.feedback_loop_enabled is True - def test_should_add_citation(self, test_activity: ConcreteTestActivity) -> None: - activity = test_activity.add_citation(0, CitationAppearance(abstract="test", name="test")) + def test_should_add_citation(self, message_activity: MessageActivityInput) -> None: + activity = message_activity.add_citation(0, CitationAppearance(abstract="test", name="test")) - assert activity.type == "test" + assert activity.type == "message" assert activity.entities and len(activity.entities) == 1 citation_entity = cast(CitationEntity, activity.entities[0]) assert citation_entity.citation and len(citation_entity.citation) == 1 - def test_should_add_citation_with_icon(self, test_activity: ConcreteTestActivity) -> None: - activity = test_activity.add_citation( + def test_should_add_citation_with_icon(self, message_activity: MessageActivityInput) -> None: + activity = message_activity.add_citation( 0, CitationAppearance(abstract="test", name="test", icon=CitationIconName.GIF) ) - assert activity.type == "test" + assert activity.type == "message" assert activity.entities and len(activity.entities) == 1 citation_entity = cast(CitationEntity, activity.entities[0]) assert citation_entity.citation and len(citation_entity.citation) == 1