#!/usr/bin/env python3 """ Synology Monitor Service ======================== Checks every CHECK_INTERVAL_HOURS hours: 1. Synology NAS server — https://enigma.seekright.ai 2. Image server — https://images.seekright.ai (simple ping, no specific paths) Alerts via: - RocketChat Incoming Webhook (always — heartbeat) - Email / SMTP (only on failures) """ import os import json import sys import time import logging import smtplib import requests import urllib3 from logging.handlers import RotatingFileHandler from datetime import datetime, timezone from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from apscheduler.schedulers.blocking import BlockingScheduler from dotenv import load_dotenv urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) load_dotenv() # ── Configuration ───────────────────────────────────────────────────────────── # Primary Synology NAS / DSM SYNOLOGY_URL = os.getenv("SYNOLOGY_URL", "https://enigma.seekright.ai").rstrip("/") # Image CDN / file server — just ping the root, no specific paths IMAGE_SERVER_URL = os.getenv("IMAGE_SERVER_URL", "https://images.seekright.ai").rstrip("/") ROCKETCHAT_WEBHOOK_URL = os.getenv("ROCKETCHAT_WEBHOOK_URL", "") ROCKETCHAT_BOT_NAME = os.getenv("ROCKETCHAT_BOT_NAME", "Enigma-Status-Bot") ROCKETCHAT_BOT_EMOJI = os.getenv("ROCKETCHAT_BOT_EMOJI", ":robot:") SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com") SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) SMTP_USERNAME = os.getenv("SMTP_USERNAME", "") SMTP_PASSWORD = os.getenv("SMTP_PASSWORD", "") EMAIL_FROM = os.getenv("EMAIL_FROM", "") EMAIL_TO = os.getenv("EMAIL_TO", "") EMAIL_CC = [e.strip() for e in os.getenv("EMAIL_CC", "").split(",") if e.strip()] CHECK_INTERVAL_HOURS = float(os.getenv("CHECK_INTERVAL_HOURS", "3")) REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT_SECONDS", "15")) LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper() LOG_FILE = os.getenv("LOG_FILE", "monitor.log") # ── Logging ─────────────────────────────────────────────────────────────────── # Rotating file log: 5 MB per file, 3 backups kept → max ~20 MB total on disk _fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") _fh = RotatingFileHandler(LOG_FILE, maxBytes=5 * 1024 * 1024, backupCount=3, encoding="utf-8") _fh.setFormatter(_fmt) # Force UTF-8 on Windows console to avoid emoji encoding errors _sh = logging.StreamHandler( open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1) if hasattr(sys.stdout, "fileno") else sys.stdout ) _sh.setFormatter(_fmt) logger = logging.getLogger("synology_monitor") logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO)) logger.addHandler(_fh) logger.addHandler(_sh) # ═════════════════════════════════════════════════════════════════════════════ # CHECK FUNCTIONS # ═════════════════════════════════════════════════════════════════════════════ def _ping(url: str, label: str) -> dict: """ Generic HTTP ping — GET the URL and check it responds. - Any 2xx / 3xx / 4xx → server is UP (we got a response) - 5xx → server error (DOWN) - Connection error → server unreachable (DOWN) - Timeout → server unreachable (DOWN) A 403 on images.seekright.ai is expected (no directory listing) — still UP. A 404 on enigma.seekright.ai means the server answered — still UP. """ try: t0 = time.monotonic() r = requests.get(url, timeout=REQUEST_TIMEOUT, verify=False, allow_redirects=True) ms = round((time.monotonic() - t0) * 1000) # Treat anything below 500 as "server is alive" ok = r.status_code < 500 return { "label": label, "url": url, "ok": ok, "status_code": r.status_code, "latency_ms": ms, "error": None if ok else f"Server error HTTP {r.status_code}", } except requests.exceptions.ConnectionError as e: return {"label": label, "url": url, "ok": False, "status_code": None, "latency_ms": None, "error": f"Connection refused / DNS failure: {e}"} except requests.exceptions.Timeout: return {"label": label, "url": url, "ok": False, "status_code": None, "latency_ms": None, "error": f"No response within {REQUEST_TIMEOUT}s"} except Exception as e: return {"label": label, "url": url, "ok": False, "status_code": None, "latency_ms": None, "error": str(e)} # ═════════════════════════════════════════════════════════════════════════════ # MESSAGE BUILDER # ═════════════════════════════════════════════════════════════════════════════ def _build_message(results: list[dict]) -> tuple[str, str, str]: """Build (plain_text, markdown_text, html_text) from a list of ping results.""" now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") all_ok = all(r["ok"] for r in results) icon = "[OK]" if all_ok else "[ALERT]" md_icon = "✅" if all_ok else "🔴" md = [f"{md_icon} *Synology Monitor Report* — {now}\n"] txt = [f"{icon} Synology Monitor Report — {now}\n"] color = "#28a745" if all_ok else "#dc3545" html = [f"
"] html.append(f"

