Implement a way to send watch events through GQL

channel.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+import gzip
 import json
 import asyncio
 import logging
@@ -11,14 +12,14 @@ from typing import Any, SupportsInt, cast, TYPE_CHECKING
 import aiohttp
 from yarl import URL
 
-from utils import Game, json_minify
+from utils import Game, json_minify, isonow
 from exceptions import MinerException, RequestException
-from constants import CALL, GQL_OPERATIONS, ONLINE_DELAY, URLType
+from constants import CALL, GQL_QUERIES, ONLINE_DELAY, URLType, GQLQuery
 
 if TYPE_CHECKING:
     from twitch import Twitch
     from gui import ChannelList
-    from constants import JsonType, GQLOperation
+    from constants import JsonType, GQLPersistedQuery
 
 
 logger = logging.getLogger("TwitchDrops")
@@ -63,6 +64,36 @@ class Stream:
         ]
         return {"data": (b64encode(json_minify(payload).encode("utf8"))).decode("utf8")}
 
+    @cached_property
+    def _gql_payload(self) -> GQLQuery:
+        payload = [
+            {
+                "event": "minute-watched",
+                "properties": {
+                    "broadcast_id": str(self.broadcast_id),
+                    "channel_id": str(self.channel.id),
+                    "channel": self.channel._login,
+                    "client_time": isonow(),
+                    "game": self.game.name if self.game is not None else "",
+                    "game_id": str(self.game.id) if self.game is not None else "",
+                    "hidden": False,
+                    "is_live": True,
+                    "live": True,
+                    "logged_in": True,
+                    "minutes_logged": 1,
+                    "muted": False,
+                    "user_id": self.channel._twitch._auth_state.user_id,
+                }
+            }
+        ]
+        return GQLQuery(
+            (
+                "\n mutation SendEvents($input: SendSpadeEventsInput!) "
+                "{\n sendSpadeEvents(input: $input) {\n statusCode\n}\n}\n"
+            ),
+            b64encode(gzip.compress(json_minify(payload).encode("utf8"))).decode("utf8")
+        )
+
     @classmethod
     def from_get_stream(cls, channel: Channel, channel_data: JsonType) -> Stream:
         stream = channel_data["stream"]
@@ -99,7 +130,7 @@ class Stream:
             return self._stream_url
         # get the stream playback access token from GQL
         playback_token_response: JsonType = await self.channel._twitch.gql_request(
-            GQL_OPERATIONS["PlaybackAccessToken"].with_variables({"login": self.channel._login})
+            GQL_QUERIES["PlaybackAccessToken"].with_variables({"login": self.channel._login})
         )
         token_data: JsonType = playback_token_response["data"]["streamPlaybackAccessToken"]
         token_value = token_data["value"]
@@ -203,8 +234,8 @@ class Channel:
         return self.id
 
     @property
-    def stream_gql(self) -> GQLOperation:
+    def stream_gql(self) -> GQLPersistedQuery:
-        return GQL_OPERATIONS["GetStreamInfo"].with_variables({"channel": self._login})
+        return GQL_QUERIES["GetStreamInfo"].with_variables({"channel": self._login})
 
     @property
     def name(self) -> str:
@@ -342,7 +373,7 @@ class Channel:
         if not stream.drops_enabled:
             try:
                 available_drops_campaigns: JsonType = await self._twitch.gql_request(
-                    GQL_OPERATIONS["AvailableDrops"].with_variables({"channelID": str(self.id)})
+                    GQL_QUERIES["AvailableDrops"].with_variables({"channelID": str(self.id)})
                 )
             except MinerException:
                 logger.log(CALL, f"AvailableDrops GQL call failed for channel: {self._login}")
@@ -408,7 +439,7 @@ class Channel:
             self.display()
 
     # NOTE: This is currently unused.
