Skip to content

Commit

Permalink
Merge pull request #3 from zusorio/v3
Browse files Browse the repository at this point in the history
Update for Overwatch 2
  • Loading branch information
zusorio authored Feb 12, 2023
2 parents 83dc882 + bae0bb9 commit 09f0284
Show file tree
Hide file tree
Showing 7 changed files with 120 additions and 157 deletions.
54 changes: 0 additions & 54 deletions .github/workflows/codeql-analysis.yml

This file was deleted.

4 changes: 2 additions & 2 deletions LICENSE.md
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
MIT License

Copyright (c) 2020 Tobias Messner
Copyright (c) 2023 Tobias Messner

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
Expand All @@ -18,4 +18,4 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
SOFTWARE.
29 changes: 8 additions & 21 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
# pyowapi
`pyowapi` is an asynchronous wrapper around an unofficial Overwatch api (https://ow-api.com) using aiohttp.

Things have changed considerably from v1 to v2, make sure to update your code before upgrading.
Things have changed considerably from v2 to v3, make sure to update your code before upgrading.

# Example usage

Expand All @@ -13,19 +13,19 @@ import pyowapi
# For a single player
player = pyowapi.get_player("Jayne#1447")
print(player.success)
print(player.actual_level)
print(player.private)
print(player.competitive_tank)

# For multiple players
player_list = ["Jayne#1447", "Krusher#9999"]
players = pyowapi.get_player(player_list)
for single_player in players:
print(player.actual_level)
print(player.competitive_tank.group)
print(player.competitive_tank.tier)

# For different platforms
player = pyowapi.get_player("Krusher99", platform="psn") # platform can be pc, xbox, ps4 and nintendo-switch
print(player.actual_level)
print(player.competitive_tank)

# If the player name is from user input sometimes it can be capitalized wrong or use the wrong discriminator
# You can pass correct_player = True and if the player name is incorrect pyowapi will attempt to find the correct one
Expand All @@ -38,35 +38,22 @@ print(player.success) # True

# If an event loop already exists you need to call get_player_async instead
player = await pyowapi.get_player_async("Jayne#1447")
print(player.actual_level)
print(player.private)
print(player.competitive_tank)

# A player has the following properties
print(player.player_name) # The name of the Player (battletag or other)
print(player.original_player_name) # If a player name was corrected this is the misspelled version
print(player.success) # If the request was successful
print(player.level) # The number in Overwatch without stars calculated in
print(player.prestige) # The number of stars in Overwatch
print(player.actual_level) # The full level with stars calculated in
print(player.private) # If the player profile is private
print(player.endorsement) # The player endorsement level
print(player.competitive_tank) # Player Tank SR. False if unplaced
print(player.competitive_damage) # Player Damage SR. False if unplaced
print(player.competitive_support) # Player Support SR. False if unplaced
print(player.competitive_tank.tier) # Player Tank Tier (5-1)
print(player.competitive_tank.group) # Player Tank Group (Bronze, Silver, Gold, Platinum, Diamond, Master, Grandmaster)
print(player.competitive_damage) # Player Damage Rating (similar to Tank)
print(player.competitive_support) # Player Support Rating (similar to Tank)
print(player.quickplay_stats) # Dictionary containing all quickplay stats
print(player.quickplay_cards)
print(player.quickplay_medals)
print(player.quickplay_medals_bronze)
print(player.quickplay_medals_silver)
print(player.quickplay_medals_gold)
print(player.quickplay_games_won)
print(player.competitive_stats) # Dictionary containing all competitive stats
print(player.competitive_cards)
print(player.competitive_medals)
print(player.competitive_medals_bronze)
print(player.competitive_medals_silver)
print(player.competitive_medals_gold)
print(player.competitive_games_played)
print(player.competitive_games_won)
```
119 changes: 69 additions & 50 deletions pyowapi/__init__.py
Original file line number Diff line number Diff line change
@@ -1,110 +1,122 @@
import asyncio
import aiohttp
from typing import List, Union, Optional
from typing import List, Union, Optional, Literal
from dataclasses import dataclass


@dataclass
class Rank:
group: Literal["Bronze", "Silver", "Gold", "Platinum", "Diamond", "Master", "Grandmaster"]
tier: Literal[1, 2, 3, 4, 5]

_group_order = ["Bronze", "Silver", "Gold", "Platinum", "Diamond", "Master", "Grandmaster"]

def __gt__(self, other):
if isinstance(other, Rank):
if self.group == other.group:
return self.tier < other.tier
else:
return self._group_order.index(self.group) > self._group_order.index(other.group)
else:
raise TypeError("Cannot compare Rank to non-Rank object")

def __lt__(self, other):
if isinstance(other, Rank):
if self.group == other.group:
return self.tier > other.tier
else:
return self._group_order.index(self.group) < self._group_order.index(other.group)
else:
raise TypeError("Cannot compare Rank to non-Rank object")

def __eq__(self, other):
if isinstance(other, Rank):
return self.group == other.group and self.tier == other.tier
else:
raise TypeError("Cannot compare Rank to non-Rank object")


class Player:
def __init__(self, player_name: str, response: dict, platform: str = "pc", original_player_name: Optional[str] = None,):
def __init__(self, player_name: str, response: dict, original_player_name: Optional[str] = None, ):
self.player_name: str = player_name
self.original_player_name: Optional[str] = original_player_name
self.platform: str = platform
self.success: bool = "error" not in response
if self.success:
self.level: Optional[int] = response["level"]
self.prestige: Optional[int] = response["prestige"]
self.actual_level: Optional[int] = self.prestige * 100 + self.level
self.private: Optional[bool] = response["private"]
self.endorsement: Optional[int] = response["endorsement"]

if not self.private:
self.quickplay_stats: Optional[dict] = response["competitiveStats"]
self.quickplay_cards: Optional[int] = self.quickplay_stats["awards"]["cards"]
self.quickplay_medals: Optional[int] = self.quickplay_stats["awards"]["medals"]
self.quickplay_medals_bronze: Optional[int] = self.quickplay_stats["awards"]["medalsBronze"]
self.quickplay_medals_silver: Optional[int] = self.quickplay_stats["awards"]["medalsSilver"]
self.quickplay_medals_gold: Optional[int] = self.quickplay_stats["awards"]["medalsGold"]
self.quickplay_games_won: Optional[int] = self.quickplay_stats["games"]["won"]

self.competitive_stats: Optional[dict] = response["competitiveStats"]
self.competitive_cards: Optional[int] = self.competitive_stats["awards"]["cards"]
self.competitive_medals: Optional[int] = self.competitive_stats["awards"]["medals"]
self.competitive_medals_bronze: Optional[int] = self.competitive_stats["awards"]["medalsBronze"]
self.competitive_medals_silver: Optional[int] = self.competitive_stats["awards"]["medalsSilver"]
self.competitive_medals_gold: Optional[int] = self.competitive_stats["awards"]["medalsGold"]
self.competitive_games_played: Optional[int] = self.competitive_stats["games"]["played"]
self.competitive_games_won: Optional[int] = self.competitive_stats["games"]["won"]

self.competitive_tank: Union[bool, int] = False
self.competitive_damage: Union[bool, int] = False
self.competitive_support: Union[bool, int] = False
self.competitive_tank: Optional[Rank] = None
self.competitive_damage: Optional[Rank] = None
self.competitive_support: Optional[Rank] = None

if response.get("ratings"):
for rating in response["ratings"]:
if rating["role"] == "tank":
self.competitive_tank = rating["level"]
self.competitive_tank = Rank(rating["group"], rating["tier"])
if rating["role"] == "damage":
self.competitive_damage = rating["level"]
self.competitive_damage = Rank(rating["group"], rating["tier"])
if rating["role"] == "support":
self.competitive_support = rating["level"]
self.competitive_support = Rank(rating["group"], rating["tier"])

def __repr__(self):
return f"<Player {self.player_name} success: {self.success}>"


async def _correct_player_internal(session: aiohttp.ClientSession, incorrect_player_name: str, platform: str = "pc") -> Optional[Player]:
async def _correct_player_internal(session: aiohttp.ClientSession, incorrect_player_name: str) -> Optional[Player]:
"""
Attempts to find the corrected player name and returns a Player object if it does
:param session: An aiohttp ClientSession
:param incorrect_player_name: The incorrect player name to correct
:param platform: Platform, can be pc, xbox, ps4 and nintendo-switch
:return: A player object if a corrected battletag is found
"""
search_terms = []
if platform == "pc":
# Use full name to correct wrong capitalization
player_name_with_discriminator = incorrect_player_name.replace('#', '%23')
# Try short version to account for a wrong discriminator
player_name_short = incorrect_player_name.split('#')[0]
search_terms.append(player_name_with_discriminator)
search_terms.append(player_name_short)
else:
# Use full player name to correct wrong capitalization, there are no discriminators on console
search_terms.append(incorrect_player_name)

# Use full name to correct wrong capitalization
player_name_with_discriminator = incorrect_player_name.replace('#', '%23')
# Try short version to account for a wrong discriminator
player_name_short = incorrect_player_name.split('#')[0]

search_terms = [player_name_with_discriminator, player_name_short]

for search_term in search_terms:
async with session.get(f"https://playoverwatch.com/en-us/search/account-by-name/{search_term}") as r:
if r.status == 200:
profiles = await r.json()
# Filter the data to only include names from the platform
profiles = [profile for profile in profiles if profile["platform"] == platform]
# If we have one match that must be it
if len(profiles) == 1:
new_player = await _get_player_internal(session, profiles[0]["name"], platform)
new_player = await _get_player_internal(session, profiles[0]["battleTag"])
if new_player.success:
new_player.original_player_name = incorrect_player_name
return new_player


async def _get_player_internal(session: aiohttp.ClientSession, player_name: str, platform: str = "pc", correct_player: bool = False) -> Player:
async def _get_player_internal(session: aiohttp.ClientSession, player_name: str,
correct_player: bool = False) -> Player:
"""
Uses an aiohttp session to get a player. This is a coroutine and must be awaited.
:param session: An aiohttp ClientSession
:param player_name: String that is the players name (battletag or other)
:param platform: Platform, can be pc, xbox, ps4 and nintendo-switch
:param correct_player: If True and the lookup fails a correction will be attempted.
:return: A Player object
"""
try:
async with session.get(
f"https://ow-api.com/v2/stats/{platform}/{player_name.replace('#', '-')}/profile") as resp:
f"https://ow-api.com/v2/stats/pc/{player_name.replace('#', '-')}/profile") as resp:
data = await resp.json()
player = Player(player_name, data)
if not correct_player:
return player
elif player.success:
return player
else:
new_player = await _correct_player_internal(session, player_name, platform)
new_player = await _correct_player_internal(session, player_name)
if new_player:
return new_player
else:
Expand All @@ -114,31 +126,38 @@ async def _get_player_internal(session: aiohttp.ClientSession, player_name: str,
return Player(player_name, {"error": "timeout"})


async def get_player_async(player_names: Union[str, List[str]], platform: str = "pc", correct_player: bool = False) -> Union[Player, List[Player]]:
async def get_player_async(player_names: Union[str, List[str]], correct_player: bool = False) -> Union[
Player, List[Player]]:
"""
This is a coroutine and must be awaited.
:param player_names: String that is the players name (battletag or other)
:param platform: Platform, can be pc, xbox, ps4 and nintendo-switch
:param correct_player: If True and the lookup fails a correction will be attempted.
:return: A Player object
"""
async with aiohttp.ClientSession() as session:
if isinstance(player_names, list):
result = await asyncio.gather(*[_get_player_internal(session, player, platform) for player in player_names])
result = await asyncio.gather(*[_get_player_internal(session, player) for player in player_names])
return result
else:
result = await _get_player_internal(session, player_names, platform, correct_player)
result = await _get_player_internal(session, player_names, correct_player)
return result


def get_player(player_names: Union[str, List[str]], platform: str = "pc", correct_player: bool = False) -> Union[Player, List[Player]]:
def get_player(player_names: Union[str, List[str]], correct_player: bool = False) -> Union[Player, List[Player]]:
"""
Automatically creates event loop. Does not work with programs that already have an event loop, await _get_player instead
:param player_names: String that is the players name (battletag or other)
:param platform: Platform, can be pc, xbox, ps4 and nintendo-switch
:param correct_player: If True and the lookup fails a correction will be attempted.
:return: A Player object
"""
loop = asyncio.get_event_loop()
result = loop.run_until_complete(get_player_async(player_names, platform, correct_player))
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

result = loop.run_until_complete(get_player_async(player_names, correct_player))

# Wait 250 ms for the underlying SSL connections to close
# See https://docs.aiohttp.org/en/stable/client_advanced.html#graceful-shutdown
loop.run_until_complete(asyncio.sleep(0.250))
loop.close()

return result
Loading

0 comments on commit 09f0284

Please sign in to comment.