diff --git a/constants.py b/constants.py
index 2cffc29..717d1fe 100644
--- a/constants.py
+++ b/constants.py
@@ -64,9 +64,9 @@ DEFAULT_LANG = "English"
BASE_URL = URL("https://twitch.tv")
CLIENT_ID = "kd1unb4b3q4t58fwlpcbzcbnm76a8fp"
# CLIENT_ID = "uo6dggojyb8d6soh92zknwmi5ej1q2"
-USER_AGENT = (
- "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
- "Chrome/105.0.0.0 Safari/537.36"
+ANDROID_USER_AGENT = (
+ "Dalvik/2.1.0 (Linux; U; Android 7.1.2; SM-G975N Build/N2G48C) "
+ "tv.twitch.android.app/13.4.1/1304010"
)
# Intervals and Delays
PING_INTERVAL = timedelta(minutes=3)
diff --git a/gui.py b/gui.py
index c8e245a..33acd6d 100644
--- a/gui.py
+++ b/gui.py
@@ -1,8 +1,6 @@
from __future__ import annotations
-import re
import sys
-import json
import ctypes
import asyncio
import logging
@@ -10,7 +8,7 @@ import webbrowser
import tkinter as tk
from collections import abc
from math import log10, ceil
-from subprocess import CREATE_NO_WINDOW
+from dataclasses import dataclass
from tkinter.font import Font, nametofont
from functools import partial, cached_property
from datetime import datetime, timedelta, timezone
@@ -23,31 +21,13 @@ import win32con
import win32gui
from yarl import URL
from PIL import Image as Image_module
-try:
- from seleniumwire.request import Request
- from selenium.common.exceptions import WebDriverException
- from seleniumwire.undetected_chromedriver import Chrome, ChromeOptions
-except ImportError as exc:
- raise ImportError(
- "You need to install Visual C++ Redist (x86 and x64): "
- "https://support.microsoft.com/en-gb/help/2977003/the-latest-supported-visual-c-downloads"
- ) from exc
from translate import _
from cache import ImageCache
+from exceptions import ExitRequest
from utils import resource_path, Game, _T
from registry import RegistryKey, ValueType
-from exceptions import MinerException, ExitRequest, LoginException
-from constants import (
- CLIENT_ID,
- SELF_PATH,
- # CACHE_PATH,
- FORMATTER,
- WS_TOPICS_LIMIT,
- MAX_WEBSOCKETS,
- WINDOW_TITLE,
- State,
-)
+from constants import SELF_PATH, FORMATTER, WS_TOPICS_LIMIT, MAX_WEBSOCKETS, WINDOW_TITLE, State
if TYPE_CHECKING:
from twitch import Twitch
@@ -424,6 +404,13 @@ class WebsocketStatus:
self._topics_var.set('\n'.join(topic_lines))
+@dataclass
+class LoginData:
+ username: str
+ password: str
+ token: str
+
+
class LoginForm:
def __init__(self, manager: GUIManager, master: ttk.Widget):
self._manager = manager
@@ -435,14 +422,14 @@ class LoginForm:
frame.rowconfigure(4, weight=1)
ttk.Label(frame, text=_("gui", "login", "labels")).grid(column=0, row=0)
ttk.Label(frame, textvariable=self._var, justify="center").grid(column=1, row=0)
- # self._login_entry = PlaceholderEntry(frame, placeholder=_("gui", "login", "username"))
- # self._login_entry.grid(column=0, row=1, columnspan=2)
- # self._pass_entry = PlaceholderEntry(
- # frame, placeholder=_("gui", "login", "password"), show='•'
- # )
- # self._pass_entry.grid(column=0, row=2, columnspan=2)
- # self._token_entry = PlaceholderEntry(frame, placeholder=_("gui", "login", "twofa_code"))
- # self._token_entry.grid(column=0, row=3, columnspan=2)
+ self._login_entry = PlaceholderEntry(frame, placeholder=_("gui", "login", "username"))
+ self._login_entry.grid(column=0, row=1, columnspan=2)
+ self._pass_entry = PlaceholderEntry(
+ frame, placeholder=_("gui", "login", "password"), show='•'
+ )
+ self._pass_entry.grid(column=0, row=2, columnspan=2)
+ self._token_entry = PlaceholderEntry(frame, placeholder=_("gui", "login", "twofa_code"))
+ self._token_entry.grid(column=0, row=3, columnspan=2)
self._confirm = asyncio.Event()
self._button = ttk.Button(
frame, text=_("gui", "login", "button"), command=self._confirm.set, state="disabled"
@@ -450,150 +437,46 @@ class LoginForm:
self._button.grid(column=0, row=4, columnspan=2)
self.update(_("gui", "login", "logged_out"), None)
- @staticmethod
- def interceptor(request: Request) -> None:
- if (
- request.method == "POST"
- and request.url == "https://passport.twitch.tv/protected_login"
- ):
- body = request.body.decode("utf-8")
- data = json.loads(body)
- data["client_id"] = CLIENT_ID
- request.body = json.dumps(data).encode("utf-8")
- del request.headers["Content-Length"]
- request.headers["Content-Length"] = str(len(request.body))
+ def clear(self, login: bool = False, password: bool = False, token: bool = False):
+ clear_all = not login and not password and not token
+ if login or clear_all:
+ self._login_entry.clear()
+ if password or clear_all:
+ self._pass_entry.clear()
+ if token or clear_all:
+ self._token_entry.clear()
- async def ask_login(self) -> str:
- self.update(_("gui", "login", "required"), None)
- self._manager.print(_("gui", "login", "request"))
- # ensure the window isn't hidden into tray when this runs
- self._manager.tray.restore()
- # the user needs to press on the login button to open the browser
+ async def wait_for_login_press(self) -> None:
self._confirm.clear()
try:
self._button.config(state="normal")
await self._manager.coro_unless_closed(self._confirm.wait())
finally:
self._button.config(state="disabled")
- self._confirm.clear()
- self._manager.print("Opening Chrome...")
- # open the chrome browser on the Twitch's login page
- # use a separate executor to void blocking the event loop
- loop = asyncio.get_running_loop()
- driver = None
- try:
- version_main = None
- for attempt in range(2):
- options = ChromeOptions()
- options.add_argument("--log-level=3")
- options.add_argument("--disable-web-security")
- options.add_argument("--allow-running-insecure-content")
- options.add_argument("--lang=en")
- options.add_argument("--no-sandbox")
- options.add_argument("--test-type")
- options.add_argument("--disable-gpu")
- options.set_capability("pageLoadStrategy", "eager")
- try:
- wire_options: dict[str, Any] = {"proxy": {}}
- if self._manager._twitch.settings.proxy:
- wire_options["proxy"]["http"] = str(self._manager._twitch.settings.proxy)
- driver_coro = loop.run_in_executor(
- None,
- lambda: Chrome(
- options=options,
- suppress_welcome=True,
- version_main=version_main,
- seleniumwire_options=wire_options,
- service_creationflags=CREATE_NO_WINDOW,
- # user_data_dir=str(CACHE_PATH.joinpath("ChromeProfile")),
- )
- )
- driver = await self._manager.coro_unless_closed(driver_coro)
- break
- except WebDriverException as exc:
- message = exc.msg
- if (
- message is not None
- and (match := re.search(
- r'Chrome version ([\d]+)\nCurrent browser version is ((\d+)\.[\d.]+)',
- message,
- )) is not None
- ):
- if not attempt:
- version_main = int(match.group(3))
- continue
- else:
- raise MinerException(
- "Your Chrome browser is out of date\n"
- f"Required version: {match.group(1)}\n"
- f"Current version: {match.group(2)}"
- ) from None
- raise MinerException(
- "An error occured while boostrapping the Chrome browser"
- ) from exc
- assert driver is not None
- driver.request_interceptor = self.interceptor
- # driver.set_page_load_timeout(30)
- page_coro = loop.run_in_executor(None, driver.get, "https://twitch.tv/login")
- await self._manager.coro_unless_closed(page_coro)
- # enable the login button once the page finishes opening
- self._button.config(state="normal")
- self._manager.print(
- "Complete the login procedure manually by pressing the Login button again."
+ async def ask_login(self) -> LoginData:
+ self.update(_("gui", "login", "required"), None)
+ # ensure the window isn't hidden into tray when this runs
+ self._manager.tray.restore()
+ while True:
+ self._manager.print(_("gui", "login", "request"))
+ await self.wait_for_login_press()
+ login_data = LoginData(
+ self._login_entry.get().strip(),
+ self._pass_entry.get(),
+ self._token_entry.get().strip(),
)
-
- # auto login
- # driver.find_element("id", "login-username").send_keys(username)
- # driver.find_element("id", "password-input").send_keys(password)
- # driver.find_element(
- # "css selector", '[data-a-target="passport-login-button"]'
- # ).click()
- # token submit button css selectors
- # Button: "screen="two_factor" target="submit_button"
- # Input:
-
- # wait for the user to navigate away from the URL, indicating successful login
- # alternatively, they can press on the login button again
- while (
- driver.current_url != "https://www.twitch.tv/?no-reload=true"
- and not self._confirm.is_set()
- ):
- if self._manager.close_requested:
- raise ExitRequest()
- await asyncio.sleep(0.5)
-
- # cookies = [
- # {
- # "domain": ".twitch.tv",
- # "expiry": 1700000000,
- # "httpOnly": False,
- # "name": "auth-token",
- # "path": "/",
- # "sameSite": "None",
- # "secure": True,
- # "value": "..."
- # },
- # ...,
- # ]
- cookies = driver.get_cookies()
- for cookie in cookies:
- if "twitch.tv" in cookie["domain"] and cookie["name"] == "auth-token":
- auth_token = cookie["value"]
- break
- else:
- raise LoginException("Unable to extract authorization token")
- except WebDriverException:
- driver = None
- raise LoginException(
- "Chrome window was closed before the login procedure could complete."
- )
- finally:
- self._button.config(state="disabled")
- if driver is not None:
- driver.quit()
- return auth_token
+ # basic input data validation
+ if len(login_data.username) < 3:
+ self.clear(login=True)
+ continue
+ if len(login_data.password) < 8:
+ self.clear(password=True)
+ continue
+ if login_data.token and len(login_data.token) < 6:
+ self.clear(token=True)
+ continue
+ return login_data
def update(self, status: str, user_id: int | None):
if user_id is not None:
diff --git a/twitch.py b/twitch.py
index 03835dc..cec421d 100644
--- a/twitch.py
+++ b/twitch.py
@@ -1,14 +1,16 @@
from __future__ import annotations
import re
-# import json
+import sys
+import json
+import random
import asyncio
import logging
from math import ceil
from time import time
from itertools import chain
from functools import partial
-# from base64 import urlsafe_b64decode
+from subprocess import CREATE_NO_WINDOW
from collections import abc, OrderedDict
from datetime import datetime, timedelta, timezone
from contextlib import suppress, asynccontextmanager
@@ -16,18 +18,34 @@ from typing import Any, Literal, Final, NoReturn, cast, TYPE_CHECKING
import aiohttp
from yarl import URL
+try:
+ from seleniumwire.request import Request
+ from selenium.common.exceptions import WebDriverException
+ from seleniumwire.undetected_chromedriver import Chrome, ChromeOptions
+except ImportError as exc:
+ raise ImportError(
+ "You need to install Visual C++ Redist (x86 and x64): "
+ "https://support.microsoft.com/en-gb/help/2977003/the-latest-supported-visual-c-downloads"
+ ) from exc
from translate import _
from gui import GUIManager
from channel import Channel
from websocket import WebsocketPool
from inventory import DropsCampaign
-from exceptions import MinerException, ExitRequest, LoginException, ReloadRequest, RequestInvalid
+from exceptions import (
+ MinerException,
+ ExitRequest,
+ LoginException,
+ ReloadRequest,
+ RequestInvalid,
+)
from utils import (
CHARS_HEX_LOWER,
timestamp,
create_nonce,
task_wrapper,
+ first_to_complete,
OrderedSet,
AwaitableValue,
ExponentialBackoff,
@@ -35,13 +53,13 @@ from utils import (
from constants import (
BASE_URL,
CLIENT_ID,
- USER_AGENT,
COOKIES_PATH,
GQL_OPERATIONS,
MAX_WEBSOCKETS,
WATCH_INTERVAL,
WS_TOPICS_LIMIT,
DROPS_ENABLED_TAG,
+ ANDROID_USER_AGENT,
State,
WebsocketTopic,
)
@@ -58,6 +76,16 @@ logger = logging.getLogger("TwitchDrops")
gql_logger = logging.getLogger("TwitchDrops.gql")
+class SkipExtraJsonDecoder(json.JSONDecoder):
+ def decode(self, s: str, *args):
+ # skip whitespace check
+ obj, end = self.raw_decode(s)
+ return obj
+
+
+SAFE_LOADS = lambda s: json.loads(s, cls=SkipExtraJsonDecoder)
+
+
class _AuthState:
def __init__(self, twitch: Twitch):
self._twitch: Twitch = twitch
@@ -98,37 +126,292 @@ class _AuthState:
)
self._logged_in.clear()
+ @staticmethod
+ def interceptor(request: Request) -> None:
+ if (
+ request.method == "POST"
+ and request.url == "https://passport.twitch.tv/protected_login"
+ ):
+ body = request.body.decode("utf-8")
+ data = json.loads(body)
+ data["client_id"] = CLIENT_ID
+ request.body = json.dumps(data).encode("utf-8")
+ del request.headers["Content-Length"]
+ request.headers["Content-Length"] = str(len(request.body))
+
+ async def _chrome_login(self) -> None:
+ gui_print = self._twitch.gui.print
+ login_form: LoginForm = self._twitch.gui.login
+ coro_unless_closed = self._twitch.gui.coro_unless_closed
+
+ # open the chrome browser on the Twitch's login page
+ # use a separate executor to void blocking the event loop
+ loop = asyncio.get_running_loop()
+ driver: Chrome | None = None
+ while True:
+ gui_print("Opening Chrome...")
+ try:
+ version_main = None
+ for attempt in range(2):
+ options = ChromeOptions()
+ options.add_argument("--log-level=3")
+ options.add_argument("--disable-web-security")
+ options.add_argument("--allow-running-insecure-content")
+ options.add_argument("--lang=en")
+ options.add_argument("--no-sandbox")
+ options.add_argument("--test-type")
+ options.add_argument("--disable-gpu")
+ options.set_capability("pageLoadStrategy", "eager")
+ try:
+ wire_options: dict[str, Any] = {"proxy": {}}
+ if self._twitch.settings.proxy:
+ wire_options["proxy"]["http"] = str(self._twitch.settings.proxy)
+ driver_coro = loop.run_in_executor(
+ None,
+ lambda: Chrome(
+ options=options,
+ suppress_welcome=True,
+ version_main=version_main,
+ seleniumwire_options=wire_options,
+ service_creationflags=CREATE_NO_WINDOW,
+ )
+ )
+ driver = await coro_unless_closed(driver_coro)
+ break
+ except WebDriverException as exc:
+ message = exc.msg
+ if (
+ message is not None
+ and (
+ match := re.search(
+ (
+ r'Chrome version ([\d]+)\n'
+ r'Current browser version is ((\d+)\.[\d.]+)'
+ ),
+ message,
+ )
+ ) is not None
+ ):
+ if not attempt:
+ version_main = int(match.group(3))
+ continue
+ else:
+ raise MinerException(
+ "Your Chrome browser is out of date\n"
+ f"Required version: {match.group(1)}\n"
+ f"Current version: {match.group(2)}"
+ ) from None
+ raise MinerException(
+ "An error occured while boostrapping the Chrome browser"
+ ) from exc
+ assert driver is not None
+ driver.request_interceptor = self.interceptor
+ # driver.set_page_load_timeout(30)
+ # page_coro = loop.run_in_executor(None, driver.get, "https://twitch.tv")
+ # await coro_unless_closed(page_coro)
+ page_coro = loop.run_in_executor(None, driver.get, "https://twitch.tv/login")
+ await coro_unless_closed(page_coro)
+
+ # auto login
+ # if login_data.username and login_data.password:
+ # driver.find_element("id", "login-username").send_keys(login_data.username)
+ # driver.find_element("id", "password-input").send_keys(login_data.password)
+ # driver.find_element(
+ # "css selector", '[data-a-target="passport-login-button"]'
+ # ).click()
+ # token submit button css selectors
+ # Button: "screen="two_factor" target="submit_button"
+ # Input:
+
+ # wait for the user to navigate away from the URL, indicating successful login
+ # alternatively, they can press on the login button again
+ async def url_waiter(driver=driver):
+ while driver.current_url != "https://www.twitch.tv/?no-reload=true":
+ await asyncio.sleep(0.5)
+
+ gui_print(
+ "Complete the login procedure manually by pressing the Login button again."
+ )
+ await first_to_complete([
+ url_waiter(),
+ coro_unless_closed(login_form.wait_for_login_press()),
+ ])
+
+ # cookies = [
+ # {
+ # "domain": ".twitch.tv",
+ # "expiry": 1700000000,
+ # "httpOnly": False,
+ # "name": "auth-token",
+ # "path": "/",
+ # "sameSite": "None",
+ # "secure": True,
+ # "value": "..."
+ # },
+ # ...,
+ # ]
+ cookies = driver.get_cookies()
+ for cookie in cookies:
+ if "twitch.tv" in cookie["domain"] and cookie["name"] == "auth-token":
+ self.access_token = cookie["value"]
+ break
+ else:
+ gui_print("Unable to extract authorization token")
+ except WebDriverException:
+ gui_print(
+ "Chrome window was closed before the login procedure could complete."
+ )
+ finally:
+ if driver is not None:
+ driver.quit()
+ driver = None
+ await coro_unless_closed(login_form.wait_for_login_press())
+
async def _login(self) -> str:
logger.debug("Login flow started")
+ gui_print = self._twitch.gui.print
login_form: LoginForm = self._twitch.gui.login
- for attempt in range(1, 6): # 5 attempts
- try:
- self.access_token = await login_form.ask_login()
- break
- except LoginException as exc:
- if attempt < 5:
- self._twitch.gui.print(exc.args[0])
- continue
- raise # reraise on and past the 5th attempt
- logger.debug("Access token granted")
- return self.access_token
+ token_kind: str = ''
+ use_chrome: bool = False
+ payload: JsonType = {
+ "client_id": CLIENT_ID, # client ID to-be associated with the access token
+ "undelete_user": False, # purpose unknown
+ "remember_me": True, # persist the session via the cookie
+ # "authy_token": str, # 2FA token
+ # "twitchguard_code": str, # email code
+ # "captcha": str, # self-fed captcha
+ # 'force_twitchguard': False, # force email code confirmation
+ }
- def gql_headers(self, *, integrity: bool) -> JsonType:
+ while True:
+ login_data = await login_form.ask_login()
+ payload["username"] = login_data.username
+ payload["password"] = login_data.password
+ # reinstate the 2FA token, if present
+ payload.pop("authy_token", None)
+ payload.pop("twitchguard_code", None)
+ if login_data.token:
+ # if there's no token kind set yet, and the user has entered a token,
+ # we can immediately assume it's an authenticator token and not an email one
+ if not token_kind:
+ token_kind = "authy"
+ if token_kind == "authy":
+ payload["authy_token"] = login_data.token
+ elif token_kind == "email":
+ payload["twitchguard_code"] = login_data.token
+
+ # use fancy headers to mimic the twitch android app
+ headers = {
+ "Accept": "application/vnd.twitchtv.v3+json",
+ "Accept-Encoding": "gzip",
+ "Accept-Language": "en-US",
+ "Client-Id": CLIENT_ID,
+ "Content-Type": "application/json; charset=UTF-8",
+ "Host": "passport.twitch.tv",
+ "User-Agent": ANDROID_USER_AGENT,
+ "X-Device-Id": self.device_id,
+ # "X-Device-Id": ''.join(random.choices('0123456789abcdef', k=32)),
+ }
+ async with self._twitch.request(
+ "POST", "https://passport.twitch.tv/login", headers=headers, json=payload
+ ) as response:
+ login_response: JsonType = await response.json(loads=SAFE_LOADS)
+
+ # Feed this back in to avoid running into CAPTCHA if possible
+ if "captcha_proof" in login_response:
+ payload["captcha"] = {"proof": login_response["captcha_proof"]}
+
+ # Error handling
+ if "error_code" in login_response:
+ error_code: int = login_response["error_code"]
+ logger.debug(f"Login error code: {error_code}")
+ if error_code == 1000:
+ logger.debug("1000: CAPTCHA is required")
+ use_chrome = True
+ break
+ elif error_code == 3001:
+ logger.debug("3001: Login failed due to incorrect username or password")
+ gui_print(_("login", "incorrect_login_pass"))
+ login_form.clear(password=True)
+ continue
+ elif error_code in (
+ 3012, # Invalid authy token
+ 3023, # Invalid email code
+ ):
+ logger.debug("3012/23: Login failed due to incorrect 2FA code")
+ if error_code == 3023:
+ token_kind = "email"
+ gui_print(_("login", "incorrect_email_code"))
+ else:
+ token_kind = "authy"
+ gui_print(_("login", "incorrect_twofa_code"))
+ login_form.clear(token=True)
+ continue
+ elif error_code in (
+ 3011, # Authy token needed
+ 3022, # Email code needed
+ ):
+ # 2FA handling
+ logger.debug("3011/22: 2FA token required")
+ # user didn't provide a token, so ask them for it
+ if error_code == 3022:
+ token_kind = "email"
+ gui_print(_("login", "email_code_required"))
+ else:
+ token_kind = "authy"
+ gui_print(_("login", "twofa_code_required"))
+ continue
+ elif error_code == 5027:
+ # client blocked from this operation - use chrome
+ logger.debug("5027: Client blocked from this operation")
+ use_chrome = True
+ break
+ else:
+ msg = str(login_response["error"])
+ logger.debug(msg)
+ raise LoginException(msg)
+ # Success handling
+ if "access_token" in login_response:
+ self.access_token = cast(str, login_response["access_token"])
+ logger.debug("Access token granted")
+ login_form.clear()
+ break
+
+ if use_chrome:
+ await self._chrome_login()
+
+ if hasattr(self, "access_token"):
+ logger.debug("Access token granted")
+ return self.access_token
+ raise MinerException("Login flow finished without setting the access token")
+
+ def headers(
+ self, *, user_agent: str = '', gql: bool = False, integrity: bool = False
+ ) -> JsonType:
headers = {
- "Authorization": f"OAuth {self.access_token}",
"Accept": "*/*",
+ "Accept-Encoding": "gzip",
+ "Accept-Language": "en-US",
"Pragma": "no-cache",
"Cache-Control": "no-cache",
- "Origin": "https://www.twitch.tv",
- "Referer": "https://www.twitch.tv/",
"Client-Id": CLIENT_ID,
- "Client-Session-Id": self.session_id,
- "Client-Version": self.client_version,
- "X-Device-Id": self.device_id,
}
- # if integrity:
- # headers["Client-Integrity"] = self.integrity_token
+ if user_agent:
+ headers["User-Agent"] = user_agent
+ if hasattr(self, "session_id"):
+ headers["Client-Session-Id"] = self.session_id
+ if hasattr(self, "client_version"):
+ headers["Client-Version"] = self.client_version
+ if hasattr(self, "device_id"):
+ headers["X-Device-Id"] = self.device_id
+ if gql:
+ headers["Origin"] = "https://www.twitch.tv"
+ headers["Referer"] = "https://www.twitch.tv/"
+ headers["Authorization"] = f"OAuth {self.access_token}"
+ if integrity:
+ headers["Client-Integrity"] = self.integrity_token
return headers
async def validate(self):
@@ -142,8 +425,11 @@ class _AuthState:
session = await self._twitch.get_session()
jar = cast(aiohttp.CookieJar, session.cookie_jar)
if not self._hasattrs("client_version", "device_id"):
- async with self._twitch.request("GET", BASE_URL) as response:
- match = re.search(r'twilightBuildID="([-a-z0-9]+)"', await response.text("utf8"))
+ async with self._twitch.request(
+ "GET", BASE_URL, headers=self.headers() # (user_agent=ANDROID_USER_AGENT)
+ ) as response:
+ page_html = await response.text("utf8")
+ match = re.search(r'twilightBuildID="([-a-z0-9]+)"', page_html)
if match is None:
raise MinerException("Unable to extract client_version")
self.client_version = match.group(1)
@@ -253,12 +539,23 @@ class Twitch:
if session.closed:
raise RuntimeError("Session is closed")
return session
+ # obtain the latest Chrome user agent from a Github project
+ async with aiohttp.request(
+ "GET", "https://jnrbsn.github.io/user-agents/user-agents.json"
+ ) as response:
+ agents = await response.json()
+ if sys.platform == "win32":
+ chrome_agent = random.choice(agents[:3])
+ elif sys.platform == "linux":
+ chrome_agent = random.choice(agents[7:11])
+ # load in cookies
cookie_jar = aiohttp.CookieJar()
if COOKIES_PATH.exists():
cookie_jar.load(COOKIES_PATH)
+ # create session
self._session = aiohttp.ClientSession(
cookie_jar=cookie_jar,
- headers={"User-Agent": USER_AGENT},
+ headers={"User-Agent": chrome_agent},
timeout=aiohttp.ClientTimeout(connect=5, total=10),
)
return self._session
@@ -974,7 +1271,7 @@ class Twitch:
"POST",
"https://gql.twitch.tv/gql",
json=op,
- headers=auth_state.gql_headers(integrity=True),
+ headers=auth_state.headers(gql=True),
invalidate_after=getattr(auth_state, "integrity_expires", None),
) as response:
response_json: JsonType = await response.json()
diff --git a/utils.py b/utils.py
index c021c78..f4d3bd7 100644
--- a/utils.py
+++ b/utils.py
@@ -38,6 +38,17 @@ _JSON_T = TypeVar("_JSON_T", bound=Mapping[Any, Any])
logger = logging.getLogger("TwitchDrops")
+async def first_to_complete(coros: abc.Iterable[abc.Coroutine[Any, Any, _T]]) -> _T:
+ # In Python 3.11, we need to explicitly wrap awaitables
+ tasks = [asyncio.ensure_future(coro) for coro in coros]
+ done: set[asyncio.Task[Any]]
+ pending: set[asyncio.Task[Any]]
+ done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+ for task in pending:
+ task.cancel()
+ return next(iter(done)).result()
+
+
def format_traceback(exc: BaseException, **kwargs: Any) -> str:
"""
Like `traceback.print_exc` but returns a string. Uses the passed-in exception.