Reimplement the old mining method (ref b47b208)

channel.py

@@ -4,13 +4,15 @@ import re
 import json
 import asyncio
 import logging
+from base64 import b64encode
+from functools import cached_property
 from typing import Any, SupportsInt, cast, TYPE_CHECKING
 
 import aiohttp
 from yarl import URL
 
-from utils import Game
-from exceptions import MinerException
+from utils import Game, json_minify
+from exceptions import MinerException, RequestException
 from constants import CALL, GQL_OPERATIONS, ONLINE_DELAY, URLType
 
 if TYPE_CHECKING:
@@ -23,10 +25,6 @@ logger = logging.getLogger("TwitchDrops")
 
 
 class Stream:
-    __slots__ = (
-        "channel", "broadcast_id", "viewers", "drops_enabled", "game", "title", "_stream_url"
-    )
-
     def __init__(
         self,
         channel: Channel,
@@ -44,6 +42,27 @@ class Stream:
         self.title: str = title
         self._stream_url: URLType | None = None
 
+    @cached_property
+    def _spade_payload(self) -> JsonType:
+        payload = [
+            {
+                "event": "minute-watched",
+                "properties": {
+                    "broadcast_id": str(self.broadcast_id),
+                    "channel_id": str(self.channel.id),
+                    "channel": self.channel._login,
+                    "hidden": False,
+                    "live": True,
+                    "location": "channel",
+                    "logged_in": True,
+                    "muted": False,
+                    "player": "site",
+                    "user_id": self.channel._twitch._auth_state.user_id,
+                }
+            }
+        ]
+        return {"data": (b64encode(json_minify(payload).encode("utf8"))).decode("utf8")}
+
     @classmethod
     def from_get_stream(cls, channel: Channel, channel_data: JsonType) -> Stream:
         stream = channel_data["stream"]
@@ -119,6 +138,11 @@ class Stream:
 
 
 class Channel:
+    __slots__ = (
+        "_twitch", "_gui_channels", "id", "_login", "_display_name", "_spade_url",
+        "_stream", "_pending_stream_up", "acl_based"
+    )
+
     def __init__(
         self,
         twitch: Twitch,
@@ -133,6 +157,7 @@ class Channel:
         self.id: int = int(id)
         self._login: str = login
         self._display_name: str | None = display_name
+        self._spade_url: URLType | None = None
         self._stream: Stream | None = None
         self._pending_stream_up: asyncio.Task[Any] | None = None
         # ACL-based channels are:
@@ -253,6 +278,34 @@ class Channel:
             self._pending_stream_up = None
         self._gui_channels.remove(self)
 
+    async def get_spade_url(self) -> URLType:
+        """
+        To get this monstrous thing, you have to walk a chain of requests.
+        Streamer page (HTML) --parse-> Streamer Settings (JavaScript) --parse-> Spade URL
+
+        For mobile view, spade_url is available immediately from the page, skipping step #2.
+        """
+        SETTINGS_PATTERN: str = (
+            r'src="(https://[\w.]+/config/settings\.[0-9a-f]{32}\.js)"'
+        )
+        SPADE_PATTERN: str = (
+            r'"spade_?url": ?"(https://video-edge-[.\w\-/]+\.ts(?:\?allow_stream=true)?)"'
+        )
+        async with self._twitch.request("GET", self.url) as response1:
+            streamer_html: str = await response1.text(encoding="utf8")
+        match = re.search(SPADE_PATTERN, streamer_html, re.I)
+        if not match:
+            match = re.search(SETTINGS_PATTERN, streamer_html, re.I)
+            if not match:
+                raise MinerException("Error while spade_url extraction: step #1")
+            streamer_settings = match.group(1)
+            async with self._twitch.request("GET", streamer_settings) as response2:
+                settings_js: str = await response2.text(encoding="utf8")
+            match = re.search(SPADE_PATTERN, settings_js, re.I)
+            if not match:
+                raise MinerException("Error while spade_url extraction: step #2")
+        return URLType(match.group(1))
+
     def external_update(self, channel_data: JsonType, available_drops: list[JsonType]):
         """
         Update stream information based on data provided externally.
@@ -354,7 +407,8 @@ class Channel:
         if needs_display:
             self.display()
 
-    async def send_watch(self) -> bool:
+    # NOTE: This is currently unused.
+    async def _send_watch(self) -> bool:
         """
         This performs a HEAD request on the stream's current playlist,
         to simulate watching the stream.
@@ -405,3 +459,16 @@ class Channel:
         # without downloading the actual stream data
         async with self._twitch.request("HEAD", stream_chunk_url) as head_response:
             return head_response.status == 200
+
+    async def send_watch(self) -> bool:
+        if self._stream is None:
+            return False
+        if self._spade_url is None:
+            self._spade_url = await self.get_spade_url()
+        try:
+            async with self._twitch.request(
+                "POST", self._spade_url, data=self._stream._spade_payload
+            ) as response:
+                return response.status == 204
+        except RequestException:
+            return False