diff --git a/channel.py b/channel.py index dd3ad3f..6330484 100644 --- a/channel.py +++ b/channel.py @@ -9,10 +9,8 @@ from functools import cached_property from datetime import datetime, timezone from typing import Any, Optional, TYPE_CHECKING -import aiohttp - -from exceptions import MinerException from utils import Game, invalidate_cache +from exceptions import MinerException, RequestException from constants import JsonType, BASE_URL, GQL_OPERATIONS, ONLINE_DELAY, DROPS_ENABLED_TAG if TYPE_CHECKING: @@ -195,9 +193,8 @@ class Channel: To get this monstrous thing, you have to walk a chain of requests. Streamer page (HTML) --parse-> Streamer Settings (JavaScript) --parse-> Spade URL """ - assert self._twitch._session is not None - async with self._twitch._session.get(self.url) as response: - streamer_html = await response.text(encoding="utf8") + async with self._twitch.request("GET", self.url) as response: + streamer_html: str = await response.text(encoding="utf8") match = re.search( r'src="(https://static\.twitchcdn\.net/config/settings\.[0-9a-f]{32}\.js)"', streamer_html, @@ -206,8 +203,8 @@ class Channel: if not match: raise MinerException("Error while spade_url extraction: step #1") streamer_settings = match.group(1) - async with self._twitch._session.get(streamer_settings) as response: - settings_js = await response.text(encoding="utf8") + async with self._twitch.request("GET", streamer_settings) as response: + settings_js: str = await response.text(encoding="utf8") match = re.search( r'"spade_url": ?"(https://video-edge-[.\w\-/]+\.ts)"', settings_js, re.I ) @@ -322,16 +319,13 @@ class Channel: """ if not self.online: return False - session = self._twitch._session - if session is None: - return False if self._spade_url is None: self._spade_url = await self.get_spade_url() logger.debug(f"Sending minute-watched to {self.name}") - for attempt in range(5): - try: - async with session.post(self._spade_url, data=self._payload) as response: - return response.status == 204 - except (aiohttp.ClientConnectionError, aiohttp.ServerTimeoutError): - continue - return False + try: + async with self._twitch.request( + "POST", self._spade_url, data=self._payload + ) as response: + return response.status == 204 + except RequestException: + return False diff --git a/twitch.py b/twitch.py index d06578c..960955c 100644 --- a/twitch.py +++ b/twitch.py @@ -8,9 +8,19 @@ from time import time from itertools import chain from datetime import datetime from functools import partial -from contextlib import suppress +from contextlib import suppress, asynccontextmanager from typing import ( - Optional, Union, List, Dict, Set, OrderedDict, Callable, Iterable, cast, TYPE_CHECKING + Optional, + Union, + List, + Dict, + Set, + OrderedDict, + Callable, + Iterable, + AsyncIterator, + cast, + TYPE_CHECKING, ) try: @@ -22,7 +32,7 @@ from gui import GUIManager from channel import Channel from websocket import WebsocketPool from inventory import DropsCampaign, TimedDrop -from exceptions import LoginException, CaptchaRequired +from exceptions import RequestException, LoginException, CaptchaRequired from utils import task_wrapper, timestamp, Game, AwaitableValue, OrderedSet from constants import ( GQL_URL, @@ -153,13 +163,11 @@ class Twitch: • Selecting a stream to watch, and watching it • Changing the stream that's being watched if necessary """ - if self._session is None: - await self.initialize() self.gui.start() + await self.check_login() if self._watching_task is not None: self._watching_task.cancel() self._watching_task = asyncio.create_task(self._watch_loop()) - await self.check_login() # Add default topics assert self._user_id is not None self.websocket.add_topics([ @@ -606,7 +614,6 @@ class Twitch: self.gui.print(f"Claimed bonus points: {points}") async def _login(self) -> str: - assert self._session is not None logger.debug("Login flow started") login_form: LoginForm = self.gui.login @@ -624,7 +631,7 @@ class Twitch: payload.pop("authy_token", None) payload.pop("twitchguard_code", None) for attempt in range(2): - async with self._session.post(f"{AUTH_URL}/login", json=payload) as response: + async with self.request("POST", f"{AUTH_URL}/login", json=payload) as response: login_response: JsonType = await response.json() # Feed this back in to avoid running into CAPTCHA if possible @@ -690,10 +697,12 @@ class Twitch: # we're all good return # looks like we're missing something - assert self._session is not None login_form: LoginForm = self.gui.login logger.debug("Checking login") login_form.update("Logging in...", None) + if self._session is None: + await self.initialize() + assert self._session is not None jar = cast(aiohttp.CookieJar, self._session.cookie_jar) for attempt in range(2): cookie = jar.filter_cookies("https://twitch.tv") # type: ignore @@ -707,7 +716,8 @@ class Twitch: self._access_token = cookie["auth-token"].value logger.debug("Restoring session from cookie") # validate our access token, by obtaining user_id - async with self._session.get( + async with self.request( + "GET", "https://id.twitch.tv/oauth2/validate", headers={"Authorization": f"OAuth {self._access_token}"} ) as response: @@ -731,23 +741,37 @@ class Twitch: jar.update_cookies(cookie, URL("https://twitch.tv")) jar.save(COOKIES_PATH) + @asynccontextmanager + async def request( + self, method: str, url: str, **kwargs + ) -> AsyncIterator[aiohttp.ClientResponse]: + if self._session is None: + await self.initialize() + session = self._session + assert session is not None + for attempt in range(5): + try: + async with session.request(method, url, **kwargs) as response: + yield response + break + except (aiohttp.ClientConnectionError, aiohttp.ServerTimeoutError): + await asyncio.sleep(0.1 * attempt) + else: + raise RequestException( + "Ran out of attempts while handling a request: " + f"(method={method.upper()}, {url=}, {kwargs=})" + ) + async def gql_request(self, op: GQLOperation) -> JsonType: - await self.check_login() - assert self._session is not None headers = { "Authorization": f"OAuth {self._access_token}", "Client-Id": CLIENT_ID, } gql_logger.debug(f"GQL Request: {op}") - for attempt in range(5): - try: - async with self._session.post(GQL_URL, json=op, headers=headers) as response: - response_json = await response.json() - gql_logger.debug(f"GQL Response: {response_json}") - return response_json - except (aiohttp.ClientConnectionError, aiohttp.ServerTimeoutError): - continue - raise RuntimeError(f"Ran out of attempts while handling a GQL request: {op}") + async with self.request("POST", GQL_URL, json=op, headers=headers) as response: + response_json: JsonType = await response.json() + gql_logger.debug(f"GQL Response: {response_json}") + return response_json async def fetch_campaign( self, campaign_id: str, claimed_benefits: Dict[str, datetime]