Skip to content
This repository has been archived by the owner on Mar 12, 2022. It is now read-only.

Commit

Permalink
feat: supports now 'all' entry in publishers.txt
Browse files Browse the repository at this point in the history 
refactor: uses now graphql response as offer supply
refactor: better logging
fix: bugfix for #7
  • Loading branch information
@sibalzer
sibalzer committed Sep 6, 2021
1 parent 55ab2b4 commit d9128fe
Show file tree
Hide file tree
Showing 4 changed files with 189 additions and 125 deletions.
2 changes: 1 addition & 1 deletion .github/workflows/ci.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,4 +24,4 @@ jobs:
flake8 primelooter.py --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Lint with pylint
run: |
pylint primelooter.py --fail-under 8
pylint primelooter.py --fail-under 10 --disable=all --enable=classes --disable=W
2 changes: 1 addition & 1 deletion README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ If you want to use the provided docker image (only linux/amd64 plattform for now

### 3. 🏢 Create a publishers.txt

Create a publishers.txt like the example file. Each line represents the publisher name used on the [https://gaming.amazon.com](https://gaming.amazon.com) website.
Create a publishers.txt like the example file. Each line represents the publisher name used on the [https://gaming.amazon.com](https://gaming.amazon.com) website (add 'all' to claim all offers).

### 4. 🏃 Run

Expand Down
308 changes: 185 additions & 123 deletions primelooter.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,9 @@
import time
import traceback
import typing
import json

from playwright.sync_api import sync_playwright, Cookie, Browser
from playwright.sync_api import sync_playwright, Cookie, Browser, Page, BrowserContext, ElementHandle


logging.basicConfig(
Expand All @@ -22,140 +23,200 @@
log = logging.getLogger()


class AuthException(Exception):
pass


def loot(cookies, publishers, headless, dump):
with sync_playwright() as playwright:
browser: Browser = playwright.firefox.launch(headless=headless)
context = browser.new_context()
context.add_cookies(cookies)
page = context.new_page()

authentication_tries = 0
authenticated = False
while not authenticated:
if authentication_tries < 5:
authentication_tries += 1
else:
raise AuthException()
page.goto("https://gaming.amazon.com/home")
page.wait_for_load_state('networkidle')

if not page.query_selector('div.sign-in'):
authenticated = True

# Ingame Loot
loot_offer_xpath = 'xpath=(//div[@data-a-target="offer-list-InGameLoot"] | (//div[@data-a-target="offer-list-undefined"])[1])//div[@data-test-selector="Offer"]'
class PrimeLooter():

def __init__(self, cookies, publishers='all', headless=True):
self.cookies = cookies
self.publishers = publishers
self.headless = headless

def __enter__(self):
self.playwright = sync_playwright()
self.browser: Browser = self.playwright.start().firefox.launch(
headless=self.headless)
self.context: BrowserContext = self.browser.new_context()
self.context.add_cookies(self.cookies)
self.page: Page = self.context.new_page()
return self

def __exit__(self, exc_type, exc_val, exc_tb):
self.page.close()
self.context.close()
self.browser.close()
self.playwright.__exit__()

@staticmethod
def exists(tab: Page, selector: str) -> bool:
if tab.query_selector(selector):
return True
return False

def auth(self) -> None:
with self.page.expect_response(lambda response: 'https://gaming.amazon.com/graphql' in response.url and 'currentUser' in response.json()['data']) as response_info:
log.debug('get auth info')
self.page.goto('https://gaming.amazon.com/home')
response = response_info.value.json()['data']['currentUser']
if not response['isSignedIn']:
Exception('Authentication: Not signed in')
elif not response['isAmazonPrime']:
Exception('Authentication: Not a valid Amazon Prime account')
elif not response['isTwitchPrime']:
Exception('Authentication: Not a valid Twitch Prime account')

def get_offers(self) -> list[dict]:
with self.page.expect_response(lambda response: 'https://gaming.amazon.com/graphql' in response.url and 'primeOffers' in response.json()['data']) as response_info:
log.debug('get offers')
self.page.goto('https://gaming.amazon.com/home')
return response_info.value.json()['data']['primeOffers']

@staticmethod
def check_eligibility(offer: dict) -> bool:
if offer['linkedJourney']:
for suboffer in offer['linkedJourney']['offers']:
if suboffer['self']['eligibility']:
return suboffer['self']['eligibility']['canClaim']
return False
elif offer['self']:
return offer['self']['eligibility']['canClaim']
else:
raise Exception(
f'Could not check offer eligibility status\n{json.dumps(offer, indent=4)}')

def claim_external(self, url, publisher):
tab = self.context.new_page()

with tab.expect_response(lambda response: 'https://gaming.amazon.com/graphql' in response.url and 'journey' in response.json()['data']) as response_info:
log.debug('get game title')
tab.goto(url)
game_name = response_info.value.json(
)['data']['journey']['assets']['title']

log.debug("Try to claim %s from %s", game_name, publisher)
tab.wait_for_selector(
'div[data-a-target=loot-card-available]')

try:
page.wait_for_selector(loot_offer_xpath, timeout=1000*30)
except Exception as ex:
log.error("Could not load loot offers. (timeout)")

if dump:
print(page.query_selector('div.home').inner_html())

elements = page.query_selector_all(loot_offer_xpath)
loot_cards = tab.query_selector_all(
'div[data-a-target=loot-card-available]')

if len(elements) == 0:
for loot_card in loot_cards:
loot_name = loot_card.query_selector(
'h3[data-a-target=LootCardSubtitle]').text_content()
log.debug("Try to claim loot %s from %s by %s",
loot_name, game_name, publisher)

claim_button = loot_card.query_selector(
'button[data-test-selector=AvailableButton]')
if not claim_button:
log.warning(
"Could not claim %s from %s by %s (in-game loot)", loot_name, game_name, publisher)
continue

claim_button.click()
tab.wait_for_load_state('networkidle')

# validate
tab.wait_for_selector(
"div[data-a-target=gms-base-modal]")

if PrimeLooter.exists(tab, 'div.gms-success-modal-container'):
log.info("Claimed %s (%s)", loot_name, game_name)

elif PrimeLooter.exists(tab, "div[data-test-selector=ProgressBarSection]"):
log.warning(
"Could not claim %s from %s by %s (account not connected)", loot_name, game_name, publisher)
else:
log.warning(
"Could not claim %s from %s by %s (unknown error)", loot_name, game_name, publisher)
if tab.query_selector('button[data-a-target=close-modal-button]'):
tab.query_selector(
'button[data-a-target=close-modal-button]').click()
except Exception as ex:
print(ex)
log.error(
"No loot offers found! Did they make some changes to the website? Please report @github if this happens multiple times.")
f"An error occured ({publisher}/{game_name})! Did they make some changes to the website? Please report @github if this happens multiple times.")
tab.close()

for i in range(len(elements)):
elem = page.query_selector_all(loot_offer_xpath)[i]
elem.scroll_into_view_if_needed()
elem.wait_for_selector('p.tw-c-text-alt-2')
publisher = elem.query_selector(
'p.tw-c-text-alt-2').text_content()
game_name = elem.query_selector('h3').text_content()
def claim_direct(self):
tab = self.context.new_page()
tab.goto('https://gaming.amazon.com/home')

if elem.query_selector('div[data-a-target=offer-claim-success]'):
log.debug("Already claimed %s by %s", game_name, publisher)
continue
if publisher not in publishers:
continue

with page.expect_navigation():
log.debug("Try to claim %s from %s", game_name, publisher)
elem.query_selector(
'button[data-a-target=ExternalOfferClaim]').click()
page.wait_for_selector(
'div[data-a-target=loot-card-available]')
try:
loot_cards = page.query_selector_all(
'div[data-a-target=loot-card-available]')

for loot_card in loot_cards:
loot_name = loot_card.query_selector(
'h3[data-a-target=LootCardSubtitle]').text_content()
log.debug("Try to claim loot %s from %s by %s",
loot_name, game_name, publisher)

claim_button = loot_card.query_selector(
'button[data-test-selector=AvailableButton]')
if not claim_button:
log.warning(
"Could not claim %s from %s by %s (in-game loot)", loot_name, game_name, publisher)
continue

claim_button.click()
page.wait_for_load_state('networkidle')

# validate
page.wait_for_selector(
"div[data-a-target=gms-base-modal]")

if page.query_selector('div.gms-success-modal-container'):
log.info("Claimed %s (%s)", loot_name, game_name)
elif page.query_selector("div[data-test-selector=ProgressBarSection]"):
log.warning(
"Could not claim %s from %s by %s (account not connected)", loot_name, game_name, publisher)
else:
log.warning(
"Could not claim %s from %s by %s (unknown error)", loot_name, game_name, publisher)
if page.query_selector('button[data-a-target=close-modal-button]'):
page.query_selector(
'button[data-a-target=close-modal-button]').click()
except Exception as ex:
print(ex)
finally:
page.goto("https://gaming.amazon.com/home")
page.wait_for_selector(loot_offer_xpath)
FGWP_XPATH = 'xpath=//button[@data-a-target="FGWPOffer"]/ancestor::div[@data-test-selector="Offer"]'

# Games
loot_offer_xpath = 'xpath=(//div[@data-a-target="offer-list-Game"] | (//div[@data-a-target="offer-list-undefined"])[2])//div[@data-test-selector="Offer"]'

try:
page.wait_for_selector(loot_offer_xpath, timeout=1000*30)
except:
log.error("Could not load game offers. (timeout)")

elements = page.query_selector_all(loot_offer_xpath)
elements = self.page.query_selector_all(FGWP_XPATH)

if len(elements) == 0:
log.error(
"No game offers found! Did they make some changes to the website? Please report @github if this happens multiple times.")
"No direct offers found! Did they make some changes to the website? Please report @github if this happens multiple times.")

for elem in elements:
elem.scroll_into_view_if_needed()
page.wait_for_load_state('networkidle')
self.page.wait_for_load_state('networkidle')

publisher = elem.query_selector(
'p.tw-c-text-alt-2').text_content()
game_name = elem.query_selector('h3').text_content()

if elem.query_selector('div[data-a-target=offer-claim-success]'):
log.debug("Already claimed %s by %s", game_name, publisher)
continue

log.debug("Try to claim %s", game_name)
log.debug("Try to claim %s by %s", game_name, publisher)
elem.query_selector("button[data-a-target=FGWPOffer]").click()
log.info("Claimed %s", game_name)
log.info("Claimed %s by %s", game_name, publisher)

context.close()
browser.close()
tab.close()

def run(self, publishers: list[str] = ['all'], dump: bool = False):
self.auth()

if dump:
print(self.page.query_selector('div.home').inner_html())
offers = self.get_offers()

not_claimable_offers = [offer for offer in offers if offer.get(
'linkedJourney') == None and offer.get('self') == None]
external_offers = [
offer for offer in offers if offer['deliveryMethod'] == 'EXTERNAL_OFFER' and offer not in not_claimable_offers and PrimeLooter.check_eligibility(offer)]
direct_offers = [
offer for offer in offers if offer['deliveryMethod'] == 'DIRECT_ENTITLEMENT' and PrimeLooter.check_eligibility(offer)]

# list non claimable offers
msg = "Can not claim these ingame offers:"
for offer in not_claimable_offers:
msg += f"\n - {offer['title']}"
msg = msg[:-1]
log.info(msg)

# claim direct offers
if direct_offers:
msg = "Claiming these direct offers:"
for offer in direct_offers:
msg += f"\n - {offer['title']}"
msg = msg[:-1]
log.info(msg)
self.claim_direct()
else:
log.info("No direct offers to Claim")

# filter publishers
if not 'all' in self.publishers:
external_offers = [offer for offer in external_offers if offer['content']
['publisher'] in self.publishers]

# claim external offers
if external_offers:
msg = "Claiming these external offers:"
for offer in external_offers:
msg += f"\n - {offer['title']}"
msg = msg[:-1]
log.info(msg)

for offer in external_offers:
try:
if PrimeLooter.check_eligibility(offer):
self.claim_external(
offer['content']['externalURL'], offer['content']['publisher'])
except Exception as ex:
log.error(ex)
else:
log.info("No external offers to Claim")


def read_cookiefile(path: str) -> typing.List[Cookie]:
Expand Down Expand Up @@ -225,11 +286,10 @@ def read_cookiefile(path: str) -> typing.List[Cookie]:
if arg['loop']:
while True:
try:
log.info("start looting cycle")
loot(cookies, publishers, headless, dump)
log.info("finished looting cycle")
except AuthException:
log.error("Authentication failed!")
with PrimeLooter(cookies, publishers, headless) as looter:
log.info("start looting cycle")
looter.run(dump)
log.info("finished looting cycle")
except Exception as ex:
log.error("Error %s", ex)
traceback.print_tb(ex.__traceback__)
Expand All @@ -238,6 +298,8 @@ def read_cookiefile(path: str) -> typing.List[Cookie]:
time.sleep(60*60*24)
else:
try:
loot(cookies, publishers, headless, dump)
except AuthException:
log.error("Authentication failed!")
with PrimeLooter(cookies, publishers, headless) as looter:
looter.run(dump)
except Exception as ex:
log.error("Error %s", ex)
traceback.print_tb(ex.__traceback__)
2 changes: 2 additions & 0 deletions publishers.example.txt
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
all
-----------------
Amanotes Pte. Ltd.
Bungie, Inc.
Deep Silver
Expand Down

0 comments on commit d9128fe

Please sign in to comment.