自动测试网络并记录脚本怎么写

本文目录导读：

1. 基础版 - Ping测试脚本
2. 进阶版 - 综合网络测试
3. 使用示例
4. 安装依赖
5. 运行脚本

我来提供一个自动测试网络并记录结果的脚本方案,这里有几个不同场景的版本：

基础版 - Ping测试脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
网络自动测试与记录脚本 - 基础版
"""
import subprocess
import platform
import datetime
import time
import os
from typing import List, Dict
class NetworkTester:
    def __init__(self, hosts: List[str], log_file: str = "network_log.txt"):
        """
        初始化网络测试器
        Args:
            hosts: 要测试的主机列表
            log_file: 日志文件路径
        """
        self.hosts = hosts
        self.log_file = log_file
        # 根据系统选择ping命令
        if platform.system().lower() == "windows":
            self.ping_cmd = ["ping", "-n", "1", "-w", "3000"]  # Windows命令
        else:
            self.ping_cmd = ["ping", "-c", "1", "-W", "3"]     # Linux/Mac命令
    def test_host(self, host: str) -> Dict:
        """
        测试单个主机的连通性
        Returns:
            包含测试结果的字典
        """
        try:
            result = subprocess.run(
                self.ping_cmd + [host],
                capture_output=True,
                text=True,
                timeout=10
            )
            success = result.returncode == 0
            return {
                "host": host,
                "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "success": success,
                "latency": self._extract_latency(result.stdout) if success else None,
                "output": result.stdout if success else result.stderr
            }
        except subprocess.TimeoutExpired:
            return {
                "host": host,
                "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "success": False,
                "latency": None,
                "output": "Timeout"
            }
        except Exception as e:
            return {
                "host": host,
                "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "success": False,
                "latency": None,
                "output": str(e)
            }
    def _extract_latency(self, output: str) -> float:
        """从ping输出中提取延迟时间"""
        try:
            if "time=" in output:
                # Linux/Mac格式
                latency_str = output.split("time=")[1].split(" ")[0]
                if latency_str.endswith("ms"):
                    return float(latency_str[:-2])
                return float(latency_str)
            elif "time<" in output:
                return 0.0
            else:
                # Windows格式
                import re
                match = re.search(r'时间[=<]\s*(\d+)(?:ms)?', output)
                if match:
                    return float(match.group(1))
        except:
            pass
        return None
    def run_test(self) -> List[Dict]:
        """运行所有主机的测试"""
        results = []
        print(f"\n{'='*50}")
        print(f"开始网络测试 - {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"{'='*50}")
        for host in self.hosts:
            print(f"Testing {host}...", end=" ", flush=True)
            result = self.test_host(host)
            results.append(result)
            status = "✓" if result["success"] else "✗"
            latency = f"({result['latency']}ms)" if result['latency'] else ""
            print(f"{status} {latency}")
        return results
    def save_results(self, results: List[Dict]):
        """保存测试结果到文件"""
        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(f"\n{'='*60}\n")
            f.write(f"测试时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write(f"{'='*60}\n")
            for result in results:
                status = "成功" if result["success"] else "失败"
                latency = f", 延迟: {result['latency']}ms" if result['latency'] else ""
                f.write(f"主机: {result['host']} - {status}{latency}\n")
            f.write(f"{'='*60}\n\n")
    def run_continuous(self, interval: int = 60, cycles: int = None):
        """
        连续运行测试
        Args:
            interval: 测试间隔（秒）
            cycles: 测试次数（None表示无限循环）
        """
        count = 0
        try:
            while cycles is None or count < cycles:
                count += 1
                print(f"\n第 {count} 轮测试")
                results = self.run_test()
                self.save_results(results)
                if cycles and count >= cycles:
                    break
                print(f"等待 {interval} 秒后进行下一轮测试...")
                time.sleep(interval)
        except KeyboardInterrupt:
            print("\n\n测试被用户中断")
# 使用示例
if __name__ == "__main__":
    # 配置要测试的主机
    test_hosts = [
        "google.com",
        "baidu.com",
        "8.8.8.8",
        "192.168.1.1"  # 路由器
    ]
    # 创建测试器
    tester = NetworkTester(test_hosts, "network_test_log.txt")
    # 运行单次测试
    results = tester.run_test()
    tester.save_results(results)
    # 或者连续测试（每隔60秒测试一次，共5次）
    # tester.run_continuous(interval=60, cycles=5)

进阶版 - 综合网络测试

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
综合网络测试脚本 - 包括Ping、HTTP请求、DNS解析等
"""
import subprocess
import requests
import socket
import dns.resolver
import json
import csv
from datetime import datetime
import time
import threading
from queue import Queue
import os
class AdvancedNetworkTester:
    def __init__(self, config_file: str = None):
        """
        初始化高级网络测试器
        """
        self.results = []
        self.log_dir = "network_logs"
        os.makedirs(self.log_dir, exist_ok=True)
        # 默认测试配置
        self.config = {
            "ping_hosts": [
                {"host": "google.com", "name": "Google"},
                {"host": "baidu.com", "name": "百度"},
                {"host": "8.8.8.8", "name": "Google DNS"},
                {"host": "114.114.114.114", "name": "114 DNS"}
            ],
            "web_hosts": [
                {"url": "https://www.google.com", "name": "Google"},
                {"url": "https://www.baidu.com", "name": "百度"}
            ],
            "dns_servers": [
                {"server": "8.8.8.8", "name": "Google DNS"},
                {"server": "114.114.114.114", "name": "114 DNS"}
            ],
            "test_interval": 300,  # 5分钟
            "max_threads": 5
        }
        if config_file and os.path.exists(config_file):
            with open(config_file, 'r') as f:
                self.config.update(json.load(f))
    def test_ping(self, host: str, count: int = 3) -> dict:
        """测试Ping连通性"""
        try:
            system = os.name
            if system == 'nt':  # Windows
                cmd = ['ping', '-n', str(count), '-w', '2000', host]
            else:  # Linux/Mac
                cmd = ['ping', '-c', str(count), '-W', '2', host]
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
            # 解析结果
            success = result.returncode == 0
            latency = None
            loss_rate = None
            if success:
                lines = result.stdout.split('\n')
                for line in lines:
                    if 'avg' in line or '平均' in line:
                        # 提取延迟
                        import re
                        numbers = re.findall(r'\d+\.?\d*', line)
                        if len(numbers) >= 2:
                            latency = float(numbers[-2])
                    if '%' in line:
                        loss_match = re.search(r'(\d+)%', line)
                        if loss_match:
                            loss_rate = int(loss_match.group(1))
            return {
                "type": "ping",
                "host": host,
                "success": success,
                "latency_ms": latency,
                "loss_rate": loss_rate,
                "output": result.stdout if success else result.stderr
            }
        except Exception as e:
            return {
                "type": "ping",
                "host": host,
                "success": False,
                "latency_ms": None,
                "loss_rate": None,
                "error": str(e)
            }
    def test_http(self, url: str, timeout: int = 10) -> dict:
        """测试HTTP请求"""
        try:
            start_time = time.time()
            response = requests.get(url, timeout=timeout, verify=False)
            response_time = (time.time() - start_time) * 1000  # 转换为毫秒
            return {
                "type": "http",
                "url": url,
                "success": True,
                "status_code": response.status_code,
                "response_time_ms": round(response_time, 2),
                "response_size": len(response.content)
            }
        except requests.exceptions.Timeout:
            return {
                "type": "http",
                "url": url,
                "success": False,
                "error": "Request timeout"
            }
        except Exception as e:
            return {
                "type": "http",
                "url": url,
                "success": False,
                "error": str(e)
            }
    def test_dns(self, domain: str = "google.com", dns_server: str = None) -> dict:
        """测试DNS解析"""
        try:
            resolver = dns.resolver.Resolver()
            if dns_server:
                resolver.nameservers = [dns_server]
            start_time = time.time()
            answers = resolver.resolve(domain, 'A')
            resolve_time = (time.time() - start_time) * 1000
            ips = [str(r) for r in answers]
            return {
                "type": "dns",
                "domain": domain,
                "dns_server": dns_server or "system default",
                "success": True,
                "resolve_time_ms": round(resolve_time, 2),
                "resolved_ips": ips
            }
        except Exception as e:
            return {
                "type": "dns",
                "domain": domain,
                "dns_server": dns_server or "system default",
                "success": False,
                "error": str(e)
            }
    def test_traceroute(self, host: str) -> dict:
        """测试路由追踪"""
        try:
            system = os.name
            if system == 'nt':
                cmd = ['tracert', '-h', '10', '-w', '1000', host]
            else:
                cmd = ['traceroute', '-m', '10', '-w', '1', host]
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
            # 解析跳数
            hops = []
            for line in result.stdout.split('\n'):
                if line.strip() and not line.startswith('traceroute') and not line.startswith('Tracing'):
                    hops.append(line.strip()[:100])  # 截取前100字符
            return {
                "type": "traceroute",
                "host": host,
                "success": result.returncode == 0,
                "hops": hops if hops else ["No route found"],
                "output": result.stdout
            }
        except Exception as e:
            return {
                "type": "traceroute",
                "host": host,
                "success": False,
                "error": str(e)
            }
    def run_all_tests(self):
        """运行所有测试"""
        test_results = {
            "timestamp": datetime.now().isoformat(),
            "tests": []
        }
        # 并行执行测试
        threads = []
        results_queue = Queue()
        # Ping测试
        for host_config in self.config['ping_hosts']:
            thread = threading.Thread(
                target=lambda q, h: q.put(self.test_ping(h['host'])),
                args=(results_queue, host_config)
            )
            threads.append(thread)
            thread.start()
        # HTTP测试
        for web_config in self.config['web_hosts']:
            thread = threading.Thread(
                target=lambda q, w: q.put(self.test_http(w['url'])),
                args=(results_queue, web_config)
            )
            threads.append(thread)
            thread.start()
        # DNS测试
        for dns_config in self.config['dns_servers']:
            thread = threading.Thread(
                target=lambda q, d: q.put(self.test_dns(dns_server=d['server'])),
                args=(results_queue, dns_config)
            )
            threads.append(thread)
            thread.start()
        # 等待所有线程完成
        for thread in threads:
            thread.join()
        # 收集结果
        while not results_queue.empty():
            test_results['tests'].append(results_queue.get())
        return test_results
    def save_to_json(self, results: dict, filename: str = None):
        """保存结果为JSON格式"""
        if not filename:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"network_test_{timestamp}.json"
        filepath = os.path.join(self.log_dir, filename)
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(results, f, indent=2, ensure_ascii=False)
        return filepath
    def save_to_csv(self, results: dict, filename: str = None):
        """保存结果为CSV格式"""
        if not filename:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"network_test_{timestamp}.csv"
        filepath = os.path.join(self.log_dir, filename)
        with open(filepath, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['Timestamp', 'Type', 'Target', 'Success', 'Details'])
            for test in results['tests']:
                details = str(test.get('latency_ms', '') or test.get('response_time_ms', '') or test.get('resolve_time_ms', '') or test.get('error', ''))
                writer.writerow([
                    results['timestamp'],
                    test['type'],
                    test.get('host') or test.get('url') or test.get('domain'),
                    test['success'],
                    details
                ])
        return filepath
