本文目录导读：
我来提供一个自动测试网络并记录结果的脚本方案,这里有几个不同场景的版本：
基础版 - 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
这些脚本会自动测试网络连接性并记录结果到文件中,方便后续分析网络质量变化。
