Skip to content

Commit

Permalink
Format with Black
Browse files Browse the repository at this point in the history
  • Loading branch information
@srhinos
srhinos committed Oct 16, 2023
1 parent 83e532f commit 1e2755a
Show file tree
Hide file tree
Showing 3 changed files with 39 additions and 117 deletions.
34 changes: 8 additions & 26 deletions experiment.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,39 +17,28 @@
}


async def claim_offer(
offer_id: str, item: dict, client: httpx.AsyncClient, headers: dict
) -> True:
async def claim_offer(offer_id: str, item: dict, client: httpx.AsyncClient, headers: dict) -> True:
if item["offers"][0]["offerSelfConnection"]["eligibility"]["isClaimed"] != True:
if (
item["offers"][0]["offerSelfConnection"]["eligibility"]["canClaim"] == False
and item["offers"][0]["offerSelfConnection"]["eligibility"][
"missingRequiredAccountLink"
]
== True
and item["offers"][0]["offerSelfConnection"]["eligibility"]["missingRequiredAccountLink"] == True
):
log.error(
f"Cannot collect game `{item['game']['assets']['title']}`, account link required."
)
log.error(f"Cannot collect game `{item['game']['assets']['title']}`, account link required.")
return
log.info(f"Collecting offer for {item['game']['assets']['title']}")
claim_payload = {
"operationName": "placeOrdersDetailPage",
"variables": {
"input": {
"offerIds": [offer_id],
"attributionChannel": '{"eventId":"ItemDetailRootPage:'
+ offer_id
+ '","page":"ItemDetailPage"}',
"attributionChannel": '{"eventId":"ItemDetailRootPage:' + offer_id + '","page":"ItemDetailPage"}',
}
},
"extensions": {},
"query": "fragment Place_Orders_Payload_Order_Information on OfferOrderInformation {\n catalogOfferId\n claimCode\n entitledAccountId\n entitledAccountName\n id\n orderDate\n orderState\n __typename\n}\n\nmutation placeOrdersDetailPage($input: PlaceOrdersInput!) {\n placeOrders(input: $input) {\n error {\n code\n __typename\n }\n orderInformation {\n ...Place_Orders_Payload_Order_Information\n __typename\n }\n __typename\n }\n}\n",
}

response = await client.post(
gql_url, headers=headers, data=json.dumps(claim_payload)
)
response = await client.post(gql_url, headers=headers, data=json.dumps(claim_payload))
if response.json()["data"]["placeOrders"]["error"] != None:
log.error(f"Error: {response.json()['data']['placeOrders']['error']}")

Expand All @@ -68,20 +57,13 @@ async def primelooter(cookie_file):
for _c in jar:
client.cookies.jar.set_cookie(_c)

html_body = (
await client.get("https://gaming.amazon.com/home", headers=base_headers)
).text
html_body = (await client.get("https://gaming.amazon.com/home", headers=base_headers)).text
matches = re.findall(r"name='csrf-key' value='(.*)'", html_body)
json_headers["csrf-token"] = matches[0]

response = await client.post(
gql_url, headers=json_headers, data=json.dumps(offers_payload)
)
response = await client.post(gql_url, headers=json_headers, data=json.dumps(offers_payload))
data = response.json()["data"]["inGameLoot"]["items"]

coros = await asyncio.gather(
*[
claim_offer(item["offers"][0]["id"], item, client, json_headers)
for item in data
]
*[claim_offer(item["offers"][0]["id"], item, client, json_headers) for item in data]
)
118 changes: 30 additions & 88 deletions legacy.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,9 @@ def __enter__(self):
self.playwright = sync_playwright()

if self.use_chrome:
self.browser: Browser = self.playwright.start().chromium.launch(
headless=self.headless
)
self.browser: Browser = self.playwright.start().chromium.launch(headless=self.headless)
else:
self.browser: Browser = self.playwright.start().firefox.launch(
headless=self.headless
)
self.browser: Browser = self.playwright.start().firefox.launch(headless=self.headless)

self.context: BrowserContext = self.browser.new_context()
self.context.add_cookies(self.cookies)
Expand All @@ -51,16 +47,10 @@ def __exit__(self, exc_type, exc_val, exc_tb):
self.playwright.__exit__()

@staticmethod
def code_to_file(
game: str, code: str, instructions: str, seperator_string: str = ""
) -> None:
seperator_string = (
seperator_string or "========================\n========================"
)
def code_to_file(game: str, code: str, instructions: str, seperator_string: str = "") -> None:
seperator_string = seperator_string or "========================\n========================"
with open("./game_codes.txt", "a") as f:
f.write(
f"{game}: {code}\n\n{instructions.replace('/n',' ')}\n{seperator_string}\n"
)
f.write(f"{game}: {code}\n\n{instructions.replace('/n',' ')}\n{seperator_string}\n")

@staticmethod
def exists(tab: Page, selector: str) -> bool:
Expand All @@ -77,9 +67,7 @@ def auth(self) -> None:
self.page.goto("https://gaming.amazon.com/home")
response = response_info.value.json()["data"]["currentUser"]
if not response["isSignedIn"]:
raise AuthException(
"Authentication: Not signed in. (Please recreate the cookie.txt file)"
)
raise AuthException("Authentication: Not signed in. (Please recreate the cookie.txt file)")
elif not response["isAmazonPrime"]:
raise AuthException(
"Authentication: Not a valid Amazon Prime account. "
Expand Down Expand Up @@ -111,9 +99,7 @@ def check_eligibility(offer: dict) -> bool:
return False
if offer["self"]:
return offer["self"]["eligibility"]["canClaim"]
raise Exception(
f"Could not check offer eligibility status\n{json.dumps(offer, indent=4)}"
)
raise Exception(f"Could not check offer eligibility status\n{json.dumps(offer, indent=4)}")

@staticmethod
def check_claim_status(offer: dict) -> bool:
Expand All @@ -125,9 +111,7 @@ def check_claim_status(offer: dict) -> bool:
return False
if offer["self"]:
return offer["self"]["eligibility"]["isClaimed"]
raise Exception(
f"Could not check offer eligibility status\n{json.dumps(offer, indent=4)}"
)
raise Exception(f"Could not check offer eligibility status\n{json.dumps(offer, indent=4)}")

def claim_external(self, url, publisher):
if "loot" not in url:
Expand All @@ -145,32 +129,20 @@ def claim_external(self, url, publisher):
) as response_info:
log.debug("get game title")
tab.goto(url)
game_name = response_info.value.json()["data"]["journey"]["assets"][
"title"
]
game_name = response_info.value.json()["data"]["journey"]["assets"]["title"]

log.debug(f"Try to claim {game_name} from {publisher}")
tab.wait_for_selector("div[data-a-target=loot-card-available]")

loot_cards = tab.query_selector_all(
"div[data-a-target=loot-card-available]"
)
loot_cards = tab.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(
f"Try to claim loot {loot_name} from {game_name} by {publisher}"
)
loot_name = loot_card.query_selector("h3[data-a-target=LootCardSubtitle]").text_content()
log.debug(f"Try to claim loot {loot_name} from {game_name} by {publisher}")

claim_button = loot_card.query_selector(
"button[data-a-target=AvailableButton]"
)
claim_button = loot_card.query_selector("button[data-a-target=AvailableButton]")
if not claim_button:
log.warning(
f"Could not claim {loot_name} from {game_name} by {publisher} (in-game loot)"
)
log.warning(f"Could not claim {loot_name} from {game_name} by {publisher} (in-game loot)")
continue

claim_button.click()
Expand All @@ -179,12 +151,8 @@ def claim_external(self, url, publisher):

# Sometimes we get forced into having to click to claim within the base modal after claiming normally
# Mainly occurs for Apex Legends claims.
if PrimeLooter.exists(
tab, 'div[class*="--current"][data-a-target="Step-3"]'
):
complete_claim_button = loot_card.query_selector(
"button[data-a-target=gms-cta]"
)
if PrimeLooter.exists(tab, 'div[class*="--current"][data-a-target="Step-3"]'):
complete_claim_button = loot_card.query_selector("button[data-a-target=gms-cta]")
if complete_claim_button:
complete_claim_button.click()

