pytrapi/pytrapi/__init__.py

import datetime
import json
import os
import ssl
import urllib.parse
import urllib.request

"""
official api documentation:
https://github.com/ToontownRewritten/api-doc/blob/master/login.md
https://github.com/ToontownRewritten/api-doc/blob/master/invasions.md
"""

INVASIONS_API_URL = 'https://www.toontownrewritten.com/api/invasions?format=json'
LOGIN_API_URL = 'https://www.toontownrewritten.com/api/login?format=json'


def _utc_from_timestamp(timestamp):
    dt = datetime.datetime.utcfromtimestamp(timestamp)
    return dt.replace(tzinfo = datetime.timezone.utc)

def api_request(url, params=None, validate_ssl_cert=True):
    resp = urllib.request.urlopen(
        url=url,
        data=urllib.parse.urlencode(params).encode('ascii')
            if params else None,
        context=None if validate_ssl_cert
            else ssl._create_unverified_context(),
    )
    return json.loads(resp.read().decode('ascii'))


class LoginSuccessful:

    def __init__(self, playcookie, gameserver):
        self.playcookie = playcookie
        self.gameserver = gameserver


class LoginDelayed:

    def __init__(self, queue_token):
        self.queue_token = queue_token


def login(username=None, password=None,
          queue_token=None, validate_ssl_cert=True):
    if username is not None and queue_token is None:
        assert password is not None
        req_params = {
            'username': username,
            'password': password,
        }
    elif username is None and queue_token is not None:
        req_params = {
            'queueToken': queue_token,
        }
    else:
        raise Exception('either specify username or queue token')
    resp_data = api_request(
        url=LOGIN_API_URL,
        params=req_params,
        validate_ssl_cert=validate_ssl_cert,
    )
    if resp_data['success'] == 'true':
        return LoginSuccessful(
            playcookie=resp_data['cookie'],
            gameserver=resp_data['gameserver'],
        )
    elif resp_data['success'] == 'delayed':
        return LoginDelayed(
            queue_token=resp_data['queueToken'],
        )
    else:
        raise Exception(repr(resp_data))


class InvasionProgress:

    def __init__(self, district, date, cog_type,
                 despawned_number, total_number):
        self.district = district
        self.date = date
        self.cog_type = cog_type
        self.despawned_number = despawned_number
        self.total_number = total_number

    @property
    def remaining_number(self):
        return self.total_number - self.despawned_number


class InvasionsResponse:

    def __init__(self, update_date, invasions):
        self.update_date = update_date
        self.invasions = invasions


def request_active_invasions(validate_ssl_certs=True):
    resp_data = api_request(INVASIONS_API_URL)
    if resp_data['error'] is not None:
        raise Exception(resp_data['error'])
    else:
        invs = {}
        for district, inv_data in resp_data['invasions'].items():
            despawned_number, total_number = inv_data['progress'].split('/')
            invs[district] = InvasionProgress(
                district=district,
                date=_utc_from_timestamp(inv_data['asOf']),
                cog_type=inv_data['type'],
                despawned_number=int(despawned_number),
                total_number=int(total_number),
            )
        return InvasionsResponse(
            update_date=_utc_from_timestamp(resp_data['lastUpdated']),
            invasions=invs,
        )