EaZyCode/twitch-miner-docker
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move ClientType into Twitch class as an attribute

Browse Source 
This allows us to use the base URL as specified in ClientInfo for
requests made to Twitch.

Previously, we'd hardcode "www.twitch.tv" as the base URL to make
channel-related requests, but because we're currently spoofing the
mobile client, that would always cause a HTTP redirect to m.twitch.tv.

This change avoids that HTTP redirect, and potentially saves a whopping
150 ms (or more 😱) when making channel-related requests.

Co-authored-by: DevilXD <4180725+DevilXD@users.noreply.github.com>
This commit is contained in:
guihkxandDevilXD authored and DevilXD committed 2024-02-14 14:51:37 +01:00
1 parent d7c2c8d5c3
commit dbd3cd82a0
3 files changed
+36 -33
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

channel.py
+2 -2
View File
@@ -9,7 +9,7 @@ from typing import Any, SupportsInt, TYPE_CHECKING
from utils import invalidate_cache, json_minify, Game from utils import invalidate_cache, json_minify, Game
from exceptions import MinerException, RequestException from exceptions import MinerException, RequestException
from constants import BASE_URL, GQL_OPERATIONS, ONLINE_DELAY, URLType from constants import GQL_OPERATIONS, ONLINE_DELAY, URLType
if TYPE_CHECKING: if TYPE_CHECKING:
from twitch import Twitch from twitch import Twitch
@@ -151,7 +151,7 @@ class Channel:
@property @property
def url(self) -> URLType: def url(self) -> URLType:
return URLType(f"{BASE_URL}/{self._login}") return URLType(f"{self._twitch._client_type.CLIENT_URL}/{self._login}")
@property @property
def iid(self) -> str: def iid(self) -> str:
constants.py
+2 -3
View File
@@ -112,7 +112,6 @@ MAX_TOPICS = (MAX_WEBSOCKETS * WS_TOPICS_LIMIT) - BASE_TOPICS
MAX_CHANNELS = MAX_TOPICS // TOPICS_PER_CHANNEL MAX_CHANNELS = MAX_TOPICS // TOPICS_PER_CHANNEL
# Misc # Misc
DEFAULT_LANG = "English" DEFAULT_LANG = "English"
BASE_URL = URL("https://www.twitch.tv")
# Intervals and Delays # Intervals and Delays
PING_INTERVAL = timedelta(minutes=3) PING_INTERVAL = timedelta(minutes=3)
PING_TIMEOUT = timedelta(seconds=10) PING_TIMEOUT = timedelta(seconds=10)
@@ -145,7 +144,7 @@ class ClientInfo:
class ClientType: class ClientType:
WEB = ClientInfo( WEB = ClientInfo(
BASE_URL, URL("https://www.twitch.tv"),
"kimne78kx3ncx6brgo4mv6wki5h1ko", "kimne78kx3ncx6brgo4mv6wki5h1ko",
( (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
@@ -187,7 +186,7 @@ class ClientType:
] ]
) )
ANDROID_APP = ClientInfo( ANDROID_APP = ClientInfo(
BASE_URL, URL("https://www.twitch.tv"),
"kd1unb4b3q4t58fwlpcbzcbnm76a8fp", "kd1unb4b3q4t58fwlpcbzcbnm76a8fp",
( (
"Dalvik/2.1.0 (Linux; U; Android 7.1.2; SM-G977N Build/LMY48Z) " "Dalvik/2.1.0 (Linux; U; Android 7.1.2; SM-G977N Build/LMY48Z) "
twitch.py
+32 -28
View File
@@ -60,7 +60,6 @@ from utils import (
) )
from constants import ( from constants import (
CALL, CALL,
BASE_URL,
COOKIES_PATH, COOKIES_PATH,
GQL_OPERATIONS, GQL_OPERATIONS,
MAX_CHANNELS, MAX_CHANNELS,
@@ -76,7 +75,7 @@ if TYPE_CHECKING:
from channel import Stream from channel import Stream
from settings import Settings from settings import Settings
from inventory import TimedDrop from inventory import TimedDrop
from constants import JsonType, GQLOperation from constants import ClientInfo, JsonType, GQLOperation
logger = logging.getLogger("TwitchDrops") logger = logging.getLogger("TwitchDrops")
@@ -90,7 +89,7 @@ class SkipExtraJsonDecoder(json.JSONDecoder):
return obj return obj
CLIENT_URL, CLIENT_ID, USER_AGENT = ClientType.MOBILE_WEB # CLIENT_URL, CLIENT_ID, USER_AGENT = ClientType.MOBILE_WEB
SAFE_LOADS = lambda s: json.loads(s, cls=SkipExtraJsonDecoder) SAFE_LOADS = lambda s: json.loads(s, cls=SkipExtraJsonDecoder)
@@ -134,15 +133,14 @@ class _AuthState:
) )
self._logged_in.clear() self._logged_in.clear()
@staticmethod def interceptor(self, request: Request) -> None:
def interceptor(request: Request) -> None:
if ( if (
request.method == "POST" request.method == "POST"
and request.url == "https://passport.twitch.tv/protected_login" and request.url == "https://passport.twitch.tv/protected_login"
): ):
body = request.body.decode("utf-8") body = request.body.decode("utf-8")
data = json.loads(body) data = json.loads(body)
data["client_id"] = CLIENT_ID data["client_id"] = self._twitch._client_type.CLIENT_ID
request.body = json.dumps(data).encode("utf-8") request.body = json.dumps(data).encode("utf-8")
del request.headers["Content-Length"] del request.headers["Content-Length"]
request.headers["Content-Length"] = str(len(request.body)) request.headers["Content-Length"] = str(len(request.body))
@@ -273,21 +271,22 @@ class _AuthState:
async def _oauth_login(self) -> str: async def _oauth_login(self) -> str:
login_form: LoginForm = self._twitch.gui.login login_form: LoginForm = self._twitch.gui.login
client_info: ClientInfo = self._twitch._client_type
headers = { headers = {
"Accept": "application/json", "Accept": "application/json",
"Accept-Encoding": "gzip", "Accept-Encoding": "gzip",
"Accept-Language": "en-US", "Accept-Language": "en-US",
"Cache-Control": "no-cache", "Cache-Control": "no-cache",
"Client-Id": CLIENT_ID, "Client-Id": client_info.CLIENT_ID,
"Host": "id.twitch.tv", "Host": "id.twitch.tv",
"Origin": str(CLIENT_URL), "Origin": str(client_info.CLIENT_URL),
"Pragma": "no-cache", "Pragma": "no-cache",
"Referer": str(CLIENT_URL), "Referer": str(client_info.CLIENT_URL),
"User-Agent": USER_AGENT, "User-Agent": client_info.USER_AGENT,
"X-Device-Id": self.device_id, "X-Device-Id": self.device_id,
} }
payload = { payload = {
"client_id": CLIENT_ID, "client_id": client_info.CLIENT_ID,
"scopes": ( "scopes": (
"channel_read chat:read user_blocks_edit " "channel_read chat:read user_blocks_edit "
"user_blocks_read user_follows_edit user_read" "user_blocks_read user_follows_edit user_read"
@@ -316,7 +315,7 @@ class _AuthState:
await login_form.ask_enter_code(user_code) await login_form.ask_enter_code(user_code)
payload = { payload = {
"client_id": CLIENT_ID, "client_id": self._twitch._client_type.CLIENT_ID,
"device_code": device_code, "device_code": device_code,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code", "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
} }
@@ -350,6 +349,7 @@ class _AuthState:
logger.info("Login flow started") logger.info("Login flow started")
gui_print = self._twitch.gui.print gui_print = self._twitch.gui.print
login_form: LoginForm = self._twitch.gui.login login_form: LoginForm = self._twitch.gui.login
client_info: ClientInfo = self._twitch._client_type
token_kind: str = '' token_kind: str = ''
use_chrome: bool = False use_chrome: bool = False
@@ -357,7 +357,8 @@ class _AuthState:
# username and password are added later # username and password are added later
# "username": str, # "username": str,
# "password": str, # "password": str,
"client_id": CLIENT_ID, # client ID to-be associated with the access token # client ID to-be associated with the access token
"client_id": client_info.CLIENT_ID,
"undelete_user": False, # purpose unknown "undelete_user": False, # purpose unknown
"remember_me": True, # persist the session via the cookie "remember_me": True, # persist the session via the cookie
# "authy_token": str, # 2FA token # "authy_token": str, # 2FA token
@@ -388,10 +389,10 @@ class _AuthState:
"Accept": "application/vnd.twitchtv.v3+json", "Accept": "application/vnd.twitchtv.v3+json",
"Accept-Encoding": "gzip", "Accept-Encoding": "gzip",
"Accept-Language": "en-US", "Accept-Language": "en-US",
"Client-Id": CLIENT_ID, "Client-Id": client_info.CLIENT_ID,
"Content-Type": "application/json; charset=UTF-8", "Content-Type": "application/json; charset=UTF-8",
"Host": "passport.twitch.tv", "Host": "passport.twitch.tv",
"User-Agent": USER_AGENT, "User-Agent": client_info.USER_AGENT,
"X-Device-Id": self.device_id, "X-Device-Id": self.device_id,
# "X-Device-Id": ''.join(random.choices('0123456789abcdef', k=32)), # "X-Device-Id": ''.join(random.choices('0123456789abcdef', k=32)),
} }
@@ -486,13 +487,14 @@ class _AuthState:
def headers( def headers(
self, *, user_agent: str = '', gql: bool = False, integrity: bool = False self, *, user_agent: str = '', gql: bool = False, integrity: bool = False
) -> JsonType: ) -> JsonType:
client_info: ClientInfo = self._twitch._client_type
headers = { headers = {
"Accept": "*/*", "Accept": "*/*",
"Accept-Encoding": "gzip", "Accept-Encoding": "gzip",
"Accept-Language": "en-US", "Accept-Language": "en-US",
"Pragma": "no-cache", "Pragma": "no-cache",
"Cache-Control": "no-cache", "Cache-Control": "no-cache",
"Client-Id": CLIENT_ID, "Client-Id": client_info.CLIENT_ID,
} }
if user_agent: if user_agent:
headers["User-Agent"] = user_agent headers["User-Agent"] = user_agent
@@ -503,8 +505,8 @@ class _AuthState:
if hasattr(self, "device_id"): if hasattr(self, "device_id"):
headers["X-Device-Id"] = self.device_id headers["X-Device-Id"] = self.device_id
if gql: if gql:
headers["Origin"] = str(BASE_URL) headers["Origin"] = str(client_info.CLIENT_URL)
headers["Referer"] = str(BASE_URL) headers["Referer"] = str(client_info.CLIENT_URL)
headers["Authorization"] = f"OAuth {self.access_token}" headers["Authorization"] = f"OAuth {self.access_token}"
if integrity: if integrity:
headers["Client-Integrity"] = self.integrity_token headers["Client-Integrity"] = self.integrity_token
@@ -520,9 +522,10 @@ class _AuthState:
if not self._hasattrs("device_id", "access_token", "user_id"): if not self._hasattrs("device_id", "access_token", "user_id"):
session = await self._twitch.get_session() session = await self._twitch.get_session()
jar = cast(aiohttp.CookieJar, session.cookie_jar) jar = cast(aiohttp.CookieJar, session.cookie_jar)
client_info: ClientInfo = self._twitch._client_type
if not self._hasattrs("device_id"): if not self._hasattrs("device_id"):
async with self._twitch.request( async with self._twitch.request(
"GET", CLIENT_URL, headers=self.headers() "GET", client_info.CLIENT_URL, headers=self.headers()
) as response: ) as response:
page_html = await response.text("utf8") page_html = await response.text("utf8")
assert page_html is not None assert page_html is not None
@@ -531,7 +534,7 @@ class _AuthState:
# raise MinerException("Unable to extract client_version") # raise MinerException("Unable to extract client_version")
# self.client_version = match.group(1) # self.client_version = match.group(1)
# doing the request ends up setting the "unique_id" value in the cookie # doing the request ends up setting the "unique_id" value in the cookie
cookie = jar.filter_cookies(CLIENT_URL) cookie = jar.filter_cookies(client_info.CLIENT_URL)
self.device_id = cookie["unique_id"].value self.device_id = cookie["unique_id"].value
if not self._hasattrs("access_token", "user_id"): if not self._hasattrs("access_token", "user_id"):
# looks like we're missing something # looks like we're missing something
@@ -539,7 +542,7 @@ class _AuthState:
logger.info("Checking login") logger.info("Checking login")
login_form.update(_("gui", "login", "logging_in"), None) login_form.update(_("gui", "login", "logging_in"), None)
for attempt in range(2): for attempt in range(2):
cookie = jar.filter_cookies(CLIENT_URL) cookie = jar.filter_cookies(client_info.CLIENT_URL)
if "auth-token" not in cookie: if "auth-token" not in cookie:
self.access_token = await self._oauth_login() self.access_token = await self._oauth_login()
cookie["auth-token"] = self.access_token cookie["auth-token"] = self.access_token
@@ -556,22 +559,22 @@ class _AuthState:
if status == 401: if status == 401:
# the access token we have is invalid - clear the cookie and reauth # the access token we have is invalid - clear the cookie and reauth
logger.info("Restored session is invalid") logger.info("Restored session is invalid")
assert CLIENT_URL.host is not None assert client_info.CLIENT_URL.host is not None
jar.clear_domain(CLIENT_URL.host) jar.clear_domain(client_info.CLIENT_URL.host)
continue continue
elif status == 200: elif status == 200:
validate_response = await response.json() validate_response = await response.json()
break break
else: else:
raise RuntimeError("Login verification failure") raise RuntimeError("Login verification failure")
if validate_response["client_id"] != CLIENT_ID: if validate_response["client_id"] != client_info.CLIENT_ID:
raise MinerException("You're using an old cookie file, please generate a new one.") raise MinerException("You're using an old cookie file, please generate a new one.")
self.user_id = int(validate_response["user_id"]) self.user_id = int(validate_response["user_id"])
cookie["persistent"] = str(self.user_id) cookie["persistent"] = str(self.user_id)
logger.info(f"Login successful, user ID: {self.user_id}") logger.info(f"Login successful, user ID: {self.user_id}")
login_form.update(_("gui", "login", "logged_in"), self.user_id) login_form.update(_("gui", "login", "logged_in"), self.user_id)
# update our cookie and save it # update our cookie and save it
jar.update_cookies(cookie, CLIENT_URL) jar.update_cookies(cookie, client_info.CLIENT_URL)
jar.save(COOKIES_PATH) jar.save(COOKIES_PATH)
# if not self._hasattrs("integrity_token") or self.integrity_expired: # if not self._hasattrs("integrity_token") or self.integrity_expired:
# async with self._twitch.request( # async with self._twitch.request(
@@ -618,7 +621,8 @@ class Twitch:
self.inventory: list[DropsCampaign] = [] self.inventory: list[DropsCampaign] = []
self._drops: dict[str, TimedDrop] = {} self._drops: dict[str, TimedDrop] = {}
self._mnt_triggers: deque[datetime] = deque() self._mnt_triggers: deque[datetime] = deque()
# Session and auth # Client type, session and auth
self._client_type: ClientInfo = ClientType.MOBILE_WEB
self._session: aiohttp.ClientSession | None = None self._session: aiohttp.ClientSession | None = None
self._auth_state: _AuthState = _AuthState(self) self._auth_state: _AuthState = _AuthState(self)
# GUI # GUI
@@ -665,7 +669,7 @@ class Twitch:
timeout=timeout, timeout=timeout,
connector=connector, connector=connector,
cookie_jar=cookie_jar, cookie_jar=cookie_jar,
headers={"User-Agent": USER_AGENT}, headers={"User-Agent": self._client_type.USER_AGENT},
) )
return self._session return self._session
@@ -1535,7 +1539,7 @@ class Twitch:
"POST", "POST",
"https://gql.twitch.tv/gql", "https://gql.twitch.tv/gql",
json=ops, json=ops,
headers=auth_state.headers(user_agent=USER_AGENT, gql=True), headers=auth_state.headers(user_agent=self._client_type.USER_AGENT, gql=True),
invalidate_after=getattr(auth_state, "integrity_expires", None), invalidate_after=getattr(auth_state, "integrity_expires", None),
) as response: ) as response:
response_json: JsonType | list[JsonType] = await response.json() response_json: JsonType | list[JsonType] = await response.json()