132 lines
4.3 KiB
Python
132 lines
4.3 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)):
|
|
response = await rcon.fetch_players()
|
|
return {"response": response}
|
|
|
|
# Kick Player Endpoint
|
|
@app.post("/kick")
|
|
async def kick_player(action: PlayerAction, rcon: rcon.AsyncRCONClient = Depends(get_rcon_manager)):
|
|
try:
|
|
response = await rcon.kick(f'{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.ban(f'{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(f'{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.whisper(f'{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)
|
|
asyncio.run(get_rcon_manager()) |