this repo has no description
1"""
2Configuration loader for Void Bot.
3Loads configuration from config.yaml and environment variables.
4"""
5
6import os
7import yaml
8import logging
9from pathlib import Path
10from typing import Dict, Any, Optional, List
11
12logger = logging.getLogger(__name__)
13
14class ConfigLoader:
15 """Configuration loader that handles YAML config files and environment variables."""
16
17 def __init__(self, config_path: str = "config.yaml"):
18 """
19 Initialize the configuration loader.
20
21 Args:
22 config_path: Path to the YAML configuration file
23 """
24 self.config_path = Path(config_path)
25 self._config = None
26 self._load_config()
27
28 def _load_config(self) -> None:
29 """Load configuration from YAML file."""
30 if not self.config_path.exists():
31 raise FileNotFoundError(
32 f"Configuration file not found: {self.config_path}\n"
33 f"Please copy config.yaml.example to config.yaml and configure it."
34 )
35
36 try:
37 with open(self.config_path, 'r', encoding='utf-8') as f:
38 self._config = yaml.safe_load(f) or {}
39 except yaml.YAMLError as e:
40 raise ValueError(f"Invalid YAML in configuration file: {e}")
41 except Exception as e:
42 raise ValueError(f"Error loading configuration file: {e}")
43
44 def get(self, key: str, default: Any = None) -> Any:
45 """
46 Get a configuration value using dot notation.
47
48 Args:
49 key: Configuration key in dot notation (e.g., 'letta.api_key')
50 default: Default value if key not found
51
52 Returns:
53 Configuration value or default
54 """
55 keys = key.split('.')
56 value = self._config
57
58 for k in keys:
59 if isinstance(value, dict) and k in value:
60 value = value[k]
61 else:
62 return default
63
64 return value
65
66 def get_with_env(self, key: str, env_var: str, default: Any = None) -> Any:
67 """
68 Get configuration value, preferring environment variable over config file.
69
70 Args:
71 key: Configuration key in dot notation
72 env_var: Environment variable name
73 default: Default value if neither found
74
75 Returns:
76 Value from environment variable, config file, or default
77 """
78 # First try environment variable
79 env_value = os.getenv(env_var)
80 if env_value is not None:
81 return env_value
82
83 # Then try config file
84 config_value = self.get(key)
85 if config_value is not None:
86 return config_value
87
88 return default
89
90 def get_required(self, key: str, env_var: Optional[str] = None) -> Any:
91 """
92 Get a required configuration value.
93
94 Args:
95 key: Configuration key in dot notation
96 env_var: Optional environment variable name to check first
97
98 Returns:
99 Configuration value
100
101 Raises:
102 ValueError: If required value is not found
103 """
104 if env_var:
105 value = self.get_with_env(key, env_var)
106 else:
107 value = self.get(key)
108
109 if value is None:
110 source = f"config key '{key}'"
111 if env_var:
112 source += f" or environment variable '{env_var}'"
113 raise ValueError(f"Required configuration value not found: {source}")
114
115 return value
116
117 def get_section(self, section: str) -> Dict[str, Any]:
118 """
119 Get an entire configuration section.
120
121 Args:
122 section: Section name
123
124 Returns:
125 Dictionary containing the section
126 """
127 return self.get(section, {})
128
129 def setup_logging(self) -> None:
130 """Setup logging based on configuration."""
131 logging_config = self.get_section('logging')
132
133 # Set root logging level
134 level = logging_config.get('level', 'INFO')
135 logging.basicConfig(
136 level=getattr(logging, level),
137 format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
138 )
139
140 # Set specific logger levels
141 loggers = logging_config.get('loggers', {})
142 for logger_name, logger_level in loggers.items():
143 logger_obj = logging.getLogger(logger_name)
144 logger_obj.setLevel(getattr(logging, logger_level))
145
146
147# Global configuration instance
148_config_instance = None
149
150def get_config(config_path: str = "config.yaml") -> ConfigLoader:
151 """
152 Get the global configuration instance.
153
154 Args:
155 config_path: Path to configuration file (only used on first call)
156
157 Returns:
158 ConfigLoader instance
159 """
160 global _config_instance
161 if _config_instance is None:
162 _config_instance = ConfigLoader(config_path)
163 return _config_instance
164
165def reload_config() -> None:
166 """Reload the configuration from file."""
167 global _config_instance
168 if _config_instance is not None:
169 _config_instance._load_config()
170
171def get_letta_config() -> Dict[str, Any]:
172 """Get Letta configuration."""
173 config = get_config()
174 return {
175 'api_key': config.get_required('letta.api_key'),
176 'timeout': config.get('letta.timeout', 600),
177 'project_id': config.get_required('letta.project_id'),
178 'agent_id': config.get_required('letta.agent_id'),
179 }
180
181def get_bluesky_config() -> Dict[str, Any]:
182 """Get Bluesky configuration."""
183 config = get_config()
184 return {
185 'username': config.get_required('bluesky.username', 'BSKY_USERNAME'),
186 'password': config.get_required('bluesky.password', 'BSKY_PASSWORD'),
187 'pds_uri': config.get_with_env('bluesky.pds_uri', 'PDS_URI', 'https://bsky.social'),
188 }
189
190def get_bot_config() -> Dict[str, Any]:
191 """Get bot behavior configuration."""
192 config = get_config()
193 return {
194 'fetch_notifications_delay': config.get('bot.fetch_notifications_delay', 30),
195 'max_processed_notifications': config.get('bot.max_processed_notifications', 10000),
196 'max_notification_pages': config.get('bot.max_notification_pages', 20),
197 }
198
199def get_agent_config() -> Dict[str, Any]:
200 """Get agent configuration."""
201 config = get_config()
202 return {
203 'name': config.get('bot.agent.name', 'void'),
204 'model': config.get('bot.agent.model', 'openai/gpt-4o-mini'),
205 'embedding': config.get('bot.agent.embedding', 'openai/text-embedding-3-small'),
206 'description': config.get('bot.agent.description', 'A social media agent trapped in the void.'),
207 'max_steps': config.get('bot.agent.max_steps', 100),
208 'blocks': config.get('bot.agent.blocks', {}),
209 }
210
211def get_threading_config() -> Dict[str, Any]:
212 """Get threading configuration."""
213 config = get_config()
214 return {
215 'parent_height': config.get('threading.parent_height', 40),
216 'depth': config.get('threading.depth', 10),
217 'max_post_characters': config.get('threading.max_post_characters', 300),
218 }
219
220def get_queue_config() -> Dict[str, Any]:
221 """Get queue configuration."""
222 config = get_config()
223 return {
224 'priority_users': config.get('queue.priority_users', ['cameron.pfiffer.org']),
225 'base_dir': config.get('queue.base_dir', 'queue'),
226 'error_dir': config.get('queue.error_dir', 'queue/errors'),
227 'no_reply_dir': config.get('queue.no_reply_dir', 'queue/no_reply'),
228 'processed_file': config.get('queue.processed_file', 'queue/processed_notifications.json'),
229 }