From 4e65872e304452316cb158aa01bedc4975975c14 Mon Sep 17 00:00:00 2001 From: Werseter Date: Fri, 28 Oct 2022 12:20:13 +0200 Subject: [PATCH 1/6] Reorder imports as per PEP8 standard --- cod_api/__init__.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/cod_api/__init__.py b/cod_api/__init__.py index e87290f..6047ff2 100644 --- a/cod_api/__init__.py +++ b/cod_api/__init__.py @@ -2,13 +2,14 @@ # Imports import asyncio -from abc import abstractmethod -from datetime import datetime import enum import json -import requests -from urllib.parse import quote import uuid +from abc import abstractmethod +from datetime import datetime +from urllib.parse import quote + +import requests # Enums From 60cc2deda3d7b88d58d752beeb82c3d89b87e74e Mon Sep 17 00:00:00 2001 From: Werseter Date: Fri, 28 Oct 2022 12:46:48 +0200 Subject: [PATCH 2/6] Cleanup dynamic __doc__ --- cod_api/__init__.py | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/cod_api/__init__.py b/cod_api/__init__.py index 6047ff2..003e3ab 100644 --- a/cod_api/__init__.py +++ b/cod_api/__init__.py @@ -48,11 +48,11 @@ def __init__(self): self.__GameDataCommons = self.__GameDataCommons() # sub classes - self.Warzone = self.__WZ(self.__GameDataCommons.__doc__) - self.ModernWarfare = self.__MW(self.__GameDataCommons.__doc__) + self.Warzone = self.__WZ() + self.ModernWarfare = self.__MW() self.Warzone2 = self.__WZ2() - self.ModernWarfare2 = self.__MW2(self.__GameDataCommons.__doc__) - self.ColdWar = self.__CW(self.__GameDataCommons.__doc__) + self.ModernWarfare2 = self.__MW2() + self.ColdWar = self.__CW() self.Vanguard = self.__VG() self.Shop = self.__SHOP() self.Me = self.__USER() @@ -401,7 +401,7 @@ class __GameDataCommons(__common): matchInfo(platform:platforms, matchId:int) returns details match details of type dict - + Async ---- fullDataAsync(platform:platforms, gamertagLstr) @@ -429,11 +429,10 @@ class __GameDataCommons(__common): returns details match details of type dict """ - def __init__(self, doc): + def __init__(self): super().__init__() - if doc is None: - doc = '' - self.__doc__ += doc + if self.__doc__ is None: + self.__doc__ = super().__doc__ @property @abstractmethod @@ -789,7 +788,7 @@ def __init__(self, platform: platforms): else: self.message = "Invalid platform, use platform class!" - + super().__init__(self.message) def __str__(self): From cfbfa5640a9c6f62754d4a54996c621b7c0f3104 Mon Sep 17 00:00:00 2001 From: Werseter Date: Fri, 28 Oct 2022 12:26:47 +0200 Subject: [PATCH 3/6] Requests overhaul --- cod_api/__init__.py | 311 ++++++++++++++++---------------------------- setup.py | 2 +- 2 files changed, 115 insertions(+), 198 deletions(-) diff --git a/cod_api/__init__.py b/cod_api/__init__.py index 003e3ab..632b2ee 100644 --- a/cod_api/__init__.py +++ b/cod_api/__init__.py @@ -9,7 +9,9 @@ from datetime import datetime from urllib.parse import quote +import aiohttp import requests +from aiohttp import ClientResponseError # Enums @@ -43,10 +45,6 @@ class friendActions(enum.Enum): class API: def __init__(self): - # common classes - self.__common = self.__common() - self.__GameDataCommons = self.__GameDataCommons() - # sub classes self.Warzone = self.__WZ() self.ModernWarfare = self.__MW() @@ -58,179 +56,101 @@ def __init__(self): self.Me = self.__USER() self.Misc = self.__ALT() + async def loginAsync(self, sso_token: str) -> None: + await API._Common.loginAsync(sso_token) + # Login def login(self, ssoToken: str): - try: - self.__common.baseHeaders["__X-XSRF-TOKEN"] = self.__common.fakeXSRF - self.__common.baseHeaders["__X-CSRF-TOKEN"] = self.__common.fakeXSRF - self.__common.baseHeaders["Atvi-Auth"] = ssoToken - self.__common.baseHeaders["ACT_SSO_COOKIE"] = ssoToken - self.__common.baseHeaders["atkn"] = ssoToken - self.__common.baseHeaders["cookie"] = f'{self.__common.baseCookie}' \ - f'ACT_SSO_COOKIE={ssoToken};' \ - f'XSRF-TOKEN={self.__common.fakeXSRF};' \ - f'API_CSRF_TOKEN={self.__common.fakeXSRF};' \ - f'ACT_SSO_EVENT="LOGIN_SUCCESS:1644346543228";' \ - f'ACT_SSO_COOKIE_EXPIRY=1645556143194;' \ - f'comid=cod;' \ - f'ssoDevId=63025d09c69f47dfa2b8d5520b5b73e4;' \ - f'tfa_enrollment_seen=true;' \ - f'gtm.custom.bot.flag=human;' - self.__common.baseSsoToken = ssoToken - self.__common.basePostHeaders["__X-XSRF-TOKEN"] = self.__common.fakeXSRF - self.__common.basePostHeaders["__X-CSRF-TOKEN"] = self.__common.fakeXSRF - self.__common.basePostHeaders["Atvi-Auth"] = ssoToken - self.__common.basePostHeaders["ACT_SSO_COOKIE"] = ssoToken - self.__common.basePostHeaders["atkn"] = ssoToken - self.__common.basePostHeaders["cookie"] = f'{self.__common.baseCookie}' \ - f'ACT_SSO_COOKIE={ssoToken};' \ - f'XSRF-TOKEN={self.__common.fakeXSRF};' \ - f'API_CSRF_TOKEN={self.__common.fakeXSRF};' \ - f'ACT_SSO_EVENT="LOGIN_SUCCESS:1644346543228";' \ - f'ACT_SSO_COOKIE_EXPIRY=1645556143194;' \ - f'comid=cod;' \ - f'ssoDevId=63025d09c69f47dfa2b8d5520b5b73e4;' \ - f'tfa_enrollment_seen=true;' \ - f'gtm.custom.bot.flag=human;' - - r = requests.get(f"{self.__common.baseUrl}{self.__common.apiPath}/crm/cod/v2/identities/{ssoToken}", - headers=self.__common.baseHeaders) - - if r.json()['status'] == 'success': - self.__common.loggedIn = True - for sub in [self.Warzone, self.Warzone2, self.ModernWarfare, self.ModernWarfare2, self.ColdWar, - self.Vanguard, self.Shop, self.Me, self.Misc]: - sub.loggedIn = self.__common.loggedIn - sub.baseSsoToken = self.__common.baseSsoToken - sub.baseHeaders = self.__common.baseHeaders - sub.basePostHeaders = self.__common.basePostHeaders - - # deleting scope data - del ssoToken, sub, r - else: - # delete scope data - del ssoToken, r - - # InvalidToken - raise InvalidToken(ssoToken) - except Exception as e: - print(e) - - # delete scope data - del ssoToken - - return e - - class __common: - customHeaders: dict = { - "__X-XSRF-TOKEN": str or None, - "__X-CSRF-TOKEN": str or None, - "Atvi-Auth": str or None, - "ACT_SSO_COOKIE": str or None, - "atkn": str or None, - "cookie": str or None, - "content-type": str or None, + API._Common.login(ssoToken) + + class _Common: + requestHeaders = { + "content-type": "application/json", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/74.0.3729.169 " + "Safari/537.36", + "Accept": "application/json", + "Connection": "Keep-Alive" } - customBody: dict = {} + cookies = {"new_SiteId": "cod", "ACT_SSO_LOCALE": "en_US", "country": "US", + "ACT_SSO_COOKIE_EXPIRY": "1645556143194"} + cachedMappings = None + + fakeXSRF = str(uuid.uuid4()) + baseUrl: str = "https://my.callofduty.com/api/papi-client" + loggedIn: bool = False + + # endPoints + + # game platform lookupType gamertag type + fullDataUrl = "/stats/cod/v1/title/%s/platform/%s/%s/%s/profile/type/%s" + # game platform lookupType gamertag type start end [?limit=n or ''] + combatHistoryUrl = "/crm/cod/v2/title/%s/platform/%s/%s/%s/matches/%s/start/%d/end/%d/details" + # game platform lookupType gamertag type start end + breakdownUrl = "/crm/cod/v2/title/%s/platform/%s/%s/%s/matches/%s/start/%d/end/%d" + # game platform lookupType gamertag + seasonLootUrl = "/loot/title/%s/platform/%s/%s/%s/status/en" + # game platform + mapListUrl = "/ce/v1/title/%s/platform/%s/gameType/mp/communityMapData/availability" + # game platform type matchId + matchInfoUrl = "/crm/cod/v2/title/%s/platform/%s/fullMatch/%s/%d/en" + + @staticmethod + async def loginAsync(sso_token: str) -> None: + API._Common.cookies["ACT_SSO_COOKIE"] = sso_token + API._Common.baseSsoToken = sso_token + r = await API._Common.__Request(f"{API._Common.baseUrl}/crm/cod/v2/identities/{sso_token}") + if r['status'] == 'success': + API._Common.loggedIn = True + else: + raise InvalidToken(sso_token) - def __init__(self): - # Variables - - # headers & cookies - self.fakeXSRF = str(uuid.uuid4()) - self.userAgent: str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " \ - "AppleWebKit/537.36 (KHTML, like Gecko) " \ - "Chrome/74.0.3729.169 " \ - "Safari/537.36" - self.baseCookie: str = "new_SiteId=cod;ACT_SSO_LOCALE=en_US;country=US;" - self.baseSsoToken: str = '' - self.baseUrl: str = "https://my.callofduty.com" - self.apiPath: str = "/api/papi-client" - self.loggedIn: bool = False - - # headers structures - self.baseHeaders: API.customHeaders = { - "content-type": 'application/json', - "cookie": self.baseCookie, - "user-agent": self.userAgent - } - - self.basePostHeaders: API.customHeaders = { - "content-type": 'text/plain', - "cookie": self.baseCookie, - "user-agent": self.userAgent - } - - # endPoints - - # game platform lookupType gamertag type - self.fullDataUrl = "/stats/cod/v1/title/%s/platform/%s/%s/%s/profile/type/%s" - # game platform lookupType gamertag type start end [?limit=n or ''] - self.combatHistoryUrl = "/crm/cod/v2/title/%s/platform/%s/%s/%s/matches/%s/start/%d/end/%d/details" - # game platform lookupType gamertag type start end - self.breakdownUrl = "/crm/cod/v2/title/%s/platform/%s/%s/%s/matches/%s/start/%d/end/%d" - # game platform lookupType gamertag - self.seasonLootUrl = "/loot/title/%s/platform/%s/%s/%s/status/en" - # game platform - self.mapListUrl = "/ce/v1/title/%s/platform/%s/gameType/mp/communityMapData/availability" - # game platform type matchId - self.matchInfoUrl = "/crm/cod/v2/title/%s/platform/%s/fullMatch/%s/%d/en" + @staticmethod + def login(sso_token: str) -> None: + API._Common.cookies["ACT_SSO_COOKIE"] = sso_token + API._Common.baseSsoToken = sso_token - # Requests - async def __Request(self, method, url): - h: self.customHeaders = self.customHeaders - b: self.customBody = self.customBody - if method == "GET": - h = self.baseHeaders - elif method == "POST": - h = self.basePostHeaders + r = requests.get(f"{API._Common.baseUrl}/crm/cod/v2/identities/{sso_token}", + headers=API._Common.requestHeaders, cookies=API._Common.cookies) - try: - r = requests.request(method=method, url=url, headers=h, data=b) + if r.json()['status'] == 'success': + API._Common.loggedIn = True + API._Common.cookies.update(r.cookies) + else: + raise InvalidToken(sso_token) - # return data - return r - except Exception as e: + @staticmethod + def sso_token() -> str: + return API._Common.cookies["ACT_SSO_COOKIE"] - return e + # Requests - async def __sendRequest(self, url: str): - if self.loggedIn: - respond = await self.__Request("GET", f"{self.baseUrl}{self.apiPath}{url}") - if type(respond) != Exception: - if respond.status_code == 200: - data = respond.json() - if data['status'] == 'success': - data = self.__mapping(data['data']) - # delete scope data - del url, respond - - # return data - return data + @staticmethod + async def __Request(url): + async with aiohttp.client.ClientSession(connector=aiohttp.TCPConnector(verify_ssl=True), + timeout=aiohttp.ClientTimeout(total=30)) as session: + try: + async with session.get(url, cookies=API._Common.cookies, + headers=API._Common.requestHeaders) as resp: + try: + resp.raise_for_status() + except ClientResponseError as err: + return {'status': 'error', 'data': {'type': type(err), 'message': err.message}} else: - # delete scope data - del url, respond - - raise StatusError() - else: - # delete scope data - del url + API._Common.cookies.update({c.key: c.value for c in session.cookie_jar}) + return await resp.json() + except asyncio.TimeoutError as err: + return {'status': 'error', 'data': {'type': type(err), 'message': str(err)}} - return respond.status_code - else: - raise NotLoggedIn() - - async def __sendPostRequest(self, url: str, body: dict): + async def __sendRequest(self, url: str): if self.loggedIn: - customBody = body - respond = await self.__Request("POST", f"{self.baseUrl}{self.apiPath}{url}") - if type(respond) != Exception: - if respond.status_code == 200: - return respond.json() - else: - return respond.status_code + response = await API._Common.__Request(f"{self.baseUrl}{url}") + if response['status'] == 'success': + response['data'] = await self.__perform_mapping(response['data']) + return response else: - raise NotLoggedIn() + raise NotLoggedIn # client name url formatter def __cleanClientName(self, gamertag): @@ -247,23 +167,22 @@ def __helper(self, platform, gamertag): platforms.XBOX]: raise InvalidPlatform(platform) else: - if platform in [platforms.Activision, platforms.Battlenet, platforms.Uno]: - gamertag = self.__cleanClientName(gamertag) + gamertag = self.__cleanClientName(gamertag) return lookUpType, gamertag, platform + async def __get_mappings(self): + if API._Common.cachedMappings is None: + API._Common.cachedMappings = ( + await API._Common.__Request('https://engineer152.github.io/wz-data/weapon-ids.json'), + await API._Common.__Request('https://engineer152.github.io/wz-data/game-modes.json'), + await API._Common.__Request('https://engineer152.github.io/wz-data/perks.json')) + return API._Common.cachedMappings + # mapping - def __mapping(self, data): - r = requests.get( - 'https://engineer152.github.io/wz-data/weapon-ids.json') - guns = r.json() - r = requests.get( - 'https://engineer152.github.io/wz-data/game-modes.json') - modes = r.json() - r = requests.get( - 'https://engineer152.github.io/wz-data/perks.json') - perks = r.json() - - # guns + async def __perform_mapping(self, data): + guns, modes, perks = await self.__get_mappings() + if not isinstance(data, list) or 'matches' not in data: + return data try: for match in data['matches']: # time stamps @@ -372,7 +291,7 @@ async def _mapListReq(self, game, platform): async def _matchInfoReq(self, game, platform, type, matchId): return await self.__sendRequest(self.matchInfoUrl % (game, platform.value, type, matchId)) - class __GameDataCommons(__common): + class __GameDataCommons(_Common): """ Methods ======= @@ -625,13 +544,11 @@ def _type(self) -> str: return "mp" # USER - class __USER(__common): + class __USER(_Common): def info(self): if self.loggedIn: - headers = self.baseHeaders - headers['Accept'] = 'application/json' - rawData = requests.get( - f"https://profile.callofduty.com/cod/userInfo/{self.baseSsoToken}", headers=headers) + rawData = requests.get(f"https://profile.callofduty.com/cod/userInfo/{self.sso_token()}", + headers=API._Common.requestHeaders) rawData = json.loads(rawData.text.replace( 'userInfo(', '').replace(');', '')) @@ -655,7 +572,7 @@ def __priv(self): async def friendFeedAsync(self): p, g = self.__priv() - data = await self._common__sendRequest( + data = await self._Common__sendRequest( f"/userfeed/v1/friendFeed/platform/{p}/gamer/{g}/friendFeedEvents/en") return data @@ -663,14 +580,14 @@ def friendFeed(self): return asyncio.run(self.friendFeedAsync()) async def eventFeedAsync(self): - data = await self._common__sendRequest(f"/userfeed/v1/friendFeed/rendered/en/{self.baseSsoToken}") + data = await self._Common__sendRequest(f"/userfeed/v1/friendFeed/rendered/en/{self.sso_token()}") return data def eventFeed(self): return asyncio.run(self.eventFeedAsync()) async def loggedInIdentitiesAsync(self): - data = await self._common__sendRequest(f"/crm/cod/v2/identities/{self.baseSsoToken}") + data = await self._Common__sendRequest(f"/crm/cod/v2/identities/{self.sso_token()}") return data def loggedInIdentities(self): @@ -678,7 +595,7 @@ def loggedInIdentities(self): async def codPointsAsync(self): p, g = self.__priv() - data = await self._common__sendRequest(f"/inventory/v1/title/mw/platform/{p}/gamer/{g}/currency") + data = await self._Common__sendRequest(f"/inventory/v1/title/mw/platform/{p}/gamer/{g}/currency") return data def codPoints(self): @@ -686,7 +603,7 @@ def codPoints(self): async def connectedAccountsAsync(self): p, g = self.__priv() - data = await self._common__sendRequest(f"/crm/cod/v2/accounts/platform/{p}/gamer/{g}") + data = await self._Common__sendRequest(f"/crm/cod/v2/accounts/platform/{p}/gamer/{g}") return data def connectedAccounts(self): @@ -694,14 +611,14 @@ def connectedAccounts(self): async def settingsAsync(self): p, g = self.__priv() - data = await self._common__sendRequest(f"/preferences/v1/platform/{p}/gamer/{g}/list") + data = await self._Common__sendRequest(f"/preferences/v1/platform/{p}/gamer/{g}/list") return data def settings(self): return asyncio.run(self.settingsAsync()) # SHOP - class __SHOP(__common): + class __SHOP(_Common): """ Shop class: A class to get bundle details and battle pass loot classCatogery: other @@ -732,21 +649,21 @@ class __SHOP(__common): """ async def purchasableItemsAsync(self, game: games): - data = await self._common__sendRequest(f"/inventory/v1/title/{game}/platform/uno/purchasable/public/en") + data = await self._Common__sendRequest(f"/inventory/v1/title/{game}/platform/uno/purchasable/public/en") return data def purchasableItems(self, game: games): return asyncio.run(self.purchasableItemsAsync(game)) async def bundleInformationAsync(self, game: games, bundleId: int): - data = await self._common__sendRequest(f"/inventory/v1/title/{game}/bundle/{bundleId}/en") + data = await self._Common__sendRequest(f"/inventory/v1/title/{game}/bundle/{bundleId}/en") return data def bundleInformation(self, game: games, bundleId: int): return asyncio.run(self.bundleInformationAsync(game, bundleId)) async def battlePassLootAsync(self, game: games, platform: platforms, season: int): - data = await self._common__sendRequest( + data = await self._Common__sendRequest( f"/loot/title/{game}/platform/{platform.value}/list/loot_season_{season}/en") return data @@ -754,11 +671,11 @@ def battlePassLoot(self, game: games, platform: platforms, season: int): return asyncio.run(self.battlePassLootAsync(game, platform, season)) # ALT - class __ALT(__common): + class __ALT(_Common): async def searchAsync(self, platform, gamertag: str): - lookUpType, gamertag, platform = self._common__helper(platform, gamertag) - data = await self._common__sendRequest(f"/crm/cod/v2/platform/{platform.value}/username/{gamertag}/search") + lookUpType, gamertag, platform = self._Common__helper(platform, gamertag) + data = await self._Common__sendRequest(f"/crm/cod/v2/platform/{platform.value}/username/{gamertag}/search") return data def search(self, platform, gamertag: str): diff --git a/setup.py b/setup.py index 4edb943..08d7c0a 100644 --- a/setup.py +++ b/setup.py @@ -1,6 +1,6 @@ from setuptools import setup -requirements = ["asyncio", "datetime", "requests", "uuid", "urllib3", "enum34"] +requirements = ["asyncio", "aiohttp", "datetime", "requests", "uuid", "urllib3", "enum34"] setup( name="cod_api", From 9260928597d8a0d78ca8cfb06e1fcf87a0f41736 Mon Sep 17 00:00:00 2001 From: Werseter Date: Fri, 28 Oct 2022 12:27:42 +0200 Subject: [PATCH 4/6] Remove random local variable deletion --- cod_api/__init__.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/cod_api/__init__.py b/cod_api/__init__.py index 632b2ee..071f37c 100644 --- a/cod_api/__init__.py +++ b/cod_api/__init__.py @@ -257,12 +257,6 @@ async def __perform_mapping(self, data): except Exception as e: print(e) - # delete scope data - try: - del guns, modes, perks, match, loadout, perk - except UnboundLocalError: - pass - # return mapped or unmapped data return data From 15546c3aa28a4e895e95ff7a37678d50564e5457 Mon Sep 17 00:00:00 2001 From: Werseter Date: Fri, 28 Oct 2022 12:28:04 +0200 Subject: [PATCH 5/6] Exception cleanup --- cod_api/__init__.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/cod_api/__init__.py b/cod_api/__init__.py index 071f37c..2fcb8d5 100644 --- a/cod_api/__init__.py +++ b/cod_api/__init__.py @@ -254,8 +254,6 @@ async def __perform_mapping(self, data): pass except KeyError: pass - except Exception as e: - print(e) # return mapped or unmapped data return data @@ -558,7 +556,7 @@ def info(self): }) return data else: - raise NotLoggedIn() + raise NotLoggedIn def __priv(self): d = self.info() From 56bdb58cf6de80227aaf744fd358311d0ffd0dbb Mon Sep 17 00:00:00 2001 From: Werseter Date: Fri, 28 Oct 2022 12:48:54 +0200 Subject: [PATCH 6/6] Fix Typos --- cod_api/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cod_api/__init__.py b/cod_api/__init__.py index 2fcb8d5..0289b9e 100644 --- a/cod_api/__init__.py +++ b/cod_api/__init__.py @@ -289,7 +289,7 @@ class __GameDataCommons(_Common): ======= Sync ---- - fullData(platform:platforms, gamertagLstr) + fullData(platform:platforms, gamertag:str) returns player's game data of type dict combatHistory(platform:platforms, gamertag:str) @@ -315,7 +315,7 @@ class __GameDataCommons(_Common): Async ---- - fullDataAsync(platform:platforms, gamertagLstr) + fullDataAsync(platform:platforms, gamertag:str) returns player's game data of type dict combatHistoryAsync(platform:platforms, gamertag:str)