Добавлены send_shoutout, send_chat_message, get_user_chat_color

src/twitchclient/api.py

@@ -913,3 +913,235 @@ class TwitchAPIClient(AioHTTPXClient):
 
             case _:
                 raise s.Error(req.status_code, 'Internal Server Error')
+
+    async def get_shared_chat_session(
+        self,
+        access_token: str,
+        broadcaster_id: int | str,
+        cache_time: int | None = None,
+    ):
+        req = await self.get(
+            '/shared_chat/session',
+            headers=self.clean_dict(
+                {
+                    'Authorization': f'Bearer {access_token}',
+                    'X-Cache-TTL': cache_time,
+                }
+            ),
+            params={'broadcaster_id': broadcaster_id},
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.SharedChatSession.model_validate(req.json()).data
+
+            case st.BAD_REQUEST | st.UNAUTHORIZED:
+                raise s.Error(req.status_code, req.json()['message'])
+
+            case _:
+                raise s.Error(req.status_code, 'Internal Server Error')
+
+    async def get_user_emotes(
+        self,
+        access_token: str,
+        user_id: int | str,
+        after: str | None = None,
+        broadcaster_id: int | str | None = None,
+        cache_time: int | None = None,
+    ):
+        req = await self.get(
+            '/chat/emotes/global',
+            headers=self.clean_dict(
+                {
+                    'Authorization': f'Bearer {access_token}',
+                    'X-Cache-TTL': cache_time,
+                }
+            ),
+            params=self.clean_dict(
+                {
+                    'user_id': user_id,
+                    'after': after,
+                    'broadcaster_id': broadcaster_id,
+                }
+            ),
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.UserEmotes.model_validate(req.json())
+
+            case st.UNAUTHORIZED:
+                raise s.Error(req.status_code, req.json()['message'])
+
+            case _:
+                raise s.Error(req.status_code, 'Internal Server Error')
+
+    async def update_chat_settings(
+        self,
+        access_token: str,
+        broadcaster_id: int | str,
+        moderator_id: int | str,
+        *,
+        emote_mode: bool | None = None,
+        follower_mode: bool | None = None,
+        follower_mode_duration: int | None = None,
+        non_moderator_chat_delay: bool | None = None,
+        non_moderator_chat_delay_duration: Literal[2, 4, 6] | None = None,
+        slow_mode: bool | None = None,
+        slow_mode_wait_time: int | None = None,
+        subscriber_mode: bool | None = None,
+        unique_chat_mode: bool | None = None,
+    ):
+        req = await self.put(
+            '/chat/settings',
+            headers={'Authorization': f'Bearer {access_token}'},
+            params={
+                'broadcaster_id': broadcaster_id,
+                'moderator_id': moderator_id,
+            },
+            json=self.clean_dict(
+                {
+                    'emote_mode': emote_mode,
+                    'follower_mode': follower_mode,
+                    'follower_mode_duration': follower_mode_duration,
+                    'non_moderator_chat_delay': non_moderator_chat_delay,
+                    'non_moderator_chat_delay_duration': (
+                        non_moderator_chat_delay_duration
+                    ),
+                    'slow_mode': slow_mode,
+                    'slow_mode_wait_time': slow_mode_wait_time,
+                    'subscriber_mode': subscriber_mode,
+                    'unique_chat_mode': unique_chat_mode,
+                }
+            ),
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.ChatSettings.model_validate(req.json()).data
+
+            case st.BAD_REQUEST | st.UNAUTHORIZED:
+                raise s.Error(req.status_code, req.json()['message'])
+
+            case _:
+                raise s.Error(req.status_code, 'Internal Server Error')
+
+    async def send_chat_announcement(
+        self,
+        access_token: str,
+        broadcaster_id: int | str,
+        moderator_id: int | str,
+        message: str,
+        color: Literal[
+            'blue', 'green', 'orange', 'purple', 'primary'
+        ] = 'primary',
+    ):
+        req = await self.post(
+            '/chat/announcements',
+            headers={'Authorization': f'Bearer {access_token}'},
+            params={
+                'broadcaster_id': broadcaster_id,
+                'moderator_id': moderator_id,
+            },
+            json=self.clean_dict({'message': message, 'color': color}),
+        )
+
+        match req.status_code:
+            case st.NO_CONTENT:
+                return True
+
+            case st.BAD_REQUEST | st.UNAUTHORIZED | st.TOO_MANY_REQUESTS:
+                raise s.Error(req.status_code, req.json()['message'])
+
+            case _:
+                raise s.Error(req.status_code, 'Internal Server Error')
+
+    async def send_shoutout(
+        self,
+        access_token: str,
+        from_broadcaster_id: int | str,
+        to_broadcaster_id: int | str,
+        moderator_id: int | str,
+    ):
+        req = await self.post(
+            '/chat/shoutouts',
+            headers={'Authorization': f'Bearer {access_token}'},
+            params={
+                'from_broadcaster_id': from_broadcaster_id,
+                'to_broadcaster_id': to_broadcaster_id,
+                'moderator_id': moderator_id,
+            },
+        )
+
+        match req.status_code:
+            case st.NO_CONTENT:
+                return True
+
+            case (
+                st.BAD_REQUEST
+                | st.UNAUTHORIZED
+                | st.FORBIDDEN
+                | st.TOO_MANY_REQUESTS
+            ):
+                raise s.Error(req.status_code, req.json()['message'])
+
+            case _:
+                raise s.Error(req.status_code, 'Internal Server Error')
+
+    async def send_chat_message(
+        self,
+        access_token: str,
+        broadcaster_id: int | str,
+        sender_id: int | str,
+        message: str,
+        *,
+        reply_parent_message_id: str | None = None,
+        for_source_only: bool | None = None,
+    ):
+        req = await self.post(
+            '/chat/messages',
+            headers={'Authorization': f'Bearer {access_token}'},
+            json=self.clean_dict(
+                {
+                    'broadcaster_id': broadcaster_id,
+                    'sender_id': sender_id,
+                    'reply_parent_message_id': reply_parent_message_id,
+                    'for_source_only': for_source_only,
+                    'message': message,
+                }
+            ),
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.Message.model_validate(req.json()).data
+
+            case (
+                st.BAD_REQUEST
+                | st.UNAUTHORIZED
+                | st.FORBIDDEN
+                | st.TOO_MANY_REQUESTS
+            ):
+                raise s.Error(req.status_code, req.json()['message'])
+
+            case _:
+                raise s.Error(req.status_code, 'Internal Server Error')
+
+    async def get_user_chat_color(
+        self, access_token: str, user_id: int | str | list[int] | list[str]
+    ):
+        req = await self.get(
+            '/chat/colors',
+            headers={'Authorization': f'Bearer {access_token}'},
+            params={'user_id': user_id},
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.UserChatColor.model_validate(req.json())
+
+            case st.BAD_REQUEST | st.UNAUTHORIZED:
+                raise s.Error(req.status_code, req.json()['message'])
+
+            case _:
+                raise s.Error(req.status_code, 'Internal Server Error')

src/twitchclient/schema.py

@@ -492,7 +492,22 @@ class ChannelEmote(BaseModel):
     name: str
     images: ChannelEmoteImages
     tier: int
-    emote_type: Literal['bitstier', 'follower', 'subscriptions']
+    emote_type: Literal[
+        'none',
+        'bitstier',
+        'follower',
+        'subscriptions',
+        'channelpoints',
+        'rewards',
+        'hypetrain',
+        'prime',
+        'turbo',
+        'smilies',
+        'globals',
+        'owl2019',
+        'twofactor',
+        'limitedtime'
+    ]
     format: list[Literal['animated', 'static']]
     scale: list[Literal['1.0', '2.0', '3.0']]
     theme_mode: list[Literal['dark', 'light']]
@@ -563,3 +578,65 @@ class ChatSettings(BaseModel):
     model_config = ConfigDict(extra='forbid')
 
     data: list[ChatSettingsData]
+
+
+class SharedChatSessionParticipant(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    broadcaster_id: int
+
+
+class SharedChatSessionData(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    session_id: str
+    host_broadcaster_id: int
+    participants: list[SharedChatSessionParticipant]
+    created_at: datetime
+    updated_at: datetime
+
+
+class SharedChatSession(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    data: list[SharedChatSessionData]
+
+
+class UserEmotes(ChannelEmotes):
+    pagination: Pagination | dict[Any, Any] | None = None
+
+
+class MessageDropReason(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    code: str
+    message: str
+
+
+class MessageData(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    message_id: str
+    is_sent: bool
+    drop_reason: MessageDropReason | None = None
+
+
+class Message(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    data: list[MessageData]
+
+
+class UserChatColorData(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    user_id: int
+    user_login: str
+    user_name: str
+    color: str
+
+
+class UserChatColor(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    data: list[UserChatColorData]