#!/usr/bin/env python3
"""
health_watchdog.py - GoldAutoTrade 主动巡查 + 自愈脚本
由 keep_alive.sh 每2分钟调用一次

检查项目：
  1. Daemon 进程是否存活
  2. Daemon 日志是否在5分钟内更新（判断是否僵死）
  3. EA terminal_status.json 是否在2分钟内更新（判断EA是否活跃）
  4. MT5持仓 vs SQLite 状态是否同步（孤儿/缺失持仓）

自愈动作：
  - DAEMON_DEAD/STALE  → 重启 daemon
  - ORPHAN_POSITION     → 自动清理 SQLite 孤儿状态
  - EA_STALE            → 自动 ping EA，失败则重启 daemon（自动修复 Bridge 通信中断）
"""

import sys
import os
import time
import json
import sqlite3
import pathlib
import subprocess
from datetime import datetime

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
from src.subprocess_util import popen as _safe_popen

# ROOT now resolved via frozen_path (PyInstaller-safe)
from frozen_path import APP_ROOT as ROOT

# ── Frozen-mode detection ────────────────────────────────────────────────────
_FROZEN = getattr(sys, 'frozen', False)

# ── Resolve app/logs dir (cross-platform, frozen-aware) ─────────────────────
if _FROZEN:
    try:
        from config_manager import get_app_dir as _get_app_dir
        _APP_DIR = pathlib.Path(_get_app_dir())
    except Exception:
        _APP_DIR = ROOT
else:
    _APP_DIR = ROOT

# env var overrides for custom deployments
_env_logs = os.environ.get("GOLD_AUTOTRADE_LOGS_DIR", "").strip()
_env_mt5 = os.environ.get("EA_BRIDGE_FILES_DIR", "").strip()

LOGS_DIR = pathlib.Path(_env_logs) if _env_logs else (_APP_DIR / "logs")
LOGS_DIR.mkdir(parents=True, exist_ok=True)

DAEMON_LOG = LOGS_DIR / "gold_trader.daemon.log"
KEEP_ALIVE_LOG = LOGS_DIR / "keep_alive.log"
PID_FILE = LOGS_DIR / "gold_trader.pid"
DB_PATH = _APP_DIR / "data" / "bridge" / "state" / "agent_state.db"

# Bridge 目录通过 MT5 dir 动态计算（跨平台）
if _env_mt5:
    _MT5_BASE = pathlib.Path(_env_mt5)
elif _FROZEN:
    try:
        from config_manager import get_mt5_dir as _get_mt5_dir
        _MT5_BASE = pathlib.Path(_get_mt5_dir())
    except Exception:
        _MT5_BASE = ROOT / "bridge" / "Files"
else:
    try:
        from config_manager import get_mt5_dir as _get_mt5_dir
        _MT5_BASE = pathlib.Path(_get_mt5_dir())
    except Exception:
        _MT5_BASE = ROOT / "bridge" / "Files"

TERMINAL_STATUS = _MT5_BASE / "gold_autotrade" / "bridge" / "terminal_status.json"
BRIDGE_TO_EA = _MT5_BASE / "gold_autotrade" / "bridge" / "to_ea"
BRIDGE_FROM_EA = _MT5_BASE / "gold_autotrade" / "bridge" / "from_ea"
MARKET_SNAPSHOT = _MT5_BASE / "gold_autotrade" / "bridge" / "market_snapshot.json"
PYTHON = sys.executable  # 使用当前运行的 Python（保持一致）


def log(msg: str):
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    line = f"[{ts}] {msg}"
    print(line)
    try:
        KEEP_ALIVE_LOG.parent.mkdir(parents=True, exist_ok=True)
        with open(KEEP_ALIVE_LOG, "a", encoding="utf-8") as f:
            f.write(line + "\n")
    except Exception:
        pass


def check_daemon() -> str:
    """检查 daemon 进程是否存活"""
    if not PID_FILE.exists():
        return "DAEMON_DEAD:no_pid_file"
    try:
        pid = int(PID_FILE.read_text(encoding="utf-8").strip())
        os.kill(pid, 0)
        return "OK"
    except (ValueError, OSError):
        return "DAEMON_DEAD"


def check_daemon_stale() -> str:
    """检查 daemon 日志是否在5分钟内更新"""
    if not DAEMON_LOG.exists():
        return "DAEMON_STALE:no_log"
    age = time.time() - DAEMON_LOG.stat().st_mtime
    if age > 300:
        return f"DAEMON_STALE:{age:.0f}s"
    return "OK"


