A community based topic aggregation platform built on atproto
bretton.dev
1"""
2Configuration Loader for Kagi News Aggregator.
3
4Loads and validates configuration from YAML files.
5"""
6import os
7import logging
8from pathlib import Path
9from typing import Dict, Any
10import yaml
11from urllib.parse import urlparse
12
13from src.models import AggregatorConfig, FeedConfig
14
15logger = logging.getLogger(__name__)
16
17
18class ConfigError(Exception):
19 """Configuration error."""
20 pass
21
22
23class ConfigLoader:
24 """
25 Loads and validates aggregator configuration.
26
27 Supports:
28 - Loading from YAML file
29 - Environment variable overrides
30 - Validation of required fields
31 - URL validation
32 """
33
34 def __init__(self, config_path: Path):
35 """
36 Initialize config loader.
37
38 Args:
39 config_path: Path to config.yaml file
40 """
41 self.config_path = Path(config_path)
42
43 def load(self) -> AggregatorConfig:
44 """
45 Load and validate configuration.
46
47 Returns:
48 AggregatorConfig object
49
50 Raises:
51 ConfigError: If config is invalid or missing
52 """
53 # Check file exists
54 if not self.config_path.exists():
55 raise ConfigError(f"Configuration file not found: {self.config_path}")
56
57 # Load YAML
58 try:
59 with open(self.config_path, 'r') as f:
60 config_data = yaml.safe_load(f)
61 except yaml.YAMLError as e:
62 raise ConfigError(f"Failed to parse YAML: {e}")
63
64 if not config_data:
65 raise ConfigError("Configuration file is empty")
66
67 # Validate and parse
68 try:
69 return self._parse_config(config_data)
70 except Exception as e:
71 raise ConfigError(f"Invalid configuration: {e}")
72
73 def _parse_config(self, data: Dict[str, Any]) -> AggregatorConfig:
74 """
75 Parse and validate configuration data.
76
77 Args:
78 data: Parsed YAML data
79
80 Returns:
81 AggregatorConfig object
82
83 Raises:
84 ConfigError: If validation fails
85 """
86 # Get coves_api_url (with env override)
87 coves_api_url = os.getenv('COVES_API_URL', data.get('coves_api_url'))
88 if not coves_api_url:
89 raise ConfigError("Missing required field: coves_api_url")
90
91 # Validate URL
92 if not self._is_valid_url(coves_api_url):
93 raise ConfigError(f"Invalid URL for coves_api_url: {coves_api_url}")
94
95 # Get log level (default to info)
96 log_level = data.get('log_level', 'info')
97
98 # Parse feeds
99 feeds_data = data.get('feeds', [])
100 if not feeds_data:
101 raise ConfigError("Configuration must include at least one feed")
102
103 feeds = []
104 for feed_data in feeds_data:
105 feed = self._parse_feed(feed_data)
106 feeds.append(feed)
107
108 logger.info(f"Loaded configuration with {len(feeds)} feeds ({sum(1 for f in feeds if f.enabled)} enabled)")
109
110 return AggregatorConfig(
111 coves_api_url=coves_api_url,
112 feeds=feeds,
113 log_level=log_level
114 )
115
116 def _parse_feed(self, data: Dict[str, Any]) -> FeedConfig:
117 """
118 Parse and validate a single feed configuration.
119
120 Args:
121 data: Feed configuration data
122
123 Returns:
124 FeedConfig object
125
126 Raises:
127 ConfigError: If validation fails
128 """
129 # Required fields
130 required_fields = ['name', 'url', 'community_handle']
131 for field in required_fields:
132 if field not in data:
133 raise ConfigError(f"Missing required field in feed config: {field}")
134
135 name = data['name']
136 url = data['url']
137 community_handle = data['community_handle']
138 enabled = data.get('enabled', True) # Default to True
139
140 # Validate URL
141 if not self._is_valid_url(url):
142 raise ConfigError(f"Invalid URL for feed '{name}': {url}")
143
144 return FeedConfig(
145 name=name,
146 url=url,
147 community_handle=community_handle,
148 enabled=enabled
149 )
150
151 def _is_valid_url(self, url: str) -> bool:
152 """
153 Validate URL format.
154
155 Args:
156 url: URL to validate
157
158 Returns:
159 True if valid, False otherwise
160 """
161 try:
162 result = urlparse(url)
163 return all([result.scheme, result.netloc])
164 except Exception:
165 return False