From 97521c11e90ba3d10f4f84e156d44317d6b74feb Mon Sep 17 00:00:00 2001 From: DevilXD Date: Fri, 10 Dec 2021 13:45:28 +0100 Subject: [PATCH] Small code cleanup --- constants.py | 8 +++++++- inventory.py | 47 ++++++++++++++++++++++------------------------- main.py | 4 ++-- twitch.py | 10 ++++------ 4 files changed, 35 insertions(+), 34 deletions(-) diff --git a/constants.py b/constants.py index 0c1c617..4ffdca5 100644 --- a/constants.py +++ b/constants.py @@ -5,21 +5,27 @@ from datetime import timedelta from typing import Any, Optional, Union, Dict, Callable +# URLs BASE_URL = "https://twitch.tv" AUTH_URL = "https://passport.twitch.tv" WEBSOCKET_URL = "wss://pubsub-edge.twitch.tv/v1" GQL_URL = "https://gql.twitch.tv/gql" +# Misc for Twitch CLIENT_ID = "kimne78kx3ncx6brgo4mv6wki5h1ko" USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/96.0.4664.45 Safari/537.36" ) +# Paths SETTINGS_PATH = "settings.json" COOKIES_PATH = "cookies.pickle" +# Intervals and Delays PING_INTERVAL = timedelta(minutes=3) ONLINE_DELAY = timedelta(seconds=30) -# tags +# Tags DROPS_ENABLED_TAG = "c2542d6d-cd10-4532-919b-3d19f30a768b" +# Strings +TERMINATED_STR = "Application Terminated.\nClose the console window to exit the application." class GQLOperation(Dict[str, Any]): diff --git a/inventory.py b/inventory.py index 007e76e..192431f 100644 --- a/inventory.py +++ b/inventory.py @@ -9,25 +9,6 @@ if TYPE_CHECKING: from twitch import Twitch -async def claim_drop(twitch: Twitch, claim_id: str) -> bool: - op = GQL_OPERATIONS["ClaimDrop"].with_variables( - {"input": {"dropInstanceID": claim_id}} - ) - response = await twitch.gql_request(op) - data = response["data"] - if "errors" in data and data["errors"]: - return False - elif "claimDropRewards" in data: - if not data["claimDropRewards"]: - return False - elif ( - data["claimDropRewards"]["status"] - in ["ELIGIBLE_FOR_ALL", "DROP_INSTANCE_ALREADY_CLAIMED"] - ): - return True - return False - - class Game: def __init__(self, data: Dict[str, Any]): self.id: int = int(data["id"]) @@ -45,7 +26,7 @@ class Game: class BaseDrop: def __init__(self, campaign: DropsCampaign, data: Dict[str, Any]): self._twitch: Twitch = campaign._twitch - self.id = data["id"] + self.id: str = data["id"] self.name: str = data["name"] self.campaign: DropsCampaign = campaign self.rewards: List[str] = [b["benefit"]["name"] for b in data["benefitEdges"]] @@ -77,9 +58,25 @@ class BaseDrop: """ if not self.can_claim: return False - assert self.claim_id is not None - self.is_claimed = await claim_drop(self._twitch, self.claim_id) - return self.is_claimed + if self.is_claimed: + return True + op = GQL_OPERATIONS["ClaimDrop"].with_variables( + {"input": {"dropInstanceID": self.claim_id}} + ) + response = await self._twitch.gql_request(op) + data = response["data"] + if "errors" in data and data["errors"]: + return False + elif "claimDropRewards" in data: + if not data["claimDropRewards"]: + return False + elif ( + data["claimDropRewards"]["status"] + in ["ELIGIBLE_FOR_ALL", "DROP_INSTANCE_ALREADY_CLAIMED"] + ): + self.is_claimed = True + return True + return False class TimedDrop(BaseDrop): @@ -92,11 +89,11 @@ class TimedDrop(BaseDrop): self.current_minutes = self.required_minutes @property - def remaining_minutes(self): + def remaining_minutes(self) -> int: return self.required_minutes - self.current_minutes @property - def progress(self): + def progress(self) -> float: return self.current_minutes / self.required_minutes def update(self, message: Dict[str, Any]): diff --git a/main.py b/main.py index 4a6a41d..c5875eb 100644 --- a/main.py +++ b/main.py @@ -13,7 +13,7 @@ import threading from typing import Any, Dict, NoReturn from twitch import Twitch -from constants import SETTINGS_PATH +from constants import SETTINGS_PATH, TERMINATED_STR try: import win32api @@ -36,7 +36,7 @@ assert sys.platform == "win32" def terminate() -> NoReturn: forever = threading.Event() - print("\nApplication Terminated.\nClose the console window to exit the application.") + print(f"\n{TERMINATED_STR}") forever.wait() raise RuntimeError("Uh oh") # this will never run, solely for MyPy diff --git a/twitch.py b/twitch.py index 29bc524..69bbff1 100644 --- a/twitch.py +++ b/twitch.py @@ -25,6 +25,7 @@ from constants import ( GQL_URL, GQL_OPERATIONS, DROPS_ENABLED_TAG, + TERMINATED_STR, GQLOperation, WebsocketTopic, get_topic, @@ -91,7 +92,6 @@ class Twitch: • Selecting a stream to watch, and watching it • Changing the stream that's being watched if necessary """ - # Start our websocket connection while True: # Claim the drops we can self.inventory = await self.get_inventory() @@ -105,14 +105,12 @@ class Twitch: games.add(campaign.game) if drop.can_claim: await drop.claim() - # games now has all games we want to farm drops for + # 'games' now has all games we want to farm drops for # if it's empty, there's no point in continuing if not games: - print( - "No active campaigns to farm drops for.\n\n" - "Application Terminated.\nClose the console window to exit the application.", - ) + print(f"No active campaigns to farm drops for.\n\n{TERMINATED_STR}") await asyncio.Future() + # Start our websocket connection, only after we confirm that there are drops to mine self.websocket.start() if not channel_names: # get a list of all channels with drops enabled