140 lines
4.7 KiB
Python
140 lines
4.7 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import json
|
|
import os
|
|
import asyncio
|
|
from typing import List, Dict, Any
|
|
from fastapi.responses import JSONResponse
|
|
import berconpy as rcon
|
|
|
|
# RCON Server Details
|
|
credentials_file = os.path.join(os.path.dirname(__file__), "credentials.json")
|
|
try:
|
|
with open(credentials_file) as f:
|
|
credentials = json.load(f)
|
|
except FileNotFoundError:
|
|
raise Exception(f"Credentials file not found: {credentials_file}")
|
|
except json.JSONDecodeError:
|
|
raise Exception(f"Error decoding credentials file: {credentials_file}")
|
|
|
|
IP_ADDR = credentials["SERVER_ADDRESS"]
|
|
PORT = credentials["SERVER_PORT"]
|
|
PASSWORD = credentials["RCON_PASSWORD"]
|
|
|
|
# Initialize FastAPI app
|
|
app = FastAPI()
|
|
|
|
# Add CORS middleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # Allow all origins
|
|
allow_credentials=True,
|
|
allow_methods=["*"], # Allow all methods
|
|
allow_headers=["*"], # Allow all headers
|
|
)
|
|
|
|
# Initialize RCON client
|
|
client = rcon.AsyncRCONClient()
|
|
|
|
# Message store
|
|
messages = []
|
|
player_connect_messages = []
|
|
|
|
# Event handler for player connect messages
|
|
@client.dispatch.on_message
|
|
async def on_player_connect(player_connect_message: str):
|
|
print('Player connected:', player_connect_message)
|
|
player_connect_messages.append(player_connect_message)
|
|
|
|
@client.dispatch.on_command
|
|
async def server_response_to_command(response: str):
|
|
print('Command response:', response)
|
|
messages.append(response)
|
|
|
|
# Background task to run the RCON client
|
|
async def run_rcon_client():
|
|
try:
|
|
async with client.connect(IP_ADDR, PORT, PASSWORD):
|
|
while True:
|
|
await asyncio.sleep(60) # Sleep for an hour and then check the connection
|
|
except Exception as e:
|
|
print(f"Error in RCON client: {e}")
|
|
finally:
|
|
await client.close()
|
|
|
|
# API endpoint to get messages
|
|
@app.get("/loginmessages", response_class=JSONResponse)
|
|
async def get_messages() -> List[str]:
|
|
return player_connect_messages
|
|
|
|
# API endpoint to get players list
|
|
@app.get("/players")
|
|
async def get_players() -> List[Dict[str, Any]]:
|
|
try:
|
|
players = await client.fetch_players()
|
|
player_list = [{"id": player.id, "name": player.name} for player in players]
|
|
return JSONResponse(content=player_list)
|
|
except Exception as e:
|
|
print(f"Error fetching players: {e}")
|
|
return JSONResponse(status_code=500, content={"message": "Error fetching players"})
|
|
|
|
# API endpoint to kick a player
|
|
@app.post("/kick")
|
|
async def kick_player(player_id: int, reason: str = ""):
|
|
try:
|
|
response = await client.kick(player_id, reason)
|
|
return JSONResponse(content={"message": response})
|
|
except Exception as e:
|
|
print(f"Error kicking player: {e}")
|
|
return JSONResponse(status_code=500, content={"message": "Error kicking player"})
|
|
|
|
# API endpoint to ban a player
|
|
@app.post("/ban")
|
|
async def ban_player(addr: str, duration: int = 0, reason: str = ""):
|
|
try:
|
|
response = await client.ban(addr, duration, reason)
|
|
return JSONResponse(content={"message": response})
|
|
except Exception as e:
|
|
print(f"Error banning player: {e}")
|
|
return JSONResponse(status_code=500, content={"message": "Error banning player"})
|
|
|
|
# API endpoint to send a global message
|
|
@app.post("/message")
|
|
async def send_global_message(message: str):
|
|
try:
|
|
response = await client.send(message)
|
|
return JSONResponse(content={"message": response})
|
|
except Exception as e:
|
|
print(f"Error sending message: {e}")
|
|
return JSONResponse(status_code=500, content={"message": "Error sending message"})
|
|
|
|
# API endpoint to unban a player
|
|
@app.post("/unban")
|
|
async def unban_player(ban_id: int):
|
|
try:
|
|
response = await client.unban(ban_id)
|
|
return JSONResponse(content={"message": response})
|
|
except Exception as e:
|
|
print(f"Error unbanning player: {e}")
|
|
return JSONResponse(status_code=500, content={"message": "Error unbanning player"})
|
|
|
|
# API endpoint to send a private message (whisper) to a player
|
|
@app.post("/whisper")
|
|
async def whisper_player(player_id: int, message: str):
|
|
try:
|
|
response = await client.whisper(player_id, message)
|
|
return JSONResponse(content={"message": response})
|
|
except Exception as e:
|
|
print(f"Error sending whisper: {e}")
|
|
return JSONResponse(status_code=500, content={"message": "Error sending whisper"})
|
|
|
|
# Start the RCON client in the background when the app starts
|
|
@app.on_event("startup")
|
|
async def startup_event():
|
|
asyncio.create_task(run_rcon_client())
|
|
|
|
# Main entry point for running the FastAPI app
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|