# -*- coding: utf-8 -*-
"""控制台与调度器的进程外保活（登录自启动，随用户会话常驻）。

职责（不承载任何业务调度规则，业务调度仍由 `gyxx schedule run` 唯一负责）：
1. 控制台：探测 http://127.0.0.1:8765/api/health——进程不在则拉起；
   进程在但探测连续失败（wedged）则结束该进程树后拉起。控制台进程内
   另有看门狗（console.py `_console_watchdog`）自检失败会主动退出，此处
   负责把退出后的空窗补上，两层互为补充。
2. 调度器：读 `var/state/scheduler/service.lock` 的 PID 判断存活，锁主
   已死则按 install.ps1 同款参数拉起 `python -m gyxx_flow schedule run`；
   调度器自身单实例锁保证不会双跑。
3. 节流：每个目标有冷却间隔；短时间反复重启触发熔断，只记录不再拉起，
   避免故障风暴。

用法：
  gyxx_keepalive.py            # 常驻循环（Startup 的 vbs 以 pythonw 启动）
  gyxx_keepalive.py --once     # 单轮探测+处置后退出，供验收/手动巡检

日志：var/logs/keepalive/keepalive.log（追加）。
自身单实例：var/state/keepalive.lock（PID 存活检测，重复启动直接退出）。
"""

from __future__ import annotations

import argparse
import ctypes
import http.client
import json
import os
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[2]
ENV_FILES = [
    Path(os.getenv("GYXX_RUNTIME_ENV_FILE", r"D:\product-collector-analyze-flow\.env")),
    Path(
        os.getenv(
            "GYXX_RUNTIME_JUSHUITAN_ENV_FILE",
            r"D:\product-collector-analyze-flow\private-secrets\jushuitan.env",
        )
    ),
]
CONSOLE_HOST = "127.0.0.1"
CONSOLE_PORT = int(os.getenv("GYXX_CONSOLE_PORT", "8765"))
LOOP_INTERVAL_SECONDS = float(os.getenv("GYXX_KEEPALIVE_INTERVAL_SECONDS", "30"))
PROBE_TIMEOUT_SECONDS = 5.0
PROBE_FAILURE_LIMIT = 3
RESTART_COOLDOWN_SECONDS = float(os.getenv("GYXX_KEEPALIVE_COOLDOWN_SECONDS", "120"))
TRIP_MAX_RESTARTS = 5
TRIP_WINDOW_SECONDS = 30 * 60
TRIP_PAUSE_SECONDS = 60 * 60

DATA_ROOT = PROJECT_ROOT / "var"
LOG_PATH = DATA_ROOT / "logs" / "keepalive" / "keepalive.log"
LOCK_PATH = DATA_ROOT / "state" / "keepalive.lock"
CONSOLE_LOG_DIR = DATA_ROOT / "logs" / "console-restarts"
SCHEDULER_LOG_DIR = DATA_ROOT / "logs" / "scheduler-service"
SCHEDULER_SERVICE_LOCK = DATA_ROOT / "state" / "scheduler" / "service.lock"


def _log(event: str, detail: str = "") -> None:
    stamp = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S")
    line = f"{stamp} {event} {detail}".rstrip()
    LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
    with LOG_PATH.open("a", encoding="utf-8", newline="\n") as stream:
        stream.write(line + "\n")
    print(line, flush=True)


def _pid_alive(pid: int) -> bool:
    if pid <= 0:
        return False
    kernel32 = ctypes.windll.kernel32
    handle = kernel32.OpenProcess(0x1000, False, pid)  # PROCESS_QUERY_LIMITED_INFORMATION
    if not handle:
        return False
    try:
        exit_code = ctypes.c_ulong()
        if kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)):
            return exit_code.value == 259  # STILL_ACTIVE
        return False
    finally:
        kernel32.CloseHandle(handle)


def _acquire_self_lock() -> bool:
    LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
    if LOCK_PATH.is_file():
        try:
            existing = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
            if _pid_alive(int(existing.get("pid", 0))):
                return False
        except Exception:
            pass
    LOCK_PATH.write_text(
        json.dumps({"pid": os.getpid(), "started_at": datetime.now(timezone.utc).isoformat()}),
        encoding="utf-8",
    )
    return True


def _console_env_files() -> list[Path]:
    return [path for path in ENV_FILES if path.is_file()]


def _spawn(argv: list[str], log_dir: Path, log_prefix: str) -> int:
    log_dir.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    log_path = log_dir / f"{log_prefix}-{stamp}.log"
    kwargs: dict[str, object] = {}
    if os.name == "nt":
        kwargs["creationflags"] = (
            subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP
        )
    else:
        kwargs["start_new_session"] = True
    with log_path.open("a", encoding="utf-8", newline="\n") as stream:
        process = subprocess.Popen(
            argv,
            cwd=str(PROJECT_ROOT),
            env=os.environ,
            stdin=subprocess.DEVNULL,
            stdout=stream,
            stderr=subprocess.STDOUT,
            shell=False,
            **kwargs,
        )
    return process.pid


def _console_argv() -> list[str]:
    argv = [
        str(PROJECT_ROOT / ".venv" / "Scripts" / "python.exe"),
        "-m",
        "gyxx_flow",
        "console",
        "--host",
        CONSOLE_HOST,
        "--port",
        str(CONSOLE_PORT),
    ]
    for env_file in _console_env_files():
        argv.extend(("--env-file", str(env_file)))
    return argv


def _scheduler_argv() -> list[str]:
    argv = [
        str(PROJECT_ROOT / ".venv" / "Scripts" / "python.exe"),
        "-m",
        "gyxx_flow",
        "schedule",
        "run",
    ]
    for env_file in _console_env_files():
        argv.extend(("--env-file", str(env_file)))
    return argv


