Python开发一个实用的网络连通性检查工具

目录

• 简介
• 功能介绍
• 应用场景
• 报错处理
• 代码实现
• 使用方法
• 安装依赖
• 基本使用
• 目标文件示例
• 命令行参数说明
• 使用示例
• 总结

网络连通性是现代计算机系统正常运行的基础。无论是访问互联网、连接内部网络还是与其他设备通信，网络连接的稳定性都至关重要。当网络出现问题时，快速诊断和定位故障点是解决问题的关键。本文将介绍一个实用的Python脚本——网络连通性检查工具，它可以全面检查网络连接状态，帮助用户快速诊断网络问题。

这个网络连通性检查工具具有以下核心功能：

• Ping测试：测试与目标主机的ICMP连通性
• 端口检查：检查指定主机的端口是否开放
• DNS解析测试：验证域名解析功能
• HTTP连通性测试：测试Web服务的可达性
• 多目标批量测试：支持同时测试多个目标
• 详细报告生成：提供详细的测试结果报告
• 定时监控：支持定期检查网络连通性
• 告警通知：在网络故障时发送告警通知

这个工具适用于以下场景：

• 网络故障排查：快速诊断网络连接问题
• 系统监控：监控关键网络服务的可用性
• 运维自动化：集成到运维流程中自动检查网络状态
• 网络性能测试：评估网络连接的质量和延迟
• 安全审计：检查网络服务的开放状态
• 灾难恢复：在网络恢复后验证连接状态

脚本包含了完善的错误处理机制：

• 超时处理：合理设置网络操作的超时时间
• 异常捕获：捕获并处理网络操作中的各种异常
• 权限验证：检查是否有足够的权限执行网络测试
• DNS解析保护：处理DNS解析失败的情况
• 连接重试：在网络不稳定时自动重试连接
• 资源清理：确保网络连接和资源得到正确释放

import os
import sys
import subprocess
import socket
import argparse
import json
import time
from datetime import datetime
import threading
import requests
from urllib.parse import urlparse

