Mojang API Wrapper with functions and classes

pymojang/user/__init__.py

@@ -0,0 +1,2 @@
+from . import api
+from .session import UserSession

pymojang/user/api.py

@@ -0,0 +1,109 @@
+import requests
+import json
+import datetime as dt
+from urllib.parse import urljoin
+from base64 import urlsafe_b64decode
+from .profile import UserProfile
+
+MOJANG_STATUS_URL = 'https://status.mojang.com/check'
+MOJANG_API_URL = 'https://api.mojang.com'
+MOJANG_SESSION_URL = 'https://sessionserver.mojang.com'
+
+def api_status():
+    result = {}
+    response = requests.get(MOJANG_STATUS_URL)
+    if response.status_code == 200:
+        data = response.json()
+
+        for status in data:
+            for key, value in status.items():
+                result[key] = value
+
+    return result
+
+def get_name_history(player_id: str):
+    url = urljoin(MOJANG_API_URL, 'user/profiles/{}/names'.format(player_id))
+    response = requests.get(url)
+
+    names = []
+    if response.status_code == 200:
+        data = response.json()
+
+        for item in data:
+            if 'changedToAt' in item:
+                item['changedToAt'] = dt.datetime.fromtimestamp(item['changedToAt'])
+            names.append((item['name'], item.get('changedToAt',None)))
+
+    return names
+
+def get_uuid(username: str, timestamp=None, only_uuid=True):
+    url = urljoin(MOJANG_API_URL, 'users/profiles/minecraft/{}'.format(username))
+    params = {'at': timestamp} if timestamp else {}
+
+    response = requests.get(url, params=params)
+    player_uuid = None
+    player_name = None
+    player_is_legacy = False
+    player_is_demo = False
+    if response.status_code == 200:
+        data = response.json()
+
+        player_uuid = data['id']
+        player_name = data['name']
+        player_is_legacy = data.get('legacy', False)
+        player_is_demo = data.get('demo', False)
+
+    if only_uuid:
+        return player_uuid
+
+    return player_uuid, player_name, player_is_legacy, player_is_demo
+
+def get_uuids(usernames: list, only_uuid=True):
+    url = urljoin(MOJANG_API_URL, 'profiles/minecraft')
+    players_data = []
+
+    if len(usernames) > 0:
+        response = requests.post(url, json=usernames)
+
+        if response.status_code == 200:
+            data = response.json()
+
+            for player_data in data:
+                player_uuid = player_data['id']
+                player_name = player_data['name']
+                player_is_legacy = player_data.get('legacy', False)
+                player_is_demo = player_data.get('demo', False)
+
+            if only_uuid:
+                players_data.append(player_uuid)
+            else:
+                players_data.append((player_uuid, player_name, player_is_legacy, player_is_demo))
+
+    return players_data
+
+def get_profile(player_id: str):
+    url = urljoin(MOJANG_SESSION_URL, 'session/minecraft/profile/{}'.format(player_id))
+    response = requests.get(url)
+    profile = UserProfile()
+
+    profile.names = get_name_history(player_id)
+
+    if response.status_code == 200:
+        data = response.json()
+
+        profile.id = data['id']
+        profile.name = data['name']
+
+        for d in data['properties']:
+            textures = json.loads(urlsafe_b64decode(d['value']))['textures']
+            if 'SKIN' in textures.keys():
+                profile.skins = [{
+                    'url': textures['SKIN']['url'],
+                    'variant': textures['SKIN'].get('metadata',{}).get('model','classic')
+                }]
+            if 'CAPE' in textures.keys():
+                profile.capes = [{
+                    'url': textures['CAPE']['url']
+                }]
+
+    return profile

pymojang/user/profile.py

@@ -0,0 +1,13 @@
+
+class UserProfile:
+
+    def __init__(self):
+        self.created_at = None
+        self.name_change_allowed = None
+
+        self.id = None
+        self.name = None
+        self.skins = []
+        self.capes = []
+
+        self.names = []

pymojang/user/session.py

@@ -0,0 +1,100 @@
+import requests
+import os
+import datetime as dt
+from urllib.parse import urljoin
+from . import api
+from .profile import UserProfile
+from ..auth import Yggdrasil, SecurityCheck
+from ..utils import TokenPair
+
+class UserSession:
+
+    MINECRAFT_SERVICE_URL = 'https://api.minecraftservices.com'
+
+    def __init__(self, username: str, password: str, token_file=None):
+        self._session = requests.Session()
+        self._session.headers.update({'Content-Type': 'application/json'})
+
+        self._username = username
+        self._password = password
+
+        self.token_pair = TokenPair(None, None)
+        if isinstance(token_file, str) and os.path.exists(token_file):
+            self.token_pair = TokenPair.from_pickle(token_file)
+
+        self._auth = Yggdrasil(self._session, self.token_pair)
+        self._security = SecurityCheck(self._session)
+        self._profile = UserProfile()
+
+        self._security_challenges = self._security.challenges
+
+    def connect(self):
+        if self.token_pair.access_token is not None:
+            if not self._auth.validate():
+                self._auth.refresh()
+        else:
+            self._auth.authenticate(self._username, self._password)
+
+        self._load_user_data()
+
+    def disconnect(self):
+        return self._auth.invalidate()
+
+    def save(self, filename: str):
+        self.token_pair.to_pickle(filename)
+
+    @property
+    def profile(self):
+        return self._profile
+
+    # Security questions/answers
+    @property
+    def must_check_security(self):
+        return not self._security.ok
+
+    @property
+    def security_challenges(self):
+        return self._security_challenges
+
+    def send_security_answers(self, answers: list):
+        return self._security.send_answers(answers)
+
+    # User data
+    def _load_user_data(self):
+        self._get_name_change()
+        self._get_profile()
+
+        self._profile.names = api.get_name_history(self._profile.id)
+
+    def _get_name_change(self):
+        name_change_url = urljoin(self.MINECRAFT_SERVICE_URL, 'minecraft/profile/namechange')
+        response = self._session.get(name_change_url)
+
+        if response.status_code == 200:
+            data = response.json()
+            self._profile.created_at = dt.datetime.strptime(data['createdAt'], '%Y-%m-%dT%H:%M:%SZ')
+            self._profile.name_change_allowed = data['nameChangeAllowed']
+        else:
+            pass
+
+    def _get_profile(self):
+        profile_url = urljoin(self.MINECRAFT_SERVICE_URL, 'minecraft/profile')
+        response = self._session.get(profile_url)
+
+        if response.status_code == 200:
+            data = response.json()
+            self._profile.id = data['id']
+            self._profile.name = data['name']
+
+            for skin in data['skins']:
+                self._profile.skins.append({
+                    'url': skin['url'],
+                    'variant': skin['variant'].lower()
+                })
+
+            for cape in data['capes']:
+                self._profile.capes.append({
+                    'url': cape['url']
+                })
+        else:
+            pass