conjurer/conanjurer_commands.py

# This Python file uses the following encoding: utf-8
"""Conan Exiles <-> Discord bridge (cog).

Follows the project convention (cog here, helpers in
``conanjurer_functions.py``).  The integration is dormant unless configured in
:mod:`constants`: with no RCON host / channels the background watchers simply
do not start and the GM commands report that RCON is unavailable, so loading
this extension is always safe even on hosts without a Conan server.
"""
import asyncio
import logging

import discord
from discord.ext import commands

from conanjurer_functions import ConanConfig, Event, RconClient, watch, watch_players
from constants import (
    CONAN_CHAT_CHANNEL_ID,
    CONAN_EVENTS_CHANNEL_ID,
    CONAN_GM_ROLE_ID,
    CONAN_JOIN_CHANNEL_ID,
    CONAN_LOG_MODE,
    CONAN_LOG_PATH,
    CONAN_PLAYER_POLL_SECONDS,
    CONAN_RCON_HOST,
    CONAN_RCON_PASSWORD,
    CONAN_RCON_PORT,
    CONAN_SFTP_HOST,
    CONAN_SFTP_PASSWORD,
    CONAN_SFTP_PORT,
    CONAN_SFTP_USER,
)


def is_gm():
    """Allow only members holding the configured Conan GM role."""

    async def predicate(ctx: commands.Context) -> bool:
        if not CONAN_GM_ROLE_ID:
            await ctx.reply(
                "⛔ Rola GM Conana nie jest skonfigurowana.", mention_author=False
            )
            return False
        ok = any(r.id == CONAN_GM_ROLE_ID for r in getattr(ctx.author, "roles", []))
        if not ok:
            await ctx.reply("⛔ Tylko GM.", mention_author=False)
        return ok

    return commands.check(predicate)


