Merge branch 'main' into ci/test-workflows

The minimal echo workflow was only to isolate the failure: it confirmed
that *every* Actions run in this repo ends in startup_failure (0s),
including a trivial echo job. The cause is account/repo-level (Actions
minutes/billing on a private free-plan repo), not the workflow file.
Restoring the real 3-job CI so it runs once Actions is unblocked.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.github/workflows/ci.yml

@@ -0,0 +1,54 @@
+name: CI
+
+on:
+  push:
+    branches: ["main"]
+  pull_request:
+    branches: ["main"]
+
+permissions:
+  contents: read
+
+jobs:
+  compile:
+    # Byte-compile every first-party .py to catch syntax errors. No deps.
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - name: py_compile first-party sources
+        run: |
+          git ls-files '*.py' | grep -vE '^(yt_dlp|spotify_dl)/' | xargs python -m py_compile
+          echo "All first-party sources compile."
+
+  unit:
+    # Pure-logic tests; tested modules guard heavy deps, so only pytest needed.
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - name: Install test deps
+        run: |
+          python -m pip install --upgrade pip
+          pip install pytest
+      - name: Run unit tests
+        run: pytest tests/unit -v
+
+  integration:
+    # Boot the Flask services and assert the X-Conjurer-Api-Key auth contract.
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - name: Install service deps
+        run: |
+          python -m pip install --upgrade pip
+          pip install pytest flask waitress requests
+      - name: Run integration tests
+        run: pytest tests/integration -v

.github/workflows/python-app.yml

@@ -1,39 +0,0 @@
-# This workflow will install Python dependencies, run tests and lint with a single version of Python
-# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
-
-name: Python application
-
-on:
-  push:
-    branches: [ "main" ]
-  pull_request:
-    branches: [ "main" ]
-
-permissions:
-  contents: read
-
-jobs:
-  build:
-
-    runs-on: ubuntu-latest
-
-    steps:
-    - uses: actions/checkout@v3
-    - name: Set up Python 3.10
-      uses: actions/setup-python@v3
-      with:
-        python-version: "3.10"
-    - name: Install dependencies
-      run: |
-        python -m pip install --upgrade pip
-        pip install flake8 pytest
-        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
-    - name: Lint with flake8
-      run: |
-        # stop the build if there are Python syntax errors or undefined names
-        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
-        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
-    - name: Test with pytest
-      run: |
-        pytest

.github/workflows/python-package.yml

@@ -1,40 +0,0 @@
-# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
-# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
-
-name: Python package
-
-on:
-  push:
-    branches: [ "main" ]
-  pull_request:
-    branches: [ "main" ]
-
-jobs:
-  build:
-
-    runs-on: ubuntu-latest
-    strategy:
-      fail-fast: false
-      matrix:
-        python-version: ["3.8", "3.9", "3.10"]
-
-    steps:
-    - uses: actions/checkout@v3
-    - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v3
-      with:
-        python-version: ${{ matrix.python-version }}
-    - name: Install dependencies
-      run: |
-        python -m pip install --upgrade pip
-        python -m pip install flake8 pytest
-        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
-    - name: Lint with flake8
-      run: |
-        # stop the build if there are Python syntax errors or undefined names
-        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
-        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
-    - name: Test with pytest
-      run: |
-        pytest

bot.py

@@ -5,6 +5,7 @@
 """
 Module of a python bot named Conjurer - used to work on BDSM discord servers.
 """
+import asyncio
 import logging
 
 # *=========================================== Standard Library Imports
@@ -70,6 +71,7 @@ async def on_ready():
     await client.load_extension("voice_recognition_commands")
     await client.load_extension("file_search_commands")
     await client.load_extension("latex_commands")
+    await client.load_extension("conanjurer_commands")
     logger.info("Sensors: online")
 
     logger.info(client.cogs)
@@ -82,22 +84,52 @@ async def on_ready():
     logger.info("All systems: operational")
 
 
+# *=========================================== Runtime orchestration
+# The legacy bootstrap used two bare threads (client.run + comm_subroutine) and
+# joined them, which made a clean shutdown impossible.  We now drive everything
+# from a single asyncio loop: the Flask comm layer still runs in its own
+# threads (via asyncio.to_thread) but is steered through a shared stop_event so
+# the bot can stop both halves cooperatively.
+
+
+async def _run_comm_subroutine(stop_event: threading.Event) -> None:
+    """Run the blocking comm subroutine in a worker thread."""
+    await asyncio.to_thread(comm_subroutine, stop_event)
+
+
+async def _run_bot(token: str, shutdown_event: asyncio.Event) -> None:
+    """Start the Discord client and flag shutdown when it returns."""
+    try:
+        await client.start(token, log_handler=None)
+    finally:
+        shutdown_event.set()
+
+
+async def main() -> None:
+    if not TOKEN:
+        logger.error("Discord token missing - set DISCORD_TOKEN or configure netrc")
+        return
+
+    logger.info("Starting discord bot")
+    shutdown_event = asyncio.Event()
+    comm_stop_event = threading.Event()
+
+    comm_task = asyncio.create_task(_run_comm_subroutine(comm_stop_event))
+    bot_task = asyncio.create_task(_run_bot(TOKEN, shutdown_event))
+
+    try:
+        await shutdown_event.wait()
+    except (KeyboardInterrupt, asyncio.CancelledError):
+        logger.info("Shutdown signal received")
+        comm_stop_event.set()
+        await client.close()
+    finally:
+        comm_stop_event.set()
+        if not client.is_closed():
+            await client.close()
+        await asyncio.gather(bot_task, comm_task, return_exceptions=True)
+
+
 # *================================== Run
 if __name__ == "__main__":
-    logger.info("Starting discord bot")
-    threads = []
-    logger.info("Starting discord bot: Creating threads")
-    threads.append(threading.Thread(target=client.run, args=(TOKEN,),kwargs={"log_handler":None}))
-    threads.append(threading.Thread(target=comm_subroutine))
-    logger.info("Starting discord bot: Starting threads")
-    WRK_CNT = 0
-    for worker in threads:
-        WRK_CNT += 1
-        logger.info("Starting discord bot: Starting thread %s", WRK_CNT)
-        worker.start()
-    logger.info("Starting discord bot: Joining threads")
-    WRK_CNT = 0
-    for worker in threads:
-        WRK_CNT += 1
-        logger.info("Starting discord bot: Joining thread %s", WRK_CNT)
-        worker.join()
+    asyncio.run(main())

communication_subroutine.py

@@ -1,17 +1,20 @@
 import json
 import logging
+import os
 import re
 import threading
 import time
 from queue import Empty, Queue
+from typing import Optional
 from urllib import request as urequest
 
-from flask import Flask, jsonify, request
+from flask import Flask, abort, jsonify, request
 from waitress import serve
 
-HOST_ADDRESS = "192.168.1.31"
-PORT_ADDRESS = 5000
-ICECAST_ADDRESS = "http://192.168.1.15:8000"
+HOST_ADDRESS = os.getenv("CONJURER_DISCORD_HOST", "192.168.1.31")
+PORT_ADDRESS = int(os.getenv("CONJURER_DISCORD_PORT", "5000"))
+ICECAST_ADDRESS = os.getenv("CONJURER_ICECAST", "http://192.168.1.15:8000")
+API_KEY = os.getenv("CONJURER_API_KEY")
 OUT_COMM_Q = Queue()
 IN_COMM_Q = Queue()
 SRCHTITLE = re.compile(rb"StreamTitle=\\*(?P<title>[^;]*);").search