{md_icon} Synology Monitor Report

") html.append(f"

{now}

") html.append("
") html.append("") html.append("
") return "\n".join(txt), "\n".join(md), "\n".join(html) # ═════════════════════════════════════════════════════════════════════════════ # NOTIFICATIONS # ═════════════════════════════════════════════════════════════════════════════ def notify_rocketchat(plain_text: str, markdown_text: str, has_failures: bool) -> None: """Send report to RocketChat via Incoming Webhook.""" if not ROCKETCHAT_WEBHOOK_URL: logger.warning("ROCKETCHAT_WEBHOOK_URL not set — skipping") return color = "#e53e3e" if has_failures else "#38a169" label = "🔴 FAILURE DETECTED" if has_failures else "✅ All systems operational" payload = { "alias": ROCKETCHAT_BOT_NAME, "icon_emoji": ROCKETCHAT_BOT_EMOJI, # 'channel' omitted — fixed in webhook admin config "text": markdown_text, "attachments": [{ "title": label, "title_link": SYNOLOGY_URL, "color": color, "text": f"Checked: `{SYNOLOGY_URL}` and `{IMAGE_SERVER_URL}`", }], } try: r = requests.post(ROCKETCHAT_WEBHOOK_URL, json=payload, timeout=10) if r.status_code == 200: logger.info("RocketChat notification sent successfully") else: logger.warning("RocketChat returned HTTP %s: %s", r.status_code, r.text[:200]) except Exception as e: logger.error("RocketChat notification failed: %s", e) def notify_email(plain_text: str, html_text: str, has_failures: bool, custom_subject: str = None) -> None: """Send alert email. Only called on failures (or explicitly via test).""" if not all([SMTP_USERNAME, SMTP_PASSWORD, EMAIL_FROM, EMAIL_TO]): logger.warning("Email credentials incomplete — skipping") return if custom_subject: subject = custom_subject else: subject = f"[{'DOWN' if has_failures else 'TEST'}] Synology Monitor | {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}" msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = EMAIL_FROM msg["To"] = EMAIL_TO if EMAIL_CC: msg["Cc"] = ", ".join(EMAIL_CC) msg.attach(MIMEText(plain_text, "plain")) msg.attach(MIMEText(html_text, "html")) recipients = [EMAIL_TO] + EMAIL_CC try: with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=15) as s: s.ehlo() s.starttls() s.login(SMTP_USERNAME, SMTP_PASSWORD) s.sendmail(EMAIL_FROM, recipients, msg.as_string()) logger.info("Alert email sent to %s", recipients) except smtplib.SMTPAuthenticationError: logger.error("SMTP auth failed — check credentials") except Exception as e: logger.error("Email failed: %s", e) # ───────────────────────────────────────────────────────────────────────────── # STATE TRACKING (prevents email spam during sustained outages) # # Saves last known status to monitor_state.json. # Email is sent ONLY when the state CHANGES: # UP → DOWN : "[ALERT] services just went DOWN" # DOWN → UP : "[RESOLVED] services are back UP" # RocketChat always gets every report (heartbeat every 3 hrs). # ───────────────────────────────────────────────────────────────────────────── STATE_FILE = "monitor_state.json" def _load_state() -> dict: """Load previous check state from disk.""" try: with open(STATE_FILE, "r", encoding="utf-8") as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {"last_was_failure": None, "failed_since": None} def _save_state(state: dict) -> None: """Persist current state to disk.""" try: with open(STATE_FILE, "w", encoding="utf-8") as f: json.dump(state, f, indent=2) except Exception as e: logger.warning("Could not save state file: %s", e) # ═════════════════════════════════════════════════════════════════════════════ # MAIN CHECK CYCLE # ═════════════════════════════════════════════════════════════════════════════ def run_checks() -> None: logger.info("=" * 60) logger.info("Starting monitoring cycle ...") results = [ _ping(SYNOLOGY_URL, "Synology NAS (enigma.seekright.ai)"), _ping(IMAGE_SERVER_URL, "Image Server (images.seekright.ai)"), ] for r in results: status = "UP" if r["ok"] else "DOWN" logger.info("[%s] %s — HTTP %s, %s ms | %s", status, r["label"], r["status_code"], r["latency_ms"], r["error"] or "ok") has_failures = any(not r["ok"] for r in results) plain_text, markdown_text, html_text = _build_message(results) # ── Load previous state ────────────────────────────────────────────────── state = _load_state() was_failure = state.get("last_was_failure") # True/False/None (first run) failed_since = state.get("failed_since") # ISO timestamp or None now_iso = datetime.now(timezone.utc).isoformat() # ── State-change logic ─────────────────────────────────────────────────── if has_failures and not was_failure: # NEW outage — first time going DOWN failed_since = now_iso logger.info("State change: UP -> DOWN — sending alert email") subject = f"[DOWN] Synology Monitor — services DOWN | {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}" notify_email(plain_text, html_text, True, custom_subject=subject) elif not has_failures and was_failure: # RECOVERY — was DOWN, now UP duration = "" if failed_since: try: t1 = datetime.fromisoformat(failed_since) t2 = datetime.fromisoformat(now_iso) diff = t2 - t1 hours, remainder = divmod(int(diff.total_seconds()), 3600) mins, _ = divmod(remainder, 60) duration = f" (was down for {hours}h {mins}m)" except Exception: pass recovery_body = plain_text + f"\n\nServices recovered{duration}." logger.info("State change: DOWN -> UP — sending recovery email%s", duration) subject = f"[RESOLVED] Synology Monitor — services back UP{duration} | {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}" notify_email(recovery_body, html_text, False, custom_subject=subject) failed_since = None elif has_failures and was_failure: # Still down — DO NOT send another email, just log it if failed_since: try: dt_down = datetime.fromisoformat(failed_since) diff = datetime.now(timezone.utc) - dt_down hours = int(diff.total_seconds() // 3600) mins = int((diff.total_seconds() % 3600) // 60) logger.info("Still DOWN (outage duration: %dh %dm) — email suppressed", hours, mins) except Exception: logger.info("Still DOWN — email suppressed to avoid spam") else: logger.info("Still DOWN — email suppressed to avoid spam") else: # All UP, was UP — nothing to do logger.info("All systems UP — no email needed") # ── Save new state ─────────────────────────────────────────────────────── _save_state({ "last_was_failure": has_failures, "failed_since": failed_since, "last_check": now_iso, }) # ── RocketChat always gets a report (heartbeat) ────────────────────────── notify_rocketchat(plain_text, markdown_text, has_failures) logger.info("Cycle complete. Failures: %s", has_failures) logger.info("=" * 60) # ═════════════════════════════════════════════════════════════════════════════ # ENTRY POINT # ═════════════════════════════════════════════════════════════════════════════ if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Synology Monitor Service") parser.add_argument("--test-email", action="store_true", help="Send a test email with current status and exit") args = parser.parse_args() logger.info("Synology Monitor starting ...") logger.info("NAS server: %s", SYNOLOGY_URL) logger.info("Image server: %s", IMAGE_SERVER_URL) if args.test_email: logger.info("Running in test-email mode...") results = [ _ping(SYNOLOGY_URL, "Synology NAS (enigma.seekright.ai)"), _ping(IMAGE_SERVER_URL, "Image Server (images.seekright.ai)"), ] has_failures = any(not r["ok"] for r in results) plain_text, markdown_text, html_text = _build_message(results) notify_email(plain_text, html_text, has_failures) logger.info("Test email sent. Exiting.") sys.exit(0) logger.info("Interval: every %.1f hour(s)", CHECK_INTERVAL_HOURS) logger.info("RocketChat: %s", bool(ROCKETCHAT_WEBHOOK_URL)) logger.info("Email alerts: %s -> %s", EMAIL_FROM, EMAIL_TO) run_checks() scheduler = BlockingScheduler() scheduler.add_job(run_checks, "interval", hours=CHECK_INTERVAL_HOURS, id="synology_monitor", name="Synology Monitor") logger.info("Scheduler running — next check in %.1f hour(s). " "Ctrl+C to stop.", CHECK_INTERVAL_HOURS) try: scheduler.start() except KeyboardInterrupt: logger.info("Monitor stopped.")