class ConanModule(commands.Cog):
    """Bridges a Conan Exiles server with Discord over RCON + log following."""

    def __init__(self, bot, logger_name):
        self.bot = bot
        self.logger = logging.getLogger(logger_name)
        self.cfg = ConanConfig(
            rcon_host=CONAN_RCON_HOST,
            rcon_port=CONAN_RCON_PORT,
            rcon_password=CONAN_RCON_PASSWORD,
            log_mode=CONAN_LOG_MODE,
            log_path=CONAN_LOG_PATH,
            sftp_host=CONAN_SFTP_HOST,
            sftp_port=CONAN_SFTP_PORT,
            sftp_user=CONAN_SFTP_USER,
            sftp_password=CONAN_SFTP_PASSWORD,
        )
        self.rcon = (
            RconClient(CONAN_RCON_HOST, CONAN_RCON_PORT, CONAN_RCON_PASSWORD)
            if self.cfg.rcon_enabled
            else None
        )
        self._log_task = None
        self._player_task = None

    async def cog_load(self):
        # Conan -> Discord chat/event mirroring (only with a log source + target)
        if self.cfg.log_enabled and (CONAN_CHAT_CHANNEL_ID or CONAN_EVENTS_CHANNEL_ID):
            self._log_task = asyncio.create_task(self._run_log_watch())
        else:
            self.logger.info("Conan: log watch disabled (not configured)")

        # Player-join notifications — disabled when the channel is not defined
        if CONAN_JOIN_CHANNEL_ID and self.rcon is not None:
            self._player_task = asyncio.create_task(self._run_player_watch())
        else:
            self.logger.info(
                "Conan: player-join notifications disabled (no channel or RCON)"
            )

    async def cog_unload(self):
        for task in (self._log_task, self._player_task):
            if task is not None:
                task.cancel()
        if self.rcon is not None:
            await self.rcon.close()

    # ---------------------------------------------------------------- watchers
    async def _run_log_watch(self):
        await self.bot.wait_until_ready()
        chat_ch = (
            self.bot.get_channel(CONAN_CHAT_CHANNEL_ID) if CONAN_CHAT_CHANNEL_ID else None
        )
        evt_ch = (
            self.bot.get_channel(CONAN_EVENTS_CHANNEL_ID)
            if CONAN_EVENTS_CHANNEL_ID
            else None
        )

        async def on_event(event: Event):
            target = chat_ch if event.kind == "chat" else evt_ch
            if target is not None:
                await target.send(
                    event.text, allowed_mentions=discord.AllowedMentions.none()
                )

        self.logger.info("Conan: starting log watch (mode=%s)", self.cfg.log_mode)
        await watch(self.cfg, on_event)

    async def _run_player_watch(self):
        """Task 2: announce on a defined channel when a player joins the server."""
        await self.bot.wait_until_ready()
        channel = self.bot.get_channel(CONAN_JOIN_CHANNEL_ID)
        if channel is None:
            self.logger.warning(
                "Conan: join channel %s not found — player notifications off",
                CONAN_JOIN_CHANNEL_ID,
            )
            return

        async def on_join(name: str):
            await channel.send(
                f"🟢 **{name}** wszedł na serwer Conan",
                allowed_mentions=discord.AllowedMentions.none(),
            )

        self.logger.info(
            "Conan: starting player-join watch (channel=%s, every %ss)",
            CONAN_JOIN_CHANNEL_ID,
            CONAN_PLAYER_POLL_SECONDS,
        )
        await watch_players(self.rcon, CONAN_PLAYER_POLL_SECONDS, on_join)

    # ----------------------------------------------------- Discord -> Conan
    async def _require_rcon(self, ctx: commands.Context):
        if self.rcon is None:
            await ctx.reply(
                "⛔ RCON Conana nie jest skonfigurowany/dostępny.",
                mention_author=False,
            )
            return None
        return self.rcon

    @commands.command(name="say")
    @is_gm()
    async def say(self, ctx: commands.Context, *, message: str):
        """Discord -> Conan: ogłoszenie widoczne dla wszystkich graczy w grze."""
        rcon = await self._require_rcon(ctx)
        if rcon is None:
            return
        resp = await rcon.command(f"broadcast {message}")
        await ctx.reply(
            f"✅ Wysłano do gry. (serwer: `{resp.strip() or 'OK'}`)",
            mention_author=False,
        )

    @commands.command(name="players")
    @is_gm()
    async def players(self, ctx: commands.Context):
        """Lista graczy online (RCON listplayers)."""
        rcon = await self._require_rcon(ctx)
        if rcon is None:
            return
        resp = await rcon.command("listplayers")
        await ctx.reply(
            f"```\n{resp.strip() or 'brak danych'}\n```", mention_author=False
        )

    @commands.command(name="kick")
    @is_gm()
    async def kick(self, ctx: commands.Context, *, who: str):
        """Wyrzuć gracza (po nazwie/charname — zależnie od wersji serwera)."""
        rcon = await self._require_rcon(ctx)
        if rcon is None:
            return
        resp = await rcon.command(f"kick {who}")
        await ctx.reply(f"👢 `{resp.strip() or 'OK'}`", mention_author=False)

    @commands.command(name="rcon")
    @is_gm()
    async def raw_rcon(self, ctx: commands.Context, *, cmd: str):
        """Surowa komenda RCON (dla zaawansowanych GM). Używaj ostrożnie."""
        rcon = await self._require_rcon(ctx)
        if rcon is None:
            return
        resp = await rcon.command(cmd)
        await ctx.reply(f"```\n{resp.strip() or 'OK'}\n```", mention_author=False)

    @commands.command(name="ogłoś", aliases=["oglos", "rp"])
    @is_gm()
    async def rp_announce(self, ctx: commands.Context, nadawca: str, *, message: str):
        """Wiadomość RP 'z nadawcą' (np. ogłoszenie w imieniu Króla Khasara).

        Na samym RCON realizujemy to jako sformatowany broadcast.
        """
        rcon = await self._require_rcon(ctx)
        if rcon is None:
            return
        resp = await rcon.command(f"broadcast [{nadawca}]: {message}")
        await ctx.reply(f"📜 Ogłoszono jako **{nadawca}**.", mention_author=False)


async def setup(bot):
    logger = logging.getLogger("discord")
    await bot.add_cog(ConanModule(bot, "discord"))
    logger.info("Loading conanjurer commands module done")