65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
def _env(key: str, default: str = "") -> str:
|
|
return (os.environ.get(key) or default).strip()
|
|
|
|
|
|
def _env_int(key: str, default: int) -> int:
|
|
raw = _env(key)
|
|
try:
|
|
return int(raw) if raw else default
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def _env_bool(key: str, default: bool) -> bool:
|
|
raw = _env(key).lower()
|
|
if not raw:
|
|
return default
|
|
return raw in {"1", "true", "yes", "y", "on"}
|
|
|
|
|
|
@dataclass
|
|
class RelayConfig:
|
|
watch_dir: Path
|
|
state_file: Path
|
|
poll_interval: int
|
|
report_globs: list[str]
|
|
alert_re: re.Pattern | None
|
|
log_file: Path
|
|
device_id: str
|
|
login_token: str
|
|
target_chat: int
|
|
hello: bool
|
|
max_lines_per_cycle: int
|
|
|
|
@classmethod
|
|
def load(cls) -> "RelayConfig":
|
|
watch_dir = Path(_env("WATCH_DIR") or "/watch")
|
|
state_dir = Path(_env("STATE_DIR") or "/data")
|
|
state_dir.mkdir(parents=True, exist_ok=True)
|
|
alert_raw = _env("ALERT_RE", r"(?i)(⚠|Сбой по расписанию|отчёт неполный|FATAL)")
|
|
try:
|
|
alert_re = re.compile(alert_raw) if alert_raw else None
|
|
except re.error as err:
|
|
raise RuntimeError(f"Некорректный ALERT_RE: {err}") from err
|
|
return cls(
|
|
watch_dir=watch_dir,
|
|
state_file=state_dir / "state.json",
|
|
poll_interval=max(5, _env_int("POLL_INTERVAL", 10)),
|
|
report_globs=[g.strip() for g in _env("REPORT_GLOBS", "otchet_*.pdf").split(",") if g.strip()],
|
|
alert_re=alert_re,
|
|
log_file=Path(_env("RELAY_LOG") or (watch_dir / "report_shot.log")),
|
|
device_id=_env("MAX_DEVICE_ID"),
|
|
login_token=_env("MAX_LOGIN_TOKEN"),
|
|
target_chat=_env_int("MAX_TARGET_CHAT", 0),
|
|
hello=_env_bool("RELAY_HELLO", True),
|
|
max_lines_per_cycle=_env_int("MAX_LINES_PER_CYCLE", 5),
|
|
)
|