external python a2s service

ext/python-a2s-rcon-api/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.10
+RUN python -m pip install git+https://github.com/Yepoleb/python-a2s git+https://github.com/conqp/rcon git+https://github.com/tiangolo/fastapi "uvicorn[standard]" && mkdir /app
+WORKDIR /app
+COPY ./ ./
+ENV PYTHONUNBUFFERED 1
+EXPOSE 8082
+ENTRYPOINT ["python", "service.py"]

ext/python-a2s-rcon-api/RCONPlayerModel.py

@@ -0,0 +1,34 @@
+from pydantic import BaseModel
+import re
+
+class BotPlayer(Exception):
+	pass
+
+class RCONPlayer:
+	#DefaultPlayer
+	name: str
+	score: int
+	id: int = 0
+	steam = None
+	#RCONPlayer
+	duration: str
+	ip: str
+	loss: int
+	ping: int
+	state: str
+	#Backcomp
+	steam2: str = ""
+
+	def __init__(self, status_line:str):
+		splited = re.split(r"\s+", status_line)
+		len = splited.__len__()
+		self.id = int(splited[1])
+		self.ip = splited[len - 1]
+		self.state = splited[len - 2]
+		if self.state == "BOT":
+			raise BotPlayer()
+		self.loss = int(splited[len - 3])
+		self.ping = int(splited[len - 4])
+		self.duration = splited[len - 5]
+		self.steam2 = splited[len - 6]
+		self.name = " ".join(splited[2:len-6])[1:-1]

ext/python-a2s-rcon-api/docker-compose.yaml

@@ -0,0 +1,5 @@
+services:
+  a2s_backend:
+    build: ./
+    ports:
+    - 8082:8082

ext/python-a2s-rcon-api/service.py

@@ -0,0 +1,120 @@
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse, Response
+import os, argparse
+from a2s import ainfo as VALVE_SERVER_INFO
+from a2s import aplayers as VALVE_SERVER_PLAYERS
+from rcon.source import rcon as VALVE_SERVER_RCON
+from time import time
+from pydantic import BaseModel
+from RCONPlayerModel import *
+import traceback
+
+class A2S_request(BaseModel):
+	server: str = None
+	
+	@property
+	def address(self):
+		return self.server.split(":")[0]
+
+	@property
+	def port(self):
+		return int(self.server.split(":")[1])
+
+	@property
+	def tuple_address(self):
+		return (self.address, self.port)
+
+class RCON_Request(A2S_request):
+	password: str = None
+	command: str = None
+
+	@property
+	def fulled_data(self):
+		return self.server and self.password and self.command
+
+class SourceBackend:
+	app = None
+
+	def __init__(self):
+		self.app = FastAPI()
+		self.setup_routes()
+
+	def run(self, host = "0.0.0.0", port = 45353):
+		import uvicorn
+		uvicorn.run(self.app, host = host, port = port)
+
+	def setup_routes(self):
+		@self.app.get("/api/ping")
+		async def get_ping():
+			return {"pong":int(time())}
+
+		@self.app.post("/api/rcon")
+		async def execute_rcon(request: RCON_Request):
+			if not request.fulled_data:
+				return Response(status_code=409)
+			try:
+				return await VALVE_SERVER_RCON(
+					request.command,
+					host = request.address,
+					port = request.port,
+					passwd = request.password
+				)
+			except:
+				return Response(status_code = 400)
+
+		@self.app.post("/api/a2s/info")
+		async def get_a2s_info(request: A2S_request):
+			if not request.address:
+				return Response(status_code=409)
+			try:
+				return JSONResponse(content=dict(await VALVE_SERVER_INFO(request.tuple_address)))
+			except:
+				traceback.print_exc()
+				return Response(status_code = 400)
+
+		@self.app.post("/api/a2s/players")
+		async def get_a2s_players(request: A2S_request):
+			if not request.address:
+				return Response(status_code=409)
+			try:
+				return JSONResponse(content=dict(await VALVE_SERVER_PLAYERS(request.tuple_address)))
+			except:
+				traceback.print_exc()
+				return Response(status_code = 400)
+
+		@self.app.post("/api/players")
+		async def get_players(request: RCON_Request):
+			a2s_players = await VALVE_SERVER_PLAYERS(request.tuple_address)
+			status_lines:str = await VALVE_SERVER_RCON(
+					"status",
+					host = request.address,
+					port = request.port,
+					passwd = request.password
+				)
+
+			try:
+				start_index = status_lines.index("# userid")
+			except:
+				return Response(status_code = 400)
+
+			skip_table_header = True
+			players:list[RCONPlayer] = []
+			for line in status_lines[start_index:].split("\n"):
+				if skip_table_header or line.__len__() < 1:
+					skip_table_header = False
+					continue
+				try:
+					players.append(RCONPlayer(line).__dict__)
+				except:
+					traceback.print_exc()
+					pass
+
+			for a2s_player in a2s_players:
+				for rcon_player in players:
+					if rcon_player["name"] == a2s_player.name:
+						rcon_player["score"] = a2s_player.score
+						break
+			return JSONResponse(content=players)
+
+if __name__ == "__main__":
+	SourceBackend().run(port = 8082)