From 40000cf2954c3ab731e7201d17a1d600e772259c Mon Sep 17 00:00:00 2001 From: DevilXD <4180725+DevilXD@users.noreply.github.com> Date: Tue, 28 Apr 2026 20:25:21 +0200 Subject: [PATCH] Implement a way to send watch events through GQL --- channel.py | 59 ++++++++++++++++++++++++++++++++++++++++++++-------- constants.py | 51 +++++++++++++++++++++++++++++---------------- inventory.py | 4 ++-- twitch.py | 18 ++++++++-------- utils.py | 4 ++++ 5 files changed, 98 insertions(+), 38 deletions(-) diff --git a/channel.py b/channel.py index b4e197f..65df658 100644 --- a/channel.py +++ b/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: - return GQL_OPERATIONS["GetStreamInfo"].with_variables({"channel": self._login}) + def stream_gql(self) -> GQLPersistedQuery: + 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 diff --git a/constants.py b/constants.py index caca306..ef900fd 100644 --- a/constants.py +++ b/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={ diff --git a/inventory.py b/inventory.py index a8d9347..bf216ce 100644 --- a/inventory.py +++ b/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}} ) ) diff --git a/twitch.py b/twitch.py index 86fd129..bd2bf9b 100644 --- a/twitch.py +++ b/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 ] diff --git a/utils.py b/utils.py index 9d5de5c..bfe5e93 100644 --- a/utils.py +++ b/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"