from logging import getLogger from typing import List, Callable from PyQt5.QtCore import pyqtSignal, QObject from PyQt5.QtWidgets import QApplication from rare.components.tabs.store.api.debug import DebugDialog from rare.components.tabs.store.constants import ( wishlist_query, search_query, wishlist_add_query, wishlist_remove_query, ) from rare.components.tabs.store.shop_models import BrowseModel from rare.utils.paths import cache_dir from rare.utils.qt_requests import QtRequests from .api.models.query import SearchStoreQuery from .api.models.response import ( DieselProduct, ResponseModel, CatalogOfferModel, ) logger = getLogger("ShopAPICore") graphql_url = "https://graphql.epicgames.com/graphql" DEBUG: Callable[[], bool] = lambda: "--debug" in QApplication.arguments() class ShopApiCore(QObject): update_wishlist = pyqtSignal() def __init__(self, token, language: str, country: str): super(ShopApiCore, self).__init__() self.token = token self.language_code: str = language self.country_code: str = country self.locale = f"{self.language_code}-{self.country_code}" self.locale = "en-US" self.manager = QtRequests(parent=self) self.authed_manager = QtRequests(token=token, parent=self) self.cached_manager = QtRequests(cache=str(cache_dir().joinpath("store")), parent=self) self.browse_active = False self.next_browse_request = tuple(()) def get_free_games(self, handle_func: callable): url = "https://store-site-backend-static-ipv4.ak.epicgames.com/freeGamesPromotions" params = { "locale": self.locale, "country": self.country_code, "allowCountries": self.country_code, } self.manager.get(url, lambda data: self.__handle_free_games(data, handle_func), params=params) @staticmethod def __handle_free_games(data, handle_func): try: response = ResponseModel.from_dict(data) results: List[CatalogOfferModel] = response.data.catalog.search_store.elements handle_func(results) except KeyError as e: if DEBUG(): raise e logger.error("Free games Api request failed") handle_func(["error", "Key error"]) return except Exception as e: if DEBUG(): raise e logger.error(f"Free games Api request failed: {e}") handle_func(["error", e]) return def get_wishlist(self, handle_func): self.authed_manager.post( graphql_url, lambda data: self.__handle_wishlist(data, handle_func), { "query": wishlist_query, "variables": { "country": self.country_code, "locale": self.locale, "withPrice": True, }, }, ) @staticmethod def __handle_wishlist(data, handle_func): try: response = ResponseModel.from_dict(data) if response.errors: logger.error(response.errors) handle_func(response.data.wishlist.wishlist_items.elements) except KeyError as e: if DEBUG(): raise e logger.error("Free games Api request failed") handle_func(["error", "Key error"]) return except Exception as e: if DEBUG(): raise e logger.error(f"Free games Api request failed: {e}") handle_func(["error", e]) return def search_game(self, name, handler): payload = { "query": search_query, "variables": { "category": "games/edition/base|bundles/games|editors|software/edition/base", "count": 20, "country": self.country_code, "keywords": name, "locale": self.locale, "sortDir": "DESC", "allowCountries": self.country_code, "start": 0, "tag": "", "withMapping": False, "withPrice": True, }, } self.manager.post(graphql_url, lambda data: self.__handle_search(data, handler), payload) @staticmethod def __handle_search(data, handler): try: response = ResponseModel.from_dict(data) handler(response.data.catalog.search_store.elements) except KeyError as e: logger.error(str(e)) if DEBUG(): raise e handler([]) except Exception as e: logger.error(f"Search Api request failed: {e}") if DEBUG(): raise e handler([]) return def browse_games(self, browse_model: SearchStoreQuery, handle_func): if self.browse_active: self.next_browse_request = (browse_model, handle_func) return self.browse_active = True payload = { "query": search_query, "variables": browse_model.to_dict() } debug = DebugDialog(payload["variables"], None) debug.exec() self.manager.post(graphql_url, lambda data: self.__handle_browse_games(data, handle_func), payload) def __handle_browse_games(self, data, handle_func): debug = DebugDialog(data, None) debug.exec() self.browse_active = False if data is None: data = {} if not self.next_browse_request: try: response = ResponseModel.from_dict(data) handle_func(response.data.catalog.search_store.elements) except KeyError as e: if DEBUG(): raise e logger.error(str(e)) handle_func([]) except Exception as e: if DEBUG(): raise e logger.error(f"Browse games Api request failed: {e}") handle_func([]) return else: self.browse_games(*self.next_browse_request) # pylint: disable=E1120 self.next_browse_request = tuple(()) def get_game(self, slug: str, is_bundle: bool, handle_func): url = "https://store-content.ak.epicgames.com/api" url += f"/{self.locale}/content/{'products' if not is_bundle else 'bundles'}/{slug}" self.manager.get(url, lambda data: self.__handle_get_game(data, handle_func)) @staticmethod def __handle_get_game(data, handle_func): debug = DebugDialog(data, None) debug.exec() try: product = DieselProduct.from_dict(data) handle_func(product) except Exception as e: if DEBUG(): raise e logger.error(str(e)) # handle_func({}) # needs a captcha def add_to_wishlist(self, namespace, offer_id, handle_func: callable): payload = { "query": wishlist_add_query, "variables": { "offerId": offer_id, "namespace": namespace, "country": self.country_code, "locale": self.locale, }, } self.authed_manager.post(graphql_url, lambda data: self._handle_add_to_wishlist(data, handle_func), payload) def _handle_add_to_wishlist(self, data, handle_func): debug = DebugDialog(data, None) debug.exec() try: response = ResponseModel.from_dict(data) data = response.data.wishlist.add_to_wishlist handle_func(data.success) except Exception as e: if DEBUG(): raise e logger.error(str(e)) handle_func(False) self.update_wishlist.emit() def remove_from_wishlist(self, namespace, offer_id, handle_func: callable): payload = { "query": wishlist_remove_query, "variables": { "offerId": offer_id, "namespace": namespace, "operation": "REMOVE", }, } self.authed_manager.post(graphql_url, lambda data: self._handle_remove_from_wishlist(data, handle_func), payload) def _handle_remove_from_wishlist(self, data, handle_func): debug = DebugDialog(data, None) debug.exec() try: response = ResponseModel.from_dict(data) data = response.data.wishlist.remove_from_wishlist handle_func(data.success) except Exception as e: if DEBUG(): raise e logger.error(str(e)) handle_func(False) self.update_wishlist.emit()