This commit is contained in:
Ayush Saini 2024-05-19 23:03:10 +05:30
parent 7d21296d63
commit 553e6caf36
20 changed files with 441 additions and 61 deletions

View file

@ -1,4 +1,4 @@
#!/usr/bin/env -S python3.12 -O
#!/usr/bin/env python3.12
# Released under the MIT License. See LICENSE for details.
#
# pylint: disable=too-many-lines
@ -12,6 +12,7 @@ import json
import signal
import tomllib
import subprocess
import platform
from pathlib import Path
from threading import Lock, Thread, current_thread
from typing import TYPE_CHECKING
@ -168,7 +169,6 @@ class ServerManagerApp:
)
# Python will handle SIGINT for us (as KeyboardInterrupt) but we
# need to register a SIGTERM handler so we have a chance to clean
# need to register a SIGTERM handler so we have a chance to
# clean up our subprocess when someone tells us to die. (and
# avoid zombie processes)

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View file

@ -41,7 +41,7 @@ from spazmod import modifyspaz
from stats import mystats
from tools import account
from tools import notification_manager
from tools import servercheck, ServerUpdate, logger, playlist, servercontroller
from tools import servercheck, server_update, logger, playlist, servercontroller
if TYPE_CHECKING:
from typing import Any
@ -63,7 +63,7 @@ class modSetup(babase.Plugin):
plus = bui.app.plus
bootstraping()
servercheck.checkserver().start()
ServerUpdate.check()
server_update.check()
# bs.apptimer(5, account.updateOwnerIps)
if settings["afk_remover"]['enable']:
afk_check.checkIdle().start()

View file