-    async def _send_watch(self) -> bool:
+    async def _send_watch_playlist(self) -> bool:
         """
         This performs a HEAD request on the stream's current playlist,
         to simulate watching the stream.
@@ -460,7 +491,8 @@ class Channel:
         async with self._twitch.request("HEAD", stream_chunk_url) as head_response:
             return head_response.status == 200
 
-    async def send_watch(self) -> bool:
+    # NOTE: This is currently unused.
+    async def _send_watch_spade(self) -> bool:
         if self._stream is None:
             return False
         if self._spade_url is None:
@@ -472,3 +504,12 @@ class Channel:
                 return response.status == 204
         except RequestException:
             return False
+
+    async def send_watch(self) -> bool:  # send_watch_gql
+        if self._stream is None:
+            return False
+        try:
+            watch_response = await self._twitch.gql_request(self._stream._gql_payload)
+            return watch_response["data"]["sendSpadeEvents"]["statusCode"] == 204
+        except RequestException:
+            return False

constants.py

@@ -110,6 +110,7 @@ SETTINGS_PATH = Path(WORKING_DIR, "settings.json")
 # Typing
 JsonType = Dict[str, Any]
 URLType = NewType("URLType", str)
+GQLOperation: TypeAlias = "GQLQuery | GQLPersistedQuery"
 TopicProcess: TypeAlias = "abc.Callable[[int, JsonType], Any]"
 # Values
 MAX_INT = sys.maxsize
@@ -264,7 +265,21 @@ class PriorityMode(Enum):
     LOW_AVBL_FIRST = 2
 
 
-class GQLOperation(JsonType):
+class GQLQuery(JsonType):
+    def __init__(self, query: str, g64data: str):
+        super().__init__(
+            query=query,
+            variables={
+                "input": {
+                    "data": g64data,
+                    "repository": "twilight",
+                    "encoding": "GZIP_B64",
+                }
+            }
+        )
+
+
+class GQLPersistedQuery(JsonType):
     def __init__(self, name: str, sha256: str, *, variables: JsonType | None = None):
         super().__init__(
             operationName=name,
@@ -278,7 +293,7 @@ class GQLOperation(JsonType):
         if variables is not None:
             self.__setitem__("variables", variables)
 
-    def with_variables(self, variables: JsonType) -> GQLOperation:
+    def with_variables(self, variables: JsonType) -> GQLPersistedQuery:
         modified = deepcopy(self)
         if "variables" in self:
             existing_variables: JsonType = modified["variables"]
@@ -288,9 +303,9 @@ class GQLOperation(JsonType):
         return modified
 
 
-GQL_OPERATIONS: dict[str, GQLOperation] = {
+GQL_QUERIES: dict[str, GQLPersistedQuery] = {
     # returns stream information for a particular channel
-    "GetStreamInfo": GQLOperation(
+    "GetStreamInfo": GQLPersistedQuery(
         "VideoPlayerStreamInfoOverlayChannel",
         "198492e0857f6aedead9665c81c5a06d67b25b58034649687124083ff288597d",
         variables={
@@ -298,7 +313,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         },
     ),
     # can be used to claim channel points
-    "ClaimCommunityPoints": GQLOperation(
+    "ClaimCommunityPoints": GQLPersistedQuery(
         "ClaimCommunityPoints",
         "46aaeebe02c99afdf4fc97c7c0cba964124bf6b0af229395f1f6d1feed05b3d0",
         variables={
@@ -309,7 +324,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         },
     ),
     # can be used to claim a drop
-    "ClaimDrop": GQLOperation(
+    "ClaimDrop": GQLPersistedQuery(
         "DropsPage_ClaimDropRewards",
         "a455deea71bdc9015b78eb49f4acfbce8baa7ccbedd28e549bb025bd0f751930",
         variables={
@@ -319,7 +334,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         },
     ),
     # returns current state of points (balance, claim available) for a particular channel
-    "ChannelPointsContext": GQLOperation(
+    "ChannelPointsContext": GQLPersistedQuery(
         "ChannelPointsContext",
         "374314de591e69925fce3ddc2bcf085796f56ebb8cad67a0daa3165c03adc345",
         variables={
@@ -327,7 +342,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         },
     ),
     # returns all in-progress campaigns
-    "Inventory": GQLOperation(
+    "Inventory": GQLPersistedQuery(
         "Inventory",
         "d86775d0ef16a63a33ad52e80eaff963b2d5b72fada7c991504a57496e1d8e4b",
         variables={
@@ -335,7 +350,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         }
     ),
     # returns current state of drops (current drop progress)
-    "CurrentDrop": GQLOperation(
+    "CurrentDrop": GQLPersistedQuery(
         "DropCurrentSessionContext",
         "4d06b702d25d652afb9ef835d2a550031f1cf762b193523a92166f40ea3d142b",
         variables={
@@ -344,7 +359,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         },
     ),
     # returns all available campaigns
-    "Campaigns": GQLOperation(
+    "Campaigns": GQLPersistedQuery(
         "ViewerDropsDashboard",
         "5a4da2ab3d5b47c9f9ce864e727b2cb346af1e3ea8b897fe8f704a97ff017619",
         variables={
@@ -352,7 +367,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         }
     ),
     # returns extended information about a particular campaign
-    "CampaignDetails": GQLOperation(
+    "CampaignDetails": GQLPersistedQuery(
         "DropCampaignDetails",
         "039277bf98f3130929262cc7c6efd9c141ca3749cb6dca442fc8ead9a53f77c1",
         variables={
@@ -361,7 +376,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         },
     ),
     # returns drops available for a particular channel
-    "AvailableDrops": GQLOperation(
+    "AvailableDrops": GQLPersistedQuery(
         "DropsHighlightService_AvailableDrops",
         "782dad0f032942260171d2d80a654f88bdd0c5a9dddc392e9bc92218a0f42d20",
         variables={
@@ -369,7 +384,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         },
     ),
     # retuns stream playback access token
-    "PlaybackAccessToken": GQLOperation(
+    "PlaybackAccessToken": GQLPersistedQuery(
         "PlaybackAccessToken",
         "ed230aa1e33e07eebb8928504583da78a5173989fadfb1ac94be06a04f3cdbe9",
         variables={
@@ -382,7 +397,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
         },
     ),
     # returns live channels for a particular game
-    "GameDirectory": GQLOperation(
+    "GameDirectory": GQLPersistedQuery(
         "DirectoryPage_Game",
         "76cb069d835b8a02914c08dc42c421d0dafda8af5b113a3f19141824b901402f",
         variables={
@@ -403,21 +418,21 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
             "sortTypeIsRecency": False,
         },
     ),
-    "SlugRedirect": GQLOperation(  # can be used to turn game name -> game slug
+    "SlugRedirect": GQLPersistedQuery(  # can be used to turn game name -> game slug
         "DirectoryGameRedirect",
         "1f0300090caceec51f33c5e20647aceff9017f740f223c3c532ba6fa59f6b6cc",
         variables={
             "name": ...,  # game name
         },
     ),
-    "NotificationsView": GQLOperation(  # unused, triggers notifications "update-summary"
+    "NotificationsView": GQLPersistedQuery(  # unused, triggers notifications "update-summary"
         "OnsiteNotifications_View",
         "e8e06193f8df73d04a1260df318585d1bd7a7bb447afa058e52095513f2bfa4f",
         variables={
             "input": {},
         },
     ),
-    "NotificationsList": GQLOperation(  # unused
+    "NotificationsList": GQLPersistedQuery(  # unused
         "OnsiteNotifications_ListNotifications",
         "11cdb54a2706c2c0b2969769907675680f02a6e77d8afe79a749180ad16bfea6",
         variables={
@@ -428,7 +443,7 @@ GQL_OPERATIONS: dict[str, GQLOperation] = {
             "shouldLoadLastBroadcast": False,
         },
     ),
-    "NotificationsDelete": GQLOperation(
+    "NotificationsDelete": GQLPersistedQuery(
         "OnsiteNotifications_DeleteNotification",
         "13d463c831f28ffe17dccf55b3148ed8b3edbbd0ebadd56352f1ff0160616816",
         variables={

inventory.py

@@ -13,7 +13,7 @@ from translate import _
 from channel import Channel
 from utils import timestamp, Game
 from exceptions import GQLException
-from constants import GQL_OPERATIONS, MAX_EXTRA_MINUTES, URLType, State
+from constants import GQL_QUERIES, MAX_EXTRA_MINUTES, URLType, State
 
 if TYPE_CHECKING:
     from collections import abc
@@ -193,7 +193,7 @@ class BaseDrop:
             return False
         try:
             response = await self._twitch.gql_request(
-                GQL_OPERATIONS["ClaimDrop"].with_variables(
+                GQL_QUERIES["ClaimDrop"].with_variables(
                     {"input": {"dropInstanceID": self.claim_id}}
                 )
             )

twitch.py

@@ -46,7 +46,7 @@ from constants import (
     DUMP_PATH,
     COOKIES_PATH,
     MAX_CHANNELS,
-    GQL_OPERATIONS,
+    GQL_QUERIES,
     WATCH_INTERVAL,
     State,
     ClientType,
@@ -906,7 +906,7 @@ class Twitch:
                 # Solution 1: use GQL to query for the currently mined drop status
                 try:
                     context = await self.gql_request(
-                        GQL_OPERATIONS["CurrentDrop"].with_variables(
+                        GQL_QUERIES["CurrentDrop"].with_variables(
                             {"channelID": str(channel.id)}
                         )
                     )
@@ -1175,7 +1175,7 @@ class Twitch:
             if watching_channel is not None:
                 for attempt in range(8):
                     context = await self.gql_request(
-                        GQL_OPERATIONS["CurrentDrop"].with_variables(
+                        GQL_QUERIES["CurrentDrop"].with_variables(
                             {"channelID": str(watching_channel.id)}
                         )
                     )
@@ -1215,7 +1215,7 @@ class Twitch:
             ):
                 self.change_state(State.INVENTORY_FETCH)
                 await self.gql_request(
-                    GQL_OPERATIONS["NotificationsDelete"].with_variables(
+                    GQL_QUERIES["NotificationsDelete"].with_variables(
                         {"input": {"id": data["id"]}}
                     )
                 )
@@ -1387,7 +1387,7 @@ class Twitch:
         auth_state = await self.get_auth()
         response_list: list[JsonType] = await self.gql_request(
             [
-                GQL_OPERATIONS["CampaignDetails"].with_variables(
+                GQL_QUERIES["CampaignDetails"].with_variables(
                     {"channelLogin": str(auth_state.user_id), "dropID": cid}
                 )
                 for cid in campaign_ids
@@ -1403,7 +1403,7 @@ class Twitch:
         status_update = self.gui.status.update
         status_update(_("gui", "status", "fetching_inventory"))
         # fetch in-progress campaigns (inventory)
-        response = await self.gql_request(GQL_OPERATIONS["Inventory"])
+        response = await self.gql_request(GQL_QUERIES["Inventory"])
         inventory: JsonType = response["data"]["currentUser"]["inventory"]
         ongoing_campaigns: list[JsonType] = inventory["dropCampaignsInProgress"] or []
         # this contains claimed benefit edge IDs, not drop IDs
@@ -1412,7 +1412,7 @@ class Twitch:
         }
         inventory_data: dict[str, JsonType] = {c["id"]: c for c in ongoing_campaigns}
         # fetch general available campaigns data (campaigns)
-        response = await self.gql_request(GQL_OPERATIONS["Campaigns"])
+        response = await self.gql_request(GQL_QUERIES["Campaigns"])
         available_list: list[JsonType] = response["data"]["currentUser"]["dropCampaigns"] or []
         applicable_statuses = ("ACTIVE", "UPCOMING")
         available_campaigns: dict[str, JsonType] = {
@@ -1546,7 +1546,7 @@ class Twitch:
             filters.append("DROPS_ENABLED")
         try:
             response = await self.gql_request(
-                GQL_OPERATIONS["GameDirectory"].with_variables({
+                GQL_QUERIES["GameDirectory"].with_variables({
                     "limit": limit,
                     "slug": game.slug,
                     "options": {
@@ -1598,7 +1598,7 @@ class Twitch:
         acl_available_drops_map: dict[int, list[JsonType]] = {}
         if self.settings.available_drops_check:
             available_gql_ops: list[GQLOperation] = [
-                GQL_OPERATIONS["AvailableDrops"].with_variables({"channelID": str(channel_id)})
+                GQL_QUERIES["AvailableDrops"].with_variables({"channelID": str(channel_id)})
                 for channel_id, channel_data in acl_streams_map.items()
                 if channel_data["stream"] is not None  # only do this for ONLINE channels
             ]

utils.py

@@ -107,6 +107,10 @@ def timestamp(string: str) -> datetime:
         return datetime.strptime(string, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
 
 
+def isonow() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", 'Z')
+
+
 CHARS_ASCII = string.ascii_letters + string.digits
 CHARS_HEX_LOWER = string.digits + "abcdef"
 CHARS_HEX_UPPER = string.digits + "ABCDEF"