101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
"""Configuration loading and validation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from monitor.models import (
|
|
AppConfig,
|
|
BrowserConfig,
|
|
MonitorConfig,
|
|
NotificationChannelConfig,
|
|
NotificationsConfig,
|
|
WebsiteConfig,
|
|
)
|
|
|
|
|
|
def _channel_config(data: dict[str, Any] | None) -> NotificationChannelConfig:
|
|
if not data:
|
|
return NotificationChannelConfig()
|
|
return NotificationChannelConfig(enabled=bool(data.get("enabled", False)))
|
|
|
|
|
|
def load_config(path: str | Path) -> AppConfig:
|
|
"""Load application configuration from a YAML file."""
|
|
config_path = Path(path)
|
|
if not config_path.exists():
|
|
raise FileNotFoundError(f"Configuration file not found: {config_path}")
|
|
|
|
with config_path.open("r", encoding="utf-8") as handle:
|
|
raw: dict[str, Any] = yaml.safe_load(handle) or {}
|
|
|
|
browser_raw = raw.get("browser", {})
|
|
monitor_raw = raw.get("monitor", {})
|
|
notifications_raw = raw.get("notifications", {})
|
|
|
|
browser = BrowserConfig(
|
|
type=str(browser_raw.get("type", "chrome")).lower(),
|
|
executable_path=browser_raw.get("executable_path"),
|
|
headless=bool(browser_raw.get("headless", False)),
|
|
profile_directory=str(browser_raw.get("profile_directory", "./profiles/default")),
|
|
)
|
|
|
|
monitor = MonitorConfig(
|
|
round_robin_delay_seconds=float(monitor_raw.get("round_robin_delay_seconds", 5)),
|
|
page_load_timeout_seconds=float(monitor_raw.get("page_load_timeout_seconds", 30)),
|
|
render_wait_seconds=float(monitor_raw.get("render_wait_seconds", 3)),
|
|
snapshot_directory=str(monitor_raw.get("snapshot_directory", "./snapshots")),
|
|
snapshot_retention_count=int(monitor_raw.get("snapshot_retention_count", 50)),
|
|
browser_restart_max_retries=int(monitor_raw.get("browser_restart_max_retries", 3)),
|
|
browser_restart_backoff_seconds=float(
|
|
monitor_raw.get("browser_restart_backoff_seconds", 5)
|
|
),
|
|
)
|
|
|
|
websites: list[WebsiteConfig] = []
|
|
for entry in raw.get("websites", []):
|
|
known_keys = {"id", "name", "url", "enabled"}
|
|
extra = {k: v for k, v in entry.items() if k not in known_keys}
|
|
websites.append(
|
|
WebsiteConfig(
|
|
id=str(entry["id"]),
|
|
name=str(entry.get("name", entry["id"])),
|
|
url=str(entry["url"]),
|
|
enabled=bool(entry.get("enabled", True)),
|
|
extra=extra,
|
|
)
|
|
)
|
|
|
|
notifications = NotificationsConfig(
|
|
sms=_channel_config(notifications_raw.get("sms")),
|
|
email=_channel_config(notifications_raw.get("email")),
|
|
)
|
|
|
|
return AppConfig(
|
|
browser=browser,
|
|
monitor=monitor,
|
|
websites=websites,
|
|
notifications=notifications,
|
|
)
|
|
|
|
|
|
def load_env_file(path: str | Path = ".env") -> None:
|
|
"""Load simple KEY=VALUE pairs from a .env file into os.environ."""
|
|
env_path = Path(path)
|
|
if not env_path.exists():
|
|
return
|
|
|
|
with env_path.open("r", encoding="utf-8") as handle:
|
|
for line in handle:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
key = key.strip()
|
|
value = value.strip().strip('"').strip("'")
|
|
if key and key not in os.environ:
|
|
os.environ[key] = value
|