steward_mirror/steward/config.py

"""Configuration management for Steward.

Supports configuration via:
1. YAML config file (passed via CONFIG_FILE env var)
2. Environment variables (STEWARD__SECTION__KEY=value format)
3. Default schema values

Environment variables take precedence over config file values.
"""

import json
import logging
import os
from pathlib import Path
from typing import Any

from omegaconf import OmegaConf
from pydantic import BaseModel, ConfigDict, Field, field_validator

logger = logging.getLogger(__name__)


def _parse_int_list(value: Any) -> Any:
    if isinstance(value, int):
        return [value]
    if not isinstance(value, str):
        return value

    text = value.strip()
    if not text:
        return []

    if len(text) >= 2 and text[0] == text[-1] and text[0] in {"'", '"'}:
        text = text[1:-1].strip()
        if not text:
            return []

    if text.startswith("["):
        try:
            parsed = json.loads(text)
        except json.JSONDecodeError:
            pass
        else:
            if isinstance(parsed, list):
                return parsed

    try:
        return [int(item.strip().strip("'\"")) for item in text.split(",") if item.strip()]
    except ValueError as exc:
        raise ValueError(
            "expected a comma-separated list of integers or a JSON integer list"
        ) from exc


class TelegramConfig(BaseModel):
    """Telegram bot configuration."""

    bot_token: str = Field(default="", description="Telegram bot token")
    allowed_user_ids: list[int] = Field(default_factory=list, description="Allowed user IDs")
    group_ids: list[int] = Field(default_factory=list, description="Allowed group/channel IDs")

    @field_validator("allowed_user_ids", "group_ids", mode="before")
    @classmethod
    def parse_int_list(cls, value: Any) -> Any:
        return _parse_int_list(value)


class OpenAIConfig(BaseModel):
    """OpenAI/LLM configuration."""

    api_key: str = Field(default="", description="OpenAI API key")
    base_url: str = Field(default="https://api.openai.com/v1", description="API base URL")
    model: str = Field(default="gpt-4o", description="Model to use")
    system_prompt: str = Field(
        default=(
            "You are Steward, a persistent, trustworthy AI-assisted personal operations platform. "
            "You reduce cognitive load by observing, remembering, planning, and proposing actions. "
            "You are conservative, transparent, and policy-aware. "
            "Always explain your reasoning."
        ),
        description="System prompt for the LLM",
    )


class AnalysisConfig(BaseModel):
    """Analysis/proposal worker configuration."""

    target_url: str = Field(default="", description="Target URL for analysis")
    target_api_key: str = Field(default="", description="API key for target")
    cron_hour: int = Field(default=8, description="Hour for cron schedule")
    cron_minute: int = Field(default=0, description="Minute for cron schedule")


class MemoryConfig(BaseModel):
    """Memory/persistence configuration."""

    thread_memory_path: str = Field(default="thread_memory.json", description="Thread memory path")


class ToolsConfig(BaseModel):
    """Tools/MCP configuration."""

    mcp_server_url: str = Field(default="", description="MCP server URL")
    mcp_server_api_key: str = Field(default="", description="MCP server API key")


class MatrixConfig(BaseModel):
    """Matrix appservice configuration."""

    homeserver_url: str = Field(default="", description="Homeserver client-server base URL")
    homeserver_domain: str = Field(default="", description="Homeserver domain (server_name)")
    as_token: str = Field(
        default="", description="Appservice token for authenticating to the homeserver"
    )
    hs_token: str = Field(
        default="", description="Homeserver token for authenticating incoming transactions"
    )
    bot_localpart: str = Field(default="steward", description="Localpart of the bot user")
    appservice_id: str = Field(default="steward", description="Unique appservice ID")
    listen_host: str = Field(
        default="0.0.0.0", description="Host the appservice HTTP server listens on"
    )
    listen_port: int = Field(default=8000, description="Port the appservice HTTP server listens on")
    allowed_room_ids: list[str] = Field(default_factory=list, description="Allowed room IDs")
    allowed_user_ids: list[str] = Field(default_factory=list, description="Allowed user MXIDs")

    @field_validator("allowed_room_ids", "allowed_user_ids", mode="before")
    @classmethod
    def parse_str_list(cls, value: Any) -> Any:
        if isinstance(value, str):
            return [item.strip() for item in value.split(",") if item.strip()]
        return value


