281 lines
8.7 KiB
Python
281 lines
8.7 KiB
Python
"""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 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)
|
|
|
|
# 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
|
|
|
|
|
|
_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()
|