conjurer/constants.py

"""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 logging
import os
from datetime import datetime
from platform import uname
from sys import platform
from typing import List, Optional, TypedDict

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",
    {
        "ctx": Optional[str],
        "queue": List[str],
        "requester": List[str],
    },
)


# *=========================================== Predefines
MASTER_TIMEOUT = datetime.now()
INITIAL_TIME_WAIT = 500
MUZYKA: Music_Config = {"ctx": None, "queue": [], "requester": []}

LOGFILE = ""
NETRC_FILE = ""
MUSIC_FOLDER = ""
MEMORY_FIVE_SIARA = ""
MEMORY_FIVE_MUZYKA = ""
SETTINGS_FILE = ""
ENCODING = "utf-8"
GRAPHICS_PATH = ""
MUZYKA_MOJEGO_LUDU_HISTORIA = 1500
MUZYKA_MOJEGO_LUDU_SLOWA_KLUCZOWE = 15
MUZYKA_MOJEGO_LUDU_PLAJLISTA = 30

GET_MP3 = "/mp3"
SEND_MP3 = "/update_mp3"
GET_PLAYLIST = "/get_music"
ADD_TO_PRIO_PLAYLIST = "/add_to_priority"
CREATE_PRIO_PLAYLIST = "/create_priority_playlist"

REQUEST_MUSIC = "/request_radio_file"
CLEAR_PRIO = "/clear_pr_pls"
SEND_QUERY = "/query"
TIME_BETWEEN_CALLS = 100000
LAST_SPONTANEOUS_CALL = datetime.now()

# *=========================================== 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 = "/"
    if "microsoft-standard" in uname().release:
        LOGFILE = "/home/mtuszowski/conjurer/discord.log"
        MEMORY_FIVE_SIARA = "/home/mtuszowski/conjurer/pamiec.json"
        SYSTEM_GPT_SETTINGS = "/home/mtuszowski/conjurer/system_gpt_settings.json"
        MEMORY_FIVE_MUZYKA = "/home/mtuszowski/conjurer/pamiec_muzyki.json"
        MUSIC_FOLDER = "/mnt/g/Muzyka/"
        SETTINGS_FILE = "/home/mtuszowski/conjurer/settings.json"
        NETRC_FILE = "/home/mtuszowski/.netrc"
        LOGSTORE = "/home/mtuszowski/conjurer/logs/"
        ACCIDENT_LOG = "/home/mtuszowski/conjurer/accident_log.json"
        ENCODING = "utf-8"
        GRAPHICS_PATH = "/home/mtuszowski/conjurer/Conjurer_graphics/"
        DIR_PATH_SADOX = "/mnt/c/Users/mtusz/OneDrive/Dokumenty/Fansadox"

    else:
        LOGFILE = "./discord.log"
        MEMORY_FIVE_SIARA = "./pamiec.json"
        SYSTEM_GPT_SETTINGS = "./system_gpt_settings.json"
        MEMORY_FIVE_MUZYKA = "./pamiec_muzyki.json"
        MUSIC_FOLDER = "./"
        SETTINGS_FILE = "./settings.json"
        NETRC_FILE = "/srv/conjurer/secrets/.netrc"
        LOGSTORE = "./logs/"
        ACCIDENT_LOG = "./accident_log.json"
        ENCODING = "utf-8"
        GRAPHICS_PATH = "./Conjurer_graphics/"
        DIR_PATH_SADOX = "./Fansadox/"


elif platform == "win32":
    LOGFILE = "discord.log"
    MEMORY_FIVE_SIARA = "pamiec.json"
    SYSTEM_GPT_SETTINGS = "system_gpt_settings.json"
    MEMORY_FIVE_MUZYKA = "pamiec_muzyki.json"
    MUSIC_FOLDER = "G:\\Muzyka\\"
    SETTINGS_FILE = "settings.json"
    NETRC_FILE = "C:\\Users\\mtusz\\.netrc"
    LOGSTORE = "C:\\Users\\mtusz\\OneDrive\\Pulpit\\Conjurer\\logs\\"
    ACCIDENT_LOG = "accident_log.json"
    ENCODING = "utf-8"
    DIR_PATH_SADOX = "C:\\Users\\mtusz\\OneDrive\\Dokumenty\\Fansadox\\"
    SEPARATOR_FILE_PATH = "\\"

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.

# Container-friendly shortcut: point CONJURER_DATA_DIR at a single mounted
# volume and every writable data file/dir is rooted under it. The per-variable
# CONJURER_* overrides below still take precedence, so granular control remains
# possible and the native (Pi) deployment is unaffected when it is unset.
_DATA_DIR = os.getenv("CONJURER_DATA_DIR")
if _DATA_DIR:
    LOGFILE = os.path.join(_DATA_DIR, "discord.log")
    SETTINGS_FILE = os.path.join(_DATA_DIR, "settings.json")
    MEMORY_FIVE_SIARA = os.path.join(_DATA_DIR, "pamiec.json")
    MEMORY_FIVE_MUZYKA = os.path.join(_DATA_DIR, "pamiec_muzyki.json")
    SYSTEM_GPT_SETTINGS = os.path.join(_DATA_DIR, "system_gpt_settings.json")
    ACCIDENT_LOG = os.path.join(_DATA_DIR, "accident_log.json")
    LOGSTORE = os.path.join(_DATA_DIR, "logs") + os.sep
    GRAPHICS_PATH = os.path.join(_DATA_DIR, "Conjurer_graphics") + os.sep
    MUSIC_FOLDER = os.path.join(_DATA_DIR, "music") + os.sep
    DIR_PATH_SADOX = os.path.join(_DATA_DIR, "Fansadox") + os.sep

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)

# Voice-recognition transcript dumps. Defaults next to the log file (so on the
# Pi it lands in /home/pi/Conjurer/transcripts/, in docker under /data).
TRANSCRIPTS_PATH = os.getenv(
    "CONJURER_TRANSCRIPTS_PATH",
    os.path.join(os.path.dirname(LOGFILE) or ".", "transcripts") + os.sep,
)

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")
# Betoniarka (radio-operator service colocated with Liquidsoap). Falls back to
# the musician address so deployments that have not split yet keep working.
RADIO_SERVICE_ADDRESS = os.getenv("CONJURER_RADIO_SERVICE", FILE_SERVICE_ADDRESS)
SKIP_TRACK = os.getenv("CONJURER_SKIP_ENDPOINT", "/skip")
LIBRARIAN_SERVICE_ADDRESS = os.getenv(
    "CONJURER_LIBRARIAN_SERVICE", "http://192.168.1.192:5001"
)
HOST_ADDRESS = os.getenv("CONJURER_DISCORD_HOST", "192.168.1.191")
PORT_ADDRESS = int(os.getenv("CONJURER_DISCORD_PORT", "5000"))

# Shared secret for authenticating internal service-to-service HTTP calls.
API_SHARED_KEY = os.getenv("CONJURER_API_KEY", "")


# *=========================================== Self-healing runtime layout
# A fresh host/volume must never kill the bot at import time. Missing
# directories are created and missing state files are seeded - first from the
# templates shipped alongside this file (repo checkout / docker image), then
# from a safe empty structure. Existing files are never touched, so preserved
# history always wins.

_TEMPLATE_DIR = os.path.dirname(os.path.abspath(__file__))


def _ensure_dir(path: str) -> None:
    if not path:
        return
    try:
        os.makedirs(path, exist_ok=True)
    except OSError as exc:
        logger.warning("Cannot create directory %s: %s", path, exc)


def _seed_file(path: str, template_name: str, empty_content: str) -> None:
    """Create *path* from the repo template (or *empty_content*) if missing."""
    if not path or os.path.exists(path):
        return
    _ensure_dir(os.path.dirname(path) or ".")
    template = os.path.join(_TEMPLATE_DIR, template_name)
    try:
        if os.path.exists(template) and os.path.abspath(template) != os.path.abspath(path):
            import shutil

            shutil.copyfile(template, path)
            logger.warning("Seeded missing %s from repo template", path)
        else:
            with open(path, "w", encoding=ENCODING) as handle:
                handle.write(empty_content)
            logger.warning("Created missing %s as empty state", path)
    except OSError as exc:
        logger.warning("Cannot seed %s: %s", path, exc)


def _ensure_runtime_layout() -> None:
    for directory in (
        os.path.dirname(LOGFILE) or ".",
        LOGSTORE,
        GRAPHICS_PATH,
        MUSIC_FOLDER,
        TRANSCRIPTS_PATH,
    ):
        _ensure_dir(directory)

    # (target path, template shipped next to this file, empty fallback)
    _seed_file(SETTINGS_FILE, "settings.json", "{}")
    _seed_file(SYSTEM_GPT_SETTINGS, "system_gpt_settings.json", "{}")
    _seed_file(MEMORY_FIVE_SIARA, "pamiec.json", "[]")
    _seed_file(MEMORY_FIVE_MUZYKA, "pamiec_muzyki.json", "[]")
    _seed_file(ACCIDENT_LOG, "accident_log.json", "[]")


_ensure_runtime_layout()


# *=========================================== 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:
    if isinstance(WORD_REACTIONS[key], list) and len(WORD_REACTIONS[key]) >= 3:
        WORD_REACTIONS[key][2] = datetime.now()

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

# Voice recognition (AssemblyAI). None = the voice cog reports and disables.
ASSEMBLYAI_API_KEY = _resolve_token("assemblyai", "ASSEMBLYAI_API_KEY")

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
LATEX_MAX_ZIP_MB =  25

OPENAI_MODEL = "gpt-4o-mini"

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