Add status pages, ntfy alerts, and license-slot-checker repo

- status/: lightweight status checker (status.yyamashita.com, /api/status)
  with ntfy DOWN/recovered alerts after consecutive failures
- uptime-kuma/: Uptime Kuma at uptime.yyamashita.com
- hetzner-admin: authenticated /status page (status-app results + docker ps -a)
- ntfy/: self-hosted ntfy at ntfy.yyamashita.com (deny-all auth, install.sh
  creates admin user and status-app token)
- Caddyfile, DNS, post-receive hook, tests/check.sh, docs updated
- license-slot-checker: add to repos.txt, hook, and Claude sessions

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 198 insertions, 0 deletions

diff --git a/status/app.py b/status/app.py
new file mode 100644
index 0000000..9cde2e9
--- /dev/null
+++ b/status/app.py
@@ -0,0 +1,198 @@
+"""軽量ステータスページ。targets.txt の各サービスを定期チェックし HTML / JSON で公開する。"""
+import json
+import os
+import socket
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections import deque
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
+from html import escape
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+TARGETS_FILE = os.environ.get("TARGETS_FILE", "/app/targets.txt")
+INTERVAL = int(os.environ.get("CHECK_INTERVAL", "60"))
+TIMEOUT = float(os.environ.get("CHECK_TIMEOUT", "10"))
+HISTORY = int(os.environ.get("HISTORY_SIZE", "90"))
+PORT = int(os.environ.get("PORT", "8080"))
+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")
+
+lock = threading.Lock()
+state = {}  # name -> {"url", "up", "detail", "ms", "checked_at", "history": deque[bool], "fails", "alerted"}
+
+
+def load_targets():
+    targets = []
+    with open(TARGETS_FILE) as f:
+        for line in f:
+            line = line.strip()
+            if not line or line.startswith("#"):
+                continue
+            name, url = line.split(None, 1)
+            targets.append((name, url.strip()))
+    return targets
+
+
+def check(url):
+    start = time.monotonic()
+    parsed = urllib.parse.urlparse(url)
+    try:
+        if parsed.scheme == "tcp":
+            with socket.create_connection((parsed.hostname, parsed.port), timeout=TIMEOUT):
+                pass
+            up, detail = True, "connected"
+        else:
+            req = urllib.request.Request(url, headers={"User-Agent": "status-check/1.0"})
+            try:
+                with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
+                    code = r.status
+            except urllib.error.HTTPError as e:
+                code = e.code
+            up, detail = code < 500, f"HTTP {code}"
+    except Exception as e:
+        up, detail = False, f"{type(e).__name__}: {e}"
+    return up, detail, int((time.monotonic() - start) * 1000)
+
+
+def notify(title, message, down):
+    if not NTFY_URL:
+        return
+    headers = {
+        "Title": title,
+        "Tags": "rotating_light" if down else "white_check_mark",
+        "Priority": "high" if down else "default",
+        "Click": PUBLIC_URL,
+    }
+    if NTFY_TOKEN:
+        headers["Authorization"] = f"Bearer {NTFY_TOKEN}"
+    req = urllib.request.Request(NTFY_URL, data=message.encode(), headers=headers, method="POST")
+    try:
+        urllib.request.urlopen(req, timeout=TIMEOUT).close()
+    except Exception as e:
+        print(f"ntfy notify failed: {e}", flush=True)
+
+
+def run_checks():
+    targets = load_targets()
+    with ThreadPoolExecutor(max_workers=max(1, len(targets))) as pool:
+        results = list(pool.map(lambda t: check(t[1]), targets))
+    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    alerts = []
+    with lock:
+        for name in set(state) - {n for n, _ in targets}:
+            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}
+            )
+            entry.update(url=url, up=up, detail=detail, ms=ms, checked_at=now)
+            entry["history"].append(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))
+            elif up and entry["alerted"]:
+                entry["alerted"] = False
+                alerts.append((f"{name} recovered", f"{url}\n{detail} ({ms} ms)", False))
+    for alert in alerts:
+        notify(*alert)
+
+
+def checker_loop():
+    while True:
+        try:
+            run_checks()
+        except Exception as e:
+            print(f"check failed: {e}", flush=True)
+        time.sleep(INTERVAL)
+
+
+def snapshot():
+    with lock:
+        services = [
+            {
+                "name": name,
+                "url": e["url"],
+                "up": e["up"],
+                "detail": e["detail"],
+                "ms": e["ms"],
+                "checked_at": e["checked_at"],
+                "uptime": round(100 * sum(e["history"]) / len(e["history"]), 1),
+                "history": list(e["history"]),
+            }
+            for name, e in state.items()
+        ]
+    return {"all_up": all(s["up"] for s in services), "services": services}
+
+
+def render_html(data):
+    rows = []
+    for s in data["services"]:
+        bars = "".join(
+            f'<i class="{"u" if ok else "d"}"></i>' for ok in s["history"]
+        )
+        rows.append(
+            f'<tr><td><span class="dot {"u" if s["up"] else "d"}"></span>{escape(s["name"])}</td>'
+            f'<td>{escape(s["detail"])}</td><td>{s["ms"]} ms</td><td>{s["uptime"]}%</td>'
+            f'<td class="bars">{bars}</td></tr>'
+        )
+    if not data["services"]:
+        banner = '<div class="banner">Checking...</div>'
+    elif data["all_up"]:
+        banner = '<div class="banner u">All systems operational</div>'
+    else:
+        banner = '<div class="banner d">Some systems are down</div>'
+    checked = data["services"][0]["checked_at"] if data["services"] else "-"
+    return f"""<!DOCTYPE html>
+<html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
+<meta http-equiv="refresh" content="{INTERVAL}"><title>Status - yyamashita.com</title>
+<style>
+body{{font-family:system-ui,sans-serif;max-width:860px;margin:2em auto;padding:0 1em;color:#222}}
+.banner{{padding:.8em 1em;border-radius:6px;color:#fff;background:#888;font-weight:600}}
+.banner.u{{background:#2a9d4b}}.banner.d{{background:#d0393e}}
+table{{width:100%;border-collapse:collapse;margin-top:1em}}td{{padding:.5em;border-bottom:1px solid #eee;white-space:nowrap}}
+.dot{{display:inline-block;width:10px;height:10px;border-radius:50%;margin-right:.5em}}
+.u{{background:#2a9d4b}}.d{{background:#d0393e}}
+.bars{{width:100%}}.bars i{{display:inline-block;width:4px;height:18px;margin-right:1px;border-radius:1px}}
+small{{color:#888}}
+</style></head><body>
+<h2>yyamashita.com status</h2>
+{banner}
+<table>{"".join(rows)}</table>
+<p><small>Last checked: {checked} (every {INTERVAL}s) · <a href="/api/status">JSON</a></small></p>
+</body></html>"""
+
+
+class Handler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        data = snapshot()
+        if self.path == "/api/status":
+            body = json.dumps(data).encode()
+            ctype = "application/json"
+        elif self.path == "/healthz":
+            body, ctype = b"ok", "text/plain"
+        elif self.path == "/":
+            body = render_html(data).encode()
+            ctype = "text/html; charset=utf-8"
+        else:
+            self.send_error(404)
+            return
+        self.send_response(200)
+        self.send_header("Content-Type", ctype)
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *args):
+        pass
+
+
+if __name__ == "__main__":
+    threading.Thread(target=checker_loop, daemon=True).start()
+    ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()