__init__.py 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141
  1. import datetime
  2. import json
  3. import os
  4. import ssl
  5. import subprocess
  6. import sys
  7. import typing
  8. import urllib.parse
  9. import urllib.request
  10. # official api documentation:
  11. # https://github.com/ToontownRewritten/api-doc/blob/master/login.md
  12. # https://github.com/ToontownRewritten/api-doc/blob/master/invasions.md
  13. _LOGIN_API_URL = "https://www.toontownrewritten.com/api/login?format=json"
  14. if sys.platform == "darwin":
  15. TOONTOWN_LIBRARY_PATH = os.path.join(
  16. os.path.expanduser("~"), "Library", "Application Support", "Toontown Rewritten",
  17. )
  18. TOONTOWN_ENGINE_DEFAULT_PATH = os.path.join(
  19. TOONTOWN_LIBRARY_PATH, "Toontown Rewritten",
  20. )
  21. else:
  22. TOONTOWN_LIBRARY_PATH = None
  23. TOONTOWN_ENGINE_DEFAULT_PATH = None
  24. def start_engine(
  25. engine_path: str, gameserver: str, playcookie: str, **popen_kwargs
  26. ) -> subprocess.Popen:
  27. env = {
  28. "TTR_GAMESERVER": gameserver,
  29. "TTR_PLAYCOOKIE": playcookie,
  30. }
  31. if sys.platform == "darwin":
  32. env["DYLD_LIBRARY_PATH"] = os.path.join(
  33. TOONTOWN_LIBRARY_PATH, "Libraries.bundle",
  34. )
  35. env["DYLD_FRAMEWORK_PATH"] = os.path.join(TOONTOWN_LIBRARY_PATH, "Frameworks",)
  36. elif sys.platform == "linux" and "XAUTHORITY" in os.environ:
  37. # Fix for TTREngine reporting:
  38. # > :display:x11display(error): Could not open display ":0.0".
  39. # > :ToonBase: Default graphics pipe is glxGraphicsPipe (OpenGL).
  40. # > :ToonBase(warning): Unable to open 'onscreen' window.
  41. # > Traceback (most recent call last):
  42. # > File "<compiled '__voltorbmain__'>", line 0, in <module>
  43. # > [...]
  44. # > File "<compiled 'direct.vlt8f63e471.ShowBase'>", line 0, in vltf05fd21b
  45. # > Exception: Could not open window.
  46. env["XAUTHORITY"] = os.environ["XAUTHORITY"]
  47. return subprocess.Popen(
  48. args=[engine_path], cwd=os.path.dirname(engine_path), env=env, **popen_kwargs,
  49. )
  50. def _api_request(
  51. url: str, params: typing.Optional[dict] = None, validate_ssl_cert: bool = True
  52. ):
  53. resp = urllib.request.urlopen(
  54. url=url,
  55. data=urllib.parse.urlencode(params).encode("ascii") if params else None,
  56. context=None if validate_ssl_cert else ssl._create_unverified_context(),
  57. )
  58. return json.loads(resp.read().decode("ascii"))
  59. class _LoginSuccessful:
  60. def __init__(self, playcookie: str, gameserver: str):
  61. self.playcookie = playcookie
  62. self.gameserver = gameserver
  63. class _LoginDelayed:
  64. def __init__(self, queue_token: str):
  65. self.queue_token = queue_token
  66. def login(
  67. username: typing.Optional[str] = None,
  68. password: typing.Optional[str] = None,
  69. queue_token: typing.Optional[str] = None,
  70. validate_ssl_cert: bool = True,
  71. ) -> typing.Union[_LoginSuccessful, _LoginDelayed]:
  72. if username is not None and queue_token is None:
  73. assert password is not None
  74. req_params = {
  75. "username": username,
  76. "password": password,
  77. }
  78. elif username is None and queue_token is not None:
  79. req_params = {
  80. "queueToken": queue_token,
  81. }
  82. else:
  83. raise Exception("either specify username or queue token")
  84. resp_data = _api_request(
  85. url=_LOGIN_API_URL, params=req_params, validate_ssl_cert=validate_ssl_cert,
  86. )
  87. if resp_data["success"] == "true":
  88. return _LoginSuccessful(
  89. playcookie=resp_data["cookie"], gameserver=resp_data["gameserver"],
  90. )
  91. if resp_data["success"] == "delayed":
  92. return _LoginDelayed(queue_token=resp_data["queueToken"],)
  93. raise Exception(repr(resp_data))
  94. def launch(
  95. engine_path: str,
  96. username: str,
  97. password: str,
  98. validate_ssl_certs: bool = True,
  99. cpu_limit_percent: typing.Optional[int] = None,
  100. ) -> None:
  101. result = login(
  102. username=username, password=password, validate_ssl_cert=validate_ssl_certs,
  103. )
  104. if isinstance(result, _LoginDelayed):
  105. result = login(
  106. queue_token=result.queue_token, validate_ssl_cert=validate_ssl_certs,
  107. )
  108. if not isinstance(result, _LoginSuccessful):
  109. raise Exception("unexpected response: {!r}".format(result))
  110. process = start_engine(
  111. engine_path=engine_path,
  112. gameserver=result.gameserver,
  113. playcookie=result.playcookie,
  114. )
  115. if cpu_limit_percent is not None:
  116. subprocess.Popen(
  117. args=[
  118. "cpulimit",
  119. "--pid",
  120. str(process.pid),
  121. "--limit",
  122. str(cpu_limit_percent),
  123. # '--verbose',
  124. ]
  125. )
  126. process.wait()