#!/usr/bin/env python3 import os import sys import subprocess import json import xml.etree.ElementTree as ET from datetime import datetime from pathlib import Path ADDON_DIR = Path("/usr/local/mgr5/addon") CONFIG_PATH = ADDON_DIR / "config.json" STATE_PATH = ADDON_DIR / "state.json" MGRCTL = "/usr/local/mgr5/sbin/mgrctl" DEFAULT_CONFIG = { "telegram_token": "", "telegram_chat_id": "", "max_token": "", "max_chat_id": "", "check_interval": 60, "timeout": 10, "retry_count": 3, "notify_telegram": False, "notify_max": False, "monitored_sites": [], } def load_config() -> dict: if CONFIG_PATH.exists(): with open(CONFIG_PATH, "r", encoding="utf-8", errors="replace") as f: return {**DEFAULT_CONFIG, **json.load(f)} return dict(DEFAULT_CONFIG) def save_config(cfg: dict): ADDON_DIR.mkdir(parents=True, exist_ok=True) with open(CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(cfg, f, indent=2, ensure_ascii=False) def load_state() -> dict: if STATE_PATH.exists(): with open(STATE_PATH, "r", encoding="utf-8", errors="replace") as f: return json.load(f) return {} def save_state(state: dict): with open(STATE_PATH, "w", encoding="utf-8") as f: json.dump(state, f, indent=2) def get_domains() -> list: try: clean_env = {k: v for k, v in os.environ.items() if not k.startswith("PARAM_")} result = subprocess.run( [MGRCTL, "-m", "ispmgr", "webdomain", "-o", "xml"], capture_output=True, text=True, timeout=10, env=clean_env ) if result.returncode != 0 or not result.stdout.strip(): return [] root = ET.fromstring(result.stdout) domains = [] for elem in root.findall(".//elem"): name_el = elem.find("name") if name_el is not None and name_el.text: domains.append(name_el.text.strip()) return domains except Exception: return [] try: PARAM_elid = os.environ.get('PARAM_elid', '') PARAM_func = os.environ.get('PARAM_func', 'sitemonitor') PARAM_sok = os.environ.get('PARAM_sok', '') PARAM_clicked_button = os.environ.get('PARAM_clicked_button', '') stdin_data = sys.stdin.read() if stdin_data.strip(): root = ET.fromstring(stdin_data) else: root = ET.Element('doc', {'func': PARAM_func}) if PARAM_func == 'sitemonitor.toggle' or (PARAM_func == 'sitemonitor' and PARAM_elid and not PARAM_sok): cfg = load_config() monitored = set(cfg.get("monitored_sites", [])) domains_to_toggle = [x.strip() for x in PARAM_elid.split(',') if x.strip()] clicked_btn = os.environ.get('PARAM_clicked_button', '') for domain in domains_to_toggle: if clicked_btn == 'toggle_on': monitored.add(domain) elif clicked_btn == 'toggle_off': monitored.discard(domain) else: if domain in monitored: monitored.discard(domain) else: monitored.add(domain) cfg["monitored_sites"] = sorted(list(monitored)) save_config(cfg) root = ET.Element('doc', {'func': 'sitemonitor'}) ET.SubElement(root, 'ok') print(ET.tostring(root, encoding='unicode')) sys.exit(0) elif PARAM_func == 'sitemonitor.settings' and (PARAM_sok == 'ok' or PARAM_clicked_button == 'save'): cfg = load_config() cfg["check_interval"] = max(10, int(os.environ.get('PARAM_check_interval', '60'))) cfg["timeout"] = max(1, int(os.environ.get('PARAM_timeout', '10'))) cfg["retry_count"] = max(1, int(os.environ.get('PARAM_retry_count', '3'))) cfg["notify_telegram"] = os.environ.get('PARAM_notify_telegram', '') == 'on' cfg["notify_max"] = os.environ.get('PARAM_notify_max', '') == 'on' tg_token = os.environ.get('PARAM_telegram_token', '').strip() if tg_token and not tg_token.startswith('***'): cfg["telegram_token"] = tg_token cfg["telegram_chat_id"] = os.environ.get('PARAM_telegram_chat_id', '').strip() max_token = os.environ.get('PARAM_max_token', '').strip() if max_token and not max_token.startswith('***'): cfg["max_token"] = max_token cfg["max_chat_id"] = os.environ.get('PARAM_max_chat_id', '').strip() save_config(cfg) out = ET.Element('doc', {'func': 'sitemonitor'}) ET.SubElement(out, 'ok') print(ET.tostring(out, encoding='unicode')) sys.exit(0) elif PARAM_func == 'sitemonitor.settings': cfg = load_config() ET.SubElement(root, 'check_interval').text = str(cfg["check_interval"]) ET.SubElement(root, 'timeout').text = str(cfg["timeout"]) ET.SubElement(root, 'retry_count').text = str(cfg["retry_count"]) ET.SubElement(root, 'notify_telegram').text = 'on' if cfg["notify_telegram"] else 'off' ET.SubElement(root, 'telegram_token').text = '********' if cfg.get('telegram_token') else '' ET.SubElement(root, 'telegram_chat_id').text = cfg.get('telegram_chat_id', '') ET.SubElement(root, 'notify_max').text = 'on' if cfg["notify_max"] else 'off' ET.SubElement(root, 'max_token').text = '********' if cfg.get('max_token') else '' ET.SubElement(root, 'max_chat_id').text = cfg.get('max_chat_id', '') print(ET.tostring(root, encoding='unicode')) sys.exit(0) else: cfg = load_config() state = load_state() monitored_sites = set(cfg.get("monitored_sites", [])) domains = get_domains() for domain in domains: site_state = state.get(domain, {}) status_raw = site_state.get("status", "unknown") status_text = "Unknown" if status_raw == "up": status_text = "Available" elif status_raw == "down": status_text = "Unavailable" rt = site_state.get("response_time") rt_text = f"{rt} ms" if rt else "—" checked = site_state.get("checked_at", "") if checked: try: dt = datetime.fromisoformat(checked) checked = dt.strftime("%d.%m %H:%M") except ValueError: pass mon_val = "on" if domain in monitored_sites else "off" elem = ET.SubElement(root, 'elem') ET.SubElement(elem, 'id').text = domain ET.SubElement(elem, 'domain').text = domain ET.SubElement(elem, 'status').text = status_text ET.SubElement(elem, 'response_time').text = rt_text ET.SubElement(elem, 'checked_at').text = checked or "—" ET.SubElement(elem, 'monitored').text = mon_val print(ET.tostring(root, encoding='unicode')) except Exception as e: root = ET.Element('doc', {'func': 'sitemonitor'}) error = ET.SubElement(root, 'error') msg = ET.SubElement(error, 'msg', {'name': 'body'}) msg.text = str(e) print(ET.tostring(root, encoding='unicode'))