class Settings(BaseModel):
    """Application settings with OmegaConf and pydantic integration."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    telegram: TelegramConfig = Field(default_factory=TelegramConfig)
    openai: OpenAIConfig = Field(default_factory=OpenAIConfig)
    analysis: AnalysisConfig = Field(default_factory=AnalysisConfig)
    memory: MemoryConfig = Field(default_factory=MemoryConfig)
    tools: ToolsConfig = Field(default_factory=ToolsConfig)
    matrix: MatrixConfig = Field(default_factory=MatrixConfig)

    # Compatibility properties for existing code
    @property
    def telegram_bot_token(self) -> str:
        """Legacy property for backward compatibility."""
        return self.telegram.bot_token

    @property
    def telegram_allowed_user_ids(self) -> list[int]:
        """Legacy property for backward compatibility."""
        return self.telegram.allowed_user_ids

    @property
    def telegram_group_ids(self) -> list[int]:
        """Legacy property for backward compatibility."""
        return self.telegram.group_ids

    @property
    def openai_api_key(self) -> str:
        """Legacy property for backward compatibility."""
        return self.openai.api_key

    @property
    def openai_base_url(self) -> str:
        """Legacy property for backward compatibility."""
        return self.openai.base_url

    @property
    def openai_model(self) -> str:
        """Legacy property for backward compatibility."""
        return self.openai.model

    @property
    def openai_system_prompt(self) -> str:
        """Legacy property for backward compatibility."""
        return self.openai.system_prompt

    @property
    def analysis_target_url(self) -> str:
        """Legacy property for backward compatibility."""
        return self.analysis.target_url

    @property
    def analysis_target_api_key(self) -> str:
        """Legacy property for backward compatibility."""
        return self.analysis.target_api_key

    @property
    def analysis_cron_hour(self) -> int:
        """Legacy property for backward compatibility."""
        return self.analysis.cron_hour

    @property
    def analysis_cron_minute(self) -> int:
        """Legacy property for backward compatibility."""
        return self.analysis.cron_minute

    @property
    def thread_memory_path(self) -> str:
        """Legacy property for backward compatibility."""
        return self.memory.thread_memory_path

    @property
    def mcp_server_url(self) -> str:
        """Legacy property for backward compatibility."""
        return self.tools.mcp_server_url

    @property
    def mcp_server_api_key(self) -> str:
        """Legacy property for backward compatibility."""
        return self.tools.mcp_server_api_key

    @property
    def matrix_enabled(self) -> bool:
        """Return True if Matrix appservice integration is configured."""
        return bool(self.matrix.homeserver_url and self.matrix.as_token and self.matrix.hs_token)


_settings_instance: Settings | None = None


def _load_config_from_file(config_path: str | Path) -> dict:
    """Load configuration from YAML file."""
    config_path = Path(config_path)
    if not config_path.exists():
        logger.warning("Config file not found: %s", config_path)
        return {}

    logger.info("Loading config from %s", config_path)
    cfg = OmegaConf.load(config_path)
    return OmegaConf.to_container(cfg, resolve=True) or {}


def _load_config_from_env() -> dict:
    """Load configuration from environment variables (STEWARD__SECTION__KEY format)."""
    cfg = {}
    prefix = "STEWARD__"

    for key, value in os.environ.items():
        if not key.startswith(prefix):
            continue

        # Parse STEWARD__SECTION__KEY=value
        parts = key[len(prefix) :].lower().split("__")
        if len(parts) < 2:
            continue

        section = parts[0]
        setting_key = "__".join(parts[1:])

        if section not in cfg:
            cfg[section] = {}

        # Try to parse value as JSON first (for lists, etc.)
        if isinstance(cfg[section], dict):
            try:
                cfg[section][setting_key] = json.loads(value)
            except (json.JSONDecodeError, ValueError):
                cfg[section][setting_key] = value

    return cfg


def get_settings() -> Settings:
    """Get or create the settings singleton.

    Configuration is loaded in order of precedence (highest to lowest):
    1. Environment variables (STEWARD__SECTION__KEY=value)
    2. YAML config file (path via CONFIG_FILE env var)
    3. Default values from schema

    Returns:
        Settings: The application settings.
    """
    global _settings_instance

    if _settings_instance is not None:
        return _settings_instance

    # Start with schema defaults
    schema_path = Path(__file__).parent / "config_schema.yaml"
    base_cfg = OmegaConf.load(schema_path)

    # Merge in config file if specified
    config_file = os.environ.get("CONFIG_FILE")
    if config_file:
        file_cfg = OmegaConf.create(_load_config_from_file(config_file))
        base_cfg = OmegaConf.merge(base_cfg, file_cfg)

    # Merge in environment variables (takes precedence)
    env_cfg = OmegaConf.create(_load_config_from_env())
    if env_cfg:
        base_cfg = OmegaConf.merge(base_cfg, env_cfg)

    # Convert to dict and create Settings instance
    config_dict = OmegaConf.to_container(base_cfg, resolve=True) or {}
    _settings_instance = Settings(**config_dict)

    logger.info("Configuration loaded successfully")
    logger.debug("Active configuration: %s", OmegaConf.to_yaml(base_cfg))

    return _settings_instance


def reload_settings() -> Settings:
    """Reload configuration from source (mainly for testing)."""
    global _settings_instance
    _settings_instance = None
    return get_settings()