diff --git a/README.md b/README.md index 4c1c296..9e3f5b7 100644 --- a/README.md +++ b/README.md @@ -6,15 +6,12 @@ This application allows you to AFK mine timed Twitch drops, without having to wo - Stream-less drop mining - save on bandwidth. - Sharded websocket connection, allowing for tracking up to `8*50-2=398` channels at the same time. +- Automatic drop campaigns discovery based on linked accounts (requires you to do [account linking](https://www.twitch.tv/drops/campaigns) yourself though) - Stream tags and drop campaign validation, so you don't end up watching a stream that can't earn you the drop. - Channel stream switching, when the one you were currently watching goes offline. - Cookie saving between sessions, so you don't need to login every time. - Mining is stopped automatically when the last available drop has been mined. -**Not managed / TODO / manual:** - -- Adding additional drop campaigns from other games to your inventory (requires you to do [account linking](https://www.twitch.tv/drops/campaigns) yourself) - **Usage:** - Download the [lastest release](https://github.com/DevilXD/TwitchDropsMiner/releases) - it's recommended to keep it in the folder it comes with. @@ -32,7 +29,6 @@ This application allows you to AFK mine timed Twitch drops, without having to wo ### "No active campaigns to mine drops for." -- Check your [inventory page](https://www.twitch.tv/drops/inventory) for in-progress campaigns. -- If there's none, use the [campaigns page](https://www.twitch.tv/drops/campaigns) to link your account to the campaign's game of choice. -- If your account is linked but the [inventory page](https://www.twitch.tv/drops/inventory) still doesn't include the campaign, go back to the [campaigns page](https://www.twitch.tv/drops/campaigns), expand the game of choice's campaign, and in the "How to Earn the Drop" section, click on "Go to a participating live channel" (or select one of the channels it lists) and make sure to watch the stream for at least one minute - this will activate the campaign. You can verify the campaign's activation on the [inventory page](https://www.twitch.tv/drops/inventory). -- If there are any in-progress campaigns, make sure that they're started and not expired. You can check when a campaign starts by hovering over the very first drop in the campaign. +- Check your [inventory page](https://www.twitch.tv/drops/inventory) for in-progress campaigns - make sure you're not looking at upcoming campaigns. You can check when a campaign starts by hovering over the very first drop in the campaign. +- Make sure to link your account on the [campaigns page](https://www.twitch.tv/drops/campaigns), to the campaign's game of choice. +- Once properly linked, the game should become available in the application's Game Selector. diff --git a/inventory.py b/inventory.py index 71748a7..1418333 100644 --- a/inventory.py +++ b/inventory.py @@ -2,7 +2,7 @@ from __future__ import annotations from datetime import datetime from functools import cached_property -from typing import Optional, List, Dict, TYPE_CHECKING +from typing import Optional, List, Dict, Set, TYPE_CHECKING from constants import JsonType, GQL_OPERATIONS @@ -39,7 +39,7 @@ class Game: class BaseDrop: - def __init__(self, campaign: DropsCampaign, data: JsonType): + def __init__(self, campaign: DropsCampaign, data: JsonType, claimed_benefits: Set[str]): self._twitch: Twitch = campaign._twitch self.id: str = data["id"] self.name: str = data["name"] @@ -47,9 +47,13 @@ class BaseDrop: self.rewards: List[str] = [b["benefit"]["name"] for b in data["benefitEdges"]] 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") - # If claim_id is not None, we can use it to claim the drop - self.claim_id: Optional[str] = data["self"]["dropInstanceID"] - self.is_claimed: bool = data["self"]["isClaimed"] + self.claim_id: Optional[str] = None + self.is_claimed: bool = False + 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"]): + self.is_claimed = True self._precondition_drops: List[str] = [d["id"] for d in (data["preconditionDrops"] or [])] def __repr__(self) -> str: @@ -101,10 +105,10 @@ class BaseDrop: """ Returns True if the claim succeeded, False otherwise. """ - if not self.can_claim: - return False 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}} @@ -125,9 +129,11 @@ class BaseDrop: class TimedDrop(BaseDrop): - def __init__(self, campaign: DropsCampaign, data: JsonType): - super().__init__(campaign, data) - self.current_minutes: int = data["self"]["currentMinutesWatched"] + def __init__(self, campaign: DropsCampaign, data: JsonType, claimed_benefits: Set[str]): + super().__init__(campaign, data, claimed_benefits) + self.current_minutes: int = 0 + if "self" in data: + self.current_minutes = data["self"]["currentMinutesWatched"] self.required_minutes: int = data["requiredMinutesWatched"] if self.is_claimed: # claimed drops report 0 current minutes, so we need to make a correction @@ -178,7 +184,7 @@ class TimedDrop(BaseDrop): class DropsCampaign: - def __init__(self, twitch: Twitch, data: JsonType): + def __init__(self, twitch: Twitch, data: JsonType, claimed_benefits: Set[str]): self._twitch: Twitch = twitch self.id: str = data["id"] self.name: str = data["name"] @@ -190,7 +196,8 @@ class DropsCampaign: if allowed["channels"] is not None: self.allowed_channels.extend(ch["name"] for ch in allowed["channels"]) self.timed_drops: Dict[str, TimedDrop] = { - d["id"]: TimedDrop(self, d) for d in data["timeBasedDrops"] + drop_data["id"]: TimedDrop(self, drop_data, claimed_benefits) + for drop_data in data["timeBasedDrops"] } def __repr__(self) -> str: diff --git a/twitch.py b/twitch.py index d263009..4efc933 100644 --- a/twitch.py +++ b/twitch.py @@ -9,7 +9,7 @@ from time import time from itertools import chain from functools import partial from typing import ( - Callable, Iterable, Optional, Union, List, Dict, Generic, TypeVar, cast, TYPE_CHECKING + Optional, Union, List, Dict, Set, Callable, Iterable, Generic, TypeVar, cast, TYPE_CHECKING ) try: @@ -697,13 +697,42 @@ class Twitch: gql_logger.debug(f"GQL Response: {response_json}") return response_json - async def fetch_inventory(self) -> None: - response = await self.gql_request(GQL_OPERATIONS["Inventory"]) - inventory = response["data"]["currentUser"]["inventory"]["dropCampaignsInProgress"] or [] - campaigns = sorted( - (DropsCampaign(self, data) for data in inventory), - key=lambda c: c.ends_at, + async def fetch_campaign(self, campaign_id: str, claimed_benefits: Set[str]) -> DropsCampaign: + response = await self.gql_request( + GQL_OPERATIONS["CampaignDetails"].with_variables( + {"channelLogin": str(self._user_id), "dropID": campaign_id} + ) ) + return DropsCampaign(self, response["data"]["user"]["dropCampaign"], claimed_benefits) + + async def fetch_inventory(self) -> None: + # 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 [] + available_campaigns: Set[str] = set( + c["id"] for c in data + if c["status"] == "ACTIVE" and c["self"]["isAccountConnected"] + ) + # fetch in-progress campaigns (inventory) + response = await self.gql_request(GQL_OPERATIONS["Inventory"]) + 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"]) + campaigns: List[DropsCampaign] = [ + DropsCampaign(self, campaign_data, claimed_benefits) + for campaign_data in ongoing_campaigns + ] + # filter out in-progress campaigns from all available campaigns, + # since we already have all information needed for them + for campaign in campaigns: + available_campaigns.discard(campaign.id) + # 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()): + campaigns.append(campaign) + campaigns.sort(key=lambda c: c.ends_at) self.inventory = campaigns def get_drop(self, drop_id: str) -> Optional[TimedDrop]: