mirror of
https://github.com/bombsquad-community/plugin-manager.git
synced 2025-10-08 14:54:36 +00:00
Add files via upload
This commit is contained in:
brostos
GitHub
parent
b90deb6a1d
commit
31a61d2c05
1 changed files with 873 additions and 0 deletions
873
plugins/utilities/discord_richpresence.py
Normal file
873
plugins/utilities/discord_richpresence.py
Normal file
|
|
@ -0,0 +1,873 @@
|
|||
# Released under the MIT and Apache License. See LICENSE for details.
|
||||
#
|
||||
"""placeholder :clown:"""
|
||||
|
||||
# ba_meta require api 8
|
||||
#!"Made to you by @brostos & @Dliwk"
|
||||
|
||||
|
||||
from __future__ import annotations
|
||||
from urllib.request import Request, urlopen, urlretrieve
|
||||
from pathlib import Path
|
||||
from os import getcwd, remove
|
||||
from zipfile import ZipFile
|
||||
from bauiv1lib.popup import PopupWindow
|
||||
from babase._mgen.enums import TimeType
|
||||
|
||||
import asyncio
|
||||
import http.client
|
||||
import ast
|
||||
import uuid
|
||||
import json
|
||||
import time
|
||||
import threading
|
||||
import shutil
|
||||
import babase
|
||||
import _babase
|
||||
import bascenev1 as bs
|
||||
import bascenev1lib
|
||||
import bauiv1 as bui
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from typing import Any, Tuple
|
||||
|
||||
|
||||
ANDROID = babase.app.classic.platform == "android"
|
||||
DIRPATH = Path(f"{_babase.app.python_directory_user}/image_id.json")
|
||||
|
||||
if ANDROID: # !can add ios in future
|
||||
|
||||
# Installing websocket
|
||||
def get_module():
|
||||
install_path = Path(f"{getcwd()}/ba_data/python") # For the guys like me on windows
|
||||
path = Path(f"{install_path}/websocket.zip")
|
||||
file_path = Path(f"{install_path}/websocket")
|
||||
if not file_path.exists():
|
||||
url = "https://github.com/brostosjoined/bombsquadrpc/releases/download/presence-1.0/websocket.zip"
|
||||
try:
|
||||
filename, headers = urlretrieve(url, filename=path)
|
||||
with ZipFile(filename) as f:
|
||||
f.extractall(install_path)
|
||||
remove(path)
|
||||
except:
|
||||
pass
|
||||
get_module()
|
||||
|
||||
import websocket
|
||||
|
||||
heartbeat_interval = int(41250)
|
||||
resume_gateway_url: str | None = None
|
||||
session_id: str | None = None
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
class PresenceUpdate:
|
||||
def __init__(self):
|
||||
self.state: str | None = "In Game"
|
||||
self.details: str | None = "Main Menu"
|
||||
self.start_timestamp = time.time()
|
||||
self.large_image_key: str | None = "bombsquadicon"
|
||||
self.large_image_text: str | None = "BombSquad Icon"
|
||||
self.small_image_key: str | None = None
|
||||
self.small_image_text: str | None = (
|
||||
f"{_babase.app.classic.platform.capitalize()}({_babase.app.version})")
|
||||
self.media_proxy = "mp:/app-assets/963434684669382696/{}.png"
|
||||
self.identify: bool = False
|
||||
self.party_id: str = str(uuid.uuid4())
|
||||
self.party_size = 1
|
||||
self.party_max = 8
|
||||
|
||||
def presence(self):
|
||||
with open(DIRPATH, "r") as maptxt:
|
||||
largetxt = json.load(maptxt)[self.large_image_key]
|
||||
with open(DIRPATH, "r") as maptxt:
|
||||
smalltxt = json.load(maptxt)[self.small_image_key]
|
||||
|
||||
presencepayload = {
|
||||
"op": 3,
|
||||
"d": {
|
||||
"since": None, # used to show how long the user went idle will add afk to work with this and then set the status to idle
|
||||
"status": "online",
|
||||
"afk": "false",
|
||||
"activities": [
|
||||
{
|
||||
"name": "BombSquad",
|
||||
"type": 0,
|
||||
"application_id": "963434684669382696",
|
||||
"state": self.state,
|
||||
"details": self.details,
|
||||
"timestamps": {
|
||||
"start": start_time
|
||||
},
|
||||
"party": {
|
||||
"id": self.party_id,
|
||||
"size": [self.party_size, self.party_max]
|
||||
},
|
||||
"assets": {
|
||||
"large_image": self.media_proxy.format(largetxt),
|
||||
"large_text": self.large_image_text,
|
||||
"small_image": self.media_proxy.format(smalltxt),
|
||||
"small_text": self.small_image_text,
|
||||
},
|
||||
"client_info": {
|
||||
"version": 0,
|
||||
"os": "android",
|
||||
"client": "mobile",
|
||||
},
|
||||
"buttons": ["Discord Server", "Download BombSquad"],
|
||||
"metadata": {
|
||||
"button_urls": [
|
||||
"https://discord.gg/bombsquad-ballistica-official-1001896771347304639",
|
||||
"https://bombsquad-community.web.app/download",
|
||||
]
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
ws.send(json.dumps(presencepayload))
|
||||
|
||||
def on_message(ws, message):
|
||||
global heartbeat_interval, resume_gateway_url, session_id
|
||||
message = json.loads(message)
|
||||
try:
|
||||
heartbeat_interval = message["d"]["heartbeat_interval"]
|
||||
except:
|
||||
pass
|
||||
try:
|
||||
resume_gateway_url = message["d"]["resume_gateway_url"]
|
||||
session_id = message["d"]["session_id"]
|
||||
except:
|
||||
pass
|
||||
|
||||
def on_error(ws, error):
|
||||
babase.print_exception(error)
|
||||
|
||||
def on_close(ws, close_status_code, close_msg):
|
||||
# print("### closed ###")
|
||||
pass
|
||||
|
||||
def on_open(ws):
|
||||
print("Connected to Discord Websocket")
|
||||
|
||||
def heartbeats():
|
||||
"""Sending heartbeats to keep the connection alive"""
|
||||
global heartbeat_interval
|
||||
if babase.do_once():
|
||||
heartbeat_payload = {
|
||||
"op": 1,
|
||||
"d": 251,
|
||||
} # step two keeping connection alive by sending heart beats and receiving opcode 11
|
||||
ws.send(json.dumps(heartbeat_payload))
|
||||
|
||||
def identify():
|
||||
"""Identifying to the gateway and enable by using user token and the intents we will be using e.g 256->For Presence"""
|
||||
with open(f"{getcwd()}/token.txt", 'r') as f:
|
||||
token = bytes.fromhex(f.read()).decode('utf-8')
|
||||
identify_payload = {
|
||||
"op": 2,
|
||||
"d": {
|
||||
"token": token,
|
||||
"properties": {
|
||||
"os": "linux",
|
||||
"browser": "Discord Android",
|
||||
"device": "android",
|
||||
},
|
||||
"intents": 256,
|
||||
},
|
||||
} # step 3 send an identify
|
||||
ws.send(json.dumps(identify_payload))
|
||||
identify()
|
||||
while True:
|
||||
heartbeat_payload = {"op": 1, "d": heartbeat_interval}
|
||||
ws.send(json.dumps(heartbeat_payload))
|
||||
time.sleep(heartbeat_interval / 1000)
|
||||
|
||||
threading.Thread(target=heartbeats, daemon=True, name="heartbeat").start()
|
||||
|
||||
# websocket.enableTrace(True)
|
||||
ws = websocket.WebSocketApp(
|
||||
"wss://gateway.discord.gg/?encoding=json&v=10",
|
||||
on_open=on_open,
|
||||
on_message=on_message,
|
||||
on_error=on_error,
|
||||
on_close=on_close,
|
||||
)
|
||||
if Path(f"{getcwd()}/token.txt").exists():
|
||||
threading.Thread(target=ws.run_forever, daemon=True, name="websocket").start()
|
||||
|
||||
|
||||
if not ANDROID:
|
||||
# installing pypresence
|
||||
def get_module():
|
||||
install_path = Path(f"{getcwd()}/ba_data/python")
|
||||
path = Path(f"{install_path}/pypresence.zip")
|
||||
file_path = Path(f"{install_path}/pypresence")
|
||||
if not file_path.exists():
|
||||
url = "https://github.com/brostosjoined/bombsquadrpc/releases/download/presence-1.0/pypresence.zip"
|
||||
try:
|
||||
filename, headers = urlretrieve(url, filename=path)
|
||||
with ZipFile(filename) as f:
|
||||
f.extractall(install_path)
|
||||
remove(path)
|
||||
except:
|
||||
pass
|
||||
get_module()
|
||||
|
||||
#Updating pypresence
|
||||
try:
|
||||
from pypresence import PipeClosed, DiscordError, DiscordNotFound
|
||||
except ImportError:
|
||||
shutil.rmtree(Path(f"{getcwd()}/ba_data/python/pypresence"))
|
||||
get_module()
|
||||
|
||||
|
||||
from pypresence.utils import get_event_loop
|
||||
import pypresence
|
||||
import socket
|
||||
|
||||
DEBUG = True
|
||||
|
||||
_last_server_addr = 'localhost'
|
||||
_last_server_port = 43210
|
||||
|
||||
def print_error(err: str, include_exception: bool = False) -> None:
|
||||
if DEBUG:
|
||||
if include_exception:
|
||||
babase.print_exception(err)
|
||||
else:
|
||||
babase.print_error(err)
|
||||
else:
|
||||
print(f"ERROR in discordrp.py: {err}")
|
||||
|
||||
def log(msg: str) -> None:
|
||||
if DEBUG:
|
||||
print(f"LOG in discordrp.py: {msg}")
|
||||
|
||||
def _run_overrides() -> None:
|
||||
old_init = bs.Activity.__init__
|
||||
|
||||
def new_init(self, *args: Any, **kwargs: Any) -> None: # type: ignore
|
||||
old_init(self, *args, **kwargs)
|
||||
self._discordrp_start_time = time.mktime(time.localtime())
|
||||
|
||||
bs.Activity.__init__ = new_init # type: ignore
|
||||
|
||||
old_connect = bs.connect_to_party
|
||||
|
||||
def new_connect(*args, **kwargs) -> None: # type: ignore
|
||||
global _last_server_addr
|
||||
global _last_server_port
|
||||
old_connect(*args, **kwargs)
|
||||
c = kwargs.get("address") or args[0]
|
||||
_last_server_port = kwargs.get("port") or args[1]
|
||||
|
||||
bs.connect_to_party = new_connect
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
class RpcThread(threading.Thread):
|
||||
def __init__(self):
|
||||
super().__init__(name="RpcThread")
|
||||
self.rpc = pypresence.Presence(963434684669382696)
|
||||
self.state: str | None = "In Game"
|
||||
self.details: str | None = "Main Menu"
|
||||
self.start_timestamp = time.mktime(time.localtime())
|
||||
self.large_image_key: str | None = "bombsquadicon"
|
||||
self.large_image_text: str | None = "BombSquad Icon"
|
||||
self.small_image_key: str | None = None
|
||||
self.small_image_text: str | None = None
|
||||
self.party_id: str = str(uuid.uuid4())
|
||||
self.party_size = 1
|
||||
self.party_max = 8
|
||||
self.join_secret: str | None = None
|
||||
self._last_update_time: float = 0
|
||||
self._last_secret_update_time: float = 0
|
||||
self._last_connect_time: float = 0
|
||||
self.should_close = False
|
||||
|
||||
@staticmethod
|
||||
def is_discord_running():
|
||||
for i in range(6463,6473):
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
s.settimeout(0.01)
|
||||
try:
|
||||
conn = s.connect_ex(('localhost', i))
|
||||
s.close()
|
||||
if (conn == 0):
|
||||
s.close()
|
||||
return(True)
|
||||
except:
|
||||
s.close()
|
||||
return(False)
|
||||
|
||||
def _generate_join_secret(self):
|
||||
# resp = requests.get('https://legacy.ballistica.net/bsAccessCheck').text
|
||||
connection_info = bs.get_connection_to_host_info()
|
||||
if connection_info:
|
||||
addr = _last_server_addr
|
||||
port = _last_server_port
|
||||
else:
|
||||
try:
|
||||
with urlopen(
|
||||
"https://legacy.ballistica.net/bsAccessCheck"
|
||||
) as resp:
|
||||
resp = resp.read().decode()
|
||||
resp = ast.literal_eval(resp)
|
||||
addr = resp["address"]
|
||||
port = 43210
|
||||
secret_dict = {
|
||||
"format_version": 1,
|
||||
"hostname": addr,
|
||||
"port": port,
|
||||
}
|
||||
self.join_secret = json.dumps(secret_dict)
|
||||
except:
|
||||
pass
|
||||
|
||||
def _update_secret(self):
|
||||
threading.Thread(target=self._generate_join_secret, daemon=True).start()
|
||||
self._last_secret_update_time = time.time()
|
||||
|
||||
def run(self) -> None:
|
||||
asyncio.set_event_loop(get_event_loop())
|
||||
while not self.should_close:
|
||||
if time.time() - self._last_update_time > 0.1:
|
||||
self._do_update_presence()
|
||||
if time.time() - self._last_secret_update_time > 15:
|
||||
self._update_secret()
|
||||
# if time.time() - self._last_connect_time > 120 and is_discord_running(): #!Eric please add module manager(pip)
|
||||
# self._reconnect()
|
||||
time.sleep(0.03)
|
||||
|
||||
def _subscribe(self, event: str, **args):
|
||||
self.rpc.send_data(
|
||||
1,
|
||||
{
|
||||
"nonce": f"{time.time():.20f}",
|
||||
"cmd": "SUBSCRIBE",
|
||||
"evt": event,
|
||||
"args": args,
|
||||
},
|
||||
)
|
||||
data = self.rpc.loop.run_until_complete(self.rpc.read_output())
|
||||
self.handle_event(data)
|
||||
|
||||
def _subscribe_events(self):
|
||||
self._subscribe("ACTIVITY_JOIN")
|
||||
self._subscribe("ACTIVITY_JOIN_REQUEST")
|
||||
|
||||
# def _update_presence(self) -> None:
|
||||
# self._last_update_time = time.time()
|
||||
# try:
|
||||
# self._do_update_presence()
|
||||
# except (AttributeError, AssertionError):
|
||||
# try:
|
||||
# self._reconnect()
|
||||
# except Exception:
|
||||
# print_error("failed to update presence", include_exception= True)
|
||||
|
||||
|
||||
def _reconnect(self) -> None:
|
||||
self.rpc.connect()
|
||||
self._subscribe_events()
|
||||
self._do_update_presence()
|
||||
self._last_connect_time = time.time()
|
||||
|
||||
def _do_update_presence(self) -> None:
|
||||
if RpcThread.is_discord_running():
|
||||
self._last_update_time = time.time()
|
||||
try:
|
||||
data = self.rpc.update(
|
||||
state=self.state or " ",
|
||||
details=self.details,
|
||||
start=start_time,
|
||||
large_image=self.large_image_key,
|
||||
large_text=self.large_image_text,
|
||||
small_image=self.small_image_key,
|
||||
small_text=self.small_image_text,
|
||||
party_id=self.party_id,
|
||||
party_size=[self.party_size, self.party_max],
|
||||
join=self.join_secret,
|
||||
# buttons = [ #!cant use buttons together with join
|
||||
# {
|
||||
# "label": "Discord Server",
|
||||
# "url": "https://ballistica.net/discord"
|
||||
# },
|
||||
# {
|
||||
# "label": "Download Bombsquad",
|
||||
# "url": "https://bombsquad.ga/download"}
|
||||
# ]
|
||||
)
|
||||
|
||||
self.handle_event(data)
|
||||
except (PipeClosed, DiscordError, AssertionError, AttributeError):
|
||||
try:
|
||||
self._reconnect()
|
||||
except (DiscordNotFound, DiscordError):
|
||||
pass
|
||||
|
||||
def handle_event(self, data):
|
||||
evt = data["evt"]
|
||||
if evt is None:
|
||||
return
|
||||
|
||||
data = data.get("data", {})
|
||||
|
||||
if evt == "ACTIVITY_JOIN":
|
||||
secret = data.get("secret")
|
||||
try:
|
||||
server = json.loads(secret)
|
||||
format_version = server["format_version"]
|
||||
except Exception:
|
||||
babase.print_exception("discordrp: unknown activity join format")
|
||||
else:
|
||||
try:
|
||||
if format_version == 1:
|
||||
hostname = server["hostname"]
|
||||
port = server["port"]
|
||||
self._connect_to_party(hostname, port)
|
||||
except Exception:
|
||||
babase.print_exception(
|
||||
f"discordrp: incorrect activity join data, {format_version=}"
|
||||
)
|
||||
|
||||
elif evt == "ACTIVITY_JOIN_REQUEST":
|
||||
user = data.get("user", {})
|
||||
uid = user.get("id")
|
||||
username = user.get("username")
|
||||
discriminator = user.get("discriminator", None)
|
||||
avatar = user.get("avatar")
|
||||
self.on_join_request(username, uid, discriminator, avatar)
|
||||
|
||||
def _connect_to_party(self, hostname, port) -> None:
|
||||
babase.pushcall(
|
||||
babase.Call(bs.connect_to_party, hostname, port), from_other_thread=True
|
||||
)
|
||||
|
||||
def on_join_request(self, username, uid, discriminator, avatar) -> None:
|
||||
del uid # unused
|
||||
del avatar # unused
|
||||
babase.pushcall(
|
||||
babase.Call(
|
||||
bui.screenmessage,
|
||||
"Discord: {} wants to join!".format(username),
|
||||
color=(0.0, 1.0, 0.0),
|
||||
),
|
||||
from_other_thread=True,
|
||||
)
|
||||
babase.pushcall(lambda: bui.getsound('bellMed').play(), from_other_thread=True)
|
||||
|
||||
|
||||
class Discordlogin(PopupWindow):
|
||||
|
||||
def __init__(self):
|
||||
# pylint: disable=too-many-locals
|
||||
_uiscale = bui.app.ui_v1.uiscale
|
||||
self._transitioning_out = False
|
||||
s = 1.25 if _uiscale is babase.UIScale.SMALL else 1.27 if _uiscale is babase.UIScale.MEDIUM else 1.3
|
||||
self._width = 380 * s
|
||||
self._height = 150 + 150 * s
|
||||
self.path = Path(f"{getcwd()}/token.txt")
|
||||
bg_color = (0.5, 0.4, 0.6)
|
||||
log_btn_colour = (0.10, 0.95, 0.10) if not self.path.exists() else (1.00, 0.15, 0.15)
|
||||
log_txt = "LOG IN" if not self.path.exists() else "LOG OUT"
|
||||
|
||||
|
||||
|
||||
|
||||
# creates our _root_widget
|
||||
PopupWindow.__init__(self,
|
||||
position=(0.0, 0.0),
|
||||
size=(self._width, self._height),
|
||||
scale=(2.1 if _uiscale is babase.UIScale.SMALL else 1.5
|
||||
if _uiscale is babase.UIScale.MEDIUM else 1.0),
|
||||
bg_color=bg_color)
|
||||
|
||||
|
||||
self._cancel_button = bui.buttonwidget(
|
||||
parent=self.root_widget,
|
||||
position=(25, self._height - 40),
|
||||
size=(50, 50),
|
||||
scale=0.58,
|
||||
label='',
|
||||
color=bg_color,
|
||||
on_activate_call=self._on_cancel_press,
|
||||
autoselect=True,
|
||||
icon=bui.gettexture('crossOut'),
|
||||
iconscale=1.2)
|
||||
|
||||
|
||||
|
||||
bui.imagewidget(parent=self.root_widget,
|
||||
position=(180, self._height - 55),
|
||||
size=(32 * s, 32 * s),
|
||||
texture=bui.gettexture("discordLogo"),
|
||||
color=(10 - 0.32, 10 - 0.39, 10 - 0.96))
|
||||
|
||||
|
||||
|
||||
self.email_widget = bui.textwidget(parent=self.root_widget,
|
||||
text="Email/Phone Number",
|
||||
size=(400, 70),
|
||||
position=(50, 180),
|
||||
h_align='left',
|
||||
v_align='center',
|
||||
editable=True,
|
||||
scale=0.8,
|
||||
autoselect=True,
|
||||
maxwidth=220)
|
||||
|
||||
|
||||
self.password_widget = bui.textwidget(parent=self.root_widget,
|
||||
text="Password",
|
||||
size=(400, 70),
|
||||
position=(50, 120),
|
||||
h_align='left',
|
||||
v_align='center',
|
||||
editable=True,
|
||||
scale=0.8,
|
||||
autoselect=True,
|
||||
maxwidth=220)
|
||||
|
||||
|
||||
bui.containerwidget(edit=self.root_widget,
|
||||
cancel_button=self._cancel_button)
|
||||
|
||||
bui.textwidget(
|
||||
parent=self.root_widget,
|
||||
position=(265, self._height - 37),
|
||||
size=(0, 0),
|
||||
h_align='center',
|
||||
v_align='center',
|
||||
scale=1.0,
|
||||
text="Discord",
|
||||
maxwidth=200,
|
||||
color=(0.80, 0.80, 0.80))
|
||||
|
||||
bui.textwidget(
|
||||
parent=self.root_widget,
|
||||
position=(265, self._height - 78),
|
||||
size=(0, 0),
|
||||
h_align='center',
|
||||
v_align='center',
|
||||
scale=1.0,
|
||||
text="💀Use at your own risk💀\n ⚠️discord account might get terminated⚠️",
|
||||
maxwidth=200,
|
||||
color=(1.00, 0.15, 0.15))
|
||||
|
||||
|
||||
self._login_button = bui.buttonwidget(
|
||||
parent=self.root_widget,
|
||||
position=(120, 65),
|
||||
size=(400, 80),
|
||||
scale=0.58,
|
||||
label=log_txt,
|
||||
color=log_btn_colour,
|
||||
on_activate_call=self.login,
|
||||
autoselect=True)
|
||||
|
||||
def _on_cancel_press(self) -> None:
|
||||
self._transition_out()
|
||||
|
||||
def _transition_out(self) -> None:
|
||||
if not self._transitioning_out:
|
||||
self._transitioning_out = True
|
||||
bui.containerwidget(edit=self.root_widget, transition='out_scale')
|
||||
|
||||
def on_bascenev1libup_cancel(self) -> None:
|
||||
bui.getsound('swish').play()
|
||||
self._transition_out()
|
||||
|
||||
def login(self):
|
||||
if not self.path.exists():
|
||||
json_data = {
|
||||
'login': bui.textwidget(query=self.email_widget),
|
||||
'password': bui.textwidget(query=self.password_widget),
|
||||
'undelete': False,
|
||||
'captcha_key': None,
|
||||
'login_source': None,
|
||||
'gift_code_sku_id': None,
|
||||
}
|
||||
headers = {
|
||||
'user-agent': "Mozilla/5.0",
|
||||
'content-type': "application/json",
|
||||
}
|
||||
|
||||
conn = http.client.HTTPSConnection("discord.com")
|
||||
|
||||
payload = json.dumps(json_data)
|
||||
# conn.request("POST", "/api/v9/auth/login", payload, headers)
|
||||
# res = conn.getresponse().read()
|
||||
|
||||
try:
|
||||
conn.request("POST", "/api/v9/auth/login", payload, headers)
|
||||
res = conn.getresponse().read()
|
||||
token = json.loads(res)['token'].encode().hex().encode()
|
||||
with open(self.path, 'wb') as f:
|
||||
f.write(token)
|
||||
bui.screenmessage("Successfully logged in", (0.21, 1.0, 0.20))
|
||||
bui.getsound('shieldUp').play()
|
||||
self.on_bascenev1libup_cancel()
|
||||
except:
|
||||
bui.screenmessage("Incorrect credentials", (1.00, 0.15, 0.15))
|
||||
bui.getsound('error').play()
|
||||
|
||||
conn.close()
|
||||
else:
|
||||
remove(self.path)
|
||||
bui.getsound('shieldDown').play()
|
||||
bui.screenmessage("Account successfully removed!!", (0.10, 0.10, 1.00))
|
||||
self.on_bascenev1libup_cancel()
|
||||
ws.close()
|
||||
|
||||
|
||||
run_once = False
|
||||
def get_once_asset():
|
||||
global run_once
|
||||
if run_once:
|
||||
return
|
||||
response = Request(
|
||||
"https://discordapp.com/api/oauth2/applications/963434684669382696/assets",
|
||||
headers={"User-Agent": "Mozilla/5.0"},
|
||||
)
|
||||
try:
|
||||
with urlopen(response) as assets:
|
||||
assets = json.loads(assets.read())
|
||||
asset = []
|
||||
asset_id = []
|
||||
for x in assets:
|
||||
dem = x["name"]
|
||||
don = x["id"]
|
||||
asset_id.append(don)
|
||||
asset.append(dem)
|
||||
asset_id_dict = dict(zip(asset, asset_id))
|
||||
|
||||
with open(DIRPATH, "w") as imagesets:
|
||||
jsonfile = json.dumps(asset_id_dict)
|
||||
json.dump(asset_id_dict, imagesets, indent=4)
|
||||
except:
|
||||
pass
|
||||
run_once = True
|
||||
|
||||
def get_class():
|
||||
if ANDROID:
|
||||
return PresenceUpdate()
|
||||
elif not ANDROID:
|
||||
return RpcThread()
|
||||
|
||||
|
||||
# ba_meta export babase.Plugin
|
||||
class DiscordRP(babase.Plugin):
|
||||
def __init__(self) -> None:
|
||||
self.update_timer: bs.Timer | None = None
|
||||
self.rpc_thread = get_class()
|
||||
self._last_server_info: str | None = None
|
||||
|
||||
if not ANDROID:
|
||||
_run_overrides()
|
||||
get_once_asset()
|
||||
|
||||
def on_app_running(self) -> None:
|
||||
if not ANDROID:
|
||||
self.rpc_thread.start()
|
||||
self.update_timer = bs.AppTimer(
|
||||
1, bs.WeakCall(self.update_status), repeat=True
|
||||
)
|
||||
if ANDROID:
|
||||
self.update_timer = bs.AppTimer(
|
||||
4, bs.WeakCall(self.update_status), repeat=True
|
||||
)
|
||||
|
||||
def has_settings_ui(self):
|
||||
return True
|
||||
|
||||
def show_settings_ui(self, button):
|
||||
if not ANDROID:
|
||||
bui.screenmessage("Nothing here achievement!!!", (0.26, 0.65, 0.94))
|
||||
bui.getsound('achievement').play()
|
||||
if ANDROID:
|
||||
Discordlogin()
|
||||
|
||||
def on_app_shutdown(self) -> None:
|
||||
if not ANDROID and self.rpc_thread.is_discord_running():
|
||||
self.rpc_thread.rpc.close()
|
||||
self.rpc_thread.should_close = True
|
||||
else:
|
||||
# stupid code
|
||||
ws.close()
|
||||
|
||||
def _get_current_activity_name(self) -> str | None:
|
||||
act = bs.get_foreground_host_activity()
|
||||
if isinstance(act, bs.GameActivity):
|
||||
return act.name
|
||||
|
||||
this = "Lobby"
|
||||
name: str | None = (
|
||||
act.__class__.__name__.replace("Activity", "")
|
||||
.replace("ScoreScreen", "Ranking")
|
||||
.replace("Coop", "")
|
||||
.replace("MultiTeam", "")
|
||||
.replace("Victory", "")
|
||||
.replace("EndSession", "")
|
||||
.replace("Transition", "")
|
||||
.replace("Draw", "")
|
||||
.replace("FreeForAll", "")
|
||||
.replace("Join", this)
|
||||
.replace("Team", "")
|
||||
.replace("Series", "")
|
||||
.replace("CustomSession", "Custom Session(mod)")
|
||||
)
|
||||
|
||||
if name == "MainMenu":
|
||||
name = "Main Menu"
|
||||
if name == this:
|
||||
self.rpc_thread.large_image_key = "lobby"
|
||||
self.rpc_thread.large_image_text = "Bombing up"
|
||||
#self.rpc_thread.small_image_key = "lobbysmall"
|
||||
if name == "Ranking":
|
||||
self.rpc_thread.large_image_key = "ranking"
|
||||
self.rpc_thread.large_image_text = "Viewing Results"
|
||||
return name
|
||||
|
||||
def _get_current_map_name(self) -> Tuple[str | None, str | None]:
|
||||
act = bs.get_foreground_host_activity()
|
||||
if isinstance(act, bs.GameActivity):
|
||||
texname = act.map.get_preview_texture_name()
|
||||
if texname:
|
||||
return act.map.name, texname.lower().removesuffix("preview")
|
||||
return None, None
|
||||
|
||||
def update_status(self) -> None:
|
||||
roster = bs.get_game_roster()
|
||||
connection_info = bs.get_connection_to_host_info()
|
||||
|
||||
self.rpc_thread.large_image_key = "bombsquadicon"
|
||||
self.rpc_thread.large_image_text = "BombSquad"
|
||||
self.rpc_thread.small_image_key = _babase.app.classic.platform
|
||||
self.rpc_thread.small_image_text = (
|
||||
f"{_babase.app.classic.platform.capitalize()}({_babase.app.version})"
|
||||
)
|
||||
connection_info = bs.get_connection_to_host_info()
|
||||
if not ANDROID:
|
||||
svinfo = str(connection_info)
|
||||
if self._last_server_info != svinfo:
|
||||
self._last_server_info = svinfo
|
||||
self.rpc_thread.party_id = str(uuid.uuid4())
|
||||
self.rpc_thread._update_secret()
|
||||
if connection_info != {}:
|
||||
servername = connection_info["name"]
|
||||
self.rpc_thread.details = "Online"
|
||||
self.rpc_thread.party_size = max(
|
||||
1, sum(len(client["players"]) for client in roster)
|
||||
)
|
||||
self.rpc_thread.party_max = max(8, self.rpc_thread.party_size)
|
||||
if len(servername) == 19 and "Private Party" in servername:
|
||||
self.rpc_thread.state = "Private Party"
|
||||
elif servername == "": # A local game joinable from the internet
|
||||
try:
|
||||
offlinename = json.loads(bs.get_game_roster()[0]["spec_string"])[
|
||||
"n"
|
||||
]
|
||||
if len(offlinename > 19):
|
||||
self.rpc_thread.state = offlinename[slice(19)] + "..."
|
||||
else:
|
||||
self.rpc_thread.state = offlinename
|
||||
except IndexError:
|
||||
pass
|
||||
else:
|
||||
if len(servername) > 19:
|
||||
self.rpc_thread.state = servername[slice(19)] + ".."
|
||||
else:
|
||||
self.rpc_thread.state = servername[slice(19)]
|
||||
|
||||
if connection_info == {}:
|
||||
self.rpc_thread.details = "Local" #! replace with something like ballistica github cause
|
||||
self.rpc_thread.state = self._get_current_activity_name()
|
||||
self.rpc_thread.party_size = max(1, len(roster))
|
||||
self.rpc_thread.party_max = max(1, bs.get_public_party_max_size())
|
||||
|
||||
if (
|
||||
bs.get_foreground_host_session() is not None
|
||||
and self.rpc_thread.details == "Local"
|
||||
):
|
||||
session = (
|
||||
bs.get_foreground_host_session()
|
||||
.__class__.__name__.replace("MainMenuSession", "")
|
||||
.replace("EndSession", "")
|
||||
.replace("FreeForAllSession", ": FFA") #! for session use small image key
|
||||
.replace("DualTeamSession", ": Teams")
|
||||
.replace("CoopSession", ": Coop")
|
||||
)
|
||||
#! self.rpc_thread.small_image_key = session.lower()
|
||||
self.rpc_thread.details = f"{self.rpc_thread.details} {session}"
|
||||
if (
|
||||
self.rpc_thread.state == "NoneType"
|
||||
): # sometimes the game just breaks which means its not really watching replay FIXME
|
||||
self.rpc_thread.state = "Watching Replay"
|
||||
self.rpc_thread.large_image_key = "replay"
|
||||
self.rpc_thread.large_image_text = "Viewing Awesomeness"
|
||||
#!self.rpc_thread.small_image_key = "replaysmall"
|
||||
|
||||
act = bs.get_foreground_host_activity()
|
||||
session = bs.get_foreground_host_session()
|
||||
if act:
|
||||
from bascenev1lib.game.elimination import EliminationGame
|
||||
from bascenev1lib.game.thelaststand import TheLastStandGame
|
||||
from bascenev1lib.game.meteorshower import MeteorShowerGame
|
||||
|
||||
# noinspection PyUnresolvedReferences,PyProtectedMember
|
||||
try:
|
||||
self.rpc_thread.start_timestamp = act._discordrp_start_time # type: ignore
|
||||
except AttributeError:
|
||||
# This can be the case if plugin launched AFTER activity
|
||||
# has been created; in that case let's assume it was
|
||||
# created just now.
|
||||
self.rpc_thread.start_timestamp = act._discordrp_start_time = time.mktime( # type: ignore
|
||||
time.localtime()
|
||||
)
|
||||
if isinstance(act, EliminationGame):
|
||||
alive_count = len([p for p in act.players if p.lives > 0])
|
||||
self.rpc_thread.details += f" ({alive_count} players left)"
|
||||
elif isinstance(act, TheLastStandGame):
|
||||
# noinspection PyProtectedMember
|
||||
points = act._score
|
||||
self.rpc_thread.details += f" ({points} points)"
|
||||
elif isinstance(act, MeteorShowerGame):
|
||||
with bs.ContextRef(act):
|
||||
sec = bs.time() - act._timer.getstarttime()
|
||||
secfmt = ""
|
||||
if sec < 60:
|
||||
secfmt = f"{sec:.2f}"
|
||||
else:
|
||||
secfmt = f"{int(sec) // 60:02}:{sec:.2f}"
|
||||
self.rpc_thread.details += f" ({secfmt})"
|
||||
|
||||
# if isinstance(session, ba.DualTeamSession):
|
||||
# scores = ':'.join([
|
||||
# str(t.customdata['score'])
|
||||
# for t in session.sessionteams
|
||||
# ])
|
||||
# self.rpc_thread.details += f' ({scores})'
|
||||
|
||||
mapname, short_map_name = self._get_current_map_name()
|
||||
if mapname:
|
||||
with open(DIRPATH, 'r') as asset_dict:
|
||||
asset_keys = json.load(asset_dict).keys()
|
||||
if short_map_name in asset_keys:
|
||||
self.rpc_thread.large_image_text = mapname
|
||||
self.rpc_thread.large_image_key = short_map_name
|
||||
self.rpc_thread.small_image_key = 'bombsquadlogo2'
|
||||
self.rpc_thread.small_image_text = 'BombSquad'
|
||||
|
||||
if _babase.get_idle_time() / (1000 * 60) % 60 >= 0.4:
|
||||
self.rpc_thread.details = f"AFK in {self.rpc_thread.details}"
|
||||
if not ANDROID:
|
||||
self.rpc_thread.large_image_key = (
|
||||
"https://media.tenor.com/uAqNn6fv7x4AAAAM/bombsquad-spaz.gif"
|
||||
)
|
||||
if ANDROID and Path(f"{getcwd()}/token.txt").exists():
|
||||
self.rpc_thread.presence()
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue