-
Notifications
You must be signed in to change notification settings - Fork 44
/
__init__.py
111 lines (94 loc) · 3.23 KB
/
__init__.py
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
"""
Presence - Get online status of friends
"""
from typing import List
from xbox.webapi.api.provider.baseprovider import BaseProvider
from xbox.webapi.api.provider.presence.models import (
PresenceBatchResponse,
PresenceItem,
PresenceLevel,
PresenceState,
)
class PresenceProvider(BaseProvider):
PRESENCE_URL = "https://userpresence.xboxlive.com"
HEADERS_PRESENCE = {"x-xbl-contract-version": "3", "Accept": "application/json"}
async def get_presence(
self,
xuid,
presence_level: PresenceLevel = PresenceLevel.USER,
**kwargs,
) -> PresenceItem:
"""
Get presence for given xuid
Args:
xuid: XUID
presence_level: Filter level
Returns:
:class:`PresenceItem`: Presence Response
"""
url = self.PRESENCE_URL + "/users/xuid(" + xuid + ")?level=" + presence_level
resp = await self.client.session.get(
url, headers=self.HEADERS_PRESENCE, **kwargs
)
resp.raise_for_status()
return PresenceItem(**resp.json())
async def get_presence_batch(
self,
xuids: List[str],
online_only: bool = False,
presence_level: PresenceLevel = PresenceLevel.USER,
**kwargs,
) -> List[PresenceItem]:
"""
Get presence for list of xuids
Args:
xuids: List of XUIDs
online_only: Only get online profiles
presence_level: Filter level
Returns: List[:class:`PresenceItem`]: List of presence items
"""
if len(xuids) > 1100:
raise Exception("Xuid list length is > 1100")
url = self.PRESENCE_URL + "/users/batch"
post_data = {
"users": [str(x) for x in xuids],
"onlineOnly": online_only,
"level": presence_level,
}
resp = await self.client.session.post(
url, json=post_data, headers=self.HEADERS_PRESENCE, **kwargs
)
resp.raise_for_status()
parsed = PresenceBatchResponse.model_validate(resp.json())
return parsed.root
async def get_presence_own(
self, presence_level: PresenceLevel = PresenceLevel.ALL, **kwargs
) -> PresenceItem:
"""
Get presence of own profile
Args:
presence_level: Filter level
Returns:
:class:`PresenceItem`: Presence Response
"""
url = self.PRESENCE_URL + "/users/me"
params = {"level": presence_level}
resp = await self.client.session.get(
url, params=params, headers=self.HEADERS_PRESENCE, **kwargs
)
resp.raise_for_status()
return PresenceItem(**resp.json())
async def set_presence_own(self, presence_state: PresenceState, **kwargs) -> bool:
"""
Set presence of own profile
Args:
presence_state: State of presence
Returns:
`True` on success, `False` otherwise
"""
url = self.PRESENCE_URL + f"/users/xuid({self.client.xuid})/state"
data = {"state": presence_state.value}
resp = await self.client.session.put(
url, json=data, headers=self.HEADERS_PRESENCE, **kwargs
)
return resp.status_code == 200