#!/usr/bin/env python3
"""通过 Paramiko 远程验收 live-console / live-edge / live.local。"""
from __future__ import annotations

import argparse
import json
import os
import sys
import time
from pathlib import Path

try:
    import paramiko
except ImportError:
    print("请先安装: pip install paramiko", file=sys.stderr)
    raise SystemExit(1)

HOST = os.environ.get("LIVE_DEPLOY_HOST", os.environ.get("LIVE_VERIFY_HOST", "192.168.21.105"))
USER = os.environ.get("LIVE_DEPLOY_USER", os.environ.get("LIVE_VERIFY_USER", "live"))
PASS = os.environ.get("LIVE_DEPLOY_PASS", os.environ.get("LIVE_VERIFY_PASS", "12345678"))
PORT = os.environ.get("LIVE_VERIFY_PORT", os.environ.get("PORT", "8001"))
MDNS_HOST = os.environ.get("LIVE_VERIFY_DOMAIN", f"{os.environ.get('LIVE_VERIFY_ALIAS', 'live')}.local")
LOG = Path(__file__).resolve().parent.parent / ".deploy_logs" / "verify_live_console.log"


def _safe_print(text: str) -> None:
    try:
        print(text)
    except UnicodeEncodeError:
        data = text.encode("utf-8", errors="replace")
        sys.stdout.buffer.write(data + b"\n")


def summarize_deploy_check(raw: str) -> list[str]:
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return ["deploy/check: invalid JSON payload"]
    lines = [
        f"deploy/check: {'READY' if data.get('ready') else 'DEGRADED'}"
        f" ({data.get('pass_count', 0)}/{data.get('total_count', 0)})",
    ]
    for item in data.get("checks") or []:
        mark = "OK" if item.get("ok") else "FAIL"
        label = item.get("label") or item.get("id") or "check"
        detail = item.get("detail") or item.get("error") or ""
        lines.append(f"{mark:4} {label}: {detail}")
    for hint in data.get("hints") or []:
        lines.append(f"hint: {hint}")
    return lines


def main() -> int:
    parser = argparse.ArgumentParser(description="远程验收 live-console / live-edge / live.local")
    parser.add_argument("--no-restart", action="store_true", help="只验收，不先重启服务")
    parser.add_argument("--wait", type=float, default=5.0, help="重启后等待秒数，默认 5")
    args = parser.parse_args()

    LOG.parent.mkdir(parents=True, exist_ok=True)
    c = paramiko.SSHClient()
    c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    c.connect(HOST, username=USER, password=PASS, timeout=30)

    def run(cmd: str, timeout: int = 90) -> tuple[int, str, str]:
        _, stdout, stderr = c.exec_command(
            "export LC_ALL=C.UTF-8 LANG=C.UTF-8; " + cmd,
            timeout=timeout,
        )
        out = stdout.read().decode("utf-8", errors="replace")
        err = stderr.read().decode("utf-8", errors="replace")
        code = stdout.channel.recv_exit_status()
        block = f"\n$ {cmd}\nexit={code}\n--- out ---\n{out}\n--- err ---\n{err}\n"
        with LOG.open("a", encoding="utf-8", errors="replace") as fh:
            fh.write(block)
        _safe_print("\n".join(block.splitlines()[-80:]))
        return code, out, err

    try:
        if not args.no_restart:
            run(
                f"echo '{PASS}' | sudo -S systemctl restart live-console.service && "
                f"(echo '{PASS}' | sudo -S systemctl restart live-edge.service || true)",
                timeout=150,
            )
            time.sleep(max(0.0, args.wait))

        run("systemctl is-active live-console.service", timeout=20)
        run("systemctl is-active live-edge.service || true", timeout=20)
        run(f"getent hosts {MDNS_HOST} || true", timeout=20)
        run(f"ss -tlnp 2>/dev/null | grep -E ':80\\b|:{PORT}\\b' || true", timeout=20)
        run(f"curl -sS -m 10 http://127.0.0.1:{PORT}/health", timeout=30)
        run("curl -sS -m 10 -o /dev/null -w '127.0.0.1:80 /health -> HTTP %{http_code}\\n' http://127.0.0.1/health || true", timeout=30)
        run(
            f"curl -sS -m 10 -o /dev/null -w '{MDNS_HOST} /health -> HTTP %{{http_code}}\\n' http://{MDNS_HOST}/health || true",
            timeout=30,
        )
        code, deploy_out, _ = run(f"curl -sS -m 12 http://127.0.0.1:{PORT}/deploy/check", timeout=40)
        if code == 0:
            _safe_print("\n".join(summarize_deploy_check(deploy_out)))
        return 0
    finally:
        c.close()


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