mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-18 07:42:09 +00:00
56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
"""Cienka, odporna na rozłączenia otoczka na klienta Source RCON.
|
|
|
|
Conan Exiles wystawia standardowy Source RCON (TCP). Używamy aiomcrcon,
|
|
który implementuje ten protokół. Reconnect na wypadek restartów serwera.
|
|
"""
|
|
from __future__ import annotations
|
|
import asyncio
|
|
import logging
|
|
|
|
from aiomcrcon import Client as _Rcon # Source RCON over TCP
|
|
|
|
log = logging.getLogger("rcon")
|
|
|
|
|
|
class RconClient:
|
|
def __init__(self, host: str, port: int, password: str):
|
|
self._host, self._port, self._pw = host, port, password
|
|
self._client: _Rcon | None = None
|
|
self._lock = asyncio.Lock()
|
|
|
|
async def _ensure(self) -> _Rcon:
|
|
if self._client is None:
|
|
c = _Rcon(self._host, self._port, self._pw)
|
|
await c.connect()
|
|
self._client = c
|
|
log.info("RCON połączony %s:%s", self._host, self._port)
|
|
return self._client
|
|
|
|
async def command(self, cmd: str) -> str:
|
|
"""Wyślij komendę do serwera Conana. Zwraca odpowiedź serwera.
|
|
|
|
To jest KANAŁ Discord -> Conan. Np. command("broadcast Witajcie!")
|
|
wyświetli komunikat wszystkim graczom w grze.
|
|
"""
|
|
async with self._lock:
|
|
for attempt in (1, 2):
|
|
try:
|
|
client = await self._ensure()
|
|
resp, _ = await client.send_cmd(cmd)
|
|
return resp
|
|
except Exception as e: # rozłączenie/restart serwera
|
|
log.warning("RCON błąd (próba %s): %s", attempt, e)
|
|
await self.close()
|
|
if attempt == 2:
|
|
raise
|
|
await asyncio.sleep(1.0)
|
|
return ""
|
|
|
|
async def close(self) -> None:
|
|
if self._client is not None:
|
|
try:
|
|
await self._client.close()
|
|
except Exception:
|
|
pass
|
|
self._client = None
|