-
-
Notifications
You must be signed in to change notification settings - Fork 173
/
inventory.py
400 lines (336 loc) · 14.3 KB
/
inventory.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
from __future__ import annotations
import re
import math
from itertools import chain
from typing import TYPE_CHECKING
from functools import cached_property
from datetime import datetime, timedelta, timezone
from channel import Channel
from constants import GQL_OPERATIONS, URLType
from utils import timestamp, invalidate_cache, Game
if TYPE_CHECKING:
from collections import abc
from twitch import Twitch
from constants import JsonType
from gui import GUIManager, InventoryOverview
DIMS_PATTERN = re.compile(r'-\d+x\d+(?=\.(?:jpg|png|gif)$)', re.I)
def remove_dimensions(url: URLType) -> URLType:
return URLType(DIMS_PATTERN.sub('', url))
class Benefit:
__slots__ = ("id", "name", "image_url")
def __init__(self, data: JsonType):
benefit_data: JsonType = data["benefit"]
self.id: str = benefit_data["id"]
self.name: str = benefit_data["name"]
self.image_url: URLType = benefit_data["imageAssetURL"]
class BaseDrop:
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"]
self.campaign: DropsCampaign = campaign
self.benefits: list[Benefit] = [Benefit(b) for b in data["benefitEdges"]]
self.starts_at: datetime = timestamp(data["startAt"])
self.ends_at: datetime = timestamp(data["endAt"])
self.claim_id: str | None = None
self.is_claimed: bool = False
if "self" in data:
self.claim_id = data["self"]["dropInstanceID"]
self.is_claimed = data["self"]["isClaimed"]
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 benefit in self.benefits
if (bid := 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 [])]
def __repr__(self) -> str:
if self.is_claimed:
additional = ", claimed=True"
elif self.can_earn():
additional = ", can_earn=True"
else:
additional = ''
return f"Drop({self.rewards_text()}{additional})"
@cached_property
def preconditions_met(self) -> bool:
campaign = self.campaign
return all(campaign.timed_drops[pid].is_claimed for pid in self._precondition_drops)
def _base_earn_conditions(self) -> bool:
# define when a drop can be earned or not
return (
self.preconditions_met # preconditions are met
and not self.is_claimed # isn't already claimed
)
def _base_can_earn(self) -> bool:
# cross-participates in can_earn and can_earn_within handling, where a timeframe is added
return (
self._base_earn_conditions()
# is within the timeframe
and self.starts_at <= datetime.now(timezone.utc) < self.ends_at
)
def can_earn(self, channel: Channel | None = None) -> bool:
return self._base_can_earn() and self.campaign._base_can_earn(channel)
def can_earn_within(self, stamp: datetime) -> bool:
return (
self._base_earn_conditions()
and self.ends_at > datetime.now(timezone.utc)
and self.starts_at < stamp
)
@property
def can_claim(self) -> bool:
# https://help.twitch.tv/s/article/mission-based-drops?language=en_US#claiming
# "If you are unable to claim the Drop in time, you will be able to claim it
# from the Drops Inventory page until 24 hours after the Drops campaign has ended."
return (
self.claim_id is not None
and not self.is_claimed
and datetime.now(timezone.utc) < self.campaign.ends_at + timedelta(hours=24)
)
def _on_claim(self) -> None:
invalidate_cache(self, "preconditions_met")
def update_claim(self, claim_id: str):
self.claim_id = claim_id
async def generate_claim(self) -> None:
# claim IDs now appear to be constructed from other IDs we have access to
# Format: UserID#CampaignID#DropID
# NOTE: This marks a drop as a ready-to-claim, so we may want to later ensure
# its mining progress is finished first
auth_state = await self.campaign._twitch.get_auth()
self.claim_id = f"{auth_state.user_id}#{self.campaign.id}#{self.id}"
def rewards_text(self, delim: str = ", ") -> str:
return delim.join(benefit.name for benefit in self.benefits)
async def claim(self) -> bool:
result = await self._claim()
if result:
self.is_claimed = result
# notify the campaign about claiming
# this will cause it to call our _on_claim, so no need to call it ourselves here
self.campaign._on_claim()
return result
async def _claim(self) -> bool:
"""
Returns True if the claim succeeded, False otherwise.
"""
if self.is_claimed:
return True
if not self.can_claim:
return False
response = await self._twitch.gql_request(
GQL_OPERATIONS["ClaimDrop"].with_variables(
{"input": {"dropInstanceID": self.claim_id}}
)
)
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 TimedDrop(BaseDrop):
def __init__(
self, campaign: DropsCampaign, data: JsonType, claimed_benefits: dict[str, datetime]
):
super().__init__(campaign, data, claimed_benefits)
self._manager: GUIManager = self._twitch.gui
self._gui_inv: InventoryOverview = self._manager.inv
self.current_minutes: int = "self" in data and data["self"]["currentMinutesWatched"] or 0
self.required_minutes: int = data["requiredMinutesWatched"]
if self.is_claimed:
# claimed drops may report inconsistent current minutes, so we need to overwrite them
self.current_minutes = self.required_minutes
def __repr__(self) -> str:
if self.is_claimed:
additional = ", claimed=True"
elif self.can_earn():
additional = ", can_earn=True"
else:
additional = ''
if 0 < self.current_minutes < self.required_minutes:
minutes = f", {self.current_minutes}/{self.required_minutes}"
else:
minutes = ''
return f"Drop({self.rewards_text()}{minutes}{additional})"
@cached_property
def remaining_minutes(self) -> int:
return self.required_minutes - self.current_minutes
@cached_property
def total_required_minutes(self) -> int:
return self.required_minutes + max(
(
self.campaign.timed_drops[pid].total_required_minutes
for pid in self._precondition_drops
),
default=0,
)
@cached_property
def total_remaining_minutes(self) -> int:
return self.remaining_minutes + max(
(
self.campaign.timed_drops[pid].total_remaining_minutes
for pid in self._precondition_drops
),
default=0,
)
@cached_property
def progress(self) -> float:
if self.current_minutes <= 0 or self.required_minutes <= 0:
return 0.0
elif self.current_minutes >= self.required_minutes:
return 1.0
return self.current_minutes / self.required_minutes
@property
def availability(self) -> float:
now = datetime.now(timezone.utc)
if self.required_minutes > 0 and self.total_remaining_minutes > 0 and now < self.ends_at:
return ((self.ends_at - now).total_seconds() / 60) / self.total_remaining_minutes
return math.inf
def _base_earn_conditions(self) -> bool:
return super()._base_earn_conditions() and self.required_minutes > 0
def _on_claim(self) -> None:
result = super()._on_claim()
self._gui_inv.update_drop(self)
return result
def _on_minutes_changed(self) -> None:
invalidate_cache(self, "progress", "remaining_minutes")
self.campaign._on_minutes_changed()
self._gui_inv.update_drop(self)
def _on_total_minutes_changed(self) -> None:
invalidate_cache(self, "total_required_minutes", "total_remaining_minutes")
async def claim(self) -> bool:
result = await super().claim()
if result:
self.current_minutes = self.required_minutes
return result
def update_minutes(self, minutes: int):
if minutes < 0:
return
elif minutes <= self.required_minutes:
self.current_minutes = minutes
else:
self.current_minutes = self.required_minutes
self._on_minutes_changed()
self.display()
def display(self, *, countdown: bool = True, subone: bool = False):
self._manager.display_drop(self, countdown=countdown, subone=subone)
def bump_minutes(self):
if self.current_minutes < self.required_minutes:
self.current_minutes += 1
self._on_minutes_changed()
self.display()
class DropsCampaign:
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"]
self.game: Game = Game(data["game"])
self.linked: bool = data["self"]["isAccountConnected"]
self.link_url: str = data["accountLinkURL"]
# campaign's image actually comes from the game object
# we use regex to get rid of the dimensions part (ex. ".../game_id-285x380.jpg")
self.image_url: URLType = remove_dimensions(data["game"]["boxArtURL"])
self.starts_at: datetime = timestamp(data["startAt"])
self.ends_at: datetime = timestamp(data["endAt"])
allowed: JsonType = data["allow"]
self.allowed_channels: list[Channel] = (
[Channel.from_acl(twitch, channel_data) for channel_data in allowed["channels"]]
if allowed["channels"] and allowed.get("isEnabled", True) else []
)
self.timed_drops: dict[str, TimedDrop] = {
drop_data["id"]: TimedDrop(self, drop_data, claimed_benefits)
for drop_data in data["timeBasedDrops"]
}
def __repr__(self) -> str:
return f"Campaign({self.game!s}, {self.name}, {self.claimed_drops}/{self.total_drops})"
@property
def drops(self) -> abc.Iterable[TimedDrop]:
return self.timed_drops.values()
@property
def time_triggers(self) -> set[datetime]:
return set(
chain(
(self.starts_at, self.ends_at),
*((d.starts_at, d.ends_at) for d in self.timed_drops.values()),
)
)
@property
def active(self) -> bool:
return self.starts_at <= datetime.now(timezone.utc) < self.ends_at
@property
def upcoming(self) -> bool:
return datetime.now(timezone.utc) < self.starts_at
@property
def expired(self) -> bool:
return self.ends_at <= datetime.now(timezone.utc)
@property
def total_drops(self) -> int:
return len(self.timed_drops)
@cached_property
def finished(self) -> bool:
return all(d.is_claimed for d in self.drops)
@cached_property
def claimed_drops(self) -> int:
return sum(d.is_claimed for d in self.drops)
@cached_property
def remaining_drops(self) -> int:
return sum(not d.is_claimed for d in self.drops)
@cached_property
def required_minutes(self) -> int:
return max(d.total_required_minutes for d in self.drops)
@cached_property
def remaining_minutes(self) -> int:
return max(d.total_remaining_minutes for d in self.drops)
@cached_property
def progress(self) -> float:
return sum(d.progress for d in self.drops) / self.total_drops
@property
def availability(self) -> float:
return min(d.availability for d in self.drops)
def _on_claim(self) -> None:
invalidate_cache(self, "finished", "claimed_drops", "remaining_drops")
for drop in self.drops:
drop._on_claim()
def _on_minutes_changed(self) -> None:
invalidate_cache(self, "progress", "required_minutes", "remaining_minutes")
for drop in self.drops:
drop._on_total_minutes_changed()
def get_drop(self, drop_id: str) -> TimedDrop | None:
return self.timed_drops.get(drop_id)
def _base_can_earn(self, channel: Channel | None = None) -> bool:
return (
self.linked # account is connected
and self.active # campaign is active
# channel isn't specified, or there's no ACL, or the channel is in the ACL
and (channel is None or not self.allowed_channels or channel in self.allowed_channels)
)
def can_earn(self, channel: Channel | None = None) -> bool:
# True if any of the containing drops can be earned
return self._base_can_earn(channel) and any(drop._base_can_earn() for drop in self.drops)
def can_earn_within(self, stamp: datetime) -> bool:
# Same as can_earn, but doesn't check the channel
# and uses a future timestamp to see if we can earn this campaign later
return (
self.linked
and self.ends_at > datetime.now(timezone.utc)
and self.starts_at < stamp
and any(drop.can_earn_within(stamp) for drop in self.drops)
)