def _probe_console() -> bool:
    try:
        connection = http.client.HTTPConnection(CONSOLE_HOST, CONSOLE_PORT, timeout=PROBE_TIMEOUT_SECONDS)
        try:
            connection.request("GET", "/api/health")
            response = connection.getresponse()
            response.read()
            return response.status == 200
        finally:
            connection.close()
    except Exception:
        return False


def _port_owner_pid(port: int) -> int | None:
    try:
        output = subprocess.run(
            ["netstat", "-ano", "-p", "tcp"],
            capture_output=True,
            text=True,
            timeout=10,
        ).stdout
    except Exception:
        return None
    needle = f"{CONSOLE_HOST}:{port}"
    for line in output.splitlines():
        parts = line.split()
        if (
            len(parts) >= 5
            and parts[0].upper() == "TCP"
            and parts[1] == needle
            and parts[3].upper() == "LISTENING"
        ):
            try:
                return int(parts[4])
            except ValueError:
                return None
    return None


def _kill_pid_tree(pid: int) -> None:
    if pid <= 0 or pid == os.getpid():
        return
    subprocess.run(
        ["taskkill", "/PID", str(pid), "/T", "/F"],
        capture_output=True,
        timeout=30,
    )


def _scheduler_lock_pid() -> int | None:
    try:
        payload = json.loads(SCHEDULER_SERVICE_LOCK.read_text(encoding="utf-8"))
        return int(payload.get("pid", 0)) or None
    except Exception:
        return None


class _TargetState:
    """记录单目标的连续失败计数、冷却与熔断状态。"""

    def __init__(self, name: str) -> None:
        self.name = name
        self.failures = 0
        self.last_restart_at = 0.0
        self.recent_restarts: list[float] = []
        self.tripped_at: float | None = None

    def tripped(self, now: float) -> bool:
        if self.tripped_at is None:
            return False
        if now - self.tripped_at >= TRIP_PAUSE_SECONDS:
            self.tripped_at = None
            self.recent_restarts.clear()
            _log("keepalive-trip-reset", f"target={self.name} 熔断窗口结束恢复守护")
            return False
        return True

    def may_restart(self, now: float) -> bool:
        if self.tripped(now):
            return False
        if now - self.last_restart_at < RESTART_COOLDOWN_SECONDS:
            return False
        self.recent_restarts = [
            stamp for stamp in self.recent_restarts if now - stamp < TRIP_WINDOW_SECONDS
        ]
        if len(self.recent_restarts) >= TRIP_MAX_RESTARTS:
            self.tripped_at = now
            _log(
                "keepalive-tripped",
                f"target={self.name} {TRIP_MAX_RESTARTS} 次重启/{TRIP_WINDOW_SECONDS // 60}分钟 "
                f"暂停拉起{TRIP_PAUSE_SECONDS // 60}分钟，请人工检查日志",
            )
            return False
        return True

    def record_restart(self, now: float) -> None:
        self.last_restart_at = now
        self.recent_restarts.append(now)


def run_cycle(console: _TargetState, scheduler: _TargetState) -> None:
    now = time.time()

    # ---- 控制台 ----
    if _probe_console():
        console.failures = 0
    else:
        console.failures += 1
        owner = _port_owner_pid(CONSOLE_PORT)
        _log(
            "console-probe-failed",
            f"{console.failures}/{PROBE_FAILURE_LIMIT} port_owner_pid={owner}",
        )
        if console.failures >= PROBE_FAILURE_LIMIT and console.may_restart(now):
            if owner is not None and _pid_alive(owner):
                _log("console-kill-wedged", f"pid={owner} 探测持续失败，结束进程树")
                _kill_pid_tree(owner)
                time.sleep(3)
            pid = _spawn(_console_argv(), CONSOLE_LOG_DIR, "keepalive-console")
            console.record_restart(time.time())
            console.failures = 0
            _log("console-restarted", f"pid={pid}")

    # ---- 调度器 ----
    scheduler_pid = _scheduler_lock_pid()
    scheduler_alive = scheduler_pid is not None and _pid_alive(scheduler_pid)
    if not scheduler_alive and scheduler.may_restart(now):
        pid = _spawn(_scheduler_argv(), SCHEDULER_LOG_DIR, "keepalive-start")
        scheduler.record_restart(time.time())
        _log(
            "scheduler-restarted",
            f"pid={pid} prev_lock_pid={scheduler_pid}",
        )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="GYXX Flow 控制台/调度器保活")
    parser.add_argument("--once", action="store_true", help="单轮探测+处置后退出")
    arguments = parser.parse_args(argv)

    if not _acquire_self_lock():
        _log("keepalive-duplicate-exit", "已有保活进程在运行")
        return 0

    console = _TargetState("console")
    scheduler = _TargetState("scheduler")
    missing_env = [str(path) for path in ENV_FILES if not path.is_file()]
    _log(
        "keepalive-start",
        f"pid={os.getpid()} once={arguments.once}"
        + (f" missing_env={missing_env}" if missing_env else ""),
    )
    try:
        while True:
            run_cycle(console, scheduler)
            if arguments.once:
                _log("keepalive-once-done", "")
                return 0
            time.sleep(LOOP_INTERVAL_SECONDS)
    except KeyboardInterrupt:
        return 0
    finally:
        try:
            if LOCK_PATH.is_file():
                payload = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
                if int(payload.get("pid", 0)) == os.getpid():
                    LOCK_PATH.unlink()
        except Exception:
            pass


if __name__ == "__main__":
    sys.exit(main())