@ -200,7 +200,7 @@ def show_player_scores(self,
is_two_team = True if len(self.session.sessionteams) == 2 else False
def _get_prec_score(p_rec: bs.PlayerRecord) -> Optional[int]:
def _get_prec_score(p_rec: bs.PlayerRecord) -> int | None:
if is_free_for_all and results is not None:
assert isinstance(results, bs.GameResults)
assert p_rec.team.activityteam is not None
@ -208,7 +208,7 @@ def show_player_scores(self,
return val
return p_rec.accumscore
def _get_prec_score_str(p_rec: bs.PlayerRecord) -> Union[str, babase.Lstr]:
def _get_prec_score_str(p_rec: bs.PlayerRecord) -> str | bs.Lstr:
if is_free_for_all and results is not None:
assert isinstance(results, bs.GameResults)
assert p_rec.team.activityteam is not None
@ -228,7 +228,7 @@ def show_player_scores(self,
valid_players = list(self.stats.get_records().items())
def _get_player_score_set_entry(
player: bs.SessionPlayer) -> Optional[bs.PlayerRecord]:
player: bs.SessionPlayer) -> bs.PlayerRecord | None:
for p_rec in valid_players:
if p_rec[1].player is player:
return p_rec[1]
@ -264,7 +264,7 @@ def show_player_scores(self,
text: babase.Lstr,
h_align: Text.HAlign = Text.HAlign.RIGHT,
extrascale: float = 1.0,
maxwidth: Optional[float] = 120.0) -> None:
maxwidth: float | None = 120.0) -> None:
Text(text,
color=(0.5, 0.5, 0.6, 0.5),
position=(ts_h_offs + xoffs * scale,
@ -324,7 +324,7 @@ def show_player_scores(self,
topkillcount = max(topkillcount, prec.accum_kill_count)
topkilledcount = min(topkilledcount, prec.accum_killed_count)
def _scoretxt(text: Union[str, babase.Lstr],
def _scoretxt(text: str | bs.Lstr,
x_offs: float,
highlight: bool,
delay2: float,

View file

@ -134,21 +134,21 @@ class textonmap:
if setti["leaderboard"]["barsBehindName"]:
self.ss1 = bs.newnode('image', attrs={'scale': (300, 30),
'texture': bs.gettexture(
'bar'),
'uiAtlas2'),
'position': (0, -80),
'attach': 'topRight',
'opacity': 0.5,
'color': (0.7, 0.1, 0)})
self.ss1 = bs.newnode('image', attrs={'scale': (300, 30),
'texture': bs.gettexture(
'bar'),
'uiAtlas2'),
'position': (0, -115),
'attach': 'topRight',
'opacity': 0.5,
'color': (0.6, 0.6, 0.6)})
self.ss1 = bs.newnode('image', attrs={'scale': (300, 30),
'texture': bs.gettexture(
'bar'),
'uiAtlas2'),
'position': (0, -150),
'attach': 'topRight',
'opacity': 0.5,

View file

@ -49,7 +49,7 @@ class FlagKale(bs.Actor):
elif isinstance(m, bs.OutOfBoundsMessage):
self.handlemessage(bs.DieMessage())
else:
super().handlemessage(msg)
super().handlemessage(m)
class Puck(bs.Actor):

View file

@ -63,7 +63,7 @@ class OnTimer(bs.Actor):
endtime = bs.time()
if self._starttime_ms is None:
logging.warning(
print(
'OnScreenTimer.stop() called without first calling start()'
)
else:

View file

@ -15,6 +15,7 @@ import random
from typing import TYPE_CHECKING
import bascenev1
import bascenev1 as bs
from bascenev1lib.actor.flag import Flag, FlagPickedUpMessage
from bascenev1lib.actor.playerspaz import PlayerSpaz
@ -42,7 +43,7 @@ class MFGame(bs.TeamGameActivity[Player, Team]):
@classmethod
def get_available_settings(
cls, sessiontype: Type[bs.Session]) -> List[babase.Setting]:
cls, sessiontype: Type[bs.Session]) -> list[bascenev1.Setting]:
settings = [
bs.IntSetting(
'Max Round Time',

View file

@ -16,7 +16,7 @@ from datetime import datetime, timedelta
import _bascenev1
import setting
from serverdata import serverdata
from tools.ServerUpdate import checkSpammer
from tools.server_update import checkSpammer
from tools.file_handle import OpenJson
from typing import TYPE_CHECKING

View file

@ -556,7 +556,10 @@ def on_begin(self, *args, **kwargs) -> None:
# ba_meta export plugin
class byHeySmoothy(babase.Plugin):
def on_app_running(self):
bui.set_party_icon_always_visible(True)
try:
bui.set_party_icon_always_visible(True)
except:
pass
bs._activity.Activity.on_begin = on_begin
# _babase.chatmessage = new_chatmessage
bascenev1lib.actor.playerspaz.PlayerSpaz = NewPlayerSpaz

View file

@ -4,11 +4,12 @@ import os
from datetime import datetime
import _babase
import bascenev1
import setting
import yaml
from babase._gameactivity import GameActivity
from playersData import pdata
from serverData import serverdata
from playersdata import pdata
from serverdata import serverdata
from stats import mystats
from typing import Type
@ -26,7 +27,7 @@ serverinfo = {}
class BsDataThread(object):
def __init__(self):
global stats
stats["name"] = _babase.app.server._config.party_name
stats["name"] = _babase.app.classic.server._config.party_name
stats["discord"] = "https://discord.gg/ucyaesh"
stats["vapidKey"] = notification_manager.get_vapid_keys()["public_key"]
@ -71,7 +72,7 @@ class BsDataThread(object):
).get_next_game_description().evaluate()
current_game_spec = bs.get_foreground_host_session()._current_game_spec
gametype: Type[GameActivity] = current_game_spec['resolved_type']
gametype: Type[bascenev1.GameActivity] = current_game_spec['resolved_type']
currentMap = gametype.get_settings_display_string(
current_game_spec).evaluate()
@ -97,7 +98,7 @@ class BsDataThread(object):
teams = session.sessionteams
for team in teams:
data[str(team.id)] = {'name': team.name if isinstance(team.name,
str) else team.name.evaluate(),
str) else team.name,
'color': list(team.color),
'score': team.customdata['score'],
'players': []
@ -265,7 +266,7 @@ def disable_kick_vote(account_id, duration):
def get_server_config():
return _babase.app.server._config.__dict__
return _babase.app.classic.server._config.__dict__
def update_server_config(config):
@ -278,7 +279,7 @@ def update_server_config(config):
def do_action(action, value):
if action == "message":
_babase.pushcall(babase.Call(_babase.chatmessage, value),
_babase.pushcall(babase.Call(bs.chatmessage, value),
from_other_thread=True)
elif action == "quit":
_babase.pushcall(babase.Call(_babase.quit), from_other_thread=True)

View file

@ -153,7 +153,7 @@ def _set_ready(self, ready: bool) -> None:
# Handle '_edit' as a special case.
if profilename == '_edit' and ready:
with _babase.Context('ui'):
with babase.ContextRef.empty():
pbrowser.ProfileBrowserWindow(in_main_menu=False)
# Give their input-device UI ownership too
@ -161,7 +161,7 @@ def _set_ready(self, ready: bool) -> None:
_babase.set_ui_input_device(self._sessionplayer.inputdevice)
return
if ready == False:
if not ready:
self._sessionplayer.assigninput(
babase.InputType.LEFT_PRESS,
Call(self.handlemessage, ChangeMessage('team', -1)))

View file

@ -315,7 +315,7 @@ class BearStore:
def __init__(self,
price: int = 1000,
value: str = '',
callback: call = None):
callback: Callable[[], None] = None):
self.price = price
self.value = value

View file

@ -12,7 +12,6 @@ import math
from typing import TYPE_CHECKING
import babase
import bascenev1 as bs
import bascenev1lib
from bascenev1lib.actor.spaz import Spaz

View file

@ -218,9 +218,9 @@ class UpdateThread(threading.Thread):
except:
stats = {}
# now add this batch of kills to our persistant stats
# now add this batch of kills to our persistent stats
for account_id, kill_count in self._account_kills.items():
# add a new entry for any accounts that dont have one
# add a new entry for any accounts that don't have one
if account_id not in stats:
# also lets ask the master-server for their account-display-str.
# (we only do this when first creating the entry to save time,
@ -228,7 +228,7 @@ class UpdateThread(threading.Thread):
# it may change)
stats[account_id] = {'rank': 0,
'name': "deafult name",
'name': "default name",
'scores': 0,
'total_damage': 0,
'kills': 0,

View file

@ -1,12 +1,11 @@
import _babase
import bascenev1
import _bascenev1
def set_speed(x):
try:
activity = bascenev1.get_foreground_host_activity()
activity = _bascenev1.get_foreground_host_activity()
with activity.context:
_babase.set_game_speed(x)
_bascenev1.set_game_speed(x)
except:
print("Error: feature available only in BCS server scripts.")

View file

@ -13,9 +13,9 @@ import shutil
import threading
import time
from dataclasses import dataclass, field
from urllib.parse import urlparse
import http.client
import _babase
import requests
import setting
from typing import TYPE_CHECKING
@ -150,8 +150,14 @@ def send_webhook_message():
headers = {
"Content-Type": "application/json"
}
response = requests.post(
WEBHOOK_URL, data=json.dumps(payload), headers=headers)
try:
url = urlparse(WEBHOOK_URL)
conn = http.client.HTTPSConnection(url.netloc)
conn.request("POST", url.path, body=json.dumps(payload), headers=headers)
response = conn.getresponse()
response_data = response.read()
except:
pass
def schedule_webhook():

View file

@ -103,7 +103,7 @@ def player_joined(pb_id):
sub, {
"notification": {
"title": f'{subscribed_players[pb_id]["name"]} is playing now',
"body": f'Join {_babase.app.server._config.party_name} server {subscribed_players[pb_id]["name"]} is waiting for you ',
"body": f'Join {_babase.app.classic.server._config.party_name} server {subscribed_players[pb_id]["name"]} is waiting for you ',
"icon": "assets/icons/icon-96x96.png",
"vibrate": [100, 50, 100],
"requireInteraction": True,

View file

@ -1,14 +1,14 @@
import _thread
import http.client
import json
import time
import urllib.request
import requests
from playersdata import pdata
from urllib.parse import urlparse
import babase
import bascenev1
from efro.terminal import Clr
from playersdata import pdata
VERSION = 71
@ -38,15 +38,22 @@ def postStatus():
link = 'https://bcsservers.ballistica.workers.dev/ping'
data = {'name': babase.app.classic.server._config.party_name,
'port': str(bascenev1.get_game_port()),
'build': babase.app.build_number,
'build': babase.app.env.engine_build_number,
'bcsversion': VERSION}
_thread.start_new_thread(postRequest, (link, data,))
def postRequest(link, data):
try:
res = requests.post(link,
json=data)
url = urlparse(link)
conn = http.client.HTTPSConnection(url.netloc)
json_payload = json.dumps(data)
headers = {
"Content-Type": "application/json"
}
conn.request("POST", url.path, body=json_payload, headers=headers)
response = conn.getresponse()
response_data = response.read()
except:
pass
@ -54,9 +61,15 @@ def postRequest(link, data):
def checkSpammer(data):
def checkMaster(data):
try:
res = requests.post(
'https://bcsservers.ballistica.workers.dev/checkspammer',
json=data)
url = urlparse('https://bcsservers.ballistica.workers.dev/checkspammer')
conn = http.client.HTTPSConnection(url.netloc)
json_payload = json.dumps(data)
headers = {
"Content-Type": "application/json"
}
conn.request("POST", url.path, body=json_payload, headers=headers)
response = conn.getresponse()
response_data = response.read()
except:
pass
# TODO handle response and kick player based on status
@ -67,15 +80,13 @@ def checkSpammer(data):
def fetchChangelogs():
url = "https://raw.githubusercontent.com/imayushsaini/Bombsquad-Ballistica-Modded-Server/public-server/dist/ba_root/mods/changelogs.json"
if 2 * 2 == 4:
try:
data = urllib.request.urlopen(url)
changelog = json.loads(data.read())
except:
return None
else:
return changelog
try:
data = urllib.request.urlopen(url)
changelog = json.loads(data.read())
except:
return None
else:
return changelog
def checkChangelog():