@@ -19,6 +22,12 @@ SRCHTITLE = re.compile(rb"StreamTitle=\\*(?P<title>[^;]*);").search
 awaiting_q = []
 incoming_q = Queue()
 app = Flask(__name__)
+
+
+def _authorize_request() -> None:
+    """Reject inbound calls lacking the shared key (no-op if key is unset)."""
+    if API_KEY and request.headers.get("X-Conjurer-Api-Key") != API_KEY:
+        abort(401)
 PREPPED_TRACKS = {
     "requests": "",
     "hit": "",
@@ -53,6 +62,7 @@ class QueryControl:
 
 @app.route("/prepped_tracks", methods=["POST"])
 def log_radio_tracks():
+    _authorize_request()
     app.logger = logging.getLogger("discord")
 
     app.logger.info(request)
@@ -85,6 +95,7 @@ def answer_external_command():
     :return: The function `answer_external_command()` is returning a JSON response with the message
     "SUCCESS".
     """
+    _authorize_request()
     logger = logging.getLogger("discord")
     logger.info(request)
     record = json.loads(request.data)
@@ -129,34 +140,42 @@ def waitress_run():
     serve(app, host=HOST_ADDRESS, port=PORT_ADDRESS)
 
 
-def scan_queue():
+def scan_queue(stop_event: Optional[threading.Event] = None):
     """
     The function `scan_queue` reads data from a queue, logs it, and appends it to another queue.
 
-    :param _logger: The `_logger` parameter is typically an instance of a logging object that is used to
-    record and store log messages. It is commonly used to track the flow of the program, record errors,
-    and provide information for debugging purposes. In this code snippet, the `_logger` object is used
-    to log the
+    A bounded ``get(timeout=1)`` is used instead of a blocking ``get()`` so the
+    worker can observe ``stop_event`` and exit cleanly during shutdown.
+
+    :param stop_event: optional :class:`threading.Event`; when set the loop
+    stops at the next iteration.
     """
     logger = logging.getLogger("discord")
     while True:
-        data = OUT_COMM_Q.get()
+        if stop_event and stop_event.is_set():
+            logger.info("scan_queue: stop requested")
+            break
+        try:
+            data = OUT_COMM_Q.get(timeout=1)
+        except Empty:
+            continue
         logger.info(data)
         awaiting_q.append(data)
 
 
-def scan_incoming():
+def scan_incoming(stop_event: Optional[threading.Event] = None):
     """
     The `scan_incoming` function continuously checks for incoming data, processes it, and logs when data
     is found.
 
-    :param _logger: The `_logger` parameter in the `scan_incoming` function is a logger object that is
-    used to log messages or information during the execution of the function. It is typically used for
-    debugging, monitoring, or tracking the flow of the program. In this case, the `_logger` is being
-    used
+    :param stop_event: optional :class:`threading.Event`; when set the loop
+    stops at the next iteration.
     """
     logger = logging.getLogger("discord")
     while True:
+        if stop_event and stop_event.is_set():
+            logger.info("scan_incoming: stop requested")
+            break
         try:
             answer = incoming_q.get(block=False)
             logger.info("DATA FOUND")
@@ -204,27 +223,45 @@ def id3(url: str) -> dict:
         return tagdata
 
 
-def comm_subroutine():
+def comm_subroutine(stop_event: Optional[threading.Event] = None):
     """
     The `comm_subroutine` function starts multiple threads to run different tasks concurrently.
 
-    :param logger: The `logger` parameter in the `comm_subroutine` function is an instance of a logger
-    object that is used for logging messages at various levels (e.g., debug, info, warning, error). In
-    the provided code snippet, the logger is used to log messages at the "info" level
+    Workers run as daemon threads and honour an optional ``stop_event`` so the
+    bot can shut the communication layer down cleanly instead of blocking
+    forever on ``join()``.
+
+    :param stop_event: optional :class:`threading.Event` shared with the caller
+    to coordinate a cooperative shutdown.
     """
     # logger.setLevel(logging.DEBUG)
     logger = logging.getLogger("discord")
     logger.info("Started comms")
     threads = []
     # threads.append(threading.Thread(target=flask_debug))
-    threads.append(threading.Thread(target=waitress_run))
-    threads.append(threading.Thread(target=scan_queue))
-    threads.append(threading.Thread(target=scan_incoming))
+    threads.append(threading.Thread(target=waitress_run, daemon=True))
+    threads.append(
+        threading.Thread(
+            target=scan_queue, kwargs={"stop_event": stop_event}, daemon=True
+        )
+    )
+    threads.append(
+        threading.Thread(
+            target=scan_incoming, kwargs={"stop_event": stop_event}, daemon=True
+        )
+    )
 
     for worker in threads:
         worker.start()
-    for worker in threads:
-        worker.join()
+
+    try:
+        while any(thread.is_alive() for thread in threads):
+            if stop_event and stop_event.is_set():
+                break
+            time.sleep(0.5)
+    finally:
+        if stop_event:
+            stop_event.set()
 
 
 if __name__ == "__main__":

conanjurer_commands.py

@@ -1,102 +1,217 @@
-# ai command cogs
-import logging
-import discord
-from discord.ext import commands
-from other_functions import discord_friendly_send, discord_friendly_reply
-from __future__ import annotations
+# 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
 
-from conanjurer_functions import watch, Event
-log = logging.getLogger("discord")
+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:
-        role_id = ctx.bot.cfg.gm_role_id
-        ok = any(r.id == role_id for r in getattr(ctx.author, "roles", []))
+        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 FromConan(commands.Cog):
-    def __init__(self, bot: commands.Bot):
+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._task: asyncio.Task | None = None
+        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):
-        self._task = asyncio.create_task(self._run())
+        # 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):
-        if self._task:
-            self._task.cancel()
+        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()
 
-    async def _run(self):
+    # ---------------------------------------------------------------- watchers
+    async def _run_log_watch(self):
         await self.bot.wait_until_ready()
-        cfg = self.bot.cfg
-        chat_ch = self.bot.get_channel(cfg.chan_chat)
-        evt_ch = self.bot.get_channel(cfg.chan_events)
+        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(e: Event):
-            target = chat_ch if e.kind == "chat" else evt_ch
+        async def on_event(event: Event):
+            target = chat_ch if event.kind == "chat" else evt_ch
             if target is not None:
-                # allowed_mentions: nie pinguj nikogo treścią z gry
-                await target.send(e.text, allowed_mentions=discord.AllowedMentions.none())
+                await target.send(
+                    event.text, allowed_mentions=discord.AllowedMentions.none()
+                )
 
-        log.info("Start obserwacji logu (tryb=%s)", cfg.log_mode)
-        await watch(cfg, on_event)
+        self.logger.info("Conan: starting log watch (mode=%s)", self.cfg.log_mode)
+        await watch(self.cfg, on_event)
 
-class ToConan(commands.Cog):
-    def __init__(self, bot: commands.Bot):
-        self.bot = bot
+    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."""
-        resp = await self.bot.rcon.command(f"broadcast {message}")
-        await ctx.reply(f"✅ Wysłano do gry. (serwer: `{resp.strip() or 'OK'}`)",
-                        mention_author=False)
+        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)."""
-        resp = await self.bot.rcon.command("listplayers")
-        await ctx.reply(f"```\n{resp.strip() or 'brak danych'}\n```",
-                        mention_author=False)
+        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)."""
-        resp = await self.bot.rcon.command(f"kick {who}")
+        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."""
-        resp = await self.bot.rcon.command(cmd)
+        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)
 
-    # --- HOOK pod przyszły mod mostu (Chat V2 API) ---
     @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. Gdy dodasz
-        własny mod na Chat V2 API z dedykowaną komendą czatu, podmień tę linię
-        na wywołanie tej komendy (np. 'tot_rp_say <nadawca> <msg>').
+        Na samym RCON realizujemy to jako sformatowany broadcast.
         """