Expand All @@ -201,32 +169,20 @@ def claim_external(self, url, publisher):
.strip()
)
instructions = (
tab.query_selector(
"div[data-a-target=gms-claim-instructions]"
)
.inner_text()
.strip()
tab.query_selector("div[data-a-target=gms-claim-instructions]").inner_text().strip()
)
PrimeLooter.code_to_file(game_name, code, instructions)
except Exception:
log.warning(
f"Could not get code for {loot_name} ({game_name}) from {publisher}"
)
log.warning(f"Could not get code for {loot_name} ({game_name}) from {publisher}")

elif PrimeLooter.exists(
tab, 'div[class*="--current"][data-a-target="Step-2"]'
) or PrimeLooter.exists(tab, "div[data-a-target=gms-progress-bar]"):
log.warning(
f"Could not claim {loot_name} from {game_name} by {publisher} (account not connected)"
)
elif PrimeLooter.exists(tab, 'div[class*="--current"][data-a-target="Step-2"]') or PrimeLooter.exists(
tab, "div[data-a-target=gms-progress-bar]"
):
log.warning(f"Could not claim {loot_name} from {game_name} by {publisher} (account not connected)")
else:
log.warning(
f"Could not claim {loot_name} from {game_name} by {publisher} (unknown error)"
)
log.warning(f"Could not claim {loot_name} from {game_name} by {publisher} (unknown error)")
if tab.query_selector("button[data-a-target=close-modal-button]"):
tab.query_selector(
"button[data-a-target=close-modal-button]"
).click()
tab.query_selector("button[data-a-target=close-modal-button]").click()
except Error as ex:
print(ex)
traceback.print_tb(ex.__traceback__)
Expand Down Expand Up @@ -264,9 +220,7 @@ def claim_direct(self):
)

for direct_offer in direct_offers:
game_body = direct_offer.query_selector(
"div[class='item-card-details__body__primary']"
)
game_body = direct_offer.query_selector("div[class='item-card-details__body__primary']")

game_name = game_body.query_selector("h3").text_content()
# publisher = game_body.query_selector("p.tw-c-text-base").text_content()
Expand All @@ -288,15 +242,10 @@ def run(self, dump: bool = False):
offers = self.get_offers()

not_claimable_offers = [
offer
for offer in offers
if offer.get("linkedJourney") is None and offer.get("self") is None
offer for offer in offers if offer.get("linkedJourney") is None and offer.get("self") is None
]
claimed_offers = [
offer
for offer in offers
if offer not in not_claimable_offers
and PrimeLooter.check_claim_status(offer)
offer for offer in offers if offer not in not_claimable_offers and PrimeLooter.check_claim_status(offer)
]
external_offers = [
offer
Expand All @@ -308,8 +257,7 @@ def run(self, dump: bool = False):
direct_offers = [
offer
for offer in offers
if offer["deliveryMethod"] == "DIRECT_ENTITLEMENT"
and PrimeLooter.check_eligibility(offer)
if offer["deliveryMethod"] == "DIRECT_ENTITLEMENT" and PrimeLooter.check_eligibility(offer)
]

# list non claimable offers
Expand Down Expand Up @@ -341,11 +289,7 @@ def run(self, dump: bool = False):
log.info("No direct offers to claim\n")
# filter publishers
if "all" not in self.publishers:
external_offers = [
offer
for offer in external_offers
if offer["content"]["publisher"] in self.publishers
]
external_offers = [offer for offer in external_offers if offer["content"]["publisher"] in self.publishers]

# claim external offers
if external_offers:
Expand All @@ -357,9 +301,7 @@ def run(self, dump: bool = False):
log.info(msg)

for offer in external_offers:
self.claim_external(
offer["content"]["externalURL"], offer["content"]["publisher"]
)
self.claim_external(offer["content"]["externalURL"], offer["content"]["publisher"])
else:
log.info("No external offers to claim\n")

Expand Down
4 changes: 1 addition & 3 deletions primelooter.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -54,9 +54,7 @@ def use_experimental_api(cookie_file):


if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Notification bot for the lower saxony vaccination portal"
)
parser = argparse.ArgumentParser(description="Notification bot for the lower saxony vaccination portal")

parser.add_argument(
"--legacy",
Expand Down

0 comments on commit 1e2755a

Please sign in to comment.