147 lines
4.9 KiB
Python
147 lines
4.9 KiB
Python
import asyncio
|
|
import logging
|
|
import uvicorn
|
|
import json
|
|
import os
|
|
from pydantic import BaseModel
|
|
import berconpy as rcon
|
|
from fastapi import FastAPI, HTTPException, Depends
|
|
|
|
# RCON Server Details
|
|
credentials_file = os.path.join(os.path.dirname(__file__), "credentials.json")
|
|
with open(credentials_file) as f:
|
|
credentials = json.load(f)
|
|
|
|
IP_ADDR = credentials["SERVER_ADDRESS"]
|
|
PORT = credentials["SERVER_PORT"]
|
|
PASSWORD = credentials["RCON_PASSWORD"]
|
|
|
|
# Logging Configuration
|
|
log = logging.getLogger("berconpy")
|
|
log.setLevel(logging.DEBUG)
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
|
|
log.addHandler(handler)
|
|
|
|
# RCON Client Initialization
|
|
client = rcon.AsyncRCONClient()
|
|
|
|
# FastAPI Initialization
|
|
app = FastAPI()
|
|
|
|
# Player Model
|
|
class Player(BaseModel):
|
|
name: str
|
|
score: int
|
|
|
|
class PlayerAction(BaseModel):
|
|
playerid: str
|
|
reason: str
|
|
duration: int = None # Optional for ban duration
|
|
|
|
# Dependency to get RCON manager
|
|
async def get_rcon_manager():
|
|
async with client.connect(IP_ADDR, PORT, PASSWORD):
|
|
yield client
|
|
|
|
# FastAPI Endpoint to List Players
|
|
@app.get("/players")
|
|
async def list_players(rcon: rcon.AsyncRCONClient = Depends(get_rcon_manager)):
|
|
try:
|
|
response = await rcon.send_command("players")
|
|
players = []
|
|
for line in response.splitlines():
|
|
parts = line.split()
|
|
if len(parts) >= 2:
|
|
name = parts[0]
|
|
try:
|
|
score = int(parts[1])
|
|
except ValueError:
|
|
score = 0
|
|
player = Player(name=name, score=score)
|
|
players.append(player)
|
|
return players
|
|
except Exception as e:
|
|
log.error(f"Error fetching players: {e}")
|
|
raise HTTPException(status_code=500, detail="Error fetching players")
|
|
|
|
# Kick Player Endpoint
|
|
@app.post("/kick")
|
|
async def kick_player(action: PlayerAction, rcon: rcon.AsyncRCONClient = Depends(get_rcon_manager)):
|
|
try:
|
|
response = await rcon.send_command(f'kick {action.playerid} {action.reason}')
|
|
return {"response": response}
|
|
except Exception as e:
|
|
log.error(f"Error kicking player: {e}")
|
|
raise HTTPException(status_code=500, detail="Error kicking player")
|
|
|
|
# Ban Player Endpoint
|
|
@app.post("/ban")
|
|
async def ban_player(action: PlayerAction, rcon: rcon.AsyncRCONClient = Depends(get_rcon_manager)):
|
|
try:
|
|
duration_str = f"{action.duration} " if action.duration else ""
|
|
response = await rcon.send_command(f'ban {action.playerid} {duration_str}"{action.reason}"')
|
|
return {"response": response}
|
|
except Exception as e:
|
|
log.error(f"Error banning player: {e}")
|
|
raise HTTPException(status_code=500, detail="Error banning player")
|
|
|
|
# Unlock Server Endpoint
|
|
@app.post("/unlock")
|
|
async def unlock_server(rcon: rcon.AsyncRCONClient = Depends(get_rcon_manager)):
|
|
try:
|
|
response = await rcon.send_command('#unlock')
|
|
return {"response": response}
|
|
except Exception as e:
|
|
log.error(f"Error unlocking server: {e}")
|
|
raise HTTPException(status_code=500, detail="Error unlocking server")
|
|
|
|
# Global Message Endpoint
|
|
@app.post("/global_message")
|
|
async def global_message(message: str, rcon: rcon.AsyncRCONClient = Depends(get_rcon_manager)):
|
|
try:
|
|
response = await rcon.send_command(f'Say -1 {message}')
|
|
return {"response": response}
|
|
except Exception as e:
|
|
log.error(f"Error sending global message: {e}")
|
|
raise HTTPException(status_code=500, detail="Error sending global message")
|
|
|
|
# Direct Message Endpoint
|
|
@app.post("/direct_message")
|
|
async def direct_message(playerid: str, message: str, rcon: rcon.AsyncRCONClient = Depends(get_rcon_manager)):
|
|
try:
|
|
response = await rcon.send_command(f'Say {playerid} {message}')
|
|
return {"response": response}
|
|
except Exception as e:
|
|
log.error(f"Error sending direct message: {e}")
|
|
raise HTTPException(status_code=500, detail="Error sending direct message")
|
|
|
|
# Lock Server Endpoint
|
|
@app.post("/lock")
|
|
async def lock_server(rcon: rcon.AsyncRCONClient = Depends(get_rcon_manager)):
|
|
try:
|
|
response = await rcon.send_command('#lock')
|
|
return {"response": response}
|
|
except Exception as e:
|
|
log.error(f"Error locking server: {e}")
|
|
raise HTTPException(status_code=500, detail="Error locking server")
|
|
|
|
# RCON Event Handlers
|
|
@client.dispatch.on_login
|
|
async def on_login():
|
|
print("Connected to RCON server.")
|
|
|
|
@client.dispatch.on_message
|
|
async def on_message(message: str):
|
|
print("Received message from server:", message)
|
|
|
|
@client.dispatch.on_command
|
|
async def server_response_to_command(response: str):
|
|
if not response:
|
|
return print("on_command: <empty>")
|
|
print("on_command:", response)
|
|
|
|
# Main Function to Run FastAPI App
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|