-        resp = await self.bot.rcon.command(f"broadcast [{nadawca}]: {message}")
+        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: commands.Bot):
-    await bot.add_cog(ToConan(bot))
-    await bot.add_cog(FromConan(bot))
-
-
+async def setup(bot):
+    logger = logging.getLogger("discord")
+    await bot.add_cog(ConanModule(bot, "discord"))
+    logger.info("Loading conanjurer commands module done")

conanjurer_functions.py

@@ -1,92 +1,103 @@
+# 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 os
-from dataclasses import dataclass
-from dotenv import load_dotenv
+
 import asyncio
 import logging
+import os
 import re
-from typing import AsyncIterator, Callable, Awaitable
-from aiomcrcon import Client as _Rcon  # Source RCON over TCP
-import asyncssh
+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")
 
-load_dotenv()
 
-log = logging.getLogger("discord")
 @dataclass
 class Event:
-    kind: str        # "chat" | "login" | "logout" | "death" | "raw"
-    text: str        # gotowy do wyświetlenia tekst
-    raw: str         # oryginalna linia (do debugowania)
+    kind: str  # "chat" | "login" | "logout" | "death" | "raw"
+    text: str  # ready-to-display text
+    raw: str   # original line (for debugging)
 
 
-# --- REGEXY DO DOSTROJENIA NA WŁASNYM LOGU ---
-# Poniżej PRZYKŁADOWE wzorce. Conan/Pippi/Tot logują różnie — sprawdź swój log
-# i dopasuj. Jeśli wzorzec nie pasuje, linia trafia jako "raw" tylko do debug.
-_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)),
-]
+@dataclass
+class ConanConfig:
+    """Runtime configuration for the bridge, built from :mod:`constants`."""
 
-@dataclass(frozen=True)
-class Config:
-    # Discord
-    guild_id: int
-    chan_chat: int
-    chan_events: int
-    gm_role_id: int
-    # RCON
     rcon_host: str
     rcon_port: int
     rcon_password: str
-    # Log
-    log_mode: str          # "local" | "sftp"
+    log_mode: str  # "local" | "sftp"
     log_path: str
-    sftp_host: str | None
+    sftp_host: str
     sftp_port: int
-    sftp_user: str | None
-    sftp_password: str | None
+    sftp_user: str
+    sftp_password: str
 
-    @staticmethod
-    def load() -> "Config":
-        mode = os.getenv("LOG_MODE", "local").lower()
-        return Config(
-            guild_id=int(_req("DISCORD_GUILD_ID")),
-            chan_chat=int(_req("CHAN_CHAT")),
-            chan_events=int(_req("CHAN_EVENTS")),
-            gm_role_id=int(_req("GM_ROLE_ID")),
-            rcon_host=_req("RCON_HOST"),
-            rcon_port=int(os.getenv("RCON_PORT", "25575")),
-            rcon_password=_req("RCON_PASSWORD"),
-            log_mode=mode,
-            log_path=_req("LOG_PATH"),
-            sftp_host=os.getenv("SFTP_HOST"),
-            sftp_port=int(os.getenv("SFTP_PORT", "22")),
-            sftp_user=os.getenv("SFTP_USER"),
-            sftp_password=os.getenv("SFTP_PASSWORD"),
-        )
+    @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: _Rcon | None = None
+        self._client: Optional["_Rcon"] = None
         self._lock = asyncio.Lock()
 
-    async def _ensure(self) -> _Rcon:
+    async def _ensure(self) -> "_Rcon":
+        if _Rcon is None:
+            raise RuntimeError("aiomcrcon not installed — RCON unavailable")
         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)
+            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:
-        """Wyślij komendę do serwera Conana. Zwraca odpowiedź serwera.
+        """Send a command to the Conan server and return its response.
 
-        To jest KANAŁ Discord -> Conan. Np. command("broadcast Witajcie!")
-        wyświetli komunikat wszystkim graczom w grze.
+        Discord -> Conan channel, e.g. ``command("broadcast Hi!")``.
         """
         async with self._lock:
             for attempt in (1, 2):
@@ -94,8 +105,8 @@ class RconClient:
                     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)
+                except Exception as exc:  # disconnect / server restart
+                    logger.warning("RCON error (attempt %s): %s", attempt, exc)
                     await self.close()
                     if attempt == 2:
                         raise
@@ -106,20 +117,19 @@ class RconClient:
         if self._client is not None:
             try:
                 await self._client.close()
-            except Exception:
+            except Exception:  # pragma: no cover - best effort
                 pass
             self._client = None
 
 
-
-def parse_line(line: str) -> Event | None:
+def parse_line(line: str) -> Optional[Event]:
     line = line.rstrip("\n")
     if not line.strip():
         return None
     for kind, pat in _PATTERNS:
-        m = pat.search(line)
-        if m:
-            g = m.groupdict()
+        match = pat.search(line)
+        if match:
+            g = match.groupdict()
             if kind == "chat":
                 return Event(kind, f"💬 **{g['who']}**: {g['msg']}", line)
             if kind == "login":
@@ -128,42 +138,98 @@ def parse_line(line: str) -> Event | None:
                 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 (nie zgadujemy)
+    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 w czystym asyncio, podąża też po rotacji pliku."""
-    import os
+    """``tail -f`` in pure asyncio, following log rotation."""
     while True:
         try:
-            with open(path, "r", encoding="utf-8", errors="replace") as f:
-                f.seek(0, os.SEEK_END)
-                inode = os.fstat(f.fileno()).st_ino
+            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 = f.readline()
+                    line = handle.readline()
                     if line:
                         yield line
                         continue
                     await asyncio.sleep(0.5)
-                    # wykryj rotację logu
                     try:
-                        if os.stat(path).st_ino != inode:
+                        if os.stat(path).st_ino != inode:  # rotation
                             break
                     except FileNotFoundError:
                         break
         except FileNotFoundError:
-            log.warning("Log nie istnieje jeszcze: %s", path)
+            logger.warning("Conan log not present yet: %s", path)
             await asyncio.sleep(3.0)
 
 
-async def _follow_sftp(cfg) -> AsyncIterator[str]:
-    """Polling przyrostowy po SFTP (Host Havoc). Czyta tylko nowe bajty."""
+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,
+                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:
@@ -171,36 +237,30 @@ async def _follow_sftp(cfg) -> AsyncIterator[str]:
                         try:
                             attrs = await sftp.stat(cfg.log_path)
                             size = attrs.size or 0
-                            if size < offset:      # rotacja
+                            if size < offset:  # rotation
                                 offset = 0
                             if size > offset:
-                                async with sftp.open(cfg.log_path, "r") as rf:
-                                    await rf.seek(offset)
-                                    chunk = await rf.read()
+                                async with sftp.open(cfg.log_path, "r") as remote:
+                                    await remote.seek(offset)
+                                    chunk = await remote.read()
                                     offset = size
-                                    for ln in chunk.splitlines():
-                                        yield ln
+                                    for line in chunk.splitlines():
+                                        yield line
                         except FileNotFoundError:
-                            log.warning("SFTP: brak logu %s", cfg.log_path)
+                            logger.warning("SFTP: missing log %s", cfg.log_path)
                         await asyncio.sleep(2.0)
-        except Exception as e:
-            log.warning("SFTP rozłączony: %s — ponawiam", e)
+        except Exception as exc:
+            logger.warning("SFTP disconnected: %s — retrying", exc)
             await asyncio.sleep(5.0)
 
 
-async def watch(cfg, on_event: Callable[[Event], Awaitable[None]]) -> None:
-    source = _follow_local(cfg.log_path) if cfg.log_mode == "local" else _follow_sftp(cfg)
+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:
-        evt = parse_line(line)
-        if evt is not None:
+        event = parse_line(line)
+        if event is not None:
             try:
-                await on_event(evt)
-            except Exception:
-                log.exception("Błąd obsługi zdarzenia")
-
-def _req(key: str) -> str:
-    val = os.getenv(key)
-    if not val or val.startswith("wklej") or val.startswith("000000"):
-        raise RuntimeError(f"Brak/placeholder w .env: {key}")
-    return val
-
+                await on_event(event)
+            except Exception:  # pragma: no cover - handler guard
+                logger.exception("Conan: event handler failed")

conftest.py

@@ -0,0 +1,14 @@
+"""Pytest bootstrap: make first-party modules importable from the tests.
+
+The bot modules live at the repository root and the musician service lives in
+``conjurer_musician/``; neither is an installable package, so we put both on
+``sys.path`` here.
+"""
+import os
+import sys
+
+_ROOT = os.path.dirname(os.path.abspath(__file__))
+
+for _path in (_ROOT, os.path.join(_ROOT, "conjurer_musician")):
+    if _path not in sys.path:
+        sys.path.insert(0, _path)

conjurer_librarian/search_bot.py

@@ -107,7 +107,7 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, no,
 
                 for item in result_list:
                     if item["DOI"] in data and not item["exists"]:
-                        print(f"HIT in {no} content {data[0]} line {data[1]} file {data[2]} {item["exists"]}")
+                        print(f"HIT in {no} content {data[0]} line {data[1]} file {data[2]} {item['exists']}")
                         _logger.info(data)
                         _logger.info("HIT")
                         item["exists"] = True

conjurer_librarian/search_bot2.py

@@ -108,7 +108,7 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, no,
                 print(f"C{no}{alive_no}\r", end="")
                 for item in result_list:
                     if item["DOI"] in data[0] and not item["exists"]:
-                        print(f"HIT in {no} content {data[0]} line {data[1]} file {data[2]} {item["exists"]}")
+                        print(f"HIT in {no} content {data[0]} line {data[1]} file {data[2]} {item['exists']}")
                         item["exists"] = True
                         live_results.append(item)
                     done_check = done_check and item["exists"]

conjurer_musician/.gitignore

@@ -0,0 +1,7 @@
+# Runtime-generated radio data — keep out of the repo
+all_playlist.playlist
+hit.playlist
+request.playlist
+priority_queue.playlist
+prio_playlist.json
+*.mp3

conjurer_musician/media_search_functions.py

@@ -1,24 +1,44 @@
 #!/usr/bin/env python3
-import argparse
+"""Share-list search/publish helpers for the musician service.
+
+Paths are environment-overridable and the share database / directory are
+accessed lazily, so importing this module has no side effects (the previous
+version ran ``SHARE_DIR.mkdir()`` and read the JSON DB at import time, which
+crashed on any host without the Pi's ``/var/www`` / ``/var/log`` layout — and
+made the service untestable).
+"""
 import json
 import os
-import sys
 import uuid
 from pathlib import Path
 
-# CONFIGURATION
-JSON_DB        = '/var/log/share_scan.json'
-SHARE_DIR      = Path('/var/www/html/share')
-BASE_URL       = 'https://czernobog.pl/share'
+# CONFIGURATION (env-overridable)
+JSON_DB = os.getenv("CONJURER_SHARE_DB", "/var/log/share_scan.json")
+SHARE_DIR = Path(os.getenv("CONJURER_SHARE_DIR", "/var/www/html/share"))
+BASE_URL = os.getenv("CONJURER_SHARE_BASE_URL", "https://czernobog.pl/share")
+
+_entries_cache = None
+
+
+def _ensure_share_dir():
+    SHARE_DIR.mkdir(parents=True, exist_ok=True)
 
-# Ensure share directory exists
-SHARE_DIR.mkdir(parents=True, exist_ok=True)
 
 def load_db():
-    with open(JSON_DB) as f:
-        return json.load(f)['entries']
+    """Load share entries, returning [] when the DB is missing/corrupt."""
+    try:
+        with open(JSON_DB) as handle:
+            return json.load(handle).get("entries", [])
+    except (FileNotFoundError, json.JSONDecodeError):
+        return []
+
+
+def _entries():
+    global _entries_cache
+    if _entries_cache is None:
+        _entries_cache = load_db()
+    return _entries_cache
 
-ENTRIES = load_db()
 
 def relevancy(path, keywords):
     score = 0
@@ -28,17 +48,20 @@ def relevancy(path, keywords):
             score += low.count(kw.lower())
     return score
 
+
 def find_matches(count, keywords):
     scored = []
-    for e in ENTRIES:
-        score = relevancy(e['path'], keywords)
+    for entry in _entries():
+        score = relevancy(entry["path"], keywords)
         if score > 0:
-            scored.append((score, e['path']))
+            scored.append((score, entry["path"]))
     scored.sort(reverse=True, key=lambda x: x[0])
     result = [p for _, p in scored]
     return result[:count]
 
+
 def publish(paths):
+    _ensure_share_dir()
     urls = []
     for path in paths:
         token = uuid.uuid4().hex
@@ -49,4 +72,3 @@ def publish(paths):
             pass
         urls.append(f"{BASE_URL}/{token}")
     return urls
-

constants.py

@@ -1,13 +1,51 @@
+"""Centralised configuration and runtime constants for Conjurer.
+
+Historically this module performed heavy filesystem and credential reads at
+import time which made the bot brittle on hosts that did not mirror the
+original paths.  This version keeps the original platform defaults (so the
+behaviour on the Raspberry Pi / WSL / Windows deployments is unchanged when no
+environment variables are set) but adds three robustness improvements ported
+from the dockerised experiment:
+
+* every path/endpoint can be overridden via an environment variable,
+* JSON state files are loaded defensively (a missing or corrupt file no longer
+  crashes the whole bot at import time),
+* optional dependencies (openai, spotipy, netrc) and credentials are guarded so
+  the bot can still start when a secondary integration is offline, and
+* a shared ``CONJURER_API_KEY`` plus ``service_headers()`` helper enables
+  authenticated internal HTTP calls.
+
+All path constants intentionally remain plain ``str`` (with their original
+trailing separators) to stay byte-for-byte compatible with the existing string
+concatenation in the command modules.
+"""
+
 import json
-import netrc
+import logging
+import os
 from datetime import datetime
 from platform import uname
 from sys import platform
 from typing import List, Optional, TypedDict
 
-import openai
-import spotipy
-from spotipy.oauth2 import SpotifyClientCredentials
+try:
+    import netrc
+except ImportError:  # pragma: no cover - standard on CPython
+    netrc = None
+
+try:
+    import openai
+except ImportError:  # pragma: no cover - optional at runtime
+    openai = None
+
+try:
+    import spotipy
+    from spotipy.oauth2 import SpotifyClientCredentials
+except ImportError:  # pragma: no cover - optional component
+    spotipy = None
+    SpotifyClientCredentials = None
+
+logger = logging.getLogger("discord")
 
 Music_Config = TypedDict(
     "Music_Config",
@@ -30,16 +68,12 @@ MUSIC_FOLDER = ""
 MEMORY_FIVE_SIARA = ""
 MEMORY_FIVE_MUZYKA = ""
 SETTINGS_FILE = ""
-ENCODING = ""
+ENCODING = "utf-8"
 GRAPHICS_PATH = ""
 MUZYKA_MOJEGO_LUDU_HISTORIA = 1500
 MUZYKA_MOJEGO_LUDU_SLOWA_KLUCZOWE = 15
 MUZYKA_MOJEGO_LUDU_PLAJLISTA = 30
 
-FILE_SERVICE_ADDRESS = "http://192.168.1.15:5000"
-RADIO_HARBOR_ADDRESS = "http://192.168.1.15:54321"
-SKIP_TRACK = "/skip"
-
 GET_MP3 = "/mp3"
 SEND_MP3 = "/update_mp3"
 GET_PLAYLIST = "/get_music"
@@ -48,14 +82,13 @@ CREATE_PRIO_PLAYLIST = "/create_priority_playlist"
 
 REQUEST_MUSIC = "/request_radio_file"
 CLEAR_PRIO = "/clear_pr_pls"
-LIBRARIAN_SERVICE_ADDRESS = "http://192.168.1.192:5001"
 SEND_QUERY = "/query"
 TIME_BETWEEN_CALLS = 100000
 LAST_SPONTANEOUS_CALL = datetime.now()
-HOST_ADDRESS = "192.168.1.191"
-PORT_ADDRESS = 5000
 
-# *=========================================== Platform Specific Predefines
+# *=========================================== Platform Specific Defaults
+# These blocks only establish *default* values.  Every constant is overridable
+# through the matching environment variable further below.
 
 if platform in ("linux", "linux2"):
     SEPARATOR_FILE_PATH = "/"
@@ -101,47 +134,158 @@ elif platform == "win32":
     ENCODING = "utf-8"
     DIR_PATH_SADOX = "C:\\Users\\mtusz\\OneDrive\\Dokumenty\\Fansadox\\"
     SEPARATOR_FILE_PATH = "\\"
-with open(SETTINGS_FILE, "r", encoding=ENCODING) as f_settings_file:
-    DATA = json.load(f_settings_file)
-REMOTE_HOST_NAME = "openai"
-netrc_mod = netrc.netrc(NETRC_FILE)
-authTokens = netrc_mod.authenticators(REMOTE_HOST_NAME)
-openai.api_key = authTokens[2]
-OPENAICLIENT = openai.AsyncOpenAI(api_key=openai.api_key)
-REMOTE_HOST_NAME = "discord"
-authTokens = netrc_mod.authenticators(REMOTE_HOST_NAME)
-TOKEN = authTokens[2]
 
-REMOTE_HOST_NAME = "spotipy"
-authTokens = netrc_mod.authenticators(REMOTE_HOST_NAME)
-SPOTIFY_CTRL = spotipy.Spotify(
-    client_credentials_manager=SpotifyClientCredentials(
-        client_id=authTokens[0],
-        client_secret=authTokens[2],
-    )
+else:
+    # Fallback for development hosts (macOS, BSD, …) that match none of the
+    # production platforms.  Everything is rooted next to this file so the
+    # module can at least be imported and unit-tested off-deployment.
+    _BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+    SEPARATOR_FILE_PATH = os.sep
+    LOGFILE = os.path.join(_BASE_DIR, "discord.log")
+    MEMORY_FIVE_SIARA = os.path.join(_BASE_DIR, "pamiec.json")
+    SYSTEM_GPT_SETTINGS = os.path.join(_BASE_DIR, "system_gpt_settings.json")
+    MEMORY_FIVE_MUZYKA = os.path.join(_BASE_DIR, "pamiec_muzyki.json")
+    MUSIC_FOLDER = os.path.join(_BASE_DIR, "music") + os.sep
+    SETTINGS_FILE = os.path.join(_BASE_DIR, "settings.json")
+    NETRC_FILE = os.path.join(os.path.expanduser("~"), ".netrc")
+    LOGSTORE = os.path.join(_BASE_DIR, "logs") + os.sep
+    ACCIDENT_LOG = os.path.join(_BASE_DIR, "accident_log.json")
+    GRAPHICS_PATH = os.path.join(_BASE_DIR, "Conjurer_graphics") + os.sep
+    DIR_PATH_SADOX = os.path.join(_BASE_DIR, "Fansadox") + os.sep
+
+
+# *=========================================== Environment overrides
+# Values stay as plain strings so existing ``PATH + filename`` concatenation in
+# the command modules keeps working unchanged.
+
+LOGFILE = os.getenv("CONJURER_LOG_FILE", LOGFILE)
+NETRC_FILE = os.getenv("CONJURER_NETRC_FILE", NETRC_FILE)
+SETTINGS_FILE = os.getenv("CONJURER_SETTINGS_FILE", SETTINGS_FILE)
+MEMORY_FIVE_SIARA = os.getenv("CONJURER_MEMORY_FILE", MEMORY_FIVE_SIARA)
+MEMORY_FIVE_MUZYKA = os.getenv("CONJURER_MUSIC_MEMORY_FILE", MEMORY_FIVE_MUZYKA)
+SYSTEM_GPT_SETTINGS = os.getenv("CONJURER_SYSTEM_GPT_SETTINGS", SYSTEM_GPT_SETTINGS)
+GRAPHICS_PATH = os.getenv("CONJURER_GRAPHICS_PATH", GRAPHICS_PATH)
+MUSIC_FOLDER = os.getenv("CONJURER_MUSIC_FOLDER", MUSIC_FOLDER)
+LOGSTORE = os.getenv("CONJURER_LOGSTORE", LOGSTORE)
+ACCIDENT_LOG = os.getenv("CONJURER_ACCIDENT_LOG", ACCIDENT_LOG)
+DIR_PATH_SADOX = os.getenv("CONJURER_SADOX_DIR", DIR_PATH_SADOX)
+ENCODING = os.getenv("CONJURER_ENCODING", ENCODING)
+SEPARATOR_FILE_PATH = os.getenv("CONJURER_PATH_SEPARATOR", SEPARATOR_FILE_PATH)
+
+FILE_SERVICE_ADDRESS = os.getenv("CONJURER_FILE_SERVICE", "http://192.168.1.15:5000")
+RADIO_HARBOR_ADDRESS = os.getenv("CONJURER_RADIO_HARBOR", "http://192.168.1.15:54321")
+SKIP_TRACK = os.getenv("CONJURER_SKIP_ENDPOINT", "/skip")
+LIBRARIAN_SERVICE_ADDRESS = os.getenv(
+    "CONJURER_LIBRARIAN_SERVICE", "http://192.168.1.192:5001"
 )
-REMOTE_HOST_NAME = "youtube"
-authTokens = netrc_mod.authenticators(REMOTE_HOST_NAME)
-YOUTUBE_AUTH = [authTokens[0],authTokens[2]]
+HOST_ADDRESS = os.getenv("CONJURER_DISCORD_HOST", "192.168.1.191")
+PORT_ADDRESS = int(os.getenv("CONJURER_DISCORD_PORT", "5000"))
 
-WORD_REACTIONS = DATA["word_reactions"]
-CYCLIC_WORDS = DATA["cyclic_words"]
+# Shared secret for authenticating internal service-to-service HTTP calls.
+API_SHARED_KEY = os.getenv("CONJURER_API_KEY", "")
+
+
+# *=========================================== Defensive state loading
+def _load_json(path: str, fallback):
+    """Load JSON from *path*, falling back gracefully on missing/corrupt files."""
+    try:
+        with open(path, "r", encoding=ENCODING) as handle:
+            return json.load(handle)
+    except FileNotFoundError:
+        logger.warning("Missing JSON file at %s - using fallback", path)
+        return fallback
+    except json.JSONDecodeError:
+        logger.warning("Corrupt JSON at %s - resetting to fallback", path)
+        return fallback
+
+
+DATA = _load_json(SETTINGS_FILE, {})
+WORD_REACTIONS = DATA.get("word_reactions", {})
+CYCLIC_WORDS = DATA.get("cyclic_words", {})
 for key in WORD_REACTIONS:
-    WORD_REACTIONS[key][2] = datetime.now()
-with open(MEMORY_FIVE_SIARA, "r+", encoding=ENCODING) as temp_memory_file:
-    # First we load existing data into a dict.
-    MESSAGE_TABLE = json.load(temp_memory_file)
+    if isinstance(WORD_REACTIONS[key], list) and len(WORD_REACTIONS[key]) >= 3:
+        WORD_REACTIONS[key][2] = datetime.now()
 
-with open(SYSTEM_GPT_SETTINGS, "r+", encoding=ENCODING) as temp_settings_file:
-    # First we load existing data into a dict.
-    GPT_SETTINGS = json.load(temp_settings_file)
-with open(MEMORY_FIVE_MUZYKA, "r+", encoding=ENCODING) as temp_music_memory_file:
-    # First we load existing data into a dict.
-    MESSAGE_TABLE_MUZYKA = json.load(temp_music_memory_file)
-SPECJALNE_ZIEMNIACZKI = GPT_SETTINGS[1]
+MESSAGE_TABLE = _load_json(MEMORY_FIVE_SIARA, [])
+GPT_SETTINGS = _load_json(SYSTEM_GPT_SETTINGS, {})
+MESSAGE_TABLE_MUZYKA = _load_json(MEMORY_FIVE_MUZYKA, [])
+
+SPECJALNE_ZIEMNIACZKI = GPT_SETTINGS[1] if isinstance(GPT_SETTINGS, list) else {}
 ASSISTANTS = {}
 
 
+# *=========================================== Credentials
+def _load_netrc_credentials(host: str):
+    """Return the netrc authenticators tuple for *host* or ``None``."""
+    if netrc is None:
+        return None
+    try:
+        parsed = netrc.netrc(NETRC_FILE)
+    except FileNotFoundError:
+        logger.warning("netrc file %s not found", NETRC_FILE)
+        return None
+    except netrc.NetrcParseError:
+        logger.warning("netrc file %s is invalid", NETRC_FILE)
+        return None
+    return parsed.authenticators(host)
+
+
+def _resolve_token(host: str, env_var: str) -> Optional[str]:
+    """Prefer an environment variable, then fall back to netrc."""
+    env_value = os.getenv(env_var)
+    if env_value:
+        return env_value
+    creds = _load_netrc_credentials(host)
+    if creds:
+        return creds[2]
+    logger.warning("Token for %s not configured", host)
+    return None
+
+
+OPENAI_API_KEY = _resolve_token("openai", "OPENAI_API_KEY")
+if openai and OPENAI_API_KEY:
+    openai.api_key = OPENAI_API_KEY
+    OPENAICLIENT = openai.AsyncOpenAI(api_key=OPENAI_API_KEY)
+else:
+    OPENAICLIENT = None
+
+TOKEN = _resolve_token("discord", "DISCORD_TOKEN")
+
+if spotipy:
+    _spotify_creds = _load_netrc_credentials("spotipy")
+    if _spotify_creds and SpotifyClientCredentials:
+        SPOTIFY_CTRL = spotipy.Spotify(
+            client_credentials_manager=SpotifyClientCredentials(
+                client_id=_spotify_creds[0],
+                client_secret=_spotify_creds[2],
+            )
+        )
+    else:
+        SPOTIFY_CTRL = None
+else:
+    SPOTIFY_CTRL = None
+
+_youtube_creds = _load_netrc_credentials("youtube")
+if _youtube_creds:
+    YOUTUBE_AUTH = [_youtube_creds[0], _youtube_creds[2]]
+else:
+    YOUTUBE_AUTH = [
+        os.getenv("YOUTUBE_USERNAME", ""),
+        os.getenv("YOUTUBE_PASSWORD", ""),
+    ]
+
+
+def service_headers():
+    """Shared header dict for internal service-to-service HTTP calls.
+
+    Returns an empty dict when no key is configured, keeping calls backward
+    compatible with deployments that do not (yet) enforce authentication.
+    """
+    if API_SHARED_KEY:
+        return {"X-Conjurer-Api-Key": API_SHARED_KEY}
+    return {}
+
+
 LATEX_TEX_ENGINE = "tectonic"
 LATEX_MAX_COMPILE_SECONDS = 45
 LATEX_MAX_ATTACH_MB = 8
@@ -153,3 +297,26 @@ ALLOWED_ROLES = ["Nocna Zmiana", "Jarl", "Thane", "Bartender"]
 GUILD_ID = 664789470779932693
 LATEST_MODEL = "gpt-4o"          # najnowszy/do rozmów (możesz zmienić w jednym miejscu)
 CHEAP_MODEL  = "gpt-4o-mini"     # najtańszy (fallback do 3.5 niżej)
+
+
+# *=========================================== Conan Exiles bridge
+# Every value is optional; the conanjurer cog stays dormant unless configured.
+# Channel/role ids default to 0 ("not defined"); RCON host/log path default to
+# "" ("disabled"). See conanjurer_commands.py / conanjurer_functions.py.
+CONAN_GM_ROLE_ID = int(os.getenv("CONAN_GM_ROLE_ID", "0"))
+CONAN_CHAT_CHANNEL_ID = int(os.getenv("CONAN_CHAT_CHANNEL_ID", "0"))
+CONAN_EVENTS_CHANNEL_ID = int(os.getenv("CONAN_EVENTS_CHANNEL_ID", "0"))
+# Player-join notifications: leave at 0 to keep the feature disabled.
+CONAN_JOIN_CHANNEL_ID = int(os.getenv("CONAN_JOIN_CHANNEL_ID", "0"))
+CONAN_PLAYER_POLL_SECONDS = int(os.getenv("CONAN_PLAYER_POLL_SECONDS", "60"))
+
+CONAN_RCON_HOST = os.getenv("CONAN_RCON_HOST", "")
+CONAN_RCON_PORT = int(os.getenv("CONAN_RCON_PORT", "25575"))
+CONAN_RCON_PASSWORD = os.getenv("CONAN_RCON_PASSWORD", "")
+
+CONAN_LOG_MODE = os.getenv("CONAN_LOG_MODE", "local")  # "local" | "sftp"
+CONAN_LOG_PATH = os.getenv("CONAN_LOG_PATH", "")
+CONAN_SFTP_HOST = os.getenv("CONAN_SFTP_HOST", "")
+CONAN_SFTP_PORT = int(os.getenv("CONAN_SFTP_PORT", "22"))
+CONAN_SFTP_USER = os.getenv("CONAN_SFTP_USER", "")
+CONAN_SFTP_PASSWORD = os.getenv("CONAN_SFTP_PASSWORD", "")

librarian_commands.py

@@ -15,7 +15,9 @@ from discord.ext import commands, tasks
 
 from ai_functions import handle_response
 from communication_subroutine import IN_COMM_Q, OUT_COMM_Q, QueryControl
-from constants import DIR_PATH_SADOX, LIBRARIAN_SERVICE_ADDRESS, SEND_QUERY
+from constants import DIR_PATH_SADOX, LIBRARIAN_SERVICE_ADDRESS, SEND_QUERY, service_headers
+
+SERVICE_HEADERS = service_headers()
 
 
 class DataModule(commands.Cog):
@@ -165,6 +167,7 @@ class DataModule(commands.Cog):
             requests.post,
             f"{LIBRARIAN_SERVICE_ADDRESS}{SEND_QUERY}",
             json=json_query,
+            headers=SERVICE_HEADERS,
             timeout=360,
         )
         await ctx.send(
@@ -260,6 +263,7 @@ class DataModule(commands.Cog):
             requests.post,
             f"{LIBRARIAN_SERVICE_ADDRESS}{SEND_QUERY}",
             json=json_query,
+            headers=SERVICE_HEADERS,
             timeout=360,
         )
         await ctx.send(

music_functions.py

@@ -17,10 +17,13 @@ from constants import (
     SEND_MP3,
     SPOTIFY_CTRL,
     YOUTUBE_AUTH,
+    service_headers,
 )
 from spotify_dl import spotify
 from spotify_dl import youtube as youtube_download
 
+SERVICE_HEADERS = service_headers()
+
 
 
 class MusicFileList(object):
@@ -43,7 +46,11 @@ class MusicFileList(object):
         """
         try:
             self.logger.info("Attempt to connect to file service")
-            response = requests.get(f"{FILE_SERVICE_ADDRESS}{GET_MP3}", timeout=360)
+            response = requests.get(
+                f"{FILE_SERVICE_ADDRESS}{GET_MP3}",
+                headers=SERVICE_HEADERS,
+                timeout=360,
+            )
             self.music_file_list = response.json()["music_file_list"]
             self.file_service_active = True
         except requests.exceptions.RequestException as e:
@@ -98,7 +105,12 @@ class MusicFileList(object):
         """
         self.music_file_list.append(item)
         post_data = {"item": str(item)}
-        requests.post(f"{FILE_SERVICE_ADDRESS}{SEND_MP3}", json=post_data, timeout=360)
+        requests.post(
+            f"{FILE_SERVICE_ADDRESS}{SEND_MP3}",
+            json=post_data,
+            headers=SERVICE_HEADERS,
+            timeout=360,
+        )
 
 
 MUSIC_FILE_LIST = MusicFileList("discord")
@@ -308,6 +320,7 @@ async def search_music(ctx, how_many=0, slowa_kluczowe=None):
         requests.post,
         f"{FILE_SERVICE_ADDRESS}{GET_PLAYLIST}",
         json=jrequest,
+        headers=SERVICE_HEADERS,
         timeout=360,
     )
     return_data = await coroutine

pytest.ini

@@ -0,0 +1,5 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
+python_functions = test_*
+addopts = -ra

radio_commands.py

@@ -8,7 +8,9 @@ import uuid
 
 import asyncio
 from datetime import datetime
-from constants import RADIO_HARBOR_ADDRESS, SKIP_TRACK, FILE_SERVICE_ADDRESS, ADD_TO_PRIO_PLAYLIST, REQUEST_MUSIC, CREATE_PRIO_PLAYLIST, CLEAR_PRIO
+from constants import RADIO_HARBOR_ADDRESS, SKIP_TRACK, FILE_SERVICE_ADDRESS, ADD_TO_PRIO_PLAYLIST, REQUEST_MUSIC, CREATE_PRIO_PLAYLIST, CLEAR_PRIO, service_headers
+
+SERVICE_HEADERS = service_headers()
 
 class RadioModule(commands.Cog):
     def __init__(self, bot, logger_name):
@@ -34,6 +36,7 @@ class RadioModule(commands.Cog):
                 coroutine = asyncio.to_thread(
                     requests.get,
                     f"{RADIO_HARBOR_ADDRESS}{SKIP_TRACK}",
+                    headers=SERVICE_HEADERS,
                     timeout=360,
                 )
                 result = await coroutine
@@ -95,6 +98,7 @@ class RadioModule(commands.Cog):
                 requests.post,
                 f"{FILE_SERVICE_ADDRESS}{ADD_TO_PRIO_PLAYLIST}",
                 json=jrequest,
+                headers=SERVICE_HEADERS,
                 timeout=360,
             )
             result = await coroutine
@@ -130,6 +134,7 @@ class RadioModule(commands.Cog):
                 requests.post,
                 f"{FILE_SERVICE_ADDRESS}{REQUEST_MUSIC}",
                 json=jrequest,
+                headers=SERVICE_HEADERS,
                 timeout=360,
             )
             result = await coroutine
@@ -167,6 +172,7 @@ class RadioModule(commands.Cog):
                 requests.post,
                 f"{FILE_SERVICE_ADDRESS}{CREATE_PRIO_PLAYLIST}",
                 json=jrequest,
+                headers=SERVICE_HEADERS,
                 timeout=360,
             )
             result = await coroutine
@@ -194,6 +200,7 @@ class RadioModule(commands.Cog):
             coroutine = asyncio.to_thread(
                 requests.get,
                 f"{FILE_SERVICE_ADDRESS}{CLEAR_PRIO}",
+                headers=SERVICE_HEADERS,
                 timeout=360,
             )
             result = await coroutine

requirements_bot.txt

@@ -17,4 +17,6 @@ PyMuPDF
 waitress
 assemblyai[extras]
 SpeechRecognition
+aiomcrcon
+asyncssh
 git+https://github.com/imayhaveborkedit/discord-ext-voice-recv

test.py

@@ -1,20 +0,0 @@
-import logging
-from logging import handlers
-
-logger = logging.getLogger("discord")
-logger.setLevel(logging.DEBUG)
-handler = handlers.RotatingFileHandler(
-    filename="test.log",
-    encoding="utf-8",
-    mode="a",
-    maxBytes=6 * 1024 * 1024,
-    backupCount=6,
-)
-formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
-handler.setFormatter(formatter)
-logger.addHandler(handler)
-
-
-logger2 = logging.getLogger("discord")
-for item in logger2.handlers:
-    print(item)

test_ai.py

@@ -1,13 +0,0 @@
-AsyncCursorPage[Message](
-    data=[Message(id='msg_3eWSdgbcU8sbCmJK2momOgQQ', 
-                  assistant_id='asst_06eZiwvYNK3MR34suFP60gvg', 
-                  attachments=[], 
-                  completed_at=None, 
-                  content=[TextContentBlock(text=Text(annotations=[], value='Cześć! Oto coś do przemyślenia: \n\n„Sukces to suma niewielkich wysiłków powtarzanych dzień po dniu.” — Robert Collier\n\nTo przypomina nam, że często to nie wielkie działania, ale konsekwentne, małe kroki prowadzą do osiągnięcia celu. Jak mogę Ci dzisiaj pomóc?'), type='text')], created_at=1731620112, incomplete_at=None, incomplete_details=None, metadata={}, object='thread.message', role='assistant', run_id='run_JGjWQTCEkDcEYpyCJnrkZU8Q', status=None, thread_id='thread_dDEjGbGm6ICfG75u0KKpoVxD'), Message(id='msg_GFcnfCFgAthGm2D3oE5d0ZkQ', assistant_id=None, attachments=[], completed_at=None, content=[TextContentBlock(text=Text(annotations=[], value='Cześć! Powiedz coś mądrego'), type='text')], created_at=1731620110, incomplete_at=None, incomplete_details=None, metadata={}, object='thread.message', role='user', run_id=None, status=None, thread_id='thread_dDEjGbGm6ICfG75u0KKpoVxD')], object='list', first_id='msg_3eWSdgbcU8sbCmJK2momOgQQ', last_id='msg_GFcnfCFgAthGm2D3oE5d0ZkQ', has_more=False)
-
-
-[TextContentBlock(
-    text=Text(annotations=[], 
-              value='Cześć! Oto coś do rozważenia: "Największą przeszkodą w naszym życiu jest brak odwagi do wprowadzenia zmian." Niezależnie od tego, jakie masz cele czy marzenia, odwaga do działania i przystosowania się do nowych sytuacji jest kluczem do osiągnięcia sukcesu. Jakie masz przemyślenia na ten temat?'), 
-              type='text')
-]