class NetworkConnectivityChecker:
    def __init__(self):
        self.results = []
        self.errors = []
        self.monitoring = False
        
    def ping_host(self, host, count=4, timeout=3):
        """Ping测试"""
        try:
            # 根据操作系统选择ping命令
            if os.name == 'nt':  # Windows
                cmd = ['ping', '-n', str(count), '-w', str(timeout * 1000), host]
            else:  # Linux/macOS
                cmd = ['ping', '-c', str(count), '-W', str(timeout), host]
                
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout * count + 5)
            
            if result.returncode == 0:
                # 解析ping输出获取统计信息
                output = result.stdout
                packet_loss = 0
                avg_rtt = 0
                
                if os.name == 'nt':  # Windows
                    # Windows ping输出解析
                    for line in output.split('n'):
                        if 'Lost' in line:
                            # 匹配 "Lost = X (X% loss)"
                            import re
                            match = re.search(r'((d+)% loss)', line)
                            if match:
                                packet_loss = int(match.group(1))
                        elif 'Average' in line:
                            # 匹配 "Average = Xms"
                            import re
                            match = re.search(r'Average = (d+)ms', line)
                            if match:
                                avg_rtt = int(match.group(1))
                else:  # Linux/macOS
                    # Unix-like系统ping输出解析
                    for line in output.split('n'):
                        if 'packet loss' in line:
                            # 匹配 "X packets transmitted, Y received, Z% packet loss"
                            import re
                            match = re.search(r'(d+)% packet loss', line)
                            if match:
                                packet_loss = int(match.group(1))
                        elif 'rtt' in line:
                            # 匹配 "rtt min/avg/max/mdev = X/Y/Z/W ms"
                            import re
                            match = re.search(r'rtt [^=]+= [^/]+/([^/]+)/', line)
                            if match:
                                avg_rtt = float(match.group(1))
                                
                return {
                    'host': host,
                    'status': 'reachable',
                    'packet_loss': packet_loss,
                    'avg_rtt': avg_rtt,
                    'output': output
                }
            else:
                return {
                    'host': host,
                    'status': 'unreachable',
                    'error': result.stderr or 'Host unreachable',
                    'output': result.stdout
                }
                
        except subprocess.TimeoutExpired:
            return {
                'host': host,
                'status': 'timeout',
                'error': f'Ping timeout after {timeout * count + 5} seconds'
            }
        except Exception as e:
            return {
                'host': host,
                'status': 'error',
                'error': str(e)
            }
            
    def check_port(self, host, port, timeout=3):
        """端口检查"""
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)
            
            result = sock.connect_ex((host, port))
            sock.close()
            
            if result == 0:
                return {
                    'host': host,
                    'port': port,
                    'status': 'open'
                }
            else:
                return {
                    'host': host,
                    'port': port,
                    'status': 'closed',
                    'error': f'Connection refused (code: {result})'
                }
                
        except socket.gaierror as e:
            return {
                'host': host,
                'port': port,
                'status': 'dns_error',
                'error': f'DNS resolution failed: {e}'
            }
        except Exception as e:
            return {
                'host': host,
                'port': port,
                'status': 'error',
                'error': str(e)
            }
            
    def resolve_dns(self, hostname):
        """DNS解析测试"""
        try:
            ip_addresses = socket.getaddrinfo(hostname, None)
            ips = []
            for addr_info in ip_addresses:
                ip = addr_info[4][0]
                if ip not in ips:
                    ips.append(ip)
                    
            return {
                'hostname': hostname,
                'status': 'resolved',
                'ip_addresses': ips
            }
        except socket.gaierror as e:
            return {
                'hostname': hostname,
                'status': 'unresolved',
                'error': str(e)
            }
        except Exception as e:
            return {
                'hostname': hostname,
                'status': 'error',
                'error': str(e)
            }
            
    def http_check(self, url, timeout=10):
        """HTTP连通性测试"""
        try:
            parsed_url = urlparse(url)
            if not parsed_url.scheme:
                url = 'http://' + url
                
            response = requests.get(url, timeout=timeout)
            
            return {
                'url': url,
                'status': 'accessible',
                'status_code': response.status_code,
                'response_time': response.elapsed.total_seconds(),
                'content_length': len(response.content)
            }
        except requests.exceptions.Timeout:
            return {
                'url': url,
                'status': 'timeout',
                'error': f'Request timeout after {timeout} seconds'
            }
        except requests.exceptions.ConnectionError as e:
            return {
                'url': url,
                'status': 'connection_error',
                'error': str(e)
            }
        except requests.exceptions.RequestException as e:
            return {
                'url': url,
                'status': 'error',
                'error': str(e)
            }
        except Exception as e:
            return {
                'url': url,
                'status': 'error',
                'error': str(e)
            }
            
    def check_target(self, target):
        """检查单个目标"""
        result = {
            'target': target,
            'timestamp': datetime.now().isoformat(),
            'tests': {}
        }
        
        # 判断目标类型
        if target.startswith('http://') or target.startswith('https://'):
            # HTTP测试
            result['tests']['http'] = self.http_check(target)
        elif ':' in target and not target.replace(':', '').replace('.', '').isdigit():
            # 主机名:端口格式
            host, port = target.rsplit(':', 1)
            if port.isdigit():
                result['tests']['ping'] = self.ping_host(host)
                result['tests']['port'] = self.check_port(host, int(port))
                result['tests']['dns'] = self.resolve_dns(host)
            else:
                result['tests']['ping'] = self.ping_host(target)
                result['tests']['dns'] = self.resolve_dns(target)
        elif target.replace('.', '').isdigit() or not target.replace('-', '').replace('.', '').isdigit():
            # IP地址或主机名
            result['tests']['ping'] = self.ping_host(target)
            result['tests']['dns'] = self.resolve_dns(target)
        else:
            # 默认进行ping测试
            result['tests']['ping'] = self.ping_host(target)
            
        return result
        
    def check_multiple_targets(self, targets):
        """检查多个目标"""
        print(f"开始检查 {len(targets)} 个目标")
        print("=" * 60)
        
        self.results = []
        
        for target in targets:
            print(f"检查目标: {target}")
            result = self.check_target(target)
            self.results.append(result)
            
            # 显示结果
            tests = result['tests']
            for test_type, test_result in tests.items():
                status_icons = {
                    'reachable': '✅',
                    'open': '✅',
                    'resolved': '✅',
                    'accessible': '✅',
                    'unreachable': '❌',
                    'closed': '❌',
                    'unresolved': '❌',
                    'timeout': '⏰',
                    'error': '⚠️',
                    'connection_error': '❌',
                    'dns_error': '❌'
                }
                
                icon = status_icons.get(test_result['status'], '❓')
                print(f"  {test_type.upper()}: {icon} {test_result['status']}")
                
                if 'error' in test_result:
                    print(f"    错误: {test_result['error']}")
                elif test_type == 'ping':
                    if 'packet_loss' in test_result:
                        print(f"    丢包率: {test_result['packet_loss']}%")
                    if 'avg_rtt' in test_result and test_result['avg_rtt'] > 0:
                        print(f"    平均延迟: {test_result['avg_rtt']}ms")
                elif test_type == 'port':
                    print(f"    端口: {test_result['port']}")
                elif test_type == 'http':
                    if 'status_code' in test_result:
                        print(f"    状态码: {test_result['status_code']}")
                    if 'response_time' in test_result:
                        print(f"    响应时间: {test_result['response_time']:.3f}s")
                elif test_type == 'dns':
                    if 'ip_addresses' in test_result:
                        print(f"    IP地址: {', '.join(test_result['ip_addresses'])}")
                        
            print()
            
        return self.results
        
    def generate_report(self):
        """生成检查报告"""
        report = []
        report.append("=" * 80)
        report.append("网络连通性检查报告")
        report.append("=" * 80)
        report.append(f"检查时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append(f"检查目标数: {len(self.results)}")
        report.append("")
        
        # 统计信息
        stats = {
            'reachable': 0,
            'unreachable': 0,
            'timeout': 0,
            'error': 0
        }
        
        for result in self.results:
            for test_result in result['tests'].values():
                status = test_result['status']
                if status in stats:
                    stats[status] += 1
                elif status not in ['open', 'resolved', 'accessible']:
                    stats['error'] += 1
                    
        report.append("状态统计:")
        report.append(f"  可达: {stats['reachable']}")
        report.append(f"  不可达: {stats['unreachable']}")
        report.append(f"  超时: {stats['timeout']}")
        report.append(f"  错误: {stats['error']}")
        report.append("")
        
        # 详细结果
        report.append("详细检查结果:")
        report.append("-" * 50)
        for result in self.results:
            report.append(f"目标: {result['target']}")
            for test_type, test_result in result['tests'].items():
                status_display = {
                    'reachable': '✅ 可达',
                    'open': '✅ 开放',
                    'resolved': '✅ 已解析',
                    'accessible': '✅ 可访问',
                    'unreachable': '❌ 不可达',
                    'closed': '❌ 关闭',
                    'unresolved': '❌ 未解析',
                    'timeout': '⏰ 超时',
                    'error': '⚠️ 错误',
                    'connection_error': '❌ 连接错误',
                    'dns_error': '❌ DNS错误'
                }
                
                status_text = status_display.get(test_result['status'], test_result['status'])
                report.append(f"  {test_type.upper()}: {status_text}")
                
                if 'error' in test_result:
                    report.append(f"    错误: {test_result['error']}")
                elif test_type == 'ping':
                    if 'packet_loss' in test_result:
                        report.append(f"    丢包率: {test_result['packet_loss']}%")
                    if 'avg_rtt' in test_result and test_result['avg_rtt'] > 0:
                        report.append(f"    平均延迟: {test_result['avg_rtt']}ms")
                elif test_type == 'port':
                    report.append(f"    端口: {test_result['port']}")
                elif test_type == 'http':
                    if 'status_code' in test_result:
                        report.append(f"    状态码: {test_result['status_code']}")
                    if 'response_time' in test_result:
                        report.append(f"    响应时间: {test_result['response_time']:.3f}s")
                elif test_type == 'dns':
                    if 'ip_addresses' in test_result:
                        report.append(f"    IP地址: {', '.join(test_result['ip_addresses'])}")
            report.append("")
            
        # 错误信息
        if self.errors:
            report.append("错误信息:")
            report.append("-" * 50)
            for error in self.errors[:10]:  # 只显示前10个错误
                report.append(f"  {error}")
            if len(self.errors) > 10:
                report.append(f"  ... 还有 {len(self.errors) - 10} 个错误")
            report.append("")
            
        report.append("=" * 80)
        return "n".join(report)
        
    def save_report(self, filename):
        """保存报告到文件"""
        try:
            report = self.generate_report()
            with open(filename, 'w', encoding='utf-8') as f:
                f.write(report)
            print(f"报告已保存到: {filename}")
            return True
        except Exception as e:
            print(f"保存报告时出错: {e}")
            return False
            
    def save_json_report(self, filename):
        """保存JSON格式报告"""
        try:
            report_data = {
                'check_time': datetime.now().isoformat(),
                'results': self.results,
                'errors': self.errors
            }
            
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(report_data, f, indent=2, ensure_ascii=False)
            print(f"JSON报告已保存到: {filename}")
            return True
        except Exception as e:
            print(f"保存JSON报告时出错: {e}")
            return False
            
    def continuous_monitor(self, targets, interval=60, max_checks=None):
        """持续监控网络连通性"""
        check_count = 0
        
        try:
            while True:
                print(f"n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 开始第 {check_count + 1} 次检查")
                self.check_multiple_targets(targets)
                
                # 显示简要状态
                success_count = 0
                total_tests = 0
                
                for result in self.results:
                    for test_result in result['tests'].values():
                        total_tests += 1
                        if test_result['status'] in ['reachable', 'open', 'resolved', 'accessible']:
                            success_count += 1
                            
                print(f"成功测试: {success_count}/{total_tests}")
                
                check_count += 1
                
                # 检查是否达到最大检查次数
                if max_checks and check_count >= max_checks:
                    break
                    
                # 等待下次检查
                time.sleep(interval)
                
        except KeyboardInterrupt:
            print("nn停止监控...")

def main():
    parser = argparse.ArgumentParser(description="网络连通性检查工具")
    parser.add_argument("targets", nargs='*', help="要检查的目标（主机名、IP地址、URL等）")
    parser.add_argument("-f", "--file", help="从文件读取目标列表")
    parser.add_argument("-o", "--output", help="保存报告到文件")
    parser.add_argument("-j", "--json", help="保存JSON报告到文件")
    parser.add_argument("-m", "--monitor", type=int, help="持续监控间隔（秒）")
    parser.add_argument("--max-checks", type=int, help="最大检查次数")
    parser.add_argument("--timeout", type=int, default=3, help="网络操作超时时间（秒，默认：3）")
    
    args = parser.parse_args()
    
    # 检查目标
    targets = []
    if args.targets:
        targets.extend(args.targets)
    if args.file:
        try:
            with open(args.file, 'r', encoding='utf-8') as f:
                file_targets = [line.strip() for line in f if line.strip()]
                targets.extend(file_targets)
        except Exception as e:
            print(f"读取目标文件时出错: {e}")
            sys.exit(1)
            
    if not targets:
        print("请指定要检查的目标")
        parser.print_help()
        sys.exit(1)
        
    # 检查requests库
    try:
        import requests
    except ImportError:
        print("警告: 缺少requests库，HTTP测试功能将不可用")
        print("安装命令: pip install requests")
        
    try:
        checker = NetworkConnectivityChecker()
        
        # 检查网络连通性
        if args.monitor:
            # 持续监控模式
            checker.continuous_monitor(targets, args.monitor, args.max_checks)
        else:
            # 单次检查模式
            checker.check_multiple_targets(targets)
            
            # 生成报告
            if args.output:
                checker.save_report(args.output)
            elif args.json:
                checker.save_json_report(args.json)
            else:
                print(checker.generate_report())
                
    except KeyboardInterrupt:
        print("nn用户中断操作")
    except Exception as e:
        print(f"程序执行出错: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

在使用此脚本之前，需要安装必要的库：

# HTTP测试需要
pip install requests

# 检查单个目标
python network_checker.py google.com

# 检查多个目标
python network_checker.py google.com github.com 8.8.8.8

# 检查特定端口
python network_checker.py google.com:443 github.com:22

# 检查HTTP服务
python network_checker.py https://www.google.com http://github.com

# 从文件读取目标列表
python network_checker.py -f targets.txt

# 保存报告到文件
python network_checker.py google.com github.com -o network_report.txt

# 保存JSON报告
python network_checker.py google.com github.com -j network_report.json

# 持续监控（每30秒检查一次）
python network_checker.py google.com github.com -m 30

# 持续监控最多10次
python network_checker.py google.com github.com -m 30 --max-checks 10

创建目标文件 targets.txt：

google.com
github.com
8.8.8.8
1.1.1.1
https://www.baidu.com
google.com:443
github.com:22

• targets: 要检查的目标（可多个）
• -f, --file: 从文件读取目标列表
• -o, --output: 保存报告到指定文件
• -j, --json: 保存JSON报告到指定文件
• -m, --monitor: 持续监控间隔（秒）
• --max-checks: 最大检查次数
• --timeout: 网络操作超时时间（秒，默认：3）

基本检查：

python network_checker.py google.com github.com 8.8.8.8

持续监控：

python network_checker.py google.com github.com -m 60 --max-checks 10 -o status.log

综合测试：

python network_checker.py -f targets.txt -o full_report.txt -j report.json

这个网络连通性检查工具提供了一个全面的网络诊断解决方案，能够测试Ping连通性、端口开放状态、DNS解析和HTTP服务可达性。它支持批量测试多个目标，提供详细的测试报告，并可以持续监控网络状态。工具适用于网络故障排查、系统监控和运维自动化等多种场景。通过这个工具，用户可以快速诊断网络问题，确保网络服务的稳定运行。

以上就是Python开发一个实用的网络连通性检查工具的详细内容，更多关于Python网络连通性检查的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• Python测试网络连通性示例【基于ping】
• python 判断网络连通的实现方法
• 基于Python编写一个网络带宽测试工具
• 基于Python开发实现网络流量分析工具详解
• 基于Python开发一个多线程IP网络扫描工具