From 5af4ee79c32365a1b19879b87024dda3b7f3b974 Mon Sep 17 00:00:00 2001 From: DevilXD Date: Wed, 30 Mar 2022 14:15:26 +0200 Subject: [PATCH] Remove Game Selector; Implement multi-game mining based on the priority list --- .gitignore | 1 + build.bat | 6 +- channel.py | 6 +- constants.py | 1 - gui.py | 166 +++++++---------------------------- inventory.py | 22 ++++- requirements.txt | 1 + twitch.py | 220 +++++++++++++++++++++++------------------------ utils.py | 2 +- 9 files changed, 164 insertions(+), 261 deletions(-) diff --git a/.gitignore b/.gitignore index 548ca0d..371d617 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,7 @@ __pycache__ /*.tmp /Twitch Drops Miner # Dev files +/env /cache cookies.jar settings.json diff --git a/build.bat b/build.bat index 84057ed..8363247 100644 --- a/build.bat +++ b/build.bat @@ -1,5 +1,5 @@ @echo off cls -pyinstaller --onefile --noconsole --name "Twitch Drops Miner (by DevilXD)" --icon pickaxe.ico ^ ---add-data pickaxe.ico;. main.py -call pack.bat +%CD%/env/scripts/pyinstaller --onefile --noconsole --name "Twitch Drops Miner (by DevilXD)" ^ +--icon pickaxe.ico --add-data pickaxe.ico;. main.py +if %ERRORLEVEL% == 0 call pack.bat diff --git a/channel.py b/channel.py index c488689..351c307 100644 --- a/channel.py +++ b/channel.py @@ -132,7 +132,7 @@ class Channel: return NotImplemented def __hash__(self) -> int: - return hash((self.__class__.__name__, self.id)) + return self.id @property def name(self) -> str: @@ -247,8 +247,8 @@ class Channel: return Stream.from_get_stream(self, stream_data) async def check_online(self) -> bool: - self._stream = stream = await self.get_stream() - if stream is None: + self._stream = await self.get_stream() + if self._stream is None: invalidate_cache(self, "_payload") return False return True diff --git a/constants.py b/constants.py index 7f46a0b..161c0a7 100644 --- a/constants.py +++ b/constants.py @@ -59,7 +59,6 @@ class State(Enum): IDLE = auto() INVENTORY_FETCH = auto() GAMES_UPDATE = auto() - GAME_SELECT = auto() CHANNELS_FETCH = auto() CHANNELS_CLEANUP = auto() CHANNEL_SWITCH = auto() diff --git a/gui.py b/gui.py index da8b377..a613c4e 100644 --- a/gui.py +++ b/gui.py @@ -6,7 +6,7 @@ import tkinter as tk from math import log10, ceil from functools import partial from tkinter.font import Font -from collections import abc, namedtuple, OrderedDict +from collections import abc, namedtuple from tkinter import Tk, ttk, StringVar, DoubleVar, IntVar from typing import Any, TypedDict, NoReturn, TYPE_CHECKING @@ -341,121 +341,6 @@ class LoginForm: self._var.set(f"{status}\n{user_str}") -class GameSelector: - def __init__(self, manager: GUIManager, master: ttk.Widget): - self._manager = manager - self._settings = manager._twitch.settings - self._var = StringVar(master) - frame = ttk.LabelFrame(master, text="Game Selector", padding=(4, 0, 4, 4)) - frame.grid(column=1, row=1, sticky="nsew", padx=2) - frame.rowconfigure(0, weight=1) - frame.columnconfigure(0, weight=1) - self._list = PaddedListbox( - frame, - height=5, - padding=(1, 0), - activestyle="none", - selectmode="single", - highlightthickness=0, - exportselection=False, - ) - self._list.grid(column=0, row=0, sticky="nsew") - self._selection: str | None = None - self._games: OrderedDict[str, tuple[int, Game]] = OrderedDict() - self._excluded_indices: set[int] = set() - self._list.bind("<>", self._on_select) - - def set_games(self, games: abc.Iterable[Game]): - self._games.clear() - self._games.update((str(game), (i, game)) for i, game in enumerate(games)) - self._list.delete(0, "end") - self._list.insert("end", *self._games.keys()) - self._list.config(width=0) # autoadjust listbox width - # gray-out excluded games and relink the selection - self._excluded_indices.clear() - self._exclude_sync() - selected_index: int | None = None - if self._selection is not None and self._selection in self._games: - selected_index = self._games[self._selection][0] - self._list.selection_clear(0, "end") - if selected_index is not None: - # reselect the currently selected item - self._list.selection_set(selected_index) - elif self._selection is not None: - # the game we've had selected isn't there anymore - clear selection - self._selection = None - - def _exclude_update(self, game_name: str, *, index: int | None = None) -> None: - # called from the settings tab or _update below - if index is None: - try: - index = self._games[game_name][0] - except KeyError: - # user added an excluded game that isn't on the list - just return - return - if ( - game_name in self._settings.exclude - or self._settings.priority_only - and game_name not in self._settings.priority - ): - if index not in self._excluded_indices: - self._excluded_indices.add(index) - self._list.itemconfig(index, foreground="gray60") - elif index in self._excluded_indices: - self._excluded_indices.discard(index) - self._list.itemconfig(index, foreground='') - - def _exclude_sync(self) -> None: - # called when priority_only state changes - for game_name, (i, _) in self._games.items(): - self._exclude_update(game_name, index=i) - - def _on_select(self, event: tk.Event[PaddedListbox]) -> None: - current: tuple[int, ...] = self._list.curselection() - if not current: - # can happen when the user clicks on an empty list - return - idx: int = current[0] - if idx in self._excluded_indices: - # user clicked on an excluded game - reselect the previous one if possible - self._list.selection_clear(0, "end") - if self._selection is not None: - for i, game_name in enumerate(self._list.get(0, "end")): - if game_name == self._selection: - self._list.selection_set(i) - break - else: - self._selection = None - return - new_selection: str = self._list.get(idx) - if new_selection != self._selection: - self._selection = new_selection - self._manager._twitch.change_state(State.GAME_SELECT) - - def get_next(self) -> Game | None: - if self._selection is not None: - return self._games[self._selection][1] - # select and return the first prioritized game from the list - self._list.selection_clear(0, "end") - for game_name in self._settings.priority: - if game_name in self._games: - # it's a prioritized game - easy choice - idx, game = self._games[game_name] - self._selection = game_name - self._list.selection_set(idx) - return game - # if priority_only is enabled, we end here - if self._settings.priority_only: - return None - # with priorities out of the way, select and return the first non-excluded game instead - for i, game_name in enumerate(self._list.get(0, "end")): - if i not in self._excluded_indices: - self._selection = game_name - self._list.selection_set(i) - return self._games[game_name][1] - return None - - class _BaseVars(TypedDict): progress: DoubleVar percentage: StringVar @@ -476,28 +361,28 @@ class _ProgressVars(TypedDict): class CampaignProgress: - BAR_LENGTH = 240 + BAR_LENGTH = 420 def __init__(self, manager: GUIManager, master: ttk.Widget): self._manager = manager self._vars: _ProgressVars = { "campaign": { - "name": StringVar(master), # campaign name + "name": StringVar(master, "..."), # campaign name "progress": DoubleVar(master), # controls the progress bar - "percentage": StringVar(master), # percentage display string - "remaining": StringVar(master), # time remaining string + "percentage": StringVar(master, "-%"), # percentage display string + "remaining": StringVar(master), # time remaining string, filled via _update_time }, "drop": { - "rewards": StringVar(master), # drop rewards + "rewards": StringVar(master, "..."), # drop rewards "progress": DoubleVar(master), # as above - "percentage": StringVar(master), # as above + "percentage": StringVar(master, "-%"), # as above "remaining": StringVar(master), # as above }, } self._frame = frame = ttk.LabelFrame( master, text="Campaign Progress", padding=(4, 0, 4, 4) ) - frame.grid(column=0, row=1, sticky="nsew", padx=2) + frame.grid(column=0, row=1, columnspan=2, sticky="nsew", padx=2) frame.columnconfigure(0, weight=2) frame.columnconfigure(1, weight=1) ttk.Label(frame, text="Campaign:").grid(column=0, row=0, columnspan=2) @@ -1100,8 +985,8 @@ class SettingsPanel: AUTOSTART_NAME: str = "TwitchDropsMiner" def __init__(self, manager: GUIManager, master: ttk.Widget): + self._twitch = manager._twitch self._settings = manager._twitch.settings - self._game_selector = manager.games self._vars: _SettingsVars = { "tray": IntVar(master, self._settings.autostart_tray), "autostart": IntVar(master, self._settings.autostart), @@ -1199,7 +1084,16 @@ class SettingsPanel: ttk.Button( exclude_frame, text="❌", command=self.exclude_delete, width=2, style="Large.TButton" ).grid(column=0, row=2, columnspan=2, sticky="ew") - self.priority_only() # update exclude section state + # Reload button + reload_frame = ttk.Frame(center_frame) + reload_frame.grid(column=0, row=1, columnspan=3, pady=4) + ttk.Label( + reload_frame, + text="Most changes require a reload to take an immediate effect: " + ).grid(column=0, row=0) + ttk.Button( + reload_frame, text="Reload", command=self._twitch.state_change(State.INVENTORY_FETCH) + ).grid(column=1, row=0) def clear_selection(self) -> None: self._priority_list.selection_clear(0, "end") @@ -1226,6 +1120,13 @@ class SettingsPanel: self._exclude_entry.config(values=games_list) self._priority_entry.config(values=games_list) + def priorities(self) -> dict[str, int]: + # NOTE: we shift the indexes so that 0 can be used as the default one + size = self._priority_list.size() + return { + game_name: i - size for i, game_name in enumerate(self._priority_list.get(0, "end")) + } + def priority_add(self) -> None: game_name: str = self._priority_entry.get() if not game_name: @@ -1240,7 +1141,6 @@ class SettingsPanel: self._priority_list.insert("end", game_name) self._priority_list.see("end") self._settings.priority.append(game_name) - self._game_selector._exclude_update(game_name) else: # already there, set the selection on it self._priority_list.selection_set(existing_idx) @@ -1272,15 +1172,11 @@ class SettingsPanel: idx: int | None = self._priority_idx() if idx is None: return - # NOTE: get the item before it's removed from the list - item: str = self._priority_list.get(idx) self._priority_list.delete(idx) del self._settings.priority[idx] - self._game_selector._exclude_update(item) def priority_only(self) -> None: self._settings.priority_only = bool(self._vars["priority_only"].get()) - self._game_selector._exclude_sync() def exclude_add(self) -> None: game_name: str = self._exclude_entry.get() @@ -1288,9 +1184,9 @@ class SettingsPanel: # prevent adding empty strings return self._exclude_entry.clear() - e = self._settings.exclude - if game_name not in e: - e.add(game_name) + exclude = self._settings.exclude + if game_name not in exclude: + exclude.add(game_name) # insert it alphabetically for i, item in enumerate(self._exclude_list.get(0, "end")): if game_name < item: @@ -1300,7 +1196,6 @@ class SettingsPanel: else: self._exclude_list.insert("end", game_name) self._exclude_list.see("end") - self._game_selector._exclude_update(game_name) else: # it was already there, select it for i, item in enumerate(self._exclude_list.get(0, "end")): @@ -1322,7 +1217,6 @@ class SettingsPanel: if item in self._settings.exclude: self._settings.exclude.discard(item) self._exclude_list.delete(idx) - self._game_selector._exclude_update(item) class GUIManager: @@ -1388,7 +1282,6 @@ class GUIManager: self.websockets = WebsocketStatus(self, main_frame) self.login = LoginForm(self, main_frame) self.progress = CampaignProgress(self, main_frame) - self.games = GameSelector(self, main_frame) self.output = ConsoleOutput(self, main_frame) self.channels = ChannelList(self, main_frame) # Inventory tab @@ -1498,7 +1391,6 @@ class GUIManager: self._cache.save() def set_games(self, games: abc.Iterable[Game]) -> None: - self.games.set_games(games) self.settings.set_games(games) def display_drop( diff --git a/inventory.py b/inventory.py index cb2bcd1..3c13595 100644 --- a/inventory.py +++ b/inventory.py @@ -75,13 +75,19 @@ class BaseDrop: campaign = self.campaign return all(campaign.timed_drops[pid].is_claimed for pid in self._precondition_drops) - def can_earn(self, channel: Channel | None = None) -> bool: + def _base_can_earn(self) -> bool: return ( self.preconditions # preconditions are met - and not self.is_claimed # drop isn't already claimed - and self.campaign.active # campaign is active - # drop is within the active timeframe + and not self.is_claimed # isn't already claimed + # 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() + # NOTE: this extends drop-only checks with campaign ones + and self.campaign.active # campaign is active # channel isn't specified, or there's no ACL, or the channel is in the ACL and ( channel is None @@ -267,3 +273,11 @@ class DropsCampaign: def get_drop(self, drop_id: str) -> TimedDrop | None: return self.timed_drops.get(drop_id) + + def can_earn(self, channel: Channel | None = None) -> bool: + # True if any of the containing drops can be earned + return ( + self.active + and (channel is None or not self.allowed_channels or channel in self.allowed_channels) + and any(drop._base_can_earn() for drop in self.drops) + ) diff --git a/requirements.txt b/requirements.txt index 1bed197..b2280b9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ aiohttp>2.0,<4.0 +Pillow pystray websockets>10.0 diff --git a/twitch.py b/twitch.py index 29e4fd1..01a27e6 100644 --- a/twitch.py +++ b/twitch.py @@ -4,7 +4,6 @@ import asyncio import logging from yarl import URL from time import time -from itertools import chain from functools import partial from collections import abc, OrderedDict from contextlib import suppress, asynccontextmanager @@ -46,20 +45,15 @@ logger = logging.getLogger("TwitchDrops") gql_logger = logging.getLogger("TwitchDrops.gql") -def viewers_key(channel: Channel) -> int: - if (viewers := channel.viewers) is not None: - return viewers - return -1 - - class Twitch: def __init__(self, settings: Settings): self.settings: Settings = settings # State management self._state: State = State.IDLE self._state_change = asyncio.Event() - self.game: Game | None = None - self.inventory: dict[Game, list[DropsCampaign]] = {} + self.games: dict[Game, int] = {} + self.inventory: list[DropsCampaign] = [] + self._drops: dict[str, TimedDrop] = {} # GUI self.gui = GUIManager(self) # Cookies, session and auth @@ -151,6 +145,22 @@ class Twitch: """ self.gui.print(*args, **kwargs) + @staticmethod + def _viewers_key(channel: Channel) -> int: + if (viewers := channel.viewers) is not None: + return viewers + return -1 + + def _game_key(self, channel: Channel) -> int: + game = channel.game + if game is None: + return 1 + elif game not in self.games: + # in case a channel is gathered from an ACL and doesn't play the expected game, + # we use the same priority as for non-prioritized games + return 0 + return self.games[game] + async def run(self): """ Main method that runs the whole client. @@ -166,6 +176,7 @@ class Twitch: except ExitRequest: # we've been requested to exit during login, most likely return + await self.websocket.start() # NOTE: watch task is explicitly restarted on each new run if self._watching_task is not None: self._watching_task.cancel() @@ -195,42 +206,35 @@ class Twitch: self.change_state(State.GAMES_UPDATE) elif self._state is State.GAMES_UPDATE: # Figure out which games to watch, and claim the drops we can - games: list[Game] = [] - for game, campaigns in self.inventory.items(): - add_game = False - for campaign in campaigns: - if not campaign.upcoming: - # claim drops from expired and active campaigns - active = campaign.active - for drop in campaign.drops: - if drop.can_claim: - await drop.claim() - # add game only for active campaigns - if active and not add_game and drop.can_earn(): - add_game = True - if add_game: - games.append(game) - # 'games' has all games we can mine drops for - # if it's empty, there's no point in continuing - if not games: - self.gui.print("No active campaigns to mine drops for.") - return - # only start the websocket after we confirm there are drops to mine - await self.websocket.start() - self.gui.set_games(games) - self.change_state(State.GAME_SELECT) - elif self._state is State.GAME_SELECT: - new_game: Game | None = self.gui.games.get_next() - if new_game != self.game: - if new_game is not None: - # restart the watch loop immediately on new game selected - self.restart_watching() - self.game = new_game - full_cleanup = True + self.games.clear() + priorities = self.gui.settings.priorities() + # claim drops from expired and active campaigns + for campaign in self.inventory: + if not campaign.upcoming: + for drop in campaign.drops: + if drop.can_claim: + await drop.claim() + # collect games from active campaigns + exclude = self.settings.exclude + priority = self.settings.priority + priority_only = self.settings.priority_only + for campaign in self.inventory: + game = campaign.game + if ( + game not in self.games # isn't already there + and ( # isn't excluded + priority_only and game.name in priority + and game.name not in exclude + ) + and campaign.can_earn() # campaign can be progressed + ): + self.games[game] = priorities.get(game.name, 0) + self.gui.set_games(self.games.keys()) + full_cleanup = True self.change_state(State.CHANNELS_CLEANUP) elif self._state is State.CHANNELS_CLEANUP: - if self.game is None or full_cleanup: - # no game selected or we're doing full cleanup: remove everything + if not self.games or full_cleanup: + # no games selected or we're doing full cleanup: remove everything to_remove: list[Channel] = list(channels.values()) else: # remove all channels that: @@ -242,7 +246,7 @@ class Twitch: and ( channel.offline # and are offline # or online but aren't streaming the game we want anymore - or (channel.game is None or channel.game != self.game) + or (channel.game is None or channel.game not in self.games) ) ) ] @@ -256,15 +260,15 @@ class Twitch: del channels[channel.id] channel.remove() del to_remove - if self.game is not None: + if self.games: self.change_state(State.CHANNELS_FETCH) else: - # with no game selected, we switch to IDLE after cleanup + # with no games available, we switch to IDLE after cleanup + self.gui.print( + "No active campaigns to mine drops for. Waiting for an active campaign..." + ) self.change_state(State.IDLE) elif self._state is State.CHANNELS_FETCH: - if self.game is None: - self.change_state(State.GAME_SELECT) - continue # pre-display the active drop without substracting a minute if (active_drop := self.get_active_drop()) is not None: active_drop.display(countdown=False) @@ -272,15 +276,15 @@ class Twitch: new_channels: OrderedSet[Channel] = OrderedSet(self.channels.values()) # gather and add ACL channels from campaigns # NOTE: we consider only campaigns that can be progressed - # NOTE: we use an other set so that we can set them online separately - no_acl = False + # NOTE: we use another set so that we can set them online separately + no_acl: set[Game] = set() acl_channels: OrderedSet[Channel] = OrderedSet() - for campaign in self.inventory[self.game]: - if any(drop.can_earn() for drop in campaign.drops): + for campaign in self.inventory: + if campaign.game in self.games and campaign.can_earn(): if campaign.allowed_channels: acl_channels.update(campaign.allowed_channels) else: - no_acl = True + no_acl.add(campaign.game) # remove all ACL channels that already exist from the other set acl_channels.difference_update(new_channels) # use the other set to set them online if possible @@ -288,17 +292,17 @@ class Twitch: await asyncio.gather(*(channel.check_online() for channel in acl_channels)) # finally, add them as new channels new_channels.update(acl_channels) - if no_acl: - # if there's at least one campaign without an ACL, + for game in no_acl: + # for every campaign without an ACL, for it's game, # add a list of live channels with drops enabled - new_channels.update(await self.get_live_streams()) - # sort them descending by viewers, - # then by priority so that prioritized ones are first + new_channels.update(await self.get_live_streams(game)) + # sort them descending by viewers, by priority and by game priority # NOTE: We can drop OrderedSet now because there's no more channels being added ordered_channels: list[Channel] = sorted( - new_channels, key=viewers_key, reverse=True + new_channels, key=self._viewers_key, reverse=True ) ordered_channels.sort(key=lambda ch: ch.priority, reverse=True) + ordered_channels.sort(key=self._game_key) # ensure that we won't end up with more channels than we can handle # NOTE: we substract 2 due to the two base topics always being added: # channel points gain and drop update subscriptions @@ -348,9 +352,6 @@ class Twitch: break self.change_state(State.CHANNEL_SWITCH) elif self._state is State.CHANNEL_SWITCH: - if self.game is None: - self.change_state(State.GAME_SELECT) - continue # Change into the selected channel, stay in the watching channel, # or select a new channel that meets the required conditions priority_channels: list[Channel] = [] @@ -362,22 +363,32 @@ class Twitch: if watching_channel is not None: priority_channels.append(watching_channel) # If there's no selected channel, change into a channel we can watch - for channel in chain(priority_channels, channels.values()): + new_watching = None + for channel in priority_channels: if self.can_watch(channel): - self.watch(channel) - # break the state change chain by clearing the flag - self._state_change.clear() + new_watching = channel break + if new_watching is None: + # NOTE: we need to sort the channels every time because one channel + # can end up streaming any game, since channels aren't game-tied + for channel in sorted(channels.values(), key=self._game_key): + if self.can_watch(channel): + new_watching = channel + break + if new_watching is not None: + self.watch(channel) + # break the state change chain by clearing the flag + self._state_change.clear() else: self.gui.print( - f"No suitable channel to watch for game: {self.game}\n" - "Waiting for an ONLINE channel..." + "No available channels to watch. Waiting for an ONLINE channel..." ) self.change_state(State.IDLE) elif self._state is State.EXIT: # we've been requested to exit the application break await self._state_change.wait() + # post-main-loop code goes here async def _watch_sleep(self, delay: float) -> None: # we use wait_for here to allow an asyncio.sleep that can be ended prematurely, @@ -417,7 +428,7 @@ class Twitch: ) if drop_data is not None: drop_id = drop_data["dropID"] - drop = self.get_drop(drop_id) + drop = self._drops.get(drop_id) if drop is None: use_active = True logger.error(f"Missing drop: {drop_id}") @@ -453,20 +464,15 @@ class Twitch: await asyncio.sleep(60) def can_watch(self, channel: Channel) -> bool: - if self.game is None: + if not self.games: return False return ( channel.online # steam online and channel.drops_enabled # drops are enabled - # it's a game we've selected - and channel.game is not None - and channel.game == self.game + # it's one of the games we've selected + and channel.game is not None and channel.game in self.games # we can progress any campaign for the selected game - and any( - drop.can_earn(channel) - for campaign in self.inventory[self.game] - for drop in campaign.drops - ) + and any(campaign.can_earn(channel) for campaign in self.inventory) ) def watch(self, channel: Channel): @@ -513,13 +519,22 @@ class Twitch: Called by a Channel when it goes online (after pending). """ logger.debug(f"{channel.name} goes ONLINE") + channel_order = self._game_key(channel) watching_channel = self.watching_channel.get_with_default(None) + if watching_channel is not None: + watching_order = self._game_key(watching_channel) + else: + watching_order = 1 if ( ( self._state is State.IDLE # we're currently idle - or channel.priority # or this channel has priority - and watching_channel is not None # and we're watching... - and not watching_channel.priority # ... a non-priority channel + or watching_channel is None # or aren't watching anything + # or this channel is higher order than the watching one + # NOTE: order is tied to the list position, so lower == higher + or channel_order < watching_order + or channel_order == watching_order # or the order is the same + and channel.priority # and this channel has priority + and not watching_channel.priority # and we're watching a non-priority channel ) and self.can_watch(channel) ): self.gui.print(f"{channel.name} goes ONLINE, switching...") @@ -546,7 +561,7 @@ class Twitch: if msg_type not in ("drop-progress", "drop-claim"): return drop_id: str = message["data"]["drop_id"] - drop: TimedDrop | None = self.get_drop(drop_id) + drop: TimedDrop | None = self._drops.get(drop_id) if msg_type == "drop-claim": if drop is None: logger.error( @@ -881,43 +896,27 @@ class Twitch: # add campaigns that remained, that can be earned but are not in-progress yet for campaign_id, available_data in available_campaigns.items(): campaign = await self.fetch_campaign(campaign_id, available_data, claimed_benefits) - if any(drop.can_earn() for drop in campaign.drops): + if campaign.can_earn(): campaigns.append(campaign) campaigns.sort(key=lambda c: c.ends_at) - self.inventory.clear() + self._drops.clear() self.gui.inv.clear() + self.inventory.clear() for campaign in campaigns: - game = campaign.game - if game not in self.inventory: - self.inventory[game] = [] + self._drops.update((drop.id, drop) for drop in campaign.drops) await self.gui.inv.add_campaign(campaign) - self.inventory[game].append(campaign) + self.inventory.append(campaign) self.gui.save() - def get_drop(self, drop_id: str) -> TimedDrop | None: - """ - Returns a drop from the inventory, based on it's ID. - """ - # try it with the currently selected game first - if self.game is not None: - for campaign in self.inventory[self.game]: - if (drop := campaign.get_drop(drop_id)) is not None: - return drop - # fallback to checking all campaigns - for campaign in chain(*self.inventory.values()): - if (drop := campaign.get_drop(drop_id)) is not None: - return drop - return None - def get_active_drop(self, channel: Channel | None = None) -> TimedDrop | None: - if self.game is None: + if not self.games: return None watching_channel = self.watching_channel.get_with_default(channel) drops = sorted( ( drop - for campaign in self.inventory[self.game] - if campaign.active + for campaign in self.inventory + if campaign.game in self.games and campaign.can_earn(watching_channel) for drop in campaign.drops if drop.can_earn(watching_channel) ), @@ -927,14 +926,11 @@ class Twitch: return drops[0] return None - async def get_live_streams(self) -> list[Channel]: - if self.game is None: - return [] - limit = 30 + async def get_live_streams(self, game: Game, *, limit: int = 30) -> list[Channel]: response = await self.gql_request( GQL_OPERATIONS["GameDirectory"].with_variables({ "limit": limit, - "name": self.game.name, + "name": game.name, "options": { "includeRestricted": ["SUB_ONLY_LIVE"], "tags": [DROPS_ENABLED_TAG], diff --git a/utils.py b/utils.py index d7fa8ac..25e5327 100644 --- a/utils.py +++ b/utils.py @@ -213,4 +213,4 @@ class Game: return NotImplemented def __hash__(self) -> int: - return hash((self.__class__.__name__, self.id)) + return self.id