From 01e7d14bf9f1cf3b9a5879ffbe4a8f0b5f553323 Mon Sep 17 00:00:00 2001
From: yyamashita <yyamashita@hetzner.yyamashita.com>
Date: Sat, 31 Oct 2026 13:34:33 +0900
Subject: status: send daily uptime summary to ntfy at 08:00 JST

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 status/app.py | 57 +++++++++++++++++++++++++++++++++++++++++++++++++--------
 1 file changed, 49 insertions(+), 8 deletions(-)

(limited to 'status/app.py')

diff --git a/status/app.py b/status/app.py
index 9cde2e9..7326b01 100644
--- a/status/app.py
+++ b/status/app.py
@@ -9,7 +9,7 @@ import urllib.parse
 import urllib.request
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from html import escape
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
@@ -22,9 +22,13 @@ NTFY_URL = os.environ.get("NTFY_URL", "")  # 例: http://ntfy/status（空なら
 NTFY_TOKEN = os.environ.get("NTFY_TOKEN", "")
 FAIL_THRESHOLD = int(os.environ.get("FAIL_THRESHOLD", "2"))  # 連続失敗この回数で DOWN 通知
 PUBLIC_URL = os.environ.get("PUBLIC_URL", "https://status.yyamashita.com")
+# 毎朝のサマリー通知（-1 で無効）。JST は DST がないので固定オフセットで扱う
+SUMMARY_HOUR = int(os.environ.get("SUMMARY_HOUR", "8"))
+LOCAL_TZ = timezone(timedelta(hours=float(os.environ.get("SUMMARY_TZ_OFFSET", "9"))))
 
 lock = threading.Lock()
-state = {}  # name -> {"url", "up", "detail", "ms", "checked_at", "history": deque[bool], "fails", "alerted"}
+state = {}  # name -> {"url", "up", "detail", "ms", "checked_at", "history": deque[bool], "fails", "alerted", "day_checks", "day_up"}
+period_start = datetime.now(LOCAL_TZ)  # サマリー集計の開始時刻
 
 
 def load_targets():
@@ -60,13 +64,13 @@ def check(url):
     return up, detail, int((time.monotonic() - start) * 1000)
 
 
-def notify(title, message, down):
+def notify(title, message, tags, priority="default"):
     if not NTFY_URL:
         return
     headers = {
         "Title": title,
-        "Tags": "rotating_light" if down else "white_check_mark",
-        "Priority": "high" if down else "default",
+        "Tags": tags,
+        "Priority": priority,
         "Click": PUBLIC_URL,
     }
     if NTFY_TOKEN:
@@ -89,25 +93,62 @@ def run_checks():
             del state[name]
         for (name, url), (up, detail, ms) in zip(targets, results):
             entry = state.setdefault(
-                name, {"history": deque(maxlen=HISTORY), "fails": 0, "alerted": False}
+                name,
+                {"history": deque(maxlen=HISTORY), "fails": 0, "alerted": False,
+                 "day_checks": 0, "day_up": 0},
             )
             entry.update(url=url, up=up, detail=detail, ms=ms, checked_at=now)
             entry["history"].append(up)
+            entry["day_checks"] += 1
+            entry["day_up"] += up
             entry["fails"] = 0 if up else entry["fails"] + 1
             if not up and entry["fails"] >= FAIL_THRESHOLD and not entry["alerted"]:
                 entry["alerted"] = True
-                alerts.append((f"{name} is DOWN", f"{url}\n{detail}", True))
+                alerts.append((f"{name} is DOWN", f"{url}\n{detail}", "rotating_light", "high"))
             elif up and entry["alerted"]:
                 entry["alerted"] = False
-                alerts.append((f"{name} recovered", f"{url}\n{detail} ({ms} ms)", False))
+                alerts.append((f"{name} recovered", f"{url}\n{detail} ({ms} ms)", "white_check_mark"))
     for alert in alerts:
         notify(*alert)
 
 
+def send_summary(now):
+    global period_start
+    with lock:
+        rows = [
+            (100 * e["day_up"] / e["day_checks"], name, e["up"])
+            for name, e in state.items()
+            if e["day_checks"]
+        ]
+        for e in state.values():
+            e["day_checks"] = e["day_up"] = 0
+        since, period_start = period_start, now
+    if not rows:
+        return
+    rows.sort()
+    down_now = [name for _, name, up in rows if not up]
+    imperfect = sum(1 for pct, _, _ in rows if pct < 100)
+    if down_now:
+        title = f"Daily status: {len(down_now)} down now"
+    elif imperfect:
+        title = f"Daily status: all up ({imperfect} had downtime)"
+    else:
+        title = f"Daily status: all {len(rows)} up, 100%"
+    lines = [f"{'DOWN ' if not up else ''}{name}: {pct:.1f}%" for pct, name, up in rows]
+    lines.append(f"since {since:%Y-%m-%d %H:%M}")
+    notify(title, "\n".join(lines), "warning" if down_now or imperfect else "sunny", "low")
+
+
 def checker_loop():
+    # 起動が当日の SUMMARY_HOUR 以降なら当日分は送らない（翌朝から）
+    last_summary = period_start.date() if period_start.hour >= SUMMARY_HOUR else None
     while True:
         try:
             run_checks()
+            now = datetime.now(LOCAL_TZ)
+            if SUMMARY_HOUR >= 0 and now.hour >= SUMMARY_HOUR and now.date() != last_summary:
+                last_summary = now.date()
+                send_summary(now)
         except Exception as e:
             print(f"check failed: {e}", flush=True)
         time.sleep(INTERVAL)
-- 
cgit v1.3