def check_ea_health() -> str:
    """检查 EA 是否活跃（terminal_status.json 120秒内更新）"""
    if not TERMINAL_STATUS.exists():
        return "EA_NO_STATUS"
    age = time.time() - TERMINAL_STATUS.stat().st_mtime
    if age > 120:
        return f"EA_STALE:{age:.0f}s"
    return "OK"


def check_market_snapshot_freshness(max_age_sec: float = 300.0) -> str:
    """检查 market_snapshot.json 是否在 max_age_sec 秒内更新（防止交易引擎使用过期行情）"""
    if not MARKET_SNAPSHOT.exists():
        return "MARKET_SNAPSHOT_MISSING"
    age = time.time() - MARKET_SNAPSHOT.stat().st_mtime
    if age > max_age_sec:
        return f"MARKET_SNAPSHOT_STALE:{age:.0f}s"
    return "OK"


def check_position_sync() -> str:
    """
    检查 MT5 持仓与 SQLite 状态是否同步。

    保守策略：
    - terminal_status.json 如果 EA 刚重启或过时，不执行自动清理
    - 只有在 EA 文件"足够新鲜"（<60s）时才信任其持仓数据
    - 防止 EA Deinit 期间清掉本还有效的持仓状态
    """
    try:
        # 检查 terminal_status.json 是否新鲜（EA 是否在活跃运行）
        ea_stale_age = None
        if TERMINAL_STATUS.exists():
            ea_age = time.time() - TERMINAL_STATUS.stat().st_mtime
            if ea_age > 60:
                return f"EA_DATA_STALE:{ea_age:.0f}s:skip_auto_heal"
            mt5_data = json.loads(TERMINAL_STATUS.read_text(encoding="utf-8"))
            mt5_tickets = {p["ticket"] for p in mt5_data.get("positions", [])}
        else:
            return "EA_NO_STATUS:skip_auto_heal"

        conn = sqlite3.connect(str(DB_PATH), timeout=10.0)
        conn.execute("PRAGMA journal_mode=WAL")
        c = conn.cursor()
        c.execute("SELECT value FROM state_kv WHERE key='position_ticket'")
        row = c.fetchone()
        db_ticket = int(row[0]) if row and row[0] else 0
        conn.close()

        if db_ticket and db_ticket not in mt5_tickets:
            return f"ORPHAN_POSITION:db={db_ticket} mt5={list(mt5_tickets)}"
        elif mt5_tickets and not db_ticket:
            return f"MISSING_POSITION:mt5={list(mt5_tickets)}"
        return "OK"
    except Exception as e:
        return f"SYNC_CHECK_ERROR:{e}"


def check_learning_integrity() -> str:
    """
    检查 learning 事件链是否存在悬空闭环。

    目标不是阻断交易，而是尽早发现：
    - execution 已发生但 closure 缺失
    - transition 长期卡在 decision-only
    - orphan execution / orphan closure
    """
    try:
        sys.path.insert(0, str(ROOT))
        from src.learning.transition_audit import run_repair_cycle

        summary = run_repair_cycle(auto_recover=True).get("audit", {})
        anomalies = summary.get("anomalies", {})
        total = int(anomalies.get("total", 0) or 0)
        critical = int(anomalies.get("critical", 0) or 0)
        coverage = summary.get("coverage", {})
        closure_coverage = float(coverage.get("closure_coverage_pct", 100.0) or 0.0)

        if critical > 0:
            return f"LEARNING_GAP:critical={critical}:closure={closure_coverage:.2f}%"
        if total > 0:
            return f"LEARNING_WARN:anomalies={total}:closure={closure_coverage:.2f}%"
        return "OK"
    except Exception as e:
        return f"LEARNING_AUDIT_ERROR:{e}"


def check_training_gate() -> str:
    """
    训练门槛检查。

    目标：
    - learner 不得在高风险脏样本阶段进入训练
    - watchdog 明确给出 TRAINING_READY / TRAINING_BLOCKED
    """
    try:
        sys.path.insert(0, str(ROOT))
        from src.learning.transition_builder import materialize_transitions
        from src.learning.transition_audit import run_repair_cycle
        from src.learning.export_replay_dataset import summarize_transitions, evaluate_training_gate

        transitions = materialize_transitions()
        summary = summarize_transitions(transitions)
        audit = run_repair_cycle(auto_recover=True).get("audit", {})
        gate = evaluate_training_gate(summary, audit)
        if gate.get("ready_for_training", False):
            return (
                "TRAINING_READY:"
                f"closure={float(gate.get('closure_coverage_pct', 0.0)):.2f}%:"
                f"recovered={float(gate.get('recovered_closed_ratio_pct', 0.0)):.2f}%"
            )
        reasons = ",".join(gate.get("blocked_reasons", [])) or "unknown"
        return (
            "TRAINING_BLOCKED:"
            f"closure={float(gate.get('closure_coverage_pct', 0.0)):.2f}%:"
            f"critical={int(gate.get('critical_anomalies', 0))}:"
            f"recovered={float(gate.get('recovered_closed_ratio_pct', 0.0)):.2f}%:"
            f"reasons={reasons}"
        )
    except Exception as e:
        return f"TRAINING_GATE_ERROR:{e}"


def heal_learning_integrity() -> str:
    try:
        sys.path.insert(0, str(ROOT))
        from src.learning.transition_audit import run_repair_cycle

        cycle = run_repair_cycle(stale_minutes=15, auto_recover=True)
        recovery = cycle.get("recovery", {})
        refreshed = cycle.get("audit", {})
        recovered = int(recovery.get("recovered", 0) or 0)
        remaining_critical = int(refreshed.get("anomalies", {}).get("critical", 0) or 0)
        total_anomalies = int(refreshed.get("anomalies", {}).get("total", 0) or 0)
        log(
            f"[heal_learning_integrity] recovered={recovered} "
            f"remaining_critical={remaining_critical} total_anomalies={total_anomalies}"
        )
        if recovered > 0:
            return "RECOVERED"
        return "NO_ACTION"
    except Exception as e:
        log(f"[heal_learning_integrity] failed: {e}")
        return f"ERROR:{e}"


def heal_restart_daemon():
    """自愈：重启 daemon (with lock to prevent concurrent restarts)"""
    heal_lock_file = LOGS_DIR / "heal_restart.lock"
    _heal_fd = None
    try:
        from src.flock_compat import flock, LOCK_EX, LOCK_NB, LOCK_UN
        _heal_fd = os.open(str(heal_lock_file), os.O_CREAT | os.O_RDWR)
        try:
            flock(_heal_fd, LOCK_EX | LOCK_NB)
        except (BlockingIOError, OSError):
            os.close(_heal_fd)
            _heal_fd = None
            log("heal_restart_daemon: another watchdog is already restarting, skipping")
            return
    except Exception:
        _heal_fd = None  # fallback: proceed without lock

    try:
        log("执行自愈: 重启 daemon")
        # 清理锁文件
        lock_file = LOGS_DIR / "gold_trader.lock"
        if lock_file.exists():
            lock_file.unlink()
            log("清理了 stale lock 文件")
        # 杀掉旧进程
        if PID_FILE.exists():
            try:
                old_pid = int(PID_FILE.read_text(encoding="utf-8").strip())
                os.kill(old_pid, 9)
                time.sleep(1)
            except (ValueError, OSError):
                pass
        # 重启
        daemon_log = ROOT / "logs" / "gold_trader.daemon.log"
        _safe_popen(
            [PYTHON, str(ROOT / "bridge" / "gold_trader.py"), "--daemon"],
            stdout=open(str(DAEMON_LOG), "a", encoding="utf-8"),
            stderr=subprocess.DEVNULL,
            cwd=str(ROOT / "bridge"),
        )
        time.sleep(3)
        if PID_FILE.exists():
            new_pid = PID_FILE.read_text(encoding="utf-8").strip()
            log(f"daemon 重启成功，PID={new_pid}")
        else:
            log("[WARN]️ daemon 重启后 PID 文件未生成，请检查")
    finally:
        if _heal_fd is not None:
            try:
                flock(_heal_fd, LOCK_UN)
                os.close(_heal_fd)
            except Exception:
                pass


def heal_clear_orphan():
    """自愈：清理 SQLite 孤儿持仓状态"""
    log("执行自愈: 清理孤儿持仓状态")
    try:
        conn = sqlite3.connect(str(DB_PATH), timeout=10.0)
        conn.execute("PRAGMA journal_mode=WAL")
        c = conn.cursor()
        c.execute(
            "DELETE FROM state_kv WHERE key IN "
            "('position_open','position_side','position_ticket',"
            "'position_entry','position_price','position_sl',"
            "'position_tp','position_vol')"
        )
        conn.commit()
        conn.close()
        log("孤儿持仓状态已从 SQLite 清除，系统将在下一轮自动恢复信号轮询")
    except Exception as e:
        log(f"清理孤儿持仓失败: {e}")


