diff --git a/.vscode/launch.json b/.vscode/launch.json index c236f32..8d01e11 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -16,7 +16,7 @@ "type": "python", "request": "launch", "program": "main.py", - "args": ["-vv"], + "args": ["--debug", "-vv"], "console": "integratedTerminal", "justMyCode": false }, @@ -25,7 +25,7 @@ "type": "python", "request": "launch", "program": "main.py", - "args": ["-vvv"], + "args": ["--debug", "-vvv"], "console": "integratedTerminal", "justMyCode": false }, @@ -34,7 +34,7 @@ "type": "python", "request": "launch", "program": "main.py", - "args": ["-vvv", "--debug-ws"], + "args": ["--debug", "-vvv", "--debug-ws"], "console": "integratedTerminal", "justMyCode": false }, diff --git a/channel.py b/channel.py index 4189e98..c3d993e 100644 --- a/channel.py +++ b/channel.py @@ -6,9 +6,9 @@ import asyncio import logging from base64 import b64encode from datetime import datetime, timezone -from typing import Any, Optional, TYPE_CHECKING +from typing import Any, Optional, SupportsInt, TYPE_CHECKING -from inventory import Game +from utils import Game from exceptions import MinerException from constants import JsonType, BASE_URL, GQL_OPERATIONS, ONLINE_DELAY, DROPS_ENABLED_TAG @@ -28,9 +28,7 @@ class Stream: self.viewers: int = stream["viewersCount"] self.drops_enabled: bool = any(tag["id"] == DROPS_ENABLED_TAG for tag in stream["tags"]) settings = data["broadcastSettings"] - self.game: Optional[Game] = None - if settings["game"] is not None: - self.game = Game(settings["game"]) + self.game: Optional[Game] = Game(settings["game"]) if settings["game"] else None self.title: str = settings["title"] self._timestamp = datetime.now(timezone.utc) @@ -39,38 +37,34 @@ class Stream: self = super().__new__(cls) self._twitch = channel._twitch self.channel = channel - self.broadcast_id = data["id"] + self.broadcast_id = int(data["id"]) self.viewers = data["viewersCount"] self.drops_enabled = any(tag["id"] == DROPS_ENABLED_TAG for tag in data["tags"]) - self.game = Game(data["game"]) + self.game = Game(data["game"]) # has to be there since we searched with it self.title = data["title"] self._timestamp = datetime.now(timezone.utc) return self class Channel: - async def __new__(cls, *args, **kwargs): - """ - Enables __init__ to be async. - The instance is returned after initialization completes. - """ - self = super().__new__(cls) - await self.__init__(*args, **kwargs) # type: ignore - return self - - async def __init__(self, twitch: Twitch, channel_name: str): # type: ignore + def __init__( + self, twitch: Twitch, channel_id: SupportsInt, channel_name: str, *, priority: bool = False + ): self._twitch: Twitch = twitch - self.id: int = 0 # temp, to be filled by get_stream + self.id: int = int(channel_id) self.name: str = channel_name self.url: str = f"{BASE_URL}/{channel_name}" self._spade_url: Optional[str] = None self.points: Optional[int] = None self._stream: Optional[Stream] = None self._pending_stream_up: Optional[asyncio.Task[Any]] = None - await self.get_stream() + # Priority channels are considered first when switching channels, + # and if we're watching a non-priority channel, + # a priority channel going up triggers a switch + self.priority: bool = priority @classmethod - def from_directory(cls, twitch: Twitch, data: JsonType): + def from_directory(cls, twitch: Twitch, data: JsonType, *, priority: bool = False) -> Channel: self = super().__new__(cls) self._twitch = twitch channel = data["broadcaster"] @@ -81,6 +75,24 @@ class Channel: self.points = None self._stream = Stream.from_directory(self, data) self._pending_stream_up = None + self.priority = priority + return self + + @classmethod + async def from_name( + cls, twitch: Twitch, channel_name: str, *, priority: bool = False + ) -> Channel: + self = super().__new__(cls) + self._twitch = twitch + # self.id to be filled by get_stream + self.name = channel_name + self.url = f"{BASE_URL}/{channel_name}" + self._spade_url = None + self.points = None + self._stream = None + self._pending_stream_up = None + self.priority = priority + await self.get_stream() return self def __repr__(self) -> str: @@ -136,10 +148,10 @@ class Channel: self.display() @property - def drops_enabled(self) -> Optional[bool]: + def drops_enabled(self) -> bool: if self._stream is not None: return self._stream.drops_enabled - return None + return False def display(self): self._twitch.gui.channels.display(self) @@ -200,9 +212,11 @@ class Channel: """ self.display() await asyncio.sleep(ONLINE_DELAY.total_seconds()) - await self.check_online() + online = await self.check_online() self._pending_stream_up = None self.display() + if online: + self._twitch.on_online(self) def set_online(self): """ @@ -228,6 +242,7 @@ class Channel: if self.online: self._stream = None self.display() + self._twitch.on_offline(self) async def claim_bonus(self): """ diff --git a/constants.py b/constants.py index 4e4999d..61a8831 100644 --- a/constants.py +++ b/constants.py @@ -48,13 +48,13 @@ FORMATTER = logging.Formatter( class State(Enum): IDLE = auto() - EXIT = auto() INVENTORY_FETCH = auto() GAMES_UPDATE = auto() GAME_SELECT = auto() CHANNELS_FETCH = auto() CHANNELS_CLEANUP = auto() CHANNEL_SWITCH = auto() + EXIT = auto() class GQLOperation(JsonType): diff --git a/inventory.py b/inventory.py index dbff0b0..b0680bc 100644 --- a/inventory.py +++ b/inventory.py @@ -2,42 +2,16 @@ from __future__ import annotations from datetime import datetime from functools import cached_property -from typing import Optional, List, Dict, Set, TYPE_CHECKING +from typing import Optional, List, Dict, Set, Iterable, TYPE_CHECKING +from channel import Channel +from utils import invalidate_cache, Game from constants import JsonType, GQL_OPERATIONS if TYPE_CHECKING: from twitch import Twitch -def _invalidate_cache(instance, *attrnames): - for name in attrnames: - try: - delattr(instance, name) - except AttributeError: - pass - - -class Game: - def __init__(self, data: JsonType): - self.id: int = int(data["id"]) - self.name: str = data["name"] - - def __str__(self) -> str: - return self.name - - def __repr__(self) -> str: - return f"Game({self.id}, {self.name})" - - def __eq__(self, other: object): - if isinstance(other, self.__class__): - return self.id == other.id - return NotImplemented - - def __hash__(self) -> int: - return hash((self.__class__.__name__, self.id)) - - class BaseDrop: def __init__(self, campaign: DropsCampaign, data: JsonType, claimed_benefits: Set[str]): self._twitch: Twitch = campaign._twitch @@ -53,6 +27,9 @@ class BaseDrop: 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 self.is_claimed = True self._precondition_drops: List[str] = [d["id"] for d in (data["preconditionDrops"] or [])] @@ -84,7 +61,7 @@ class BaseDrop: return self.claim_id is not None def _on_claim(self) -> None: - _invalidate_cache(self, "preconditions") + invalidate_cache(self, "preconditions") def update_claim(self, claim_id: str): self.claim_id = claim_id @@ -161,7 +138,7 @@ class TimedDrop(BaseDrop): return self.current_minutes / self.required_minutes def _on_minutes_changed(self) -> None: - _invalidate_cache(self, "progress", "remaining_minutes") + invalidate_cache(self, "progress", "remaining_minutes") self.campaign._on_minutes_changed() async def claim(self) -> bool: @@ -192,9 +169,13 @@ class DropsCampaign: self.starts_at: datetime = datetime.strptime(data["startAt"], "%Y-%m-%dT%H:%M:%SZ") self.ends_at: datetime = datetime.strptime(data["endAt"], "%Y-%m-%dT%H:%M:%SZ") allowed = data["allow"] - self.allowed_channels: List[str] = [] - if allowed["channels"] is not None: - self.allowed_channels.extend(ch["name"] for ch in allowed["channels"]) + self.allowed_channels: List[Channel] = ( + [ + Channel(twitch, ch["id"], ch["displayName"], priority=True) + for ch in allowed["channels"] + ] + if allowed["channels"] and allowed["isEnabled"] else [] + ) self.timed_drops: Dict[str, TimedDrop] = { drop_data["id"]: TimedDrop(self, drop_data, claimed_benefits) for drop_data in data["timeBasedDrops"] @@ -203,6 +184,10 @@ class DropsCampaign: def __repr__(self) -> str: return f"Campaign({self.name}({self.game!s}), {self.claimed_drops}/{self.total_drops})" + @property + def drops(self) -> Iterable[TimedDrop]: + return self.timed_drops.values() + @property def active(self): return self.starts_at <= datetime.utcnow() < self.ends_at @@ -221,11 +206,11 @@ class DropsCampaign: @cached_property def claimed_drops(self) -> int: - return sum(d.is_claimed for d in self.timed_drops.values()) + 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.timed_drops.values()) + return sum(not d.is_claimed for d in self.drops) @cached_property def remaining_minutes(self) -> int: @@ -233,12 +218,12 @@ class DropsCampaign: @cached_property def progress(self) -> float: - return sum(d.progress for d in self.timed_drops.values()) / self.total_drops + return sum(d.progress for d in self.drops) / self.total_drops def _on_claim(self) -> None: - _invalidate_cache(self, "claimed_drops", "remaining_drops") - for drop in self.timed_drops.values(): + invalidate_cache(self, "claimed_drops", "remaining_drops") + for drop in self.drops: drop._on_claim() def _on_minutes_changed(self) -> None: - _invalidate_cache(self, "progress", "remaining_minutes") + invalidate_cache(self, "progress", "remaining_minutes") diff --git a/main.py b/main.py index 940ab37..7611e61 100644 --- a/main.py +++ b/main.py @@ -21,6 +21,7 @@ class ParsedArgs(argparse.Namespace): _debug_gql: bool log: bool tray: bool + debug: bool game: Optional[str] @property @@ -54,15 +55,6 @@ class ParsedArgs(argparse.Namespace): return logging.NOTSET -# check if we're not already running -try: - exists = ctypes.windll.user32.FindWindowW(None, WINDOW_TITLE) -except AttributeError: - # we're not on Windows - continue - exists = False -if exists: - # already running - exit - sys.exit() # handle input parameters parser = argparse.ArgumentParser( "Twitch Drops Miner (by DevilXD).exe", @@ -70,12 +62,22 @@ parser = argparse.ArgumentParser( ) parser.add_argument("-V", "--version", action="version", version=f"v{__version__}") parser.add_argument("-v", dest="_verbose", action="count", default=0) +parser.add_argument("--debug", dest="debug", action="store_true") parser.add_argument("--debug-ws", dest="_debug_ws", action="store_true") parser.add_argument("--debug-gql", dest="_debug_gql", action="store_true") parser.add_argument("-g", "--game", default=None) parser.add_argument("--tray", action="store_true") parser.add_argument("-l", "--log", action="store_true") options: ParsedArgs = parser.parse_args(namespace=ParsedArgs()) +# check if we're not already running +try: + exists = ctypes.windll.user32.FindWindowW(None, WINDOW_TITLE) +except AttributeError: + # we're not on Windows - continue + exists = False +if exists and not options.debug: + # already running - exit + sys.exit() # handle logging stuff if options.logging_level > logging.DEBUG: # redirect the root logger into a NullHandler, effectively ignoring all logging calls diff --git a/twitch.py b/twitch.py index c979e00..b1e4fd4 100644 --- a/twitch.py +++ b/twitch.py @@ -8,7 +8,7 @@ from time import time from itertools import chain from functools import partial from typing import ( - Optional, Union, List, Dict, Set, Callable, Iterable, Generic, TypeVar, cast, TYPE_CHECKING + Optional, Union, List, Dict, Set, OrderedDict, Callable, Iterable, cast, TYPE_CHECKING ) try: @@ -17,10 +17,11 @@ except ModuleNotFoundError as exc: raise ImportError("You have to run 'pip install aiohttp' first") from exc from channel import Channel +from websocket import WebsocketPool from gui import GUIManager, LoginData -from websocket import WebsocketPool, task_wrapper from inventory import DropsCampaign, Game, TimedDrop from exceptions import LoginException, CaptchaRequired +from utils import AwaitableValue, OrderedSet, task_wrapper from constants import ( State, JsonType, @@ -44,35 +45,6 @@ logger = logging.getLogger("TwitchDrops") gql_logger = logging.getLogger("TwitchDrops.gql") -_V = TypeVar("_V") -_D = TypeVar("_D") - - -class _AwaitableValue(Generic[_V]): - def __init__(self): - self._value: _V - self._event = asyncio.Event() - - def has_value(self) -> bool: - return self._event.is_set() - - def get_with_default(self, default: _D) -> Union[_D, _V]: - if self._event.is_set(): - return self._value - return default - - async def get(self) -> _V: - await self._event.wait() - return self._value - - def set(self, value: _V) -> None: - self._value = value - self._event.set() - - def clear(self) -> None: - self._event.clear() - - class Twitch: def __init__(self, loop: asyncio.AbstractEventLoop, options: ParsedArgs): self._loop = loop @@ -95,10 +67,11 @@ class Twitch: # State management self._state: State = State.INVENTORY_FETCH self._state_change = asyncio.Event() - self.inventory: List[DropsCampaign] = [] + self.inventory: Dict[Game, List[DropsCampaign]] = {} + self.game: Optional[Game] = None # Storing and watching channels - self.channels: Dict[int, Channel] = {} - self._watching_channel: _AwaitableValue[Channel] = _AwaitableValue() + self.channels: OrderedDict[int, Channel] = OrderedDict() + self.watching_channel: AwaitableValue[Channel] = AwaitableValue() self._watching_task: Optional[asyncio.Task[None]] = None self._watching_restart = asyncio.Event() self._drop_update: Optional[asyncio.Future[bool]] = None @@ -166,7 +139,7 @@ class Twitch: await asyncio.sleep(start_time + 1 - time()) def is_watching(self, channel: Channel) -> bool: - watching_channel = self._watching_channel.get_with_default(None) + watching_channel = self.watching_channel.get_with_default(None) return watching_channel is not None and watching_channel == channel async def run(self): @@ -188,7 +161,6 @@ class Twitch: WebsocketTopic("User", "CommunityPoints", self._user_id, self.process_points), ]) games: List[Game] = [] - selected_game: Optional[Game] = None self.change_state(State.INVENTORY_FETCH) while True: if self._state is State.IDLE: @@ -200,17 +172,23 @@ class Twitch: elif self._state is State.GAMES_UPDATE: # Figure out which games to watch, and claim the drops we can games.clear() - for campaign in self.inventory: - # we have no use in processing upcoming campaigns here - if not campaign.upcoming: - for drop in campaign.timed_drops.values(): - if drop.can_claim: - await drop.claim() - if drop.can_earn and (game := campaign.game) not in games: - games.append(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) self.change_state(State.GAME_SELECT) elif self._state is State.GAME_SELECT: - # 'games' has all games we want to mine drops for + # '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.") @@ -218,20 +196,25 @@ class Twitch: # only start the websocket after we confirm there are drops to mine await self.websocket.start() self.gui.games.set_games(games) - selected_game = self.gui.games.get_selection() + self.game = self.gui.games.get_selection() # pre-display the active drop without a countdown - active_drop = self.get_active_drop(selected_game) + active_drop = self.get_active_drop() if active_drop is not None: active_drop.display(countdown=False) + self.restart_watching() self.change_state(State.CHANNELS_CLEANUP) elif self._state is State.CHANNELS_CLEANUP: # remove all channels that are offline, # or aren't streaming the game we want anymore - to_remove: List[Channel] = [ - channel for channel in self.channels.values() - if not (channel.online or channel.pending_online) - or channel.game is None or channel.game != selected_game - ] + if self.game is None: + # remove everything + to_remove: List[Channel] = list(self.channels.values()) + else: + to_remove = [ + channel for channel in self.channels.values() + if not (channel.online or channel.pending_online) + or channel.game is None or channel.game != self.game + ] self.websocket.remove_topics( WebsocketTopic.as_str("Channel", "VideoPlayback", channel.id) for channel in to_remove @@ -245,26 +228,42 @@ class Twitch: self.gui.channels.shrink() self.change_state(State.CHANNELS_FETCH) elif self._state is State.CHANNELS_FETCH: - if selected_game is None: + if self.game is None: self.change_state(State.GAME_SELECT) else: - # get a list of all live channels with drops enabled - live_streams: List[Channel] = await self.get_live_streams( - selected_game, [DROPS_ENABLED_TAG] - ) - if live_streams: + # pre-display the active drop without substracting a minute + active_drop = self.get_active_drop() + if active_drop is not None: + active_drop.display(countdown=False) + # gather ACLs from campaigns + no_acl = False + new_channels: OrderedSet[Channel] = OrderedSet() + for campaign in self.inventory[self.game]: + acls = campaign.allowed_channels + if acls: + for channel in acls: + new_channels.add(channel) + else: + no_acl = True + # set them online if possible + await asyncio.gather(*(channel.get_stream() for channel in new_channels)) + if no_acl: + # if there's at least one game without an ACL, + # get a list of all live channels with drops enabled + live_streams: List[Channel] = await self.get_live_streams() + for channel in live_streams: + new_channels.add(channel) + if new_channels: # there are online streams, so let's pre-display the active drop again, # but this time with a substracted minute - active_drop = self.get_active_drop(selected_game) + active_drop = self.get_active_drop() if active_drop is not None: active_drop.display(countdown=False, subone=True) # add them, filtering out ones we already have - for channel in live_streams: + for channel in new_channels: if channel.id not in self.channels: self.channels[channel.id] = channel channel.display() - # load points - # asyncio.gather(*(channel.claim_bonus() for channel in live_streams)) # Sub to these channel updates topics: List[WebsocketTopic] = [ WebsocketTopic( @@ -275,7 +274,7 @@ class Twitch: self.websocket.add_topics(topics) self.change_state(State.CHANNEL_SWITCH) elif self._state is State.CHANNEL_SWITCH: - if selected_game is None: + if self.game is None: self.change_state(State.GAME_SELECT) else: # Change into the selected channel, stay in the watching channel, @@ -286,24 +285,20 @@ class Twitch: if selected_channel is not None: self.gui.channels.clear_selection() priority_channels.append(selected_channel) - watching_channel = self._watching_channel.get_with_default(None) + watching_channel = self.watching_channel.get_with_default(None) if watching_channel is not None: priority_channels.append(watching_channel) channels = chain(priority_channels, self.channels.values()) # If there's no selected channel, change into a channel we can watch for channel in channels: - if ( - channel.online # steam online - and channel.drops_enabled # drops are enabled - and channel.game == selected_game # it's a game we've selected - ): + if self.can_watch(channel): self.watch(channel) # break the state change chain by clearing the flag self._state_change.clear() break else: self.stop_watching() - self.gui.print(f"No suitable channel to watch for game: {selected_game}") + self.gui.print(f"No suitable channel to watch for game: {self.game}") self.change_state(State.IDLE) elif self._state is State.EXIT: # we've been requested to exit the application @@ -315,7 +310,7 @@ class Twitch: interval = WATCH_INTERVAL.total_seconds() i = 1 while True: - channel = await self._watching_channel.get() + channel = await self.watching_channel.get() await channel.send_watch() last_watch = time() self._drop_update = asyncio.Future() @@ -347,8 +342,7 @@ class Twitch: # Sometimes, even GQL fails to give us the correct drop. # In that case, we can use the locally cached inventory to try # and put together the drop that we're actually mining right now - selected_game = self.gui.games.get_selection() - drop = self.get_active_drop(selected_game) + drop = self.get_active_drop() if drop is not None: drop.bump_minutes() drop.display() @@ -371,27 +365,37 @@ class Twitch: except asyncio.TimeoutError: pass + def can_watch(self, channel: Channel) -> bool: + if self.game is None: + return False + return ( + channel.online # steam online + and channel.drops_enabled # drops are enabled + and channel.game == self.game # it's a game we've selected + # we can progress any campaign for the selected game + and any( + drop.can_earn + for campaign in self.inventory[self.game] + if not campaign.allowed_channels or channel in campaign.allowed_channels + for drop in campaign.drops + ) + ) + def watch(self, channel: Channel): if self.is_watching(channel): # we're already watching the same channel, so there's no point switching return self.gui.channels.set_watching(channel) - self._watching_channel.set(channel) + self.watching_channel.set(channel) def stop_watching(self): self.gui.progress.stop_timer() self.gui.channels.clear_watching() - self._watching_channel.clear() + self.watching_channel.clear() - def restart_watching(self, channel: Optional[Channel] = None): - # this forcibly re-sends the watching payload to the specified or currently watched channel - if channel is None: - channel = self._watching_channel.get_with_default(None) - if channel is not None: - self.gui.progress.stop_timer() - self._watching_channel.set(channel) - self.gui.channels.set_watching(channel) - self._watching_restart.set() + def restart_watching(self): + self.gui.progress.stop_timer() + self._watching_restart.set() @task_wrapper async def process_stream_state(self, channel_id: int, message: JsonType): @@ -401,14 +405,8 @@ class Twitch: logger.error(f"Stream state change for a non-existing channel: {channel_id}") return if msg_type == "stream-down": - logger.info(f"{channel.name} goes OFFLINE") channel.set_offline() - if self.is_watching(channel): - self.gui.print(f"{channel.name} goes OFFLINE, switching...") - # change the channel if we're currently watching it - self.change_state(State.CHANNEL_SWITCH) elif msg_type == "stream-up": - logger.info(f"{channel.name} goes ONLINE") channel.set_online() elif msg_type == "viewcount": if not channel.online: @@ -417,7 +415,28 @@ class Twitch: else: viewers = message["viewers"] channel.viewers = viewers - logger.debug(f"{channel.name} viewers: {viewers}") + # logger.debug(f"{channel.name} viewers: {viewers}") + + def on_online(self, channel: Channel): + """ + Called by a Channel when it goes online (after pending). + """ + logger.debug(f"{channel.name} goes ONLINE") + if channel.priority: + wch = self.watching_channel.get_with_default(None) + if wch is not None and not wch.priority and self.can_watch(channel): + self.watch(channel) + + def on_offline(self, channel: Channel): + """ + Called by a Channel when it goes offline. + """ + # change the channel if we're currently watching it + if self.is_watching(channel): + self.gui.print(f"{channel.name} goes OFFLINE, switching...") + self.change_state(State.CHANNEL_SWITCH) + else: + logger.debug(f"{channel.name} goes OFFLINE") @task_wrapper async def process_drops(self, user_id: int, message: JsonType): @@ -706,9 +725,10 @@ class Twitch: # fetch all available campaign IDs, that are currently ACTIVE and account is connected response = await self.gql_request(GQL_OPERATIONS["Campaigns"]) data = response["data"]["currentUser"]["dropCampaigns"] or [] + applicable_statuses = ("ACTIVE", "UPCOMING") available_campaigns: Set[str] = set( c["id"] for c in data - if c["status"] == "ACTIVE" and c["self"]["isAccountConnected"] + if c["status"] in applicable_statuses and c["self"]["isAccountConnected"] ) # fetch in-progress campaigns (inventory) response = await self.gql_request(GQL_OPERATIONS["Inventory"]) @@ -727,26 +747,36 @@ class Twitch: # add campaigns that remained, that can be earned but are not in-progress yet for campaign_id in available_campaigns: campaign = await self.fetch_campaign(campaign_id, claimed_benefits) - if any(drop.can_earn for drop in campaign.timed_drops.values()): + if any(drop.can_earn for drop in campaign.drops): campaigns.append(campaign) campaigns.sort(key=lambda c: c.ends_at) - self.inventory = campaigns + self.inventory.clear() + for campaign in campaigns: + game = campaign.game + if game not in self.inventory: + self.inventory[game] = [] + self.inventory[game].append(campaign) def get_drop(self, drop_id: str) -> Optional[TimedDrop]: - for campaign in self.inventory: + for campaign in chain(*self.inventory.values()): drop = campaign.timed_drops.get(drop_id) if drop is not None: return drop return None - def get_active_drop(self, game: Game) -> Optional[TimedDrop]: + def get_active_drop(self) -> Optional[TimedDrop]: + if self.game is None: + return None + watching_channel = self.watching_channel.get_with_default(None) drops = sorted( ( drop - for campaign in self.inventory - if campaign.active and campaign.game == game - for drop in campaign.timed_drops.values() - if drop.can_earn + for campaign in self.inventory[self.game] + if campaign.active + for drop in campaign.drops + if drop.can_earn and ( + watching_channel is None or watching_channel in campaign.allowed_channels + ) ), key=lambda d: d.remaining_minutes, ) @@ -754,15 +784,17 @@ class Twitch: return drops[0] return None - async def get_live_streams(self, game: Game, tag_ids: List[str]) -> List[Channel]: + async def get_live_streams(self) -> List[Channel]: + if self.game is None: + return [] limit = 45 response = await self.gql_request( GQL_OPERATIONS["GameDirectory"].with_variables({ "limit": limit, - "name": game.name, + "name": self.game.name, "options": { "includeRestricted": ["SUB_ONLY_LIVE"], - "tags": tag_ids, + "tags": [DROPS_ENABLED_TAG], }, }) ) diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..74211c9 --- /dev/null +++ b/utils.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import random +import string +import asyncio +import logging +from functools import wraps +from contextlib import suppress +from collections import OrderedDict +from typing import Union, List, MutableSet, Iterable, Iterator, Generic, TypeVar + +from constants import JsonType + + +_V = TypeVar("_V") +_D = TypeVar("_D") +logger = logging.getLogger("TwitchDrops") +NONCE_CHARS = string.ascii_letters + string.digits + + +def create_nonce(length: int = 30) -> str: + return ''.join(random.choices(NONCE_CHARS, k=length)) + + +def deduplicate(iterable: Iterable[_V]) -> List[_V]: + return list(OrderedDict.fromkeys(iterable).keys()) + + +def task_wrapper(afunc): + @wraps(afunc) + async def wrapper(self, *args, **kwargs): + try: + await afunc(self, *args, **kwargs) + except Exception: + logger.exception("Exception in task") + raise # raise up to the wrapping task + return wrapper + + +def invalidate_cache(instance, *attrnames): + """ + To be used to invalidate `functools.cached_property`. + """ + for name in attrnames: + with suppress(AttributeError): + delattr(instance, name) + + +class OrderedSet(MutableSet[_V]): + """ + Implementation of a set that preserves insertion order, + based on OrderedDict with values set to None. + """ + def __init__(self, iterable: Iterable[_V] = []): + self._items: OrderedDict[_V, None] = OrderedDict((item, None) for item in iterable) + + def __repr__(self) -> str: + return f"{self.__class__.__name__}([{', '.join(map(repr, self._items))}])" + + def __contains__(self, x: object) -> bool: + return x in self._items + + def __iter__(self) -> Iterator[_V]: + return iter(self._items) + + def __len__(self) -> int: + return len(self._items) + + def add(self, item: _V) -> None: + self._items[item] = None + + def discard(self, item: _V) -> None: + with suppress(KeyError): + del self._items[item] + + +class AwaitableValue(Generic[_V]): + def __init__(self): + self._value: _V + self._event = asyncio.Event() + + def has_value(self) -> bool: + return self._event.is_set() + + def get_with_default(self, default: _D) -> Union[_D, _V]: + if self._event.is_set(): + return self._value + return default + + async def get(self) -> _V: + await self._event.wait() + return self._value + + def set(self, value: _V) -> None: + self._value = value + self._event.set() + + def clear(self) -> None: + self._event.clear() + + +class Game: + def __init__(self, data: JsonType): + self.id: int = int(data["id"]) + self.name: str = data["name"] + + def __str__(self) -> str: + return self.name + + def __repr__(self) -> str: + return f"Game({self.id}, {self.name})" + + def __eq__(self, other: object): + if isinstance(other, self.__class__): + return self.id == other.id + return NotImplemented + + def __hash__(self) -> int: + return hash((self.__class__.__name__, self.id)) diff --git a/websocket.py b/websocket.py index b2476c7..7615492 100644 --- a/websocket.py +++ b/websocket.py @@ -1,12 +1,9 @@ from __future__ import annotations import json -import random -import string import asyncio import logging from time import time -from functools import wraps from contextlib import suppress from typing import Optional, List, Dict, Set, Iterable, TYPE_CHECKING @@ -14,6 +11,7 @@ from websockets.exceptions import ConnectionClosed, ConnectionClosedOK from websockets.client import WebSocketClientProtocol, connect as websocket_connect from exceptions import MinerException +from utils import task_wrapper, create_nonce from constants import ( JsonType, WebsocketTopic, @@ -30,22 +28,6 @@ if TYPE_CHECKING: logger = logging.getLogger("TwitchDrops") ws_logger = logging.getLogger("TwitchDrops.websocket") -NONCE_CHARS = string.ascii_letters + string.digits - - -def create_nonce(length: int = 30) -> str: - return ''.join(random.choices(NONCE_CHARS, k=length)) - - -def task_wrapper(afunc): - @wraps(afunc) - async def wrapper(self, *args, **kwargs): - try: - await afunc(self, *args, **kwargs) - except Exception: - logger.exception("Exception in task") - raise # raise up to the wrapping task - return wrapper class Websocket: