"""设备/系统健康日报(每日一次):内存/磁盘/GPU 使用率 + 系统级错误检查。

对应 device-check deployment,每天 09:00(可调)。检查项:
- 内存/swap、宿主机磁盘(经 /:/host:ro 挂载)、GPU(nvidia-smi)——使用率 ≥ 阈值(默认 80%)标 🔴
- Docker 容器状态(经 /var/run/docker.sock):应运行但退出 / restarting / unhealthy / OOMKilled
- Prefect 近 24h FAILED/CRASHED flow runs、OCR 服务探活、PG 连通性、系统负载

依赖 docker-compose 给 prefect-worker 挂载 /:/host:ro、/var/run/docker.sock、GPU;
未挂载时对应检查项自动降级为"不可用",不影响其余检查。
"""

from __future__ import annotations

import os
import subprocess
from datetime import datetime, timedelta

import httpx
import psutil
from prefect import flow

from flows.shared import SyncDB, emit_sync_summary, send_markdown

# (容器内路径, 展示名);/host 为 compose 挂载的宿主机根目录
DISK_PATHS = [("/host", "/"), ("/host/home", "/home")]
DOCKER_SOCK = "/var/run/docker.sock"

_OCR_BASE = "http://192.168.124.92:9801"


def _gb(nbytes: float) -> str:
    return f"{nbytes / (1 << 30):.1f}G"


def _check_memory() -> dict:
    vm = psutil.virtual_memory()
    sw = psutil.swap_memory()
    return {
        "pct": vm.percent, "used": _gb(vm.used), "total": _gb(vm.total),
        "swapPct": sw.percent, "swapUsed": _gb(sw.used), "swapTotal": _gb(sw.total),
    }


def _check_disks() -> list[dict]:
    paths = DISK_PATHS if os.path.isdir(DISK_PATHS[0][0]) else [("/", "/")]
    out = []
    for raw, label in paths:
        try:
            du = psutil.disk_usage(raw)
            out.append({"label": label, "pct": du.percent, "used": _gb(du.used), "total": _gb(du.total)})
        except Exception as err:  # noqa: BLE001
            out.append({"label": label, "error": str(err)[:80]})
    return out


def _check_gpus() -> list[dict] | str:
    """nvidia-smi 查 GPU 利用率与显存。容器无 GPU 访问时返回说明文字。"""
    try:
        proc = subprocess.run(
            ["nvidia-smi", "--query-gpu=index,name,utilization.gpu,memory.used,memory.total",
             "--format=csv,noheader,nounits"],
            capture_output=True, text=True, timeout=15,
        )
        if proc.returncode != 0:
            return f"nvidia-smi 退出码 {proc.returncode}: {proc.stderr.strip()[:100]}"
        gpus = []
        for line in proc.stdout.strip().splitlines():
            parts = [p.strip() for p in line.split(",")]
            if len(parts) != 5:
                continue
            idx, name, util, mem_used, mem_total = parts
            mem_pct = float(mem_used) / float(mem_total) * 100 if float(mem_total) > 0 else 0.0
            gpus.append({"name": f"GPU{idx} {name[:24]}", "utilPct": float(util), "memPct": mem_pct,
                         "memUsed": f"{float(mem_used)/1024:.1f}G", "memTotal": f"{float(mem_total)/1024:.1f}G"})
        return gpus or "未检测到 GPU"
    except FileNotFoundError:
        return "不可用(容器内无 nvidia-smi,需 compose 配置 GPU)"
    except Exception as err:  # noqa: BLE001
        return f"检查失败: {str(err)[:80]}"


def _inspect_container(client: httpx.Client, cid: str) -> dict:
    try:
        resp = client.get(f"http://localhost/containers/{cid}/json", timeout=10)
        resp.raise_for_status()
        return resp.json()
    except Exception:  # noqa: BLE001
        return {}


def _check_docker() -> dict:
    """容器状态汇总 + 异常容器列表(应运行但退出/restarting/unhealthy/OOMKilled)。"""
    if not os.path.exists(DOCKER_SOCK):
        return {"running": None, "issues": [{"name": "Docker", "issue": f"不可用(未挂载 {DOCKER_SOCK})"}]}
    with httpx.Client(transport=httpx.HTTPTransport(uds=DOCKER_SOCK), timeout=10) as client:
        resp = client.get("http://localhost/containers/json", params={"all": "1"})
        resp.raise_for_status()
        containers = resp.json()
        running = sum(1 for c in containers if c.get("State") == "running")
        issues = []
        stopped = 0
        for c in containers:
            name = (c.get("Names") or ["?"])[0].lstrip("/")
            state, status = c.get("State", ""), c.get("Status", "")
            if state == "restarting":
                issues.append({"name": name, "issue": f"反复重启 ({status})"})
            elif state == "running" and "(unhealthy)" in status:
                issues.append({"name": name, "issue": f"健康检查失败 ({status})"})
            elif state in ("exited", "dead"):
                detail = _inspect_container(client, c["Id"])
                policy = ((detail.get("HostConfig") or {}).get("RestartPolicy") or {}).get("Name", "no")
                oom = (detail.get("State") or {}).get("OOMKilled", False)
                finished_at = (detail.get("State") or {}).get("FinishedAt", "")
                if oom and _exited_recently(finished_at):
                    issues.append({"name": name, "issue": f"被 OOM 杀死 ({status})"})
                elif policy in ("always", "unless-stopped"):
                    if _exited_recently(finished_at):
                        issues.append({"name": name, "issue": f"应常驻但刚退出 ({status}, restart={policy})"})
                    else:
                        stopped += 1  # 数周前人为停用的旧服务,不算异常
        return {"running": running, "total": len(containers), "issues": issues, "stopped": stopped}


def _exited_recently(finished_at: str, hours: int = 24) -> bool:
    try:
        from datetime import timezone
        dt = datetime.fromisoformat(finished_at.replace("Z", "+00:00"))
        return (datetime.now(timezone.utc) - dt).total_seconds() < hours * 3600
    except (ValueError, TypeError):
        return True  # 解析失败按新退出处理,宁可多报


def check_prefect_failures(hours: int = 24) -> list[dict]:
    api_url = os.environ.get("PREFECT_API_URL", "http://localhost:4200/api")
    since = (datetime.now().astimezone() - timedelta(hours=hours)).isoformat()
    resp = httpx.post(
        f"{api_url}/flow_runs/filter",
        json={
            "flow_runs": {"state": {"type": {"any_": ["FAILED", "CRASHED"]}}, "start_time": {"after_": since}},
            "limit": 50,
            "sort": "START_TIME_DESC",
        },
        timeout=15,
    )
    resp.raise_for_status()
    out = []
    for r in resp.json():
        start = (r.get("start_time") or "")[:16].replace("T", " ")
        out.append({"name": r.get("name", "?"), "state": (r.get("state") or {}).get("type", "?"),
                    "message": ((r.get("state") or {}).get("message") or "").strip()[:100], "start": start})
    return out


def _check_services() -> dict:
    from api.ocr import OcrClient

    ocr_ok = False
    try:
        ocr_ok = OcrClient().health()
    except Exception:  # noqa: BLE001
        pass
    pg_ok = False
    try:
        pg_ok = bool(SyncDB().ping())
    except Exception:  # noqa: BLE001
        pass
    return {"ocr": ocr_ok, "pg": pg_ok}


def _mark(pct: float | None, threshold: int) -> str:
    if pct is None:
        return "—"
    return f"{pct:.0f}%" + (" 🔴" if pct >= threshold else "")


@flow(name="device-check", log_prints=True)
def device_check(alarm_threshold: int = 80) -> dict:
    """设备/系统健康日报:检查并直接发企微群。单项检查失败不影响其余项。"""
    checks: dict[str, object] = {}
    for name, fn in [
        ("memory", _check_memory), ("disks", _check_disks), ("gpus", _check_gpus),
        ("docker", _check_docker), ("prefectFailures", check_prefect_failures), ("services", _check_services),
    ]:
        try:
            checks[name] = fn()
        except Exception as err:  # noqa: BLE001
            checks[name] = None
            print(f"[device-check] {name} 检查失败: {err}")

    threshold = alarm_threshold
    lines = [f"🖥 **设备健康日报** {datetime.now().strftime('%Y-%m-%d %H:%M')}", ""]

    mem = checks.get("memory") or {}
    if mem:
        lines.append(f"**内存**: {_mark(mem['pct'], threshold)}({mem['used']}/{mem['total']})"
                     f" | swap {_mark(mem['swapPct'], threshold)}({mem['swapUsed']}/{mem['swapTotal']})")
    else:
        lines.append("**内存**: 检查失败")

    lines.append("**磁盘**:")
    lines.append("| 挂载点 | 使用率 | 已用/总量 |")
    lines.append("| --- | --- | --- |")
    for d in checks.get("disks") or []:
        if "error" in d:
            lines.append(f"| {d['label']} | 检查失败 | {d['error']} |")
        else:
            lines.append(f"| {d['label']} | {_mark(d['pct'], threshold)} | {d['used']}/{d['total']} |")

    lines.append("**GPU**:")
    gpus = checks.get("gpus")
    if isinstance(gpus, list) and gpus:
        lines.append("| GPU | 利用率 | 显存 |")
        lines.append("| --- | --- | --- |")
        for g in gpus:
            lines.append(f"| {g['name']} | {_mark(g['utilPct'], threshold)} | "
                         f"{_mark(g['memPct'], threshold)}({g['memUsed']}/{g['memTotal']}) |")
    else:
        lines.append(f"> {gpus if isinstance(gpus, str) else '未检测到'}")

    docker = checks.get("docker") or {}
    issues = docker.get("issues") or []
    if docker.get("running") is None:
        lines.append(f"**Docker**: {issues[0]['issue'] if issues else '不可用'}")
    else:
        lines.append(f"**Docker 容器**: 运行中 {docker['running']}/{docker.get('total', '?')}"
                     + (f",已停用 {docker['stopped']}" if docker.get("stopped") else "")
                     + (f",⚠️ 异常 {len(issues)} 个" if issues else ",全部正常"))
        for it in issues[:10]:
            lines.append(f"- 🔴 {it['name']}: {it['issue']}")

    failures = checks.get("prefectFailures") or []
    lines.append(f"**Prefect 近24h 失败**: {len(failures)} 个 flow run"
                 + ("" if failures else " ✅"))
    for f in failures[:8]:
        lines.append(f"- 🔴 {f['name']}({f['state']}) {f['start']} {f['message'][:60]}")

    svc = checks.get("services") or {}
    lines.append(f"**服务探活**: OCR {'✅' if svc.get('ocr') else '🔴'} | "
                 f"PG {'✅' if svc.get('pg') else '🔴'} | 负载 {os.getloadavg()[0]:.1f}")

    alarms = (
        (mem and (mem["pct"] >= threshold or mem["swapPct"] >= threshold))
        or any(d.get("pct", 0) >= threshold for d in (checks.get("disks") or []))
        or (isinstance(gpus, list) and any(g["utilPct"] >= threshold or g["memPct"] >= threshold for g in gpus))
        or bool(issues) or bool(failures) or not svc.get("ocr") or not svc.get("pg")
    )
    lines += ["", ("🔴 **有项目需要处理,见上**" if alarms else "✅ **一切正常**")]

    content = "\n".join(lines)
    sent = send_markdown(content)
    emit_sync_summary("设备健康日报", [
        ("内存", f"{_mark(mem['pct'], threshold)}" if mem else "—"),
        ("Docker 异常", len(issues)),
        ("Prefect 失败(24h)", len(failures)),
        ("OCR", "✅" if svc.get("ocr") else "🔴"), ("PG", "✅" if svc.get("pg") else "🔴"),
        ("已发企微", "是" if sent else "否"),
    ])
    return {"alarms": bool(alarms), "sent": sent, "dockerIssues": len(issues), "prefectFailures": len(failures)}


if __name__ == "__main__":
    device_check.serve(name="device-check", cron="0 9 * * *")