personal-experiment/ping_executor.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Ping执行器模块
提供不同操作系统的Ping实现策略
"""

import subprocess
import platform
from abc import ABC, abstractmethod
from typing import Tuple, Optional
from dataclasses import dataclass


@dataclass
class PingResult:
    """Ping结果数据类"""
    success: bool
    latency_ms: float
    error_message: Optional[str] = None
    
    def __str__(self) -> str:
        if self.success:
            return f"✓ 成功 (延迟: {self.latency_ms:.2f}ms)"
        else:
            error = self.error_message or "未知错误"
            return f"✗ 失败 ({error})"


class PingExecutor(ABC):
    """Ping执行器抽象基类（策略模式）"""
    
    @abstractmethod
    def execute(self, target: str, timeout: float = 5.0) -> PingResult:
        """
        执行ping操作
        
        Args:
            target: 目标地址
            timeout: 超时时间（秒）
            
        Returns:
            Ping结果对象
        """
        pass
    
    @abstractmethod
    def get_command(self, target: str, timeout: float) -> list:
        """
        获取ping命令
        
        Args:
            target: 目标地址
            timeout: 超时时间（秒）
            
        Returns:
            命令列表
        """
        pass


class WindowsPingExecutor(PingExecutor):
    """Windows平台Ping执行器"""
    
    def execute(self, target: str, timeout: float = 5.0) -> PingResult:
        """执行Windows ping命令"""
        command = self.get_command(target, timeout)
        
        try:
            import time
            start_time = time.time()
            
            result = subprocess.run(
                command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                timeout=timeout + 1
            )
            
            latency_ms = (time.time() - start_time) * 1000
            
            return PingResult(
                success=result.returncode == 0,
                latency_ms=latency_ms
            )
        
        except subprocess.TimeoutExpired:
            return PingResult(
                success=False,
                latency_ms=timeout * 1000,
                error_message="超时"
            )
        except Exception as e:
            return PingResult(
                success=False,
                latency_ms=0.0,
                error_message=str(e)
            )
    
    def get_command(self, target: str, timeout: float) -> list:
        """获取Windows ping命令"""
        timeout_ms = int(timeout * 1000)
        return ["ping", "-n", "1", "-w", str(timeout_ms), target]


class LinuxPingExecutor(PingExecutor):
    """Linux平台Ping执行器"""
    
    def execute(self, target: str, timeout: float = 5.0) -> PingResult:
        """执行Linux ping命令"""
        command = self.get_command(target, timeout)
        
        try:
            import time
            start_time = time.time()
            
            result = subprocess.run(
                command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                timeout=timeout + 1
            )
            
            latency_ms = (time.time() - start_time) * 1000
            
            return PingResult(
                success=result.returncode == 0,
                latency_ms=latency_ms
            )
        
        except subprocess.TimeoutExpired:
            return PingResult(
                success=False,
                latency_ms=timeout * 1000,
                error_message="超时"
            )
        except Exception as e:
            return PingResult(
                success=False,
                latency_ms=0.0,
                error_message=str(e)
            )
    
    def get_command(self, target: str, timeout: float) -> list:
        """获取Linux ping命令"""
        return ["ping", "-c", "1", "-W", str(int(timeout)), target]


class MacOSPingExecutor(PingExecutor):
    """macOS平台Ping执行器"""
    
    def execute(self, target: str, timeout: float = 5.0) -> PingResult:
        """执行macOS ping命令"""
        command = self.get_command(target, timeout)
        
        try:
            import time
            start_time = time.time()
            
            result = subprocess.run(
                command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                timeout=timeout + 1
            )
            
            latency_ms = (time.time() - start_time) * 1000
            
            return PingResult(
                success=result.returncode == 0,
                latency_ms=latency_ms
            )
        
        except subprocess.TimeoutExpired:
            return PingResult(
                success=False,
                latency_ms=timeout * 1000,
                error_message="超时"
            )
        except Exception as e:
            return PingResult(
                success=False,
                latency_ms=0.0,
                error_message=str(e)
            )
    
    def get_command(self, target: str, timeout: float) -> list:
        """获取macOS ping命令"""
        return ["ping", "-c", "1", "-W", str(int(timeout * 1000)), target]


class PingExecutorFactory:
    """Ping执行器工厂类（工厂模式）"""
    
    _executors: dict = {
        'windows': WindowsPingExecutor,
        'linux': LinuxPingExecutor,
        'darwin': MacOSPingExecutor
    }
    
    @classmethod
    def create(cls) -> PingExecutor:
        """
        根据当前操作系统创建Ping执行器
        
        Returns:
            适配当前系统的Ping执行器实例
            
        Raises:
            RuntimeError: 不支持的操作系统
        """
        os_name = platform.system().lower()
        
        executor_class = cls._executors.get(os_name)
        
        if executor_class is None:
            raise RuntimeError(f"不支持的操作系统: {os_name}")
        
        return executor_class()
    
    @classmethod
    def register_executor(cls, os_name: str, executor_class: type) -> None:
        """
        注册自定义Ping执行器
        
        Args:
            os_name: 操作系统名称
            executor_class: 执行器类
        """
        cls._executors[os_name] = executor_class