""" Arma3RemoteAdmin — implements the RemoteAdmin protocol using BERConClient. """ from __future__ import annotations import logging from adapters.arma3.rcon_client import BERConClient from adapters.exceptions import RemoteAdminError logger = logging.getLogger(__name__) class Arma3RemoteAdmin: """ RemoteAdmin protocol implementation for Arma3 BattlEye RCon. Args: server_id: Database server ID. host: RCon host (usually 127.0.0.1). port: RCon port (usually game_port + 3). password: RCon password. """ def __init__( self, server_id: int, host: str, port: int, password: str, ) -> None: self._server_id = server_id self._client = BERConClient(host=host, port=port, password=password) # ── RemoteAdmin protocol ── def connect(self) -> None: """Connect to RCon. Raises RemoteAdminError on failure.""" try: self._client.connect() except ConnectionError as exc: raise RemoteAdminError(str(exc)) from exc def disconnect(self) -> None: self._client.disconnect() def is_connected(self) -> bool: return self._client.is_connected def get_players(self) -> list[dict]: """Fetch current player list.""" try: return self._client.get_players() except Exception as exc: raise RemoteAdminError(f"get_players failed: {exc}") from exc def send_command(self, command: str, timeout: float = 5.0) -> str | None: """Send an arbitrary RCon command.""" try: return self._client.send_command(command) except Exception as exc: raise RemoteAdminError(f"send_command failed: {exc}") from exc def kick_player(self, player_number: int, reason: str = "") -> bool: """Kick a player by their in-game slot number.""" command = f"kick {player_number}" if reason: command += f" {reason}" try: self._client.send_command(command) return True except Exception as exc: logger.warning("[%s] kick_player failed for player %d: %s", self._server_id, player_number, exc) return False def ban_player(self, player_uid: str, duration_minutes: int = 0, reason: str = "") -> bool: """Add a GUID ban. duration_minutes=0 means permanent.""" duration = duration_minutes if duration_minutes > 0 else 0 command = f"addBan {player_uid} {duration} {reason}" try: self._client.send_command(command) return True except Exception as exc: logger.warning("[%s] ban_player failed: %s", self._server_id, exc) return False def say_all(self, message: str) -> bool: """Broadcast a message to all players.""" try: self._client.send_command(f"say -1 {message}") return True except Exception as exc: logger.warning("[%s] say_all failed: %s", self._server_id, exc) return False def shutdown(self) -> bool: """Shutdown the game server via RCon.""" try: self._client.send_command("#shutdown") return True except Exception as exc: logger.warning("[%s] shutdown failed: %s", self._server_id, exc) return False def keepalive(self) -> None: """Send keepalive if idle.""" self._client.keepalive() class Arma3RemoteAdminFactory: """ RemoteAdmin factory for Arma3. Implements the RemoteAdmin protocol (create_client, get_startup_delay, etc.). """ def create_client(self, host: str, port: int, password: str) -> Arma3RemoteAdmin: """Create a new Arma3RemoteAdmin client instance.""" return Arma3RemoteAdmin( server_id=0, # Will be set by caller host=host, port=port, password=password, ) def get_startup_delay(self) -> float: """Seconds to wait after server start before connecting.""" return 30.0 def get_poll_interval(self) -> float: """Seconds between player list polls.""" return 10.0 def get_player_data_schema(self): """Pydantic model for players.game_data JSON.""" return None