Remove channel. from EventSub dispatched events

Author: Eviee Py

twitchio/eventsub/websockets.py

@@ -48,6 +48,7 @@
     MISSING,
     _from_json,  # type: ignore
 )
+from .subscriptions import _SUB_MAPPING
 
 
 if TYPE_CHECKING:
@@ -408,11 +409,11 @@ async def _process_revocation(self, data: RevocationMessage) -> None:
         self._subscriptions.pop(payload.id, None)
 
     async def _process_notification(self, data: NotificationMessage) -> None:
-        subscription_type = data["metadata"]["subscription_type"]
-        event: str = subscription_type.replace("channel.channel_", "channel.").replace(".", "_")
+        sub_type = data["metadata"]["subscription_type"]
+        event = _SUB_MAPPING.get(sub_type, sub_type.removeprefix("channel."))
 
         try:
-            payload_class = create_event_instance(subscription_type, data["payload"]["event"], http=self._http)
+            payload_class = create_event_instance(sub_type, data["payload"]["event"], http=self._http)
         except ValueError:
             logger.warning("Websocket '%s' received an unhandled eventsub event: '%s'.", self, event)
             return

twitchio/web/aio_adapter.py

@@ -34,6 +34,7 @@
 from aiohttp import web
 
 from ..authentication import Scopes
+from ..eventsub.subscriptions import _SUB_MAPPING
 from ..models.eventsub_ import SubscriptionRevoked, create_event_instance
 from ..types_.eventsub import EventSubHeaders
 from ..utils import _from_json, parse_timestamp  # type: ignore
@@ -176,11 +177,11 @@ async def eventsub_callback(self, request: web.Request) -> web.Response:
             return web.Response(text=data["challenge"], status=200, headers={"Content-Type": "text/plain"})
 
         elif msg_type == "notification":
-            subscription_type: str = data["subscription"]["type"]
-            event: str = subscription_type.replace("channel.channel_", "channel.").replace(".", "_")
+            sub_type: str = data["subscription"]["type"]
+            event = _SUB_MAPPING.get(sub_type, sub_type.removeprefix("channel."))
 
             try:
-                payload_class = create_event_instance(subscription_type, data["event"], http=self.client._http)
+                payload_class = create_event_instance(sub_type, data["event"], http=self.client._http)
             except ValueError:
                 logger.warning("Webhook '%s' received an unhandled eventsub event: '%s'.", self, event)
                 return web.Response(status=200)

twitchio/web/starlette_adapter.py

@@ -37,6 +37,7 @@
 from starlette.routing import Route
 
 from ..authentication import Scopes
+from ..eventsub.subscriptions import _SUB_MAPPING
 from ..models.eventsub_ import SubscriptionRevoked, create_event_instance
 from ..types_.eventsub import EventSubHeaders
 from ..utils import _from_json, parse_timestamp  # type: ignore
@@ -189,11 +190,11 @@ async def eventsub_callback(self, request: Request) -> Response:
             return Response(data["challenge"], status_code=200, headers={"Content-Type": "text/plain"})
 
         elif msg_type == "notification":
-            subscription_type: str = data["subscription"]["type"]
-            event: str = subscription_type.replace("channel.channel_", "channel.").replace(".", "_")
+            sub_type: str = data["subscription"]["type"]
+            event = _SUB_MAPPING.get(sub_type, sub_type.removeprefix("channel."))
 
             try:
-                payload_class = create_event_instance(subscription_type, data["event"], http=self.client._http)
+                payload_class = create_event_instance(sub_type, data["event"], http=self.client._http)
             except ValueError:
                 logger.warning("Webhook '%s' received an unhandled eventsub event: '%s'.", self, event)
                 return Response(status_code=200)