conjurer/conanjurer_functions.py

# 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<who>.+?):\s*(?P<msg>.+)$", re.I)),
    ("login", re.compile(r"(?P<who>.+?)\s+(joined|connected|logged in)", re.I)),
    ("logout", re.compile(r"(?P<who>.+?)\s+(left|disconnected|logged out)", re.I)),
    ("death", re.compile(r"(?P<who>.+?)\s+was killed by\s+(?P<by>.+)$", 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")