124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from types import MappingProxyType
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
|
|
class ConfigurationError(ValueError):
|
|
"""Raised when application configuration is missing or invalid."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Config:
|
|
bargheman_token: str
|
|
eitaayar_token: str
|
|
bill_chat_ids: Mapping[str, tuple[str, ...]]
|
|
state_file: Path
|
|
lookahead_days: int = 5
|
|
request_timeout_seconds: float = 15.0
|
|
max_events_per_bill: int = 25
|
|
|
|
@property
|
|
def bill_ids(self) -> tuple[str, ...]:
|
|
return tuple(self.bill_chat_ids)
|
|
|
|
@classmethod
|
|
def from_env(cls) -> Config:
|
|
load_dotenv(override=False)
|
|
|
|
values = {
|
|
"BARGHEMAN_TOKEN": os.getenv("BARGHEMAN_TOKEN", "").strip(),
|
|
"EITAAYAR_TOKEN": os.getenv("EITAAYAR_TOKEN", "").strip(),
|
|
}
|
|
missing = [name for name, value in values.items() if not value]
|
|
|
|
raw_routes = os.getenv("BILL_CHAT_MAP", "").strip()
|
|
if raw_routes:
|
|
bill_chat_ids = _bill_chat_map(raw_routes)
|
|
else:
|
|
chat_id = os.getenv("CHAT_ID", "").strip()
|
|
raw_bill_ids = os.getenv("BILL_IDS", "")
|
|
bill_ids = tuple(
|
|
dict.fromkeys(part.strip() for part in raw_bill_ids.split(",") if part.strip())
|
|
)
|
|
if not chat_id:
|
|
missing.append("CHAT_ID")
|
|
if not bill_ids:
|
|
missing.append("BILL_IDS")
|
|
elif invalid := [bill_id for bill_id in bill_ids if not bill_id.isdecimal()]:
|
|
raise ConfigurationError(
|
|
f"BILL_IDS must contain only decimal IDs: {', '.join(invalid)}"
|
|
)
|
|
bill_chat_ids = {bill_id: (chat_id,) for bill_id in bill_ids}
|
|
|
|
if missing:
|
|
raise ConfigurationError(
|
|
f"Missing required environment variables: {', '.join(missing)}"
|
|
)
|
|
|
|
return cls(
|
|
bargheman_token=values["BARGHEMAN_TOKEN"],
|
|
eitaayar_token=values["EITAAYAR_TOKEN"],
|
|
bill_chat_ids=MappingProxyType(bill_chat_ids),
|
|
state_file=Path(os.getenv("STATE_FILE", "state/outages.json").strip()),
|
|
lookahead_days=_integer_setting("LOOKAHEAD_DAYS", 5, minimum=1, maximum=30),
|
|
request_timeout_seconds=_float_setting(
|
|
"REQUEST_TIMEOUT_SECONDS", 15.0, minimum=1.0, maximum=120.0
|
|
),
|
|
max_events_per_bill=_integer_setting("MAX_EVENTS_PER_BILL", 25, minimum=1, maximum=100),
|
|
)
|
|
|
|
|
|
def _bill_chat_map(raw_routes: str) -> dict[str, tuple[str, ...]]:
|
|
routes: dict[str, list[str]] = {}
|
|
malformed: list[str] = []
|
|
for raw_route in raw_routes.split(","):
|
|
route = raw_route.strip()
|
|
if not route:
|
|
continue
|
|
bill_id, separator, chat_id = route.partition(":")
|
|
bill_id = bill_id.strip()
|
|
chat_id = chat_id.strip()
|
|
if not separator or not bill_id or not chat_id:
|
|
malformed.append(route)
|
|
continue
|
|
if not bill_id.isdecimal():
|
|
raise ConfigurationError(f"BILL_CHAT_MAP bill IDs must be decimal: {bill_id}")
|
|
chat_ids = routes.setdefault(bill_id, [])
|
|
if chat_id not in chat_ids:
|
|
chat_ids.append(chat_id)
|
|
if malformed:
|
|
raise ConfigurationError(
|
|
"BILL_CHAT_MAP entries must use BILL_ID:CHAT_ID format: " + ", ".join(malformed)
|
|
)
|
|
if not routes:
|
|
raise ConfigurationError("BILL_CHAT_MAP must contain at least one route")
|
|
return {bill_id: tuple(chat_ids) for bill_id, chat_ids in routes.items()}
|
|
|
|
|
|
def _integer_setting(name: str, default: int, *, minimum: int, maximum: int) -> int:
|
|
raw_value = os.getenv(name, str(default)).strip()
|
|
try:
|
|
value = int(raw_value)
|
|
except ValueError as error:
|
|
raise ConfigurationError(f"{name} must be an integer") from error
|
|
if not minimum <= value <= maximum:
|
|
raise ConfigurationError(f"{name} must be between {minimum} and {maximum}")
|
|
return value
|
|
|
|
|
|
def _float_setting(name: str, default: float, *, minimum: float, maximum: float) -> float:
|
|
raw_value = os.getenv(name, str(default)).strip()
|
|
try:
|
|
value = float(raw_value)
|
|
except ValueError as error:
|
|
raise ConfigurationError(f"{name} must be a number") from error
|
|
if not minimum <= value <= maximum:
|
|
raise ConfigurationError(f"{name} must be between {minimum:g} and {maximum:g}")
|
|
return value
|