diff --git a/discordgsm/protocols/__init__.py b/discordgsm/protocols/__init__.py index af04394..2eed280 100644 --- a/discordgsm/protocols/__init__.py +++ b/discordgsm/protocols/__init__.py @@ -8,6 +8,7 @@ from .eco import Eco from .factorio import Factorio from .fivem import FiveM +from .front import Front from .gamespy1 import GameSpy1 from .gamespy2 import GameSpy2 from .gamespy3 import GameSpy3 diff --git a/discordgsm/protocols/front.py b/discordgsm/protocols/front.py new file mode 100644 index 0000000..4f82e77 --- /dev/null +++ b/discordgsm/protocols/front.py @@ -0,0 +1,106 @@ +from dataclasses import dataclass +import json +import time +from typing import TYPE_CHECKING + +import aiohttp +import opengsq +from opengsq.socket_async import SocketAsync + +if __name__ == '__main__': + from protocol import Protocol +else: + from discordgsm.protocols.protocol import Protocol + +if TYPE_CHECKING: + from discordgsm.gamedig import GamedigResult + + +@dataclass +class FrontServer: + server_name: str + district_id: int + server_id: int + type: int + addr: str + port: int + info: dict + online: int + status: int + owner_type: int + + +class Front(Protocol): + pre_query_required = True + name = 'front' + master_servers = None + + async def pre_query(self): + url = 'https://privatelist.playthefront.com/private_list' + + async with aiohttp.ClientSession() as session: + async with session.get(url) as response: + data = await response.read() + json_data = json.loads(data) + server_list = json_data['server_list'] + + # Convert server_list to a list of FrontServer objects + Front.master_servers = {f"{server['addr']}:{server['port']}": FrontServer( + server_name=server['server_name'], + district_id=server['district_id'], + server_id=server['server_id'], + type=server['type'], + addr=server['addr'], + port=server['port'], + info=json.loads(server['info']), + online=server['online'], + status=server['status'], + owner_type=server['owner_type'] + ) for server in server_list} + + return Front.master_servers + + async def query(self): + if Front.master_servers is None: + await self.pre_query() + assert Front.master_servers is not None, "Front.master_servers is still None after pre_query" + + host, port = str(self.kv['host']), int(str(self.kv['port'])) + start = time.time() + source = opengsq.Source(host, port, self.timeout) + info = await source.get_info() + ping = int((time.time() - start) * 1000) + + ip = SocketAsync.gethostbyname(host) + host_address = f'{ip}:{info["GamePort"]}' + + if host_address not in Front.master_servers: + raise Exception('Server not found') + + server = Front.master_servers[host_address] + + result: GamedigResult = { + 'name': server.server_name, + 'map': server.info.get('game_map'), + 'password': info['Visibility'] != 0, + 'numplayers': server.online, + 'numbots': 0, + 'maxplayers': server.info.get('maxplayer'), + 'players': [], + 'bots': [], + 'connect': host_address, + 'ping': ping, + 'raw': server.__dict__ + } + + return result + + +if __name__ == '__main__': + import asyncio + + async def main(): + front = Front({'host': '', 'port': 27015}) + print(await front.query()) + + asyncio.run(main())