PingWatch/backend/app/services/pinger.py

"""
异步 Ping 引擎

核心逻辑：
1. 每轮从数据库加载所有启用设备，批量 fping
2. 记录每台设备本轮 ping 结果（存活/响应时间）
3. 状态机管理设备状态，判断是否从 online→offline 或 offline→online
4. 结果通过回调或队列通知 alerter
"""

import asyncio
import subprocess
import time
import logging
from datetime import datetime
from typing import Optional, Callable, Awaitable
from collections import defaultdict

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.models.device import Device, DeviceTypeEnum
from app.models.ping_record import PingRecord

logger = logging.getLogger("pingwatch.pinger")


class PingResult:
    """单台设备一轮 ping 的结果"""
    def __init__(self, device_id: int, is_alive: bool, response_time_ms: Optional[float] = None):
        self.device_id = device_id
        self.is_alive = is_alive
        self.response_time_ms = response_time_ms


class DeviceStateChange:
    """设备状态变化事件"""
    def __init__(self, device: Device, old_status: str, new_status: str, consecutive_failures: int):
        self.device = device
        self.old_status = old_status
        self.new_status = new_status
        self.consecutive_failures = consecutive_failures


class Pinger:
    """
    Ping 引擎，使用 fping 批量并发检测。
    对所有设备进 ping，返回存活状态和响应时间。
    """

    def __init__(self):
        self._round_num = 0
        self._on_state_change: Optional[Callable[[DeviceStateChange], Awaitable[None]]] = None

    def on_state_change(self, callback: Callable[[DeviceStateChange], Awaitable[None]]):
        """注册状态变化回调"""
        self._on_state_change = callback

    async def run_one_round(self, db: AsyncSession) -> list[PingResult]:
        """
        执行一轮 ping 检测：
        1. 加载所有启用设备
        2. 批量 fping
        3. 记录结果
        4. 更新设备状态
        """
        self._round_num += 1
        round_num = self._round_num

        # 1. 加载启用设备
        result = await db.execute(
            select(Device).where(Device.is_enabled == True)
        )
        devices = list(result.scalars().all())

        if not devices:
            logger.info(f"[Round {round_num}] 没有启用的设备")
            return []

        logger.info(f"[Round {round_num}] 开始检测 {len(devices)} 台设备")

        # 2. 批量 ping
        start_time = time.time()
        ip_to_device = {d.ip: d for d in devices}
        ip_list = list(ip_to_device.keys())

        ping_results_map = await self._batch_ping(ip_list)

        # 3. 构造结果
        results: list[PingResult] = []
        for ip, dev in ip_to_device.items():
            is_alive, rtt = ping_results_map.get(ip, (False, None))
            results.append(PingResult(device_id=dev.id, is_alive=is_alive, response_time_ms=rtt))

        elapsed = time.time() - start_time
        alive_count = sum(1 for r in results if r.is_alive)

        # 4. 批量写入 ping_records
        now = datetime.now()
        records = [
            PingRecord(
                device_id=r.device_id,
                is_alive=r.is_alive,
                response_time_ms=r.response_time_ms,
                round_num=round_num,
                created_at=now,
            )
            for r in results
        ]
        db.add_all(records)
        await db.flush()

        # 5. 更新设备状态（状态机）
        device_map = {d.id: d for d in devices}
        for r in results:
            dev = device_map.get(r.device_id)
            if not dev:
                continue

            old_status = dev.current_status
            if r.is_alive:
                dev.consecutive_failures = 0
                dev.last_ping_time = now
                dev.last_online_time = now
                dev.current_status = "online"
            else:
                dev.consecutive_failures = (dev.consecutive_failures or 0) + 1
                dev.last_ping_time = now
                if dev.consecutive_failures >= dev.alert_threshold:
                    if dev.current_status != "offline":
                        dev.current_status = "offline"
                        dev.last_offline_time = now
                else:
                    if dev.current_status == "online":
                        dev.current_status = "checking"

            # 状态变化回调
            if old_status != dev.current_status and self._on_state_change:
                change = DeviceStateChange(
                    device=dev,
                    old_status=old_status,
                    new_status=dev.current_status,
                    consecutive_failures=dev.consecutive_failures,
                )
                await self._on_state_change(change)

        await db.commit()

        logger.info(
            f"[Round {round_num}] 完成: {alive_count}/{len(devices)} 在线, "
            f"耗时 {elapsed:.2f}s"
        )
        return results

    async def _batch_ping(self, ip_list: list[str]) -> dict[str, tuple[bool, Optional[float]]]:
        """
        使用 fping 批量 ping
        返回: { ip: (is_alive, response_time_ms) }
        """
        if not ip_list:
            return {}

        try:
            # fping 一次性 ping 多个 IP
            # -c 1: 每个 IP 发 1 个包
            # -t: 超时毫秒
            timeout_ms = int(settings.PING_TIMEOUT_SECONDS * 1000)
            cmd = [
                settings.FPING_PATH,
                "-c", "1",
                "-t", str(timeout_ms),
                "-e",  # 显示响应时间
            ] + ip_list

            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            stdout, stderr = await proc.communicate()

            result_map: dict[str, tuple[bool, Optional[float]]] = {}

            # fping 标准输出逐行: "IP : xmt/rcv/%loss = 1/1/0%, rtt min/avg/max = 0.12/0.12/0.12"
            # 或 "IP : xmt/rcv/%loss = 1/0/100%"
            for line in stdout.decode("utf-8", errors="replace").splitlines():
                line = line.strip()
                if ":" not in line:
                    continue
                ip = line.split(":")[0].strip()
                # 解析响应时间
                if "rtt" in line:
                    try:
                        # 提取 avg rtt
                        rtt_part = line.split("rtt")[1]
                        # 格式: min/avg/max = 0.12/0.12/0.12
                        if "=" in rtt_part:
                            avg_rtt_str = rtt_part.split("=")[1].strip().split("/")[1]
                            rtt_ms = float(avg_rtt_str)
                        else:
                            rtt_ms = None
                    except (IndexError, ValueError):
                        rtt_ms = None
                    result_map[ip] = (True, rtt_ms)
                else:
                    result_map[ip] = (False, None)

            return result_map

        except FileNotFoundError:
            logger.warning("fping 未找到，回退到系统 ping (串行)")
            return await self._fallback_ping(ip_list)
        except Exception as e:
            logger.error(f"fping 异常: {e}")
            return await self._fallback_ping(ip_list)

    async def _fallback_ping(self, ip_list: list[str]) -> dict[str, tuple[bool, Optional[float]]]:
        """回退方案：使用系统 ping，并发执行"""
        async def ping_one(ip: str) -> tuple[str, bool, Optional[float]]:
            try:
                timeout = settings.PING_TIMEOUT_SECONDS
                cmd = ["ping", "-c", "1", "-W", str(int(timeout)), ip]
                start = time.time()
                proc = await asyncio.create_subprocess_exec(
                    *cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                )
                await proc.wait()
                elapsed = (time.time() - start) * 1000
                return ip, proc.returncode == 0, round(elapsed, 2)
            except Exception:
                return ip, False, None

        tasks = [ping_one(ip) for ip in ip_list]
        sem = asyncio.Semaphore(settings.PING_CONCURRENCY)

        async def bounded_ping(ip: str):
            async with sem:
                return await ping_one(ip)

        results = await asyncio.gather(*[bounded_ping(ip) for ip in ip_list])
        return {ip: (alive, rtt) for ip, alive, rtt in results}