def main():
    """主函数"""
    tester = AdvancedNetworkTester()
    print("="*50)
    print("网络自动测试系统")
    print("="*50)
    print("\n运行测试中...")
    results = tester.run_all_tests()
    # 保存结果
    json_file = tester.save_to_json(results)
    csv_file = tester.save_to_csv(results)
    print(f"\n测试完成！")
    print(f"JSON结果保存至: {json_file}")
    print(f"CSV结果保存至: {csv_file}")
    # 显示摘要
    print("\n测试摘要:")
    for test in results['tests']:
        status = "✓" if test['success'] else "✗"
        target = test.get('host') or test.get('url') or test.get('domain')
        print(f"  {status} {test['type'].upper()}: {target}")
if __name__ == "__main__":
    main()

使用示例

创建配置文件 network_config.json:

{
    "ping_hosts": [
        {"host": "google.com", "name": "Google"},
        {"host": "baidu.com", "name": "百度"}
    ],
    "web_hosts": [
        {"url": "https://www.google.com", "name": "Google"},
        {"url": "https://www.baidu.com", "name": "百度"}
    ],
    "dns_servers": [
        {"server": "8.8.8.8", "name": "Google DNS"},
        {"server": "114.114.114.114", "name": "114 DNS"}
    ],
    "test_interval": 300,
    "max_threads": 5
}

安装依赖

pip install requests dnspython

运行脚本

# 基础版
python network_tester_basic.py
# 进阶版
python network_tester_advanced.py
# 进阶版使用配置文件
python network_tester_advanced.py --config network_config.json

这些脚本会自动测试网络连接性并记录结果到文件中,方便后续分析网络质量变化。