conjurer/constants.py

"""Centralised configuration and runtime constants for Conjurer services.

This module used to perform heavy filesystem and credential reads at import
time which made the project brittle on hosts that did not mirror the original
paths.  The current implementation defers that work, reads configuration from
environment variables (with sensible fallbacks), and protects optional
dependencies so the main bot can start even when a secondary service is
offline.
"""

import json
import logging
import os
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple, 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

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": []}

logger = logging.getLogger("discord")


def _env_path(var_name: str, fallback: Path) -> Path:
    value = os.getenv(var_name)
    if value:
        return Path(value).expanduser().resolve()
    return fallback


def _env(var_name: str, fallback: str) -> str:
    return os.getenv(var_name, fallback)


BASE_DIR = Path(os.getenv("CONJURER_BASE_DIR", Path(__file__).resolve().parent))

LOGFILE = _env_path("CONJURER_LOG_FILE", BASE_DIR / "discord.log")
NETRC_FILE = _env_path("CONJURER_NETRC_FILE", Path.home() / ".netrc")
SETTINGS_FILE = _env_path("CONJURER_SETTINGS_FILE", BASE_DIR / "settings.json")
MEMORY_FIVE_SIARA = _env_path("CONJURER_MEMORY_FILE", BASE_DIR / "pamiec.json")
MEMORY_FIVE_MUZYKA = _env_path(
    "CONJURER_MUSIC_MEMORY_FILE", BASE_DIR / "pamiec_muzyki.json"
)
GRAPHICS_PATH = _env_path(
    "CONJURER_GRAPHICS_PATH", BASE_DIR / "Conjurer_graphics"
)
MUSIC_FOLDER = _env_path("CONJURER_MUSIC_FOLDER", BASE_DIR / "music")
ENCODING = _env("CONJURER_ENCODING", "utf-8")

API_SHARED_KEY = os.getenv("CONJURER_API_KEY", "")
MUZYKA_MOJEGO_LUDU_HISTORIA = 1500
MUZYKA_MOJEGO_LUDU_SLOWA_KLUCZOWE = 15
MUZYKA_MOJEGO_LUDU_PLAJLISTA = 30

FILE_SERVICE_ADDRESS = _env("CONJURER_FILE_SERVICE", "http://127.0.0.1:5000")
RADIO_HARBOR_ADDRESS = _env("CONJURER_RADIO_HARBOR", "http://127.0.0.1:54321")
SKIP_TRACK = _env("CONJURER_SKIP_ENDPOINT", "/skip")

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"
LIBRARIAN_SERVICE_ADDRESS = _env(
    "CONJURER_LIBRARIAN_SERVICE", "http://127.0.0.1:5001"
)
SEND_QUERY = _env("CONJURER_LIBRARIAN_QUERY_ENDPOINT", "/query")
TIME_BETWEEN_CALLS = 100000
LAST_SPONTANEOUS_CALL = datetime.now()
HOST_ADDRESS = _env("CONJURER_DISCORD_HOST", "0.0.0.0")
PORT_ADDRESS = int(_env("CONJURER_DISCORD_PORT", "5000"))

# *=========================================== Platform Specific Predefines

SEPARATOR_FILE_PATH = _env("CONJURER_PATH_SEPARATOR", os.sep)

DIR_PATH_SADOX = _env_path(
    "CONJURER_SADOX_DIR", BASE_DIR / "Fansadox"
)

SYSTEM_GPT_SETTINGS = _env_path(
    "CONJURER_SYSTEM_GPT_SETTINGS", BASE_DIR / "system_gpt_settings.json"
)

LOGSTORE = _env_path("CONJURER_LOGSTORE", BASE_DIR / "logs")
ACCIDENT_LOG = _env_path(
    "CONJURER_ACCIDENT_LOG", BASE_DIR / "accident_log.json"
)


def _load_json(path: Path, fallback) -> object:
    try:
        with path.open("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: Dict[str, Tuple[str, str, int, object]] = {}


def _load_netrc_credentials(host: str) -> Optional[Tuple[str, str, str]]:
    if netrc is None:
        return None
    try:
        parsed = netrc.netrc(str(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]:
    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() -> Dict[str, str]:
    """Shared header dict for internal HTTP calls."""
    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)