# This Python file uses the following encoding: utf-8 """Helper logic for the Conan Exiles <-> Discord bridge. Mirrors the project layout: the cog lives in ``conanjurer_commands.py`` and the reusable logic lives here. Configuration comes from :mod:`constants` (env-var overridable); optional third-party dependencies (``aiomcrcon``/``asyncssh``) are imported defensively so the main bot can still load the extension when the Conan integration is not installed or not in use. """ from __future__ import annotations import asyncio import logging import os import re from dataclasses import dataclass from typing import AsyncIterator, Awaitable, Callable, Optional, Set try: from aiomcrcon import Client as _Rcon # Source RCON over TCP except ImportError: # pragma: no cover - optional component _Rcon = None try: import asyncssh except ImportError: # pragma: no cover - optional component asyncssh = None logger = logging.getLogger("discord") @dataclass class Event: kind: str # "chat" | "login" | "logout" | "death" | "raw" text: str # ready-to-display text raw: str # original line (for debugging) @dataclass class ConanConfig: """Runtime configuration for the bridge, built from :mod:`constants`.""" rcon_host: str rcon_port: int rcon_password: str log_mode: str # "local" | "sftp" log_path: str sftp_host: str sftp_port: int sftp_user: str sftp_password: str @property def rcon_enabled(self) -> bool: """RCON usable only when host+password are set and the lib is present.""" return bool(self.rcon_host and self.rcon_password and _Rcon is not None) @property def log_enabled(self) -> bool: """Log following usable only when its prerequisites are configured.""" if not self.log_path: return False if self.log_mode == "sftp": return bool(self.sftp_host and asyncssh is not None) return True # --- REGEXY DO DOSTROJENIA NA WŁASNYM LOGU --- # Conan/Pippi/Tot logują różnie — dopasuj do swojego logu. Linie niepasujące są # ignorowane (nie zgadujemy). _PATTERNS: list[tuple[str, re.Pattern]] = [ ("chat", re.compile(r"Chat:\s*(?P.+?):\s*(?P.+)$", re.I)), ("login", re.compile(r"(?P.+?)\s+(joined|connected|logged in)", re.I)), ("logout", re.compile(r"(?P.+?)\s+(left|disconnected|logged out)", re.I)), ("death", re.compile(r"(?P.+?)\s+was killed by\s+(?P.+)$", re.I)), ] class RconClient: """Thin async wrapper around a Source-RCON connection to the Conan server.""" def __init__(self, host: str, port: int, password: str): self._host, self._port, self._pw = host, port, password self._client: Optional["_Rcon"] = None self._lock = asyncio.Lock() async def _ensure(self) -> "_Rcon": if _Rcon is None: raise RuntimeError("aiomcrcon not installed — RCON unavailable") if self._client is None: client = _Rcon(self._host, self._port, self._pw) await client.connect() self._client = client logger.info("RCON connected %s:%s", self._host, self._port) return self._client async def command(self, cmd: str) -> str: """Send a command to the Conan server and return its response. Discord -> Conan channel, e.g. ``command("broadcast Hi!")``. """ 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 exc: # disconnect / server restart logger.warning("RCON error (attempt %s): %s", attempt, exc) 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: # pragma: no cover - best effort pass self._client = None def parse_line(line: str) -> Optional[Event]: line = line.rstrip("\n") if not line.strip(): return None for kind, pat in _PATTERNS: match = pat.search(line) if match: g = match.groupdict() if kind == "chat": return Event(kind, f"💬 **{g['who']}**: {g['msg']}", line) if kind == "login": return Event(kind, f"🟢 **{g['who']}** dołączył do gry", line) if kind == "logout": return Event(kind, f"⚪ **{g['who']}** opuścił grę", line) if kind == "death": return Event(kind, f"💀 **{g['who']}** zginął z ręki {g['by']}", line) return None # nierozpoznane -> ignoruj def parse_players(listplayers_output: str) -> Set[str]: """Extract the set of connected player char-names from RCON ``listplayers``. Conan's table is roughly:: Idx | Char name | Player name | User ID | Platform ID | Platform Name 0 | Conan | SomeUser | 12345 | 765... | Steam The char-name column (index 1) is used. The exact format varies between server builds, so this is best-effort and intentionally tunable. """ players: Set[str] = set() for raw in listplayers_output.splitlines(): line = raw.strip() if not line or "|" not in line: continue cols = [c.strip() for c in line.split("|")] head = cols[0].lower() # skip the header row and any separator rows (e.g. "---|---") if head in ("idx", "") or set(cols[0]) <= set("-"): continue if len(cols) >= 2 and cols[1]: players.add(cols[1]) return players async def watch_players( rcon: RconClient, interval: float, on_join: Callable[[str], Awaitable[None]], ) -> None: """Poll RCON ``listplayers`` and call *on_join* for each new player. The first poll seeds the known-player set without announcing, so restarting the bot does not re-announce everyone already online. RCON failures are logged and retried on the next tick rather than killing the task. """ known: Optional[Set[str]] = None while True: try: response = await rcon.command("listplayers") current = parse_players(response) if known is None: known = current else: for name in current - known: try: await on_join(name) except Exception: # pragma: no cover - handler guard logger.exception("Conan: on_join handler failed") known = current except Exception as exc: logger.warning("Conan: player poll failed: %s", exc) await asyncio.sleep(interval) async def _follow_local(path: str) -> AsyncIterator[str]: """``tail -f`` in pure asyncio, following log rotation.""" while True: try: with open(path, "r", encoding="utf-8", errors="replace") as handle: handle.seek(0, os.SEEK_END) inode = os.fstat(handle.fileno()).st_ino while True: line = handle.readline() if line: yield line continue await asyncio.sleep(0.5) try: if os.stat(path).st_ino != inode: # rotation break except FileNotFoundError: break except FileNotFoundError: logger.warning("Conan log not present yet: %s", path) await asyncio.sleep(3.0) async def _follow_sftp(cfg: ConanConfig) -> AsyncIterator[str]: """Incremental SFTP polling (e.g. Host Havoc): reads only new bytes.""" offset = 0 while True: try: async with asyncssh.connect( cfg.sftp_host, port=cfg.sftp_port, username=cfg.sftp_user, password=cfg.sftp_password, known_hosts=None, ) as conn: async with conn.start_sftp_client() as sftp: while True: try: attrs = await sftp.stat(cfg.log_path) size = attrs.size or 0 if size < offset: # rotation offset = 0 if size > offset: async with sftp.open(cfg.log_path, "r") as remote: await remote.seek(offset) chunk = await remote.read() offset = size for line in chunk.splitlines(): yield line except FileNotFoundError: logger.warning("SFTP: missing log %s", cfg.log_path) await asyncio.sleep(2.0) except Exception as exc: logger.warning("SFTP disconnected: %s — retrying", exc) await asyncio.sleep(5.0) async def watch(cfg: ConanConfig, on_event: Callable[[Event], Awaitable[None]]) -> None: """Follow the Conan log and dispatch recognised lines to *on_event*.""" source = _follow_local(cfg.log_path) if cfg.log_mode != "sftp" else _follow_sftp(cfg) async for line in source: event = parse_line(line) if event is not None: try: await on_event(event) except Exception: # pragma: no cover - handler guard logger.exception("Conan: event handler failed")