workphone-sdk/sdk/scripts/run_stability_longrun.py

#!/usr/bin/env python3
"""持续记录工作手机 SDK、WS 与 Hook 稳定性，供 24h/长期验收使用。"""

from __future__ import annotations

import argparse
import json
import signal
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from urllib.parse import urlencode
from urllib.request import Request, urlopen


def request_json(url: str, timeout: float) -> dict[str, Any]:
    request = Request(url, headers={"Accept": "application/json", "User-Agent": "workphone-stability/1.0"})
    with urlopen(request, timeout=timeout) as response:
        return json.loads(response.read().decode("utf-8"))


def write_json(path: Path, data: dict[str, Any]) -> None:
    temp = path.with_suffix(path.suffix + ".tmp")
    temp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    temp.replace(path)


def restore_samples(samples_path: Path) -> tuple[datetime, int, int, int, int, int, int, int, bool | None, str]:
    started_at = datetime.now().astimezone()
    total = ok = failures = offline_events = reconnects = 0
    latency_total = max_latency = 0
    last_ok: bool | None = None
    last_error = ""
    if not samples_path.exists():
        return (
            started_at,
            total,
            ok,
            failures,
            offline_events,
            reconnects,
            latency_total,
            max_latency,
            last_ok,
            last_error,
        )

    for line in samples_path.read_text(encoding="utf-8").splitlines():
        try:
            row = json.loads(line)
        except json.JSONDecodeError:
            continue
        if not isinstance(row, dict):
            continue
        try:
            timestamp = datetime.fromisoformat(str(row.get("timestamp", "")))
            if total == 0:
                started_at = timestamp
        except ValueError:
            pass
        row_ok = bool(row.get("ok"))
        total += 1
        ok += int(row_ok)
        failures += int(not row_ok)
        elapsed_ms = int(row.get("request_elapsed_ms") or 0)
        latency_total += elapsed_ms
        max_latency = max(max_latency, elapsed_ms)
        if row_ok:
            if last_ok is False:
                reconnects += 1
            last_error = ""
        else:
            if last_ok is not False:
                offline_events += 1
            last_error = str(row.get("error") or "sdk/ws/hook not ready")
        last_ok = row_ok

    return (
        started_at,
        total,
        ok,
        failures,
        offline_events,
        reconnects,
        latency_total,
        max_latency,
        last_ok,
        last_error,
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--base", default="http://127.0.0.1:8899")
    parser.add_argument("--device-id", default="xgfe65eimrrofyws")
    parser.add_argument("--interval-seconds", type=float, default=60)
    parser.add_argument("--timeout-seconds", type=float, default=45)
    parser.add_argument("--output-dir", required=True)
    args = parser.parse_args()

    output_dir = Path(args.output_dir).expanduser().resolve()
    output_dir.mkdir(parents=True, exist_ok=True)
    samples_path = output_dir / "stability_samples.jsonl"
    summary_path = output_dir / "stability_latest_summary.json"
    stop = False

    def handle_stop(_signum: int, _frame: Any) -> None:
        nonlocal stop
        stop = True

    signal.signal(signal.SIGTERM, handle_stop)
    signal.signal(signal.SIGINT, handle_stop)

    (
        started_at,
        total,
        ok,
        failures,
        offline_events,
        reconnects,
        latency_total,
        max_latency,
        last_ok,
        last_error,
    ) = restore_samples(samples_path)

    while not stop:
        sample_started = time.time()
        now = datetime.now().astimezone()
        row: dict[str, Any] = {
            "timestamp": now.isoformat(),
            "device_id": args.device_id,
            "ok": False,
        }
        try:
            health = request_json(f"{args.base}/health", args.timeout_seconds)
            query = urlencode({"device_id": args.device_id, "samples": 1, "interval_seconds": 0})
            watch = request_json(f"{args.base}/api/v3/stability/watch?{query}", args.timeout_seconds)
            watch_data = watch.get("data") if isinstance(watch.get("data"), dict) else {}
            samples = watch_data.get("samples") if isinstance(watch_data.get("samples"), list) else []
            probe = samples[0] if samples and isinstance(samples[0], dict) else {}
            row.update(
                {
                    "sdk_healthy": health.get("status") == "healthy",
                    "devices_online": health.get("devices_online"),
                    "ws_online": bool(probe.get("ws_online")),
                    "adb_online": bool(probe.get("adb_online")),
                    "hook_ok": bool(probe.get("hook_ok")),
                    "wechat_version": probe.get("wechat_version", ""),
                    "probe_latency_ms": probe.get("latency_ms", 0),
                    "error": probe.get("error", ""),
                }
            )
            row["ok"] = bool(row["sdk_healthy"] and row["ws_online"] and row["hook_ok"])
        except Exception as exc:
            row["error"] = str(exc)[:500]

        row["request_elapsed_ms"] = int((time.time() - sample_started) * 1000)
        total += 1
        latency_total += row["request_elapsed_ms"]
        max_latency = max(max_latency, row["request_elapsed_ms"])
        if row["ok"]:
            ok += 1
            if last_ok is False:
                reconnects += 1
            last_error = ""
        else:
            failures += 1
            last_error = str(row.get("error") or "sdk/ws/hook not ready")
            if last_ok is not False:
                offline_events += 1
        last_ok = bool(row["ok"])

        with samples_path.open("a", encoding="utf-8") as stream:
            stream.write(json.dumps(row, ensure_ascii=False) + "\n")

        elapsed_seconds = max((now - started_at).total_seconds(), 0)
        summary = {
            "started_at": started_at.isoformat(),
            "updated_at": now.isoformat(),
            "elapsed_seconds": int(elapsed_seconds),
            "elapsed_hours": round(elapsed_seconds / 3600, 3),
            "device_id": args.device_id,
            "base": args.base,
            "interval_seconds": args.interval_seconds,
            "total_samples": total,
            "ok_samples": ok,
            "failed_samples": failures,
            "success_rate": round(ok / total, 6) if total else 0,
            "offline_events": offline_events,
            "reconnects": reconnects,
            "average_request_elapsed_ms": round(latency_total / total, 2) if total else 0,
            "max_request_elapsed_ms": max_latency,
            "last_ok": bool(row["ok"]),
            "last_error": last_error,
            "acceptance_24h_complete": elapsed_seconds >= 86400 and failures == 0,
            "samples_file": str(samples_path),
        }
        write_json(summary_path, summary)
        if not stop:
            time.sleep(max(args.interval_seconds, 5))

    return 0


if __name__ == "__main__":
    raise SystemExit(main())