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