예제 소스

probe_mariago_uptime.py

#!/usr/bin/env python3
"""IDC(또는 외부망)에서 MariaGO 홈 도달성 연속 점검.

집 Mac이 아니라 **IDC에서** 돌리면, 외부→집 경로(공인 IP/NAT/절전) 이상을
휴대폰과 비슷한 시점에서 잡을 수 있습니다.

  # IDC에서 (stdlib only)
  python3 probe_mariago_uptime.py --url https://baduk.mariago.kr/login --interval 30

  # 백그라운드
  nohup python3 probe_mariago_uptime.py --interval 30 --out-dir ./mariago_uptime \\
    > mariago_uptime/stdout.log 2>&1 &

결과:
  out-dir/probes.jsonl  — 매회 1줄
  out-dir/STATUS.md     — 최근 요약 (ok율, 연속실패, 최근 실패)
"""

from __future__ import annotations

import argparse
import json
import ssl
import time
import urllib.error
import urllib.request
from collections import deque
from datetime import datetime, timezone
from pathlib import Path


def iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def probe_one(url: str, timeout_s: float) -> dict:
    t0 = time.monotonic()
    ctx = ssl.create_default_context()
    req = urllib.request.Request(
        url,
        method="GET",
        headers={
            "User-Agent": "MariaGO-uptime-probe/1.0 (+idc)",
            "Accept": "text/html,*/*",
            "Connection": "close",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout_s, context=ctx) as resp:
            body = resp.read(2048)
            dt_ms = (time.monotonic() - t0) * 1000
            code = int(resp.status)
            ok = 200 <= code < 400
            return {
                "ts": iso(),
                "url": url,
                "ok": ok,
                "status": code,
                "ms": round(dt_ms, 1),
                "bytes": len(body),
                "error": None,
            }
    except Exception as e:
        dt_ms = (time.monotonic() - t0) * 1000
        return {
            "ts": iso(),
            "url": url,
            "ok": False,
            "status": None,
            "ms": round(dt_ms, 1),
            "bytes": 0,
            "error": f"{type(e).__name__}: {e}",
        }


def write_status(
    path: Path,
    *,
    url: str,
    interval: float,
    window: deque,
    fail_streak: int,
    total_ok: int,
    total_n: int,
) -> None:
    recent_fail = [r for r in list(window)[-20:] if not r["ok"]]
    lines = [
        f"# MariaGO uptime probe — {iso()}",
        "",
        f"- **url**: `{url}`",
        f"- **interval_s**: `{interval}`",
        f"- **total**: `{total_n}` (ok `{total_ok}` = `{100.0 * total_ok / max(1, total_n):.1f}%`)",
        f"- **fail_streak**: `{fail_streak}`",
        f"- **window_ok**: `{sum(1 for r in window if r['ok'])}/{len(window)}`",
        "",
        "## recent failures",
        "",
    ]
    if recent_fail:
        for r in recent_fail[-10:]:
            lines.append(
                f"- `{r['ts']}` status=`{r['status']}` ms=`{r['ms']}` err=`{r['error']}`"
            )
    else:
        lines.append("- (없음)")
    lines.append("")
    path.write_text("\n".join(lines), encoding="utf-8")


def main() -> int:
    p = argparse.ArgumentParser(description="External uptime probe for MariaGO")
    p.add_argument("--url", default="https://baduk.mariago.kr/login")
    p.add_argument("--interval", type=float, default=30.0, help="초 단위")
    p.add_argument("--timeout", type=float, default=15.0)
    p.add_argument("--out-dir", default="mariago_uptime")
    p.add_argument("--window", type=int, default=120, help="STATUS 요약용 최근 N회")
    args = p.parse_args()

    out = Path(args.out_dir)
    out.mkdir(parents=True, exist_ok=True)
    jsonl = out / "probes.jsonl"
    status = out / "STATUS.md"

    window: deque = deque(maxlen=max(10, args.window))
    fail_streak = 0
    total_ok = total_n = 0

    print(
        f"[probe] url={args.url} interval={args.interval}s out={out.resolve()}",
        flush=True,
    )

    while True:
        r = probe_one(args.url, args.timeout)
        total_n += 1
        if r["ok"]:
            total_ok += 1
            fail_streak = 0
            tag = "OK"
        else:
            fail_streak += 1
            tag = "FAIL"
        window.append(r)
        with jsonl.open("a", encoding="utf-8") as f:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
        write_status(
            status,
            url=args.url,
            interval=args.interval,
            window=window,
            fail_streak=fail_streak,
            total_ok=total_ok,
            total_n=total_n,
        )
        print(
            f"[{tag}] {r['ts']} status={r['status']} {r['ms']}ms "
            f"streak={fail_streak} err={r['error']}",
            flush=True,
        )
        time.sleep(max(1.0, args.interval))


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except KeyboardInterrupt:
        print("[probe] stopped", flush=True)
        raise SystemExit(0)