path: root/status/app.py

"""軽量ステータスページ。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, timedelta, timezone
from html import escape
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

TARGETS_FILE = os.environ.get("TARGETS_FILE", "/app/targets.txt")
# hetzner-admin で停止したサービス名（1 行 1 つ）。停止中で応答がなければ DOWN ではなく STOPPED 扱い
STOPPED_FILE = os.path.join(os.path.dirname(TARGETS_FILE), "stopped.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")
# ntfy アプリで通知トピックを開くディープリンク（空ならページに出さない）
NTFY_DEEPLINK = os.environ.get("NTFY_DEEPLINK", "ntfy://ntfy.yyamashita.com/status")
# 毎朝のサマリー通知（-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", "stopped", "detail", "ms", "checked_at", "history": deque[bool|None], "fails", "alerted", "day_checks", "day_up"}
period_start = datetime.now(LOCAL_TZ)  # サマリー集計の開始時刻


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 load_stopped():
    try:
        with open(STOPPED_FILE) as f:
            return {line.strip() for line in f if line.strip()}
    except FileNotFoundError:
        return set()


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, tags, priority="default"):
    if not NTFY_URL:
        return
    headers = {
        "Title": title,
        "Tags": tags,
        "Priority": priority,
        "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()
    stopped_names = load_stopped()
    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,
                 "day_checks": 0, "day_up": 0},
            )
            # 停止中でも応答があれば（push で再デプロイされた等）普通に UP として扱う
            stopped = name in stopped_names and not up
            if stopped:
                detail = "Stopped from admin"
            entry.update(url=url, up=up, stopped=stopped, detail=detail, ms=ms, checked_at=now)
            if stopped:
                # 稼働率・DOWN 通知の対象外。履歴は灰色で残す
                entry["history"].append(None)
                entry["fails"] = 0
                continue
            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}", "rotating_light", "high"))
            elif up and entry["alerted"]:
                entry["alerted"] = 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"] or e["stopped"])
            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)


def snapshot():
    with lock:
        services = []
        for name, e in state.items():
            checked = [h for h in e["history"] if h is not None]
            services.append({
                "name": name,
                "url": e["url"],
                "up": e["up"],
                "stopped": e["stopped"],
                "state": "stopped" if e["stopped"] else "up" if e["up"] else "down",
                "detail": e["detail"],
                "ms": e["ms"],
                "checked_at": e["checked_at"],
                "uptime": round(100 * sum(checked) / len(checked), 1) if checked else None,
                "history": list(e["history"]),
            })
    return {"all_up": all(s["up"] or s["stopped"] for s in services), "services": services}


def render_html(data):
    rows = []
    for s in data["services"]:
        bars = "".join(
            f'<i class="{"s" if ok is None else "u" if ok else "d"}"></i>' for ok in s["history"]
        )
        dot = {"up": "u", "down": "d", "stopped": "s"}[s["state"]]
        uptime = "-" if s["uptime"] is None else f'{s["uptime"]}%'
        ms = "-" if s["stopped"] else f'{s["ms"]} ms'
        rows.append(
            f'<tr><td><span class="dot {dot}"></span>{escape(s["name"])}</td>'
            f'<td>{escape(s["detail"])}</td><td>{ms}</td><td>{uptime}</td>'
            f'<td class="bars">{bars}</td></tr>'
        )
    if not data["services"]:
        banner = '<div class="banner">Checking...</div>'
    elif data["all_up"]:
        n_stopped = sum(s["stopped"] for s in data["services"])
        note = f" ({n_stopped} stopped)" if n_stopped else ""
        banner = f'<div class="banner u">All systems operational{note}</div>'
    else:
        banner = '<div class="banner d">Some systems are down</div>'
    checked = data["services"][0]["checked_at"] if data["services"] else "-"
    subscribe = (
        f' · <a href="{escape(NTFY_DEEPLINK)}">Subscribe in ntfy app</a>' if NTFY_DEEPLINK 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}}.s{{background:#bbb}}
.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>{subscribe}</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()