test_time.py

@@ -1,8 +0,0 @@
-import time
-
-first_time = time.time_ns()
-
-time.sleep(1)
-time_diff = time.time_ns() - first_time
-print(time_diff)
-# 2000149433

tests/integration/test_comm_auth.py

@@ -0,0 +1,56 @@
+"""Integration: the bot's communication Flask layer enforces the shared key,
+and the key the bot would *send* (constants.service_headers) is accepted.
+"""
+import constants
+import communication_subroutine as cs
+
+
+def _client(key="test-secret"):
+    cs.API_KEY = key
+    return cs.app.test_client()
+
+
+def test_prepped_tracks_rejected_without_key():
+    client = _client()
+    resp = client.post(
+        "/prepped_tracks", data='["all", "x"]', content_type="application/json"
+    )
+    assert resp.status_code == 401
+
+
+def test_prepped_tracks_accepted_with_key():
+    client = _client()
+    resp = client.post(
+        "/prepped_tracks",
+        data='["all", "x"]',
+        headers={"X-Conjurer-Api-Key": "test-secret"},
+        content_type="application/json",
+    )
+    assert resp.status_code == 200
+
+
+def test_conjurer_get_is_open():
+    client = _client()
+    assert client.get("/conjurer").status_code == 200
+
+
+def test_open_when_key_unset():
+    client = _client(key=None)
+    resp = client.post(
+        "/prepped_tracks", data='["all", "x"]', content_type="application/json"
+    )
+    assert resp.status_code == 200
+
+
+def test_bot_service_headers_accepted_by_service(monkeypatch):
+    # End-to-end auth contract: the header constants.service_headers() produces
+    # is exactly what communication_subroutine._authorize_request() expects.
+    monkeypatch.setattr(constants, "API_SHARED_KEY", "shared-xyz")
+    cs.API_KEY = "shared-xyz"
+    resp = cs.app.test_client().post(
+        "/prepped_tracks",
+        data='["all", "x"]',
+        headers=constants.service_headers(),
+        content_type="application/json",
+    )
+    assert resp.status_code == 200

tests/integration/test_musician_auth.py

@@ -0,0 +1,34 @@
+"""Integration: the musician Flask service enforces the shared key on its
+authenticated endpoints while leaving the open ones reachable.
+"""
+import conjurer_musician as m
+
+
+def _client(key="test-secret"):
+    m.API_KEY = key
+    return m.app.test_client()
+
+
+def test_clear_pr_pls_rejected_without_key():
+    client = _client()
+    assert client.get("/clear_pr_pls").status_code == 401
+
+
+def test_clear_pr_pls_accepted_with_key():
+    client = _client()
+    resp = client.get(
+        "/clear_pr_pls", headers={"X-Conjurer-Api-Key": "test-secret"}
+    )
+    assert resp.status_code == 200
+
+
+def test_mp3_list_is_open():
+    client = _client()
+    resp = client.get("/mp3")
+    assert resp.status_code == 200
+    assert "music_file_list" in resp.get_json()
+
+
+def test_open_when_key_unset():
+    client = _client(key=None)
+    assert client.get("/clear_pr_pls").status_code == 200

tests/unit/test_conanjurer_functions.py

@@ -0,0 +1,63 @@
+"""Unit tests for the Conan Exiles bridge helpers (no Discord/RCON needed)."""
+import conanjurer_functions as cf
+
+
+def test_parse_players_basic():
+    out = (
+        "Idx | Char name | Player name | User ID\n"
+        "0 | Conan | SteamGuy | 1\n"
+        "1 | Khasar | OtherGuy | 2\n"
+        "--- | --- | --- | ---"
+    )
+    assert cf.parse_players(out) == {"Conan", "Khasar"}
+
+
+def test_parse_players_empty_inputs():
+    assert cf.parse_players("No players connected.") == set()
+    assert cf.parse_players("") == set()
+
+
+def test_parse_line_login():
+    event = cf.parse_line("SomeGuy joined the server")
+    assert event is not None
+    assert event.kind == "login"
+    assert "SomeGuy" in event.text
+
+
+def test_parse_line_chat():
+    event = cf.parse_line("Chat: Bob: hello there")
+    assert event is not None
+    assert event.kind == "chat"
+    assert "Bob" in event.text and "hello" in event.text
+
+
+def test_parse_line_unrecognised_is_ignored():
+    assert cf.parse_line("random server noise") is None
+
+
+def test_conanconfig_disabled_when_unconfigured():
+    cfg = cf.ConanConfig("", 25575, "", "local", "", "", 22, "", "")
+    assert cfg.rcon_enabled is False
+    assert cfg.log_enabled is False
+
+
+def test_conanconfig_local_log_enabled():
+    cfg = cf.ConanConfig("", 0, "", "local", "/tmp/conan.log", "", 22, "", "")
+    assert cfg.log_enabled is True
+
+
+def test_conanconfig_rcon_enabled_requires_lib(monkeypatch):
+    # Without aiomcrcon installed, rcon stays disabled even when host+pw are set.
+    monkeypatch.setattr(cf, "_Rcon", None)
+    cfg = cf.ConanConfig("1.2.3.4", 25575, "pw", "local", "", "", 22, "", "")
+    assert cfg.rcon_enabled is False
+    # With the lib present, it enables.
+    monkeypatch.setattr(cf, "_Rcon", object)
+    cfg2 = cf.ConanConfig("1.2.3.4", 25575, "pw", "local", "", "", 22, "", "")
+    assert cfg2.rcon_enabled is True
+
+
+def test_conanconfig_sftp_log_requires_asyncssh(monkeypatch):
+    monkeypatch.setattr(cf, "asyncssh", None)
+    cfg = cf.ConanConfig("", 0, "", "sftp", "/log", "sftp.host", 22, "u", "p")
+    assert cfg.log_enabled is False

tests/unit/test_constants.py

@@ -0,0 +1,35 @@
+"""Unit tests for constants helpers (defensive config / auth headers)."""
+import constants
+
+
+def test_service_headers_empty_when_no_key(monkeypatch):
+    monkeypatch.setattr(constants, "API_SHARED_KEY", "")
+    assert constants.service_headers() == {}
+
+
+def test_service_headers_with_key(monkeypatch):
+    monkeypatch.setattr(constants, "API_SHARED_KEY", "s3cr3t")
+    assert constants.service_headers() == {"X-Conjurer-Api-Key": "s3cr3t"}
+
+
+def test_load_json_missing_returns_fallback(tmp_path):
+    missing = tmp_path / "nope.json"
+    assert constants._load_json(str(missing), {"fallback": 1}) == {"fallback": 1}
+
+
+def test_load_json_valid(tmp_path):
+    good = tmp_path / "ok.json"
+    good.write_text('{"a": 2}', encoding="utf-8")
+    assert constants._load_json(str(good), {}) == {"a": 2}
+
+
+def test_load_json_corrupt_returns_fallback(tmp_path):
+    bad = tmp_path / "bad.json"
+    bad.write_text("{ not valid json", encoding="utf-8")
+    assert constants._load_json(str(bad), []) == []
+
+
+def test_conan_defaults_present():
+    # Feature toggles default to "off" so the bridge stays dormant.
+    assert constants.CONAN_JOIN_CHANNEL_ID == 0
+    assert constants.CONAN_RCON_HOST == ""