Netdata.cloud bot for Zulip
anil.recoil.org
1"""Configuration management for the Netdata Zulip bot."""
2
3import os
4from pathlib import Path
5from typing import Optional
6
7import structlog
8from dotenv import load_dotenv
9
10from .models import ZulipConfig, ServerConfig
11
12logger = structlog.get_logger()
13
14
15def load_config() -> tuple[ZulipConfig, ServerConfig]:
16 """Load configuration from environment variables and .env files."""
17
18 # Load .env file if present
19 env_file = Path(".env")
20 if env_file.exists():
21 load_dotenv(env_file)
22 logger.info("Loaded configuration from .env file")
23
24 # Load Zulip configuration (optional from env, main source should be zuliprc)
25 zulip_config = ZulipConfig(
26 site=os.getenv("ZULIP_SITE"),
27 email=os.getenv("ZULIP_EMAIL"),
28 api_key=os.getenv("ZULIP_API_KEY"),
29 stream=os.getenv("ZULIP_STREAM", "netdata-alerts"),
30 )
31
32 # Load server configuration
33 server_config = ServerConfig(
34 host=os.getenv("SERVER_HOST", "0.0.0.0"),
35 port=int(os.getenv("SERVER_PORT", "8080")),
36 challenge_secret=os.getenv("SERVER_CHALLENGE_SECRET"),
37 )
38
39 logger.info(
40 "Configuration loaded",
41 zulip_site=zulip_config.site or "(from zuliprc)",
42 zulip_email=zulip_config.email or "(from zuliprc)",
43 zulip_stream=zulip_config.stream,
44 server_host=server_config.host,
45 server_port=server_config.port,
46 )
47
48 return zulip_config, server_config
49
50
51def load_zuliprc_config(zuliprc_path: Optional[str] = None) -> ZulipConfig:
52 """Load Zulip configuration from a zuliprc file.
53
54 Args:
55 zuliprc_path: Path to zuliprc file. If None, looks for ~/.zuliprc
56
57 Returns:
58 ZulipConfig instance
59 """
60 if zuliprc_path is None:
61 zuliprc_path = Path.home() / ".zuliprc"
62 else:
63 zuliprc_path = Path(zuliprc_path)
64
65 if not zuliprc_path.exists():
66 raise FileNotFoundError(f"Zuliprc file not found: {zuliprc_path}")
67
68 config = {}
69 with open(zuliprc_path, 'r') as f:
70 for line in f:
71 line = line.strip()
72 if line and not line.startswith('#') and '=' in line:
73 key, value = line.split('=', 1)
74 config[key.strip()] = value.strip()
75
76 # Map zuliprc keys to our config
77 zulip_config = ZulipConfig(
78 site=config.get('site', ''),
79 email=config.get('email', ''),
80 api_key=config.get('key', ''),
81 stream=config.get('stream', 'netdata-alerts'),
82 )
83
84 # Validate required fields from zuliprc
85 if not all([zulip_config.site, zulip_config.email, zulip_config.api_key]):
86 missing = []
87 if not zulip_config.site: missing.append('site')
88 if not zulip_config.email: missing.append('email')
89 if not zulip_config.api_key: missing.append('key')
90 raise ValueError(
91 f"Missing required Zulip configuration in {zuliprc_path}: {', '.join(missing)}"
92 )
93
94 logger.info(
95 "Loaded Zulip configuration from zuliprc",
96 path=str(zuliprc_path),
97 site=zulip_config.site,
98 email=zulip_config.email,
99 stream=zulip_config.stream
100 )
101
102 return zulip_config