Добавлен get_chat_settings

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "twitchclient"
-version = "0.1.0"
+version = "0.2.0"
 description = "Client for Twitch API"
 readme = "README.md"
 authors = [

src/twitchclient/api.py

@@ -29,7 +29,9 @@ class TwitchAPIClient(AioHTTPXClient):
             logger='Twitch API',
         )
 
-    async def start_commercial(self, access_token: str, broadcaster_id: int):
+    async def start_commercial(
+        self, access_token: str, broadcaster_id: int | str
+    ):
         req = await self.get(
             '/channels/commercial',
             headers={'Authorization': f'Bearer {access_token}'},
@@ -53,7 +55,9 @@ class TwitchAPIClient(AioHTTPXClient):
             case _:
                 raise s.Error(req.status_code, 'Internal Server Error')
 
-    async def get_ad_schedule(self, access_token: str, broadcaster_id: int):
+    async def get_ad_schedule(
+        self, access_token: str, broadcaster_id: int | str
+    ):
         req = await self.get(
             '/channels/ads',
             headers={'Authorization': f'Bearer {access_token}'},
@@ -72,7 +76,9 @@ class TwitchAPIClient(AioHTTPXClient):
             case _:
                 raise s.Error(req.status_code, 'Internal Server Error')
 
-    async def snooze_next_ad(self, access_token: str, broadcaster_id: int):
+    async def snooze_next_ad(
+        self, access_token: str, broadcaster_id: int | str
+    ):
         req = await self.post(
             '/channels/ads/schedule/snooze',
             headers={'Authorization': f'Bearer {access_token}'},
@@ -193,7 +199,7 @@ class TwitchAPIClient(AioHTTPXClient):
                 raise s.Error(req.status_code, 'Internal Server Error')
 
     async def get_cheermotes(
-        self, access_token: str, broadcaster_id: int | None = None
+        self, access_token: str, broadcaster_id: int | str | None = None
     ):
         req = await self.get(
             '/bits/cheermotes',
@@ -243,7 +249,9 @@ class TwitchAPIClient(AioHTTPXClient):
                 raise s.Error(req.status_code, 'Internal Server Error')
 
     async def get_channel_information(
-        self, access_token: str, broadcaster_id: int | list[int]
+        self,
+        access_token: str,
+        broadcaster_id: int | list[int] | str | list[str],
     ):
         req = await self.get(
             '/channels',
@@ -264,9 +272,9 @@ class TwitchAPIClient(AioHTTPXClient):
     async def modify_channel_information(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         *,
-        game_id: int | None = None,
+        game_id: int | str | None = None,
         broadcaster_language: str | None = None,
         title: str | None = None,
         delay: int | None = None,
@@ -309,7 +317,7 @@ class TwitchAPIClient(AioHTTPXClient):
                 raise s.Error(req.status_code, 'Internal Server Error')
 
     async def get_channel_editors(
-        self, access_token: str, broadcaster_id: int
+        self, access_token: str, broadcaster_id: int | str
     ):
         req = await self.get(
             '/channels/editors',
@@ -330,7 +338,7 @@ class TwitchAPIClient(AioHTTPXClient):
     async def get_followed_channels(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         *,
         first: int = 20,
         after: str | None = None,
@@ -360,7 +368,7 @@ class TwitchAPIClient(AioHTTPXClient):
     async def get_channel_followers(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         *,
         user_id: int | None = None,
         first: int = 20,
@@ -392,7 +400,7 @@ class TwitchAPIClient(AioHTTPXClient):
     async def create_custom_rewards(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         title: str,
         cost: int,
         *,
@@ -451,7 +459,7 @@ class TwitchAPIClient(AioHTTPXClient):
                 raise s.Error(req.status_code, 'Internal Server Error')
 
     async def delete_custom_reward(
-        self, access_token: str, broadcaster_id: str, reward_id: str
+        self, access_token: str, broadcaster_id: int | str, reward_id: str
     ):
         req = await self.delete(
             '/channel_points/custom_rewards',
@@ -478,7 +486,7 @@ class TwitchAPIClient(AioHTTPXClient):
     async def get_custom_rewards(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         *,
         reward_id: str | None = None,
         only_manageable_rewards: bool = False,
@@ -514,7 +522,7 @@ class TwitchAPIClient(AioHTTPXClient):
     async def get_custom_reward_redemption(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         reward_id: str,
         status: Literal['CANCELED', 'FULFILLED', 'UNFULFILLED'],
         *,
@@ -559,7 +567,7 @@ class TwitchAPIClient(AioHTTPXClient):
     async def update_custom_reward(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         reward_id: str,
         *,
         title: str | None = None,
@@ -623,7 +631,7 @@ class TwitchAPIClient(AioHTTPXClient):
     async def update_redemption_status(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         reward_id: int,
         redemption_id: int | list[int],
         status: Literal['CANCELED', 'FULFILLED'],
@@ -656,7 +664,7 @@ class TwitchAPIClient(AioHTTPXClient):
                 raise s.Error(req.status_code, 'Internal Server Error')
 
     async def get_charity_campaign(
-        self, access_token: str, broadcaster_id: int
+        self, access_token: str, broadcaster_id: int | str
     ):
         req = await self.get(
             '/charity/campaigns',
@@ -677,7 +685,7 @@ class TwitchAPIClient(AioHTTPXClient):
     async def get_charity_campaign_donations(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         first: int = 20,
         after: str | None = None,
         cache_time: int | None = None,
@@ -712,7 +720,7 @@ class TwitchAPIClient(AioHTTPXClient):
     async def get_chatters(
         self,
         access_token: str,
-        broadcaster_id: int,
+        broadcaster_id: int | str,
         moderator_id: int,
         first: int = 20,
         after: str | None = None,
@@ -745,3 +753,163 @@ class TwitchAPIClient(AioHTTPXClient):
 
             case _:
                 raise s.Error(req.status_code, 'Internal Server Error')
+
+    async def get_channel_emotes(
+        self,
+        access_token: str,
+        broadcaster_id: int | str,
+        cache_time: int | None = None,
+    ):
+        req = await self.get(
+            '/chat/emotes',
+            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.ChannelEmotes.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')
+
+    async def get_global_emotes(
+        self, access_token: str, 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,
+                }
+            ),
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.GlobalEmotes.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 get_emote_sets(
+        self,
+        access_token: str,
+        emote_set_id: int,
+        cache_time: int | None = None,
+    ):
+        req = await self.get(
+            '/chat/emotes/set',
+            headers=self.clean_dict(
+                {
+                    'Authorization': f'Bearer {access_token}',
+                    'X-Cache-TTL': cache_time,
+                }
+            ),
+            params={'emote_set_id': emote_set_id},
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.EmoteSets.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')
+
+    async def get_channel_chat_badges(
+        self,
+        access_token: str,
+        broadcaster_id: int | str,
+        cache_time: int | None = None,
+    ):
+        req = await self.get(
+            '/chat/badges',
+            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.ChannelChatBadges.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')
+
+    async def get_global_chat_badges(
+        self, access_token: str, cache_time: int | None = None
+    ):
+        req = await self.get(
+            '/chat/badges/global',
+            headers=self.clean_dict(
+                {
+                    'Authorization': f'Bearer {access_token}',
+                    'X-Cache-TTL': cache_time,
+                }
+            ),
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.GlobalChatBadges.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 get_chat_settings(
+        self,
+        access_token: str,
+        broadcaster_id: int | str,
+        moderator_id: int | str | None = None,
+        cache_time: int | None = None,
+    ):
+        req = await self.get(
+            '/chat/settings',
+            headers=self.clean_dict(
+                {
+                    'Authorization': f'Bearer {access_token}',
+                    'X-Cache-TTL': cache_time,
+                }
+            ),
+            params=self.clean_dict(
+                {
+                    'broadcaster_id': broadcaster_id,
+                    'moderator_id': moderator_id,
+                }
+            ),
+        )
+
+        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')

src/twitchclient/schema.py

@@ -475,3 +475,91 @@ class Chatters(BaseModel):
     data: list[ChattersData]
     pagination: Pagination | dict[Any, Any] | None = None
     total: int
+
+
+class ChannelEmoteImages(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    url_1x: str
+    url_2x: str
+    url_4x: str
+
+
+class ChannelEmote(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    id: int
+    name: str
+    images: ChannelEmoteImages
+    tier: int
+    emote_type: Literal['bitstier', 'follower', 'subscriptions']
+    format: list[Literal['animated', 'static']]
+    scale: list[Literal['1.0', '2.0', '3.0']]
+    theme_mode: list[Literal['dark', 'light']]
+
+
+class ChannelEmotes(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    data: list[ChannelEmote]
+    template: str
+
+
+class GlobalEmotes(ChannelEmotes):
+    pass
+
+
+class EmoteSets(ChannelEmotes):
+    pass
+
+
+class ChannelChatBadgesVersion(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    id: str
+    image_url_1x: str
+    image_url_2x: str
+    image_url_4x: str
+    title: str
+    description: str
+    click_action: str | None
+    click_url: str | None
+
+
+class ChannelChatBadgesData(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    set_id: str
+    versions: list[ChannelChatBadgesVersion]
+
+
+class ChannelChatBadges(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    data: list[ChannelChatBadgesData]
+
+
+class GlobalChatBadges(ChannelChatBadges):
+    pass
+
+
+class ChatSettingsData(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    broadcaster_id: int
+    emote_mode: bool
+    follower_mode: bool
+    follower_mode_duration: int | None
+    moderator_id: int | None = None
+    non_moderator_chat_delay: bool | None = None
+    non_moderator_chat_delay_duration: int | None = None
+    slow_mode: bool
+    slow_mode_wait_time: int | None
+    subscriber_mode: bool
+    unique_chat_mode: bool
+
+
+class ChatSettings(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    data: list[ChatSettingsData]