Small code cleanup

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]):

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]):

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
 

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