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