mirror of
https://github.com/imayushsaini/Bombsquad-Ballistica-Modded-Server.git
synced 2025-10-20 00:00:39 +00:00
87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
# ba_meta require api 6
|
|
from __future__ import annotations
|
|
|
|
import ba
|
|
import _ba
|
|
import bacommon.cloud
|
|
import logging
|
|
from efro.error import CommunicationError
|
|
from playersData import pdata
|
|
|
|
|
|
STATUS_CHECK_INTERVAL_SECONDS = 2.0
|
|
|
|
|
|
class AccountUtil:
|
|
def __init__(self):
|
|
self._proxyid: str | None = None
|
|
self._proxykey: str | None = None
|
|
ba.internal.sign_out_v1()
|
|
ba.app.cloud.send_message_cb(bacommon.cloud.LoginProxyRequestMessage(),
|
|
on_response=ba.Call(self._on_proxy_request_response))
|
|
|
|
def _on_proxy_request_response(self, response: bacommon.cloud.LoginProxyRequestResponse | Exception) -> None:
|
|
if isinstance(response, Exception):
|
|
logging.error("error occured")
|
|
logging.critical("Falling back to V1 account")
|
|
ba.internal.sign_in_v1('Local')
|
|
return
|
|
address = ba.internal.get_master_server_address(
|
|
version=2) + response.url
|
|
logging.debug("Copy this URL to your browser : " + address)
|
|
self._proxyid = response.proxyid
|
|
self._proxykey = response.proxykey
|
|
ba.timer(STATUS_CHECK_INTERVAL_SECONDS,
|
|
ba.Call(self._ask_for_status))
|
|
|
|
def _ask_for_status(self) -> None:
|
|
assert self._proxyid is not None
|
|
assert self._proxykey is not None
|
|
ba.app.cloud.send_message_cb(
|
|
bacommon.cloud.LoginProxyStateQueryMessage(
|
|
proxyid=self._proxyid, proxykey=self._proxykey),
|
|
on_response=ba.Call(self._got_status))
|
|
|
|
def _got_status(
|
|
self, response: bacommon.cloud.LoginProxyStateQueryResponse | Exception
|
|
) -> None:
|
|
# For now, if anything goes wrong on the server-side, just abort
|
|
# with a vague error message. Can be more verbose later if need be.
|
|
if (isinstance(response, bacommon.cloud.LoginProxyStateQueryResponse)
|
|
and response.state is response.State.FAIL):
|
|
logging.error("error occured ..terminating login request")
|
|
logging.critical("Falling back to V1 account")
|
|
ba.internal.sign_in_v1('Local')
|
|
|
|
# If we got a token, set ourself as signed in. Hooray!
|
|
if (isinstance(response, bacommon.cloud.LoginProxyStateQueryResponse)
|
|
and response.state is response.State.SUCCESS):
|
|
assert response.credentials is not None
|
|
ba.app.accounts_v2.set_primary_credentials(response.credentials)
|
|
|
|
ba.timer(3, self._logged_in)
|
|
|
|
return
|
|
|
|
# If we're still waiting, ask again soon.
|
|
if (isinstance(response, Exception)
|
|
or response.state is response.State.WAITING):
|
|
ba.timer(STATUS_CHECK_INTERVAL_SECONDS,
|
|
ba.Call(self._ask_for_status))
|
|
|
|
def _logged_in(self):
|
|
logging.info("Logged in as: " +
|
|
ba.internal.get_v1_account_display_string())
|
|
|
|
|
|
def updateOwnerIps():
|
|
if "owner" in pdata.get_roles():
|
|
accountIds = pdata.get_roles()["owner"]["ids"]
|
|
profiles = pdata.get_profiles()
|
|
|
|
for account_id in accountIds:
|
|
if account_id in profiles:
|
|
profile = profiles[account_id]
|
|
if "lastIP" in profile:
|
|
ip = profile["lastIP"]
|
|
_ba.append_owner_ip(ip)
|