CSLTL/api_dispatch.py at master · boboququ/CSLTL · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
"""Steam + OpenDota API Interaction."""
from urllib.error import HTTPError

import aiohttp
import requests

# Magic number (from steam) :O
INDIVIDUAL_CONSTANT = 0x0110000100000000

# Conversion factor between steam32 ID and account number
CONVERSION_FACTOR = 76561197960265728


def convert_text_to_32id(steam_id):
    """Convert steam ID text to steam 32 ID."""
    steam_id = steam_id[6:]
    # print(steam_id)
    steam_id = steam_id.split(":")
    instance = int(steam_id[1])
    account_number = int(steam_id[2]) * 2 + instance + INDIVIDUAL_CONSTANT
    return account_number - CONVERSION_FACTOR


def convert_32id_to_account(steam_id):
    """Convert steam 32 ID to account number."""
    return steam_id + CONVERSION_FACTOR


def get_account_info_sync(id_32):
    """
    Get account details synchronously with requests.

    Parameters
    ----------
    id_32: int
        The Steam32 ID of a player

    Returns
    -------
    player_resp: dict
        Dict containing player information

    """
    api = "https://api.opendota.com/api/players/" + str(id_32)
    req = requests.get(api)
    if req.status_code != 200:
        raise HTTPError(url=api, code=req.status_code, hdrs=[], fp=None,
                        msg="bad call to api in get_account_info_sync")
    else:
        return req.json()


async def get_account_info_async(session, id_32):
    """
    Get account details asynchronously with aiohttp.

    Parameters
    ----------
    session: aiohttp.ClientSession
        Client session to manage outgoing requests

    id_32: int
        The Steam32 ID of a player

    Returns
    -------
    player_resp: dict
        Dict containing player information

    """
    api = "https://api.opendota.com/api/players/" + str(id_32)
    async with session.get(api) as req:
        if req.status != 200:
            raise HTTPError(url=api, code=req.status, hdrs=[], fp=None,
                            msg="bad call to api in get_account_info_async")
        else:
            return await req.json()


def get_account_heroes_sync(id_32, matches_limit=100, lobby_only=False):
    """
    Get the most played heroes for this player synchronously with requests.

    For more information about the API endpoint used here, see the docs at:
    https://docs.opendota.com/#tag/players%2Fpaths%2F~1players~1%7Baccount_id%7D~1heroes%2Fget

    Additionally, we add the following key-value pairs to the dictionary:
        loc_name, containing the hero's English name
        winrate, containing the player's winrate with the hero

    Parameters
    ----------
    id_32: int
        The Steam32 ID of a player

    matches_limit: int (default 100)
        The number of recent matches to consider for this player.

    lobby_only: bool (default False)
        Limit match results to lobby matches only?

    Returns
    -------
    played_heroes: list of dicts
        The list of num_heroes heroes that the player has played the most.

    """
    api = "https://api.opendota.com/api/players/" + str(id_32) + "/heroes"
    api_params = dict(limit=matches_limit)
    if lobby_only:
        api_params['lobby_type'] = 1
    req = requests.get(api, params=api_params)
    if req.status_code != 200:
        raise HTTPError(url=api, code=req.status_code, hdrs=[], fp=None,
                        msg="bad call to api in get_account_heroes_sync")
    else:
        return req.json()


async def get_account_heroes_async(session: aiohttp.ClientSession, id_32,
                                   matches_limit=100, lobby_only=False):
    """
    Get the most played heroes for this player asynchronously with aiohttp.

    For more information about the API endpoint used here, see the docs at:
    https://docs.opendota.com/#tag/players%2Fpaths%2F~1players~1%7Baccount_id%7D~1heroes%2Fget

    Additionally, we add the following key-value pairs to the dictionary:
        loc_name, containing the hero's English name
        winrate, containing the player's winrate with the hero

    Parameters
    ----------
    session: aiohttp.ClientSession
        Client session to manage outgoing requests

    id_32: int
        The Steam32 ID of a player

    matches_limit: int (default 100)
        The number of recent matches to consider for this player.

    lobby_only: bool (default False)
        Limit match results to lobby matches only?

    Returns
    -------
    played_heroes: list of dicts
        The list of num_heroes heroes that the player has played the most.

    """
    api = "https://api.opendota.com/api/players/" + str(id_32) + "/heroes"
    api_params = dict(limit=matches_limit)
    if lobby_only:
        api_params['lobby_type'] = 1
    async with session.get(api, params=api_params) as req:
        if req.status != 200:
            raise HTTPError(url=api, code=req.status, hdrs=[], fp=None,
                            msg="bad call to api in get_account_heroes_async")
        else:
            return await req.json()


if __name__ == '__main__':
    steam_id = "STEAM_0:1:51900704"
    steam_32 = convert_text_to_32id(steam_id)
    print(steam_32)
    account_info = get_account_info_sync(steam_32)
    account_matches = get_account_heroes_sync(steam_32)
    print(account_info)
    print(account_matches)