def heal_ping_ea(timeout=10) -> bool:
    """自愈：向 EA 发送 ping 请求，验证 Bridge 通信是否畅通"""
    import uuid
    sig_id = f"ping_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}"
    req_file = BRIDGE_TO_EA / f"{sig_id}.request.json"
    resp_file = BRIDGE_FROM_EA / f"{sig_id}.pong.json"

    try:
        BRIDGE_TO_EA.mkdir(parents=True, exist_ok=True)
        req = {
            "signal_id": sig_id,
            "action": "PING",
            "symbol": "XAUUSD",
            "timeframe": "M5",
            "count": 1,
        }
        req_file.write_text(json.dumps(req))
    except Exception as e:
        log(f"[heal_ping_ea] 写入 ping 请求失败: {e}")
        return False

    for _ in range(timeout * 10):
        try:
            if resp_file.exists():
                raw = resp_file.read_text(encoding="utf-8")
                data = json.loads(raw)
                if data.get("status") == "pong" or data.get("terminal"):
                    resp_file.unlink(missing_ok=True)
                    req_file.unlink(missing_ok=True)
                    log(f"[heal_ping_ea] EA 响应正常 [OK]")
                    return True
        except Exception:
            pass
        time.sleep(0.1)

    # 超时：清理并判定失败
    try:
        req_file.unlink(missing_ok=True)
        resp_file.unlink(missing_ok=True)
    except Exception:
        pass
    log(f"[heal_ping_ea] EA 无响应（{timeout}s），将触发重启")
    return False


def main():
    log("=== 主动巡查开始 ===")

    results = {
        "daemon": check_daemon(),
        "daemon_stale": check_daemon_stale(),
        "ea_health": check_ea_health(),
        "market_snapshot": check_market_snapshot_freshness(),
        "position_sync": check_position_sync(),
        "learning_integrity": check_learning_integrity(),
        "training_gate": check_training_gate(),
    }

    for key, val in results.items():
        log(f"  {key}: {val}")

    # 决策
    actions = []
    for key, val in results.items():
        if val != "OK":
            if key in ("daemon", "daemon_stale"):
                actions.append("RESTART_DAEMON")
            elif key == "position_sync" and "ORPHAN" in val:
                actions.append("CLEAR_ORPHAN")
            elif key == "ea_health":
                actions.append("RESTART_EA")
            elif key == "market_snapshot":
                actions.append("ALERT_MARKET_SNAPSHOT")
            elif key == "learning_integrity" and str(val).startswith("LEARNING_"):
                actions.append("RECOVER_LEARNING")
            elif key == "training_gate" and not str(val).startswith("TRAINING_READY"):
                actions.append("BLOCK_TRAINING")

    if not actions:
        log("状态: 健康 [OK]")
        log("=== 巡查结束 ===")
        return

    # 执行自愈
    if "RESTART_DAEMON" in actions:
        heal_restart_daemon()
    if "CLEAR_ORPHAN" in actions:
        heal_clear_orphan()
    if "RESTART_EA" in actions:
        if heal_ping_ea(timeout=10):
            log("EA ping 成功，Bridge 通信正常，无需重启")
        else:
            log("[WARN]️ EA 无响应，执行自动修复：重启 daemon")
            heal_restart_daemon()
    if "ALERT_MARKET_SNAPSHOT" in actions:
        val = results.get("market_snapshot", "")
        log(f"[WARN]️ 行情快照异常: {val}，尝试 ping EA 诊断通信链路")
        if heal_ping_ea(timeout=10):
            log("EA ping 成功，行情快照异常可能由 EA 侧数据源引起，保持观察")
        else:
            log("[WARN]️ EA 无响应 + 行情快照异常，执行自动修复：重启 daemon")
            heal_restart_daemon()
    if "RECOVER_LEARNING" in actions:
        heal_learning_integrity()
    if "BLOCK_TRAINING" in actions:
        log("[WARN]️ 训练门槛未满足，当前禁止进入 learner 训练阶段")

    log("=== 巡查结束 ===")


if __name__ == "__main__":
    main()
