#!/usr/bin/env python3
"""verify_reassess_pipeline.py — 重评推送管道审计 + 全局cron失败监控

检查：
1. price_monitor 每2分正常跑
2. zone breach检测正常
3. holding_strategies有数据
4. XMPP bridge在线
5. reassess模块可导入
6. 【新增】所有关键cron job状态（是否有failed）

输出：正常时 [SILENT]，有异常时推XMPP
"""
import json, os, sys, subprocess, sqlite3
from pathlib import Path
from datetime import datetime, timedelta
from urllib.request import Request, urlopen

# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
try:
    from messenger import install_stdio_hook as _msh
    _msh()
except Exception:
    pass

BASE = Path(__file__).parent.parent
sys.path.insert(0, str(BASE))
sys.path.insert(0, "/home/hmo/MoFin")

XMPP_BRIDGE = "http://127.0.0.1:5805/"
XMPP_USER = "hmo@yoin.fun"

def xmpp_push(text):
    try:
        payload = json.dumps({"to": XMPP_USER, "body": text, "type": "chat"}).encode()
        req = Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"})
        urlopen(req, timeout=5)
    except Exception as e:
        print(f"[XMPP推送失败] {e}", file=sys.stderr)

def scan_cron_failures():
    """扫描两个cron jobs.json看是否有failed状态的关键job"""
    failures = []
    jobs_files = [
        "/home/hmo/.hermes/cron/jobs.json",
        "/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json",
    ]
    for jf in jobs_files:
        try:
            data = json.load(open(jf))
            for job in data.get("jobs", []):
                jid = job.get("id", "?")
                name = job.get("name", "") or jid[:12]
                status = job.get("last_status", "")
                enabled = job.get("enabled", True)
                if not enabled:
                    continue
                # 关键job：价格监控、重评、盘前中监控
                key_job = any(kw in name.lower() for kw in [
                    "price_monitor", "monitor", "盘前中", "reassess",
                    "重评", "自选买入", "stale_push", "管道审计",
                    "宏观风险", "策略时效"
                ])
                if not key_job:
                    continue
                if status == "failed":
                    last_run = job.get("last_run_at", "?")
                    failures.append(f"  ❌ {name} ({jid[:8]}) last_run={last_run}")
        except Exception:
            pass
    return failures

def check_cron_jobs():
    """另法：直接查cron数据库"""
    issues = []
    for db_path in [
        BASE / "cron" / "cron.db",
        Path("/home/hmo/.hermes/cron/cron.db"),
    ]:
        if not db_path.exists():
            continue
        try:
            c = sqlite3.connect(str(db_path))
            for row in c.execute("""
                SELECT id, name, last_status, last_run_at, enabled
                FROM cron_jobs WHERE enabled=1
                ORDER BY last_run_at DESC
            """).fetchall():
                jid, name, status, last_run, enabled = row
                if status == "failed":
                    issues.append(f"  ❌ {name}({jid[:8]}) last_run={last_run}")
            c.close()
        except Exception:
            pass
    return issues

def run():
    ok = True
    alerts = []
    checks = []

    # 1. price_monitor 最近运行时间
    try:
        # 开盘 grace：price_monitor 09:00 才启动，09:15 前隔夜数据属正常，不误报
        from datetime import datetime as _dt
        _now = _dt.now()
        _morning_grace = _now.weekday() < 5 and _now.hour == 9 and _now.minute < 15
        conn = None
        last_err = None
        # malformed 可能是 I/O 风暴下的瞬态 WAL 损坏（2026-07-21 事件）：
        # checkpoint 后自愈。重试一次再告警，避免误报轰炸
        for _attempt in range(2):
            try:
                conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db")
                conn.execute("SELECT 1 FROM live_prices LIMIT 1").fetchone()
                break
            except Exception as e:
                last_err = e
                import time as _t
                _t.sleep(3)
        if conn is None:
            raise last_err
        lp = conn.execute("SELECT MAX(updated_at) FROM live_prices").fetchone()[0]
        if lp:
            lp_dt = datetime.fromisoformat(lp) if isinstance(lp, str) else lp
            if hasattr(lp_dt, 'tzinfo') and lp_dt.tzinfo is None:
                if isinstance(lp, str) and '+' not in lp:
                    lp_dt = lp_dt.replace(tzinfo=None)
            mins_ago = (datetime.now() - lp_dt).total_seconds() / 60
            status = "ok" if mins_ago < 10 else "warn"
            if mins_ago > 15 and not _morning_grace:
                status = "fail"
                ok = False
                alerts.append(f"price_monitor {mins_ago:.0f}分未更新")
            checks.append({"check":"price_monitor","status":status,"detail":f"最后更新{mins_ago:.0f}分前" + ("(开盘grace)" if _morning_grace and mins_ago > 15 else "")})
        else:
            checks.append({"check":"price_monitor","status":"warn","detail":"live_prices无数据"})
    except Exception as e:
        checks.append({"check":"price_monitor","status":"fail","detail":str(e)})
        ok = False
        alerts.append(f"price_monitor异常: {e}")

    # 2. 策略评估活动（reassess_with_context写strategy_evaluations，不是holding_strategies）
    try:
        today_se = conn.execute("SELECT COUNT(*) FROM strategy_evaluations WHERE date(created_at)=date('now')").fetchone()[0]
        total_se = conn.execute("SELECT COUNT(*) FROM strategy_evaluations").fetchone()[0]
        # 也尝试查holding_strategies（如果存在并有数据）
        hs_exists = conn.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='holding_strategies'").fetchone()[0]
        hs = 0
        if hs_exists:
            hs = conn.execute("SELECT COUNT(*) FROM holding_strategies").fetchone()[0]
        detail = f"今日{today_se}次评估, 累计{total_se}条"
        if hs > 0:
            detail += f", holding_strategies{hs}条"
        checks.append({"check":"strategy_activity","status":"ok","detail":detail})
    except Exception as e:
        checks.append({"check":"strategies","status":"fail","detail":str(e)})
        ok = False

    # 3. XMPP bridge 是否在线（TCP端口检测，不发消息到Dad）
    try:
        import socket
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        result = sock.connect_ex(("127.0.0.1", 5805))
        sock.close()
        bridge_ok = (result == 0)
        if not bridge_ok:
            ok = False
            alerts.append("XMPP bridge(5805)端口无响应")
        checks.append({"check":"xmpp_bridge","status":"ok" if bridge_ok else "fail","detail":"在线" if bridge_ok else "端口无响应"})
    except Exception as e:
        checks.append({"check":"xmpp_bridge","status":"fail","detail":str(e)})
        ok = False
        alerts.append(f"XMPP bridge不可达: {e}")

    # 4. reassess模块可导入
    try:
        from strategy_lifecycle import reassess_with_context
        checks.append({"check":"reassess_module","status":"ok","detail":"可导入"})
    except Exception as e:
        checks.append({"check":"reassess_module","status":"fail","detail":str(e)})
        ok = False
        alerts.append(f"reassess模块导入失败: {e}")

    # 5. cron job失败检测
    cron_issues = scan_cron_failures() + check_cron_jobs()
    if cron_issues:
        ok = False
        alerts.append(f"{len(cron_issues)}个cron job失败")
        for issue in cron_issues[:5]:
            alerts.append(issue)
        checks.append({"check":"cron_jobs","status":"fail","detail":"; ".join(cron_issues[:3])})
    else:
        checks.append({"check":"cron_jobs","status":"ok","detail":"所有关键job正常"})

    conn.close()

    # 输出
    result = {
        "pipeline": "ok" if ok else "degraded",
        "checked_at": datetime.now().isoformat(),
        "checks": checks,
        "alerts": alerts
    }

    if ok:
        print("[SILENT]")
    else:
        msg = "🔴 重评管道异常:\n" + "\n".join(alerts)
        print(json.dumps(result, ensure_ascii=False, indent=2))
        # 有异常时主动推XMPP（取代静默）
        xmpp_push(msg)
        print(f"\n已推送XMPP: {len(alerts)}条告警", file=sys.stderr)

if __name__ == "__main__":
    run()