Improve figuring out if new campaigns are claimed or not

inventory.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from functools import cached_property
 from datetime import datetime, timezone
-from typing import Optional, List, Dict, Set, Iterable, TYPE_CHECKING
+from typing import Optional, List, Dict, Iterable, TYPE_CHECKING
 
 from channel import Channel
 from constants import JsonType, GQL_OPERATIONS
@@ -13,7 +13,9 @@ if TYPE_CHECKING:
 
 
 class BaseDrop:
-    def __init__(self, campaign: DropsCampaign, data: JsonType, claimed_benefits: Set[str]):
+    def __init__(
+        self, campaign: DropsCampaign, data: JsonType, claimed_benefits: Dict[str, datetime]
+    ):
         self._twitch: Twitch = campaign._twitch
         self.id: str = data["id"]
         self.name: str = data["name"]
@@ -26,10 +28,22 @@ class BaseDrop:
         if "self" in data:
             self.claim_id = data["self"]["dropInstanceID"]
             self.is_claimed = data["self"]["isClaimed"]
-        elif all(b["benefit"]["id"] in claimed_benefits for b in data["benefitEdges"]):
-            # NOTE: this may erroneously mark drops from an unprogressed campaign as claimed
-            # if the benefits repeat, but shouldn't cause a problem
-            # once the campaign is in-progress
+        elif (
+            # If there's no self edge available, we can use claimed_benefits to determine
+            # (with pretty good certainty) if this drop has been claimed or not.
+            # To do this, we check if the benefitEdges appear in claimed_benefits, and then
+            # deref their "lastAwardedAt" timestamps into a list to check against.
+            # If the benefits were claimed while the drop was active,
+            # the drop has been claimed too.
+            (
+                dts := [
+                    claimed_benefits[bid]
+                    for b in data["benefitEdges"]
+                    if (bid := b["benefit"]["id"]) in claimed_benefits
+                ]
+            )
+            and all(self.starts_at <= dt < self.ends_at for dt in dts)
+        ):
             self.is_claimed = True
         self._precondition_drops: List[str] = [d["id"] for d in (data["preconditionDrops"] or [])]
 
@@ -109,7 +123,9 @@ class BaseDrop:
 
 
 class TimedDrop(BaseDrop):
-    def __init__(self, campaign: DropsCampaign, data: JsonType, claimed_benefits: Set[str]):
+    def __init__(
+        self, campaign: DropsCampaign, data: JsonType, claimed_benefits: Dict[str, datetime]
+    ):
         super().__init__(campaign, data, claimed_benefits)
         self.current_minutes: int = 0
         if "self" in data:
@@ -164,7 +180,7 @@ class TimedDrop(BaseDrop):
 
 
 class DropsCampaign:
-    def __init__(self, twitch: Twitch, data: JsonType, claimed_benefits: Set[str]):
+    def __init__(self, twitch: Twitch, data: JsonType, claimed_benefits: Dict[str, datetime]):
         self._twitch: Twitch = twitch
         self.id: str = data["id"]
         self.name: str = data["name"]

twitch.py

@@ -6,6 +6,7 @@ import logging
 from yarl import URL
 from time import time
 from itertools import chain
+from datetime import datetime
 from functools import partial
 from contextlib import suppress
 from typing import (
@@ -22,7 +23,7 @@ from websocket import WebsocketPool
 from gui import GUIManager, LoginData
 from inventory import DropsCampaign, TimedDrop
 from exceptions import LoginException, CaptchaRequired
-from utils import Game, AwaitableValue, OrderedSet, task_wrapper
+from utils import Game, AwaitableValue, OrderedSet, task_wrapper, timestamp
 from constants import (
     State,
     JsonType,
@@ -734,7 +735,9 @@ class Twitch:
                 continue
         raise RuntimeError(f"Ran out of attempts while handling a GQL request: {op}")
 
-    async def fetch_campaign(self, campaign_id: str, claimed_benefits: Set[str]) -> DropsCampaign:
+    async def fetch_campaign(
+        self, campaign_id: str, claimed_benefits: Dict[str, datetime]
+    ) -> DropsCampaign:
         response = await self.gql_request(
             GQL_OPERATIONS["CampaignDetails"].with_variables(
                 {"channelLogin": str(self._user_id), "dropID": campaign_id}
@@ -756,7 +759,9 @@ class Twitch:
         inventory = response["data"]["currentUser"]["inventory"]
         ongoing_campaigns = inventory["dropCampaignsInProgress"] or []
         # this contains claimed benefit edge IDs, not drop IDs
-        claimed_benefits: Set[str] = set(d["id"] for d in inventory["gameEventDrops"])
+        claimed_benefits: Dict[str, datetime] = {
+            b["id"]: timestamp(b["lastAwardedAt"]) for b in inventory["gameEventDrops"]
+        }
         campaigns: List[DropsCampaign] = [
             DropsCampaign(self, campaign_data, claimed_benefits)
             for campaign_data in ongoing_campaigns