__version__ = "2.0.1" # Imports import asyncio import enum import json import uuid import os from abc import abstractmethod from datetime import datetime from urllib.parse import quote import aiohttp import requests from aiohttp import ClientResponseError # Enums class platforms(enum.Enum): All = 'all' Activision = 'acti' Battlenet = 'battle' PSN = 'psn' Steam = 'steam' Uno = 'uno' XBOX = 'xbl' class games(enum.Enum): ColdWar = 'cw' ModernWarfare = 'mw' ModernWarfare2 = 'mw2' Vanguard = 'vg' Warzone = 'wz' Warzone2 = 'wz2' class friendActions(enum.Enum): Invite = "invite" Uninvite = "uninvite" Remove = "remove" Block = "block" Unblock = "unblock" class API: """ Call Of Duty API Wrapper Developed by Todo Lodo & Engineer152 Contributors - Werseter Source Code: https://github.com/TodoLodo/cod-python-api """ def __init__(self): # sub classes self.Warzone = self.__WZ() self.ModernWarfare = self.__MW() self.Warzone2 = self.__WZ2() self.ModernWarfare2 = self.__MW2() self.ColdWar = self.__CW() self.Vanguard = self.__VG() self.Shop = self.__SHOP() 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): 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" } 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) @staticmethod def login(sso_token: str) -> None: API._Common.cookies["ACT_SSO_COOKIE"] = sso_token API._Common.baseSsoToken = sso_token r = requests.get(f"{API._Common.baseUrl}/crm/cod/v2/identities/{sso_token}", headers=API._Common.requestHeaders, cookies=API._Common.cookies) if r.json()['status'] == 'success': API._Common.loggedIn = True API._Common.cookies.update(r.cookies) else: raise InvalidToken(sso_token) @staticmethod def sso_token() -> str: return API._Common.cookies["ACT_SSO_COOKIE"] # Requests @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: 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)}} async def __sendRequest(self, url: str): if self.loggedIn: 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 # client name url formatter def __cleanClientName(self, gamertag): return quote(gamertag.encode("utf-8")) # helper def __helper(self, platform, gamertag): lookUpType = "gamer" if platform == platforms.Uno: lookUpType = "id" if platform == platforms.Activision: platform = platforms.Uno if platform not in [platforms.Activision, platforms.Battlenet, platforms.Uno, platforms.All, platforms.PSN, platforms.XBOX]: raise InvalidPlatform(platform) else: 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 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 try: match['utcStartDateTime'] = datetime.fromtimestamp( match['utcStartSeconds']).strftime("%A, %B %d, %Y, %I:%M:%S") match['utcEndDateTime'] = datetime.fromtimestamp( match['utcEndSeconds']).strftime("%A, %B %d, %Y, %I:%M:%S") except KeyError: pass # loadouts list for loadout in match['player']['loadouts']: # weapons if loadout['primaryWeapon']['label'] is None: try: loadout['primaryWeapon']['label'] = guns[loadout['primaryWeapon']['name']] except KeyError: pass if loadout['secondaryWeapon']['label'] is None: try: loadout['secondaryWeapon']['label'] = guns[loadout['secondaryWeapon']['name']] except KeyError: pass # perks list for perk in loadout['perks']: if perk['label'] is None: try: perk['label'] = perks[perk['name']] except KeyError: pass # extra perks list for perk in loadout['extraPerks']: if perk['label'] is None: try: perk['label'] = perks[perk['name']] except KeyError: pass # loadout list for loadout in match['player']['loadout']: if loadout['primaryWeapon']['label'] is None: try: loadout['primaryWeapon']['label'] = guns[loadout['primaryWeapon']['name']] except KeyError: pass if loadout['secondaryWeapon']['label'] is None: try: loadout['secondaryWeapon']['label'] = guns[loadout['secondaryWeapon']['name']] except KeyError: pass # perks list for perk in loadout['perks']: if perk['label'] is None: try: perk['label'] = perks[perk['name']] except KeyError: pass # extra perks list for perk in loadout['extraPerks']: if perk['label'] is None: try: perk['label'] = perks[perk['name']] except KeyError: pass except KeyError: pass # return mapped or unmapped data return data # API Requests async def _fullDataReq(self, game, platform, gamertag, type): lookUpType, gamertag, platform = self.__helper(platform, gamertag) return await self.__sendRequest(self.fullDataUrl % (game, platform.value, lookUpType, gamertag, type)) async def _combatHistoryReq(self, game, platform, gamertag, type, start, end): lookUpType, gamertag, platform = self.__helper(platform, gamertag) return await self.__sendRequest( self.combatHistoryUrl % (game, platform.value, lookUpType, gamertag, type, start, end)) async def _breakdownReq(self, game, platform, gamertag, type, start, end): lookUpType, gamertag, platform = self.__helper(platform, gamertag) return await self.__sendRequest( self.breakdownUrl % (game, platform.value, lookUpType, gamertag, type, start, end)) async def _seasonLootReq(self, game, platform, gamertag): lookUpType, gamertag, platform = self.__helper(platform, gamertag) return await self.__sendRequest(self.seasonLootUrl % (game, platform.value, lookUpType, gamertag)) async def _mapListReq(self, game, platform): return await self.__sendRequest(self.mapListUrl % (game, platform.value)) async def _matchInfoReq(self, game, platform, type, matchId): return await self.__sendRequest(self.matchInfoUrl % (game, platform.value, type, matchId)) class __GameDataCommons(_Common): """ Methods ======= Sync ---- fullData(platform:platforms, gamertag:str) returns player's game data of type dict combatHistory(platform:platforms, gamertag:str) returns player's combat history of type dict combatHistoryWithDate(platform:platforms, gamertag:str, start:int, end:int) returns player's combat history within the specified timeline of type dict breakdown(platform:platforms, gamertag:str) returns player's combat history breakdown of type dict breakdownWithDate(platform:platforms, gamertag:str, start:int, end:int) returns player's combat history breakdown within the specified timeline of type dict seasonLoot(platform:platforms, gamertag:str) returns player's season loot mapList(platform:platforms) returns available maps and available modes for each matchInfo(platform:platforms, matchId:int) returns details match details of type dict Async ---- fullDataAsync(platform:platforms, gamertag:str) returns player's game data of type dict combatHistoryAsync(platform:platforms, gamertag:str) returns player's combat history of type dict combatHistoryWithDateAsync(platform:platforms, gamertag:str, start:int, end:int) returns player's combat history within the specified timeline of type dict breakdownAsync(platform:platforms, gamertag:str) returns player's combat history breakdown of type dict breakdownWithDateAsync(platform:platforms, gamertag:str, start:int, end:int) returns player's combat history breakdown within the specified timeline of type dict seasonLootAsync(platform:platforms, gamertag:str) returns player's season loot mapListAsync(platform:platforms) returns available maps and available modes for each matchInfoAsync(platform:platforms, matchId:int) returns details match details of type dict """ def __init_subclass__(cls, **kwargs): cls.__doc__ = cls.__doc__ + super(cls, cls).__doc__ @property @abstractmethod def _game(self) -> str: raise NotImplementedError @property @abstractmethod def _type(self) -> str: raise NotImplementedError async def fullDataAsync(self, platform: platforms, gamertag: str): data = await self._fullDataReq(self._game, platform, gamertag, self._type) return data def fullData(self, platform: platforms, gamertag: str): return asyncio.run(self.fullDataAsync(platform, gamertag)) async def combatHistoryAsync(self, platform: platforms, gamertag: str): data = await self._combatHistoryReq(self._game, platform, gamertag, self._type, 0, 0) return data def combatHistory(self, platform: platforms, gamertag: str): return asyncio.run(self.combatHistoryAsync(platform, gamertag)) async def combatHistoryWithDateAsync(self, platform, gamertag: str, start: int, end: int): data = await self._combatHistoryReq(self._game, platform, gamertag, self._type, start, end) return data def combatHistoryWithDate(self, platform, gamertag: str, start: int, end: int): return asyncio.run(self.combatHistoryWithDateAsync(platform, gamertag, start, end)) async def breakdownAsync(self, platform, gamertag: str): data = await self._breakdownReq(self._game, platform, gamertag, self._type, 0, 0) return data def breakdown(self, platform, gamertag: str): return asyncio.run(self.breakdownAsync(platform, gamertag)) async def breakdownWithDateAsync(self, platform, gamertag: str, start: int, end: int): data = await self._breakdownReq(self._game, platform, gamertag, self._type, start, end) return data def breakdownWithDate(self, platform, gamertag: str, start: int, end: int): return asyncio.run(self.breakdownWithDateAsync(platform, gamertag, start, end)) async def matchInfoAsync(self, platform, matchId: int): data = await self._matchInfoReq(self._game, platform, self._type, matchId) return data def matchInfo(self, platform, matchId: int): return asyncio.run(self.matchInfoAsync(platform, matchId)) async def seasonLootAsync(self, platform, gamertag): data = await self._seasonLootReq(self._game, platform, gamertag) return data def seasonLoot(self, platform, gamertag): return asyncio.run(self.seasonLootAsync(platform, gamertag)) async def mapListAsync(self, platform): data = await self._mapListReq(self._game, platform) return data def mapList(self, platform): return asyncio.run(self.mapListAsync(platform)) # WZ class __WZ(__GameDataCommons): """ Warzone class: A class to get players warzone stats, warzone combat history and specific warzone match details classCategory: game gameId/gameTitle: mw or wz gameType: wz """ @property def _game(self) -> str: return "mw" @property def _type(self) -> str: return "wz" async def seasonLootAsync(self, platform, gamertag): raise InvalidEndpoint async def mapListAsync(self, platform): raise InvalidEndpoint # WZ2 class __WZ2(__GameDataCommons): """ Warzone 2 class: A class to get players warzone 2 stats, warzone 2 combat history and specific warzone 2 match details classCategory: game gameId/gameTitle: mw or wz gameType: wz2 """ @property def _game(self) -> str: return "mw2" @property def _type(self) -> str: return "wz2" async def seasonLootAsync(self, platform, gamertag): raise InvalidEndpoint async def mapListAsync(self, platform): raise InvalidEndpoint # MW class __MW(__GameDataCommons): """ ModernWarfare class: A class to get players modernwarfare stats, modernwarfare combat history, a player's modernwarfare season loot, modernwarfare map list and specific modernwarfare match details classCategory: game gameId/gameTitle: mw gameType: mp """ @property def _game(self) -> str: return "mw" @property def _type(self) -> str: return "mp" # CW class __CW(__GameDataCommons): """ ColdWar class: A class to get players coldwar stats, coldwar combat history, a player's coldwar season loot, coldwar map list and specific coldwar match details classCategory: game gameId/gameTitle: cw gameType: mp """ @property def _game(self) -> str: return "cw" @property def _type(self) -> str: return "mp" # VG class __VG(__GameDataCommons): """ Vanguard class: A class to get players vanguard stats, vanguard combat history, a player's vanguard season loot, vanguard map list and specific vanguard match details classCategory: game gameId/gameTitle: vg gameType: pm """ @property def _game(self) -> str: return "vg" @property def _type(self) -> str: return "mp" # MW2 class __MW2(__GameDataCommons): """ ModernWarfare 2 class: A class to get players modernwarfare 2 stats, modernwarfare 2 combat history, a player's modernwarfare 2 season loot, modernwarfare 2 map list and specific modernwarfare 2 match details classCategory: game gameId/gameTitle: mw gameType: mp """ @property def _game(self) -> str: return "mw2" @property def _type(self) -> str: return "mp" # USER class __USER(_Common): def info(self): if self.loggedIn: # First, try to load the user info from a local file if it exists. if os.path.exists('userInfo.json'): with open('userInfo.json', 'r') as file: rawData = json.load(file) else: # If the file doesn't exist, make the HTTP request. rawData = requests.get(f"https://profile.callofduty.com/cod/userInfo/{self.sso_token()}", headers=API._Common.requestHeaders) # Process the raw data (either from the file or HTTP response) data = {'userName': rawData['userInfo']['userName'], 'identities': []} for i in rawData['identities']: data['identities'].append({ 'platform': i['provider'], 'gamertag': i['username'], 'accountID': i['accountID'] }) return data else: raise NotLoggedIn def __priv(self): d = self.info() return d['identities'][0]['platform'], quote(d['identities'][0]['gamertag'].encode("utf-8")) async def friendFeedAsync(self): p, g = self.__priv() data = await self._Common__sendRequest( f"/userfeed/v1/friendFeed/platform/{p}/gamer/{g}/friendFeedEvents/en") return data def friendFeed(self): return asyncio.run(self.friendFeedAsync()) async def eventFeedAsync(self): 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.sso_token()}") return data def loggedInIdentities(self): return asyncio.run(self.loggedInIdentitiesAsync()) async def codPointsAsync(self): p, g = self.__priv() data = await self._Common__sendRequest(f"/inventory/v1/title/mw/platform/{p}/gamer/{g}/currency") return data def codPoints(self): return asyncio.run(self.codPointsAsync()) async def connectedAccountsAsync(self): p, g = self.__priv() data = await self._Common__sendRequest(f"/crm/cod/v2/accounts/platform/{p}/gamer/{g}") return data def connectedAccounts(self): return asyncio.run(self.connectedAccountsAsync()) async def settingsAsync(self): p, g = self.__priv() 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): """ Shop class: A class to get bundle details and battle pass loot classCategory: other Methods ======= Sync ---- purchasableItems(game: games) returns purchasable items for a specific gameId/gameTitle bundleInformation(game: games, bundleId: int) returns bundle details for the specific gameId/gameTitle and bundleId battlePassLoot(game: games, platform: platforms, season: int) returns battle pass loot for specific game and season on given platform Async ---- purchasableItemsAsync(game: games) returns purchasable items for a specific gameId/gameTitle bundleInformationAsync(game: games, bundleId: int) returns bundle details for the specific gameId/gameTitle and bundleId battlePassLootAsync(game: games, platform: platforms, season: int) returns battle pass loot for specific game and season on given platform """ async def purchasableItemsAsync(self, game: games): data = await self._Common__sendRequest(f"/inventory/v1/title/{game.value}/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.value}/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( f"/loot/title/{game.value}/platform/{platform.value}/list/loot_season_{season}/en") return data def battlePassLoot(self, game: games, platform: platforms, season: int): return asyncio.run(self.battlePassLootAsync(game, platform, season)) # ALT 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") return data def search(self, platform, gamertag: str): return asyncio.run(self.searchAsync(platform, gamertag)) # Exceptions class NotLoggedIn(Exception): def __str__(self): return "Not logged in!" class InvalidToken(Exception): def __init__(self, token): self.token = token def __str__(self): return f"Token is invalid, token: {self.token}" class InvalidPlatform(Exception): def __init__(self, platform: platforms): self.message: str if platform == platforms.Steam: self.message = "Steam cannot be used till further updates." else: self.message = "Invalid platform, use platform class!" super().__init__(self.message) def __str__(self): return self.message class InvalidEndpoint(Exception): def __str__(self): return "This endpoint is not available for selected title" class StatusError(Exception): def __str__(self): return "Status Error, Check if your sso token is valid or try again later."