languard-servers-manager/backend/adapters/arma3/remote_admin.py

"""
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