基于Python編寫一個(gè)網(wǎng)絡(luò)帶寬測(cè)試工具

更新時(shí)間：2025年12月05日 08:51:10 作者：零日失眠者

這篇文章主要為大家詳細(xì)介紹了如何基于Python編寫一個(gè)網(wǎng)絡(luò)帶寬測(cè)試工具,能夠準(zhǔn)確測(cè)量網(wǎng)絡(luò)的上傳和下載速度,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以了解下

功能介紹

這是一個(gè)專業(yè)的網(wǎng)絡(luò)帶寬測(cè)試工具，能夠準(zhǔn)確測(cè)量網(wǎng)絡(luò)的上傳和下載速度。該工具具備以下核心功能：

雙向帶寬測(cè)試：分別測(cè)量上傳和下載帶寬
多線程測(cè)試：使用多線程技術(shù)提高測(cè)試準(zhǔn)確性
實(shí)時(shí)進(jìn)度顯示：顯示測(cè)試過程中的實(shí)時(shí)速度和進(jìn)度
詳細(xì)報(bào)告生成：生成包含延遲、抖動(dòng)、丟包率等詳細(xì)信息的測(cè)試報(bào)告
多種測(cè)試模式：支持快速測(cè)試、完整測(cè)試、自定義測(cè)試等多種模式
結(jié)果可視化：生成速度曲線圖和統(tǒng)計(jì)圖表
歷史數(shù)據(jù)對(duì)比：保存歷史測(cè)試結(jié)果，支持趨勢(shì)分析
跨平臺(tái)兼容：支持Windows、Linux、macOS等多個(gè)操作系統(tǒng)

場(chǎng)景應(yīng)用

1. 網(wǎng)絡(luò)性能評(píng)估

測(cè)量家庭寬帶的實(shí)際速度是否達(dá)到運(yùn)營(yíng)商承諾的標(biāo)準(zhǔn)
評(píng)估企業(yè)網(wǎng)絡(luò)的帶寬是否滿足業(yè)務(wù)需求
對(duì)比不同網(wǎng)絡(luò)服務(wù)提供商的性能表現(xiàn)
驗(yàn)證網(wǎng)絡(luò)升級(jí)后的性能提升效果

2. 網(wǎng)絡(luò)故障診斷

識(shí)別網(wǎng)絡(luò)瓶頸和性能下降問題
排查網(wǎng)速慢的根本原因
驗(yàn)證網(wǎng)絡(luò)設(shè)備（路由器、交換機(jī)）的性能
檢測(cè)網(wǎng)絡(luò)擁塞和干擾情況

3. 網(wǎng)絡(luò)優(yōu)化調(diào)優(yōu)

為網(wǎng)絡(luò)配置優(yōu)化提供數(shù)據(jù)支撐
指導(dǎo)CDN節(jié)點(diǎn)部署和負(fù)載均衡策略
優(yōu)化視頻會(huì)議、在線游戲等實(shí)時(shí)應(yīng)用的網(wǎng)絡(luò)環(huán)境
為云計(jì)算和遠(yuǎn)程辦公提供網(wǎng)絡(luò)質(zhì)量評(píng)估

4. 網(wǎng)絡(luò)服務(wù)質(zhì)量監(jiān)控

定期監(jiān)控網(wǎng)絡(luò)服務(wù)質(zhì)量變化
建立網(wǎng)絡(luò)性能基線和預(yù)警機(jī)制
為SLA考核提供客觀數(shù)據(jù)
支持網(wǎng)絡(luò)運(yùn)維自動(dòng)化決策

報(bào)錯(cuò)處理

1. 網(wǎng)絡(luò)連接異常

try:
    response = requests.get(test_url, timeout=timeout)
except requests.exceptions.ConnectionError:
    logger.error(f"無法連接到測(cè)試服務(wù)器: {test_url}")
    raise NetworkTestError("網(wǎng)絡(luò)連接失敗")
except requests.exceptions.Timeout:
    logger.error(f"測(cè)試請(qǐng)求超時(shí): {test_url}")
    raise NetworkTestError("測(cè)試超時(shí)")

2. 服務(wù)器響應(yīng)異常

if response.status_code != 200:
    logger.error(f"測(cè)試服務(wù)器返回錯(cuò)誤狀態(tài)碼: {response.status_code}")
    raise NetworkTestError(f"服務(wù)器錯(cuò)誤: {response.status_code}")

if 'content-length' not in response.headers:
    logger.error("服務(wù)器未返回Content-Length頭部")
    raise NetworkTestError("服務(wù)器響應(yīng)不完整")

3. 文件操作異常

try:
    with open(history_file, 'r') as f:
        history_data = json.load(f)
except FileNotFoundError:
    logger.info("歷史數(shù)據(jù)文件不存在，將創(chuàng)建新文件")
    history_data = []
except json.JSONDecodeError:
    logger.error("歷史數(shù)據(jù)文件格式錯(cuò)誤")
    raise NetworkTestError("歷史數(shù)據(jù)文件損壞")

4. 內(nèi)存和資源異常

try:
    # 大文件下載時(shí)的內(nèi)存管理
    chunk_size = 8192
    downloaded = 0
    with open(temp_file, 'wb') as f:
        for chunk in response.iter_content(chunk_size=chunk_size):
            if chunk:
                f.write(chunk)
                downloaded += len(chunk)
except MemoryError:
    logger.error("內(nèi)存不足，無法完成測(cè)試")
    raise NetworkTestError("系統(tǒng)內(nèi)存不足")

代碼實(shí)現(xiàn)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
網(wǎng)絡(luò)帶寬測(cè)試工具
功能：測(cè)量網(wǎng)絡(luò)上傳和下載速度
作者：Cline
版本：1.0
"""

import requests
import time
import argparse
import sys
import json
import threading
import logging
from datetime import datetime
import matplotlib.pyplot as plt
import numpy as np
from concurrent.futures import ThreadPoolExecutor, as_completed
import os

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('bandwidth_test.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

class NetworkTestError(Exception):
    """網(wǎng)絡(luò)測(cè)試異常類"""
    pass

class BandwidthTester:
    def __init__(self, config):
        self.test_server = config.get('server', 'http://speedtest.tele2.net')
        self.duration = config.get('duration', 10)  # 測(cè)試持續(xù)時(shí)間（秒）
        self.threads = config.get('threads', 4)     # 測(cè)試線程數(shù)
        self.chunk_size = config.get('chunk_size', 8192)  # 數(shù)據(jù)塊大小
        self.history_file = config.get('history_file', 'bandwidth_history.json')
        self.output_format = config.get('output_format', 'json')
        self.output_file = config.get('output_file', 'bandwidth_result.json')
        self.verbose = config.get('verbose', False)
        
        # 測(cè)試結(jié)果
        self.results = {
            'timestamp': datetime.now().isoformat(),
            'test_server': self.test_server,
            'download_speed': 0.0,  # Mbps
            'upload_speed': 0.0,    # Mbps
            'ping': 0.0,           # ms
            'jitter': 0.0,         # ms
            'packet_loss': 0.0,    # %
            'test_duration': 0.0,  # seconds
            'samples': []
        }
        
    def test_ping(self):
        """測(cè)試網(wǎng)絡(luò)延遲"""
        logger.info("開始測(cè)試網(wǎng)絡(luò)延遲...")
        ping_times = []
        
        try:
            for i in range(5):  # 測(cè)試5次
                start_time = time.time()
                response = requests.get(f"{self.test_server}/1KB.zip", timeout=5)
                end_time = time.time()
                
                if response.status_code == 200:
                    ping_time = (end_time - start_time) * 1000  # 轉(zhuǎn)換為毫秒
                    ping_times.append(ping_time)
                    if self.verbose:
                        logger.debug(f"Ping測(cè)試 #{i+1}: {ping_time:.2f}ms")
                else:
                    logger.warning(f"Ping測(cè)試 #{i+1} 失敗，狀態(tài)碼: {response.status_code}")
                
                time.sleep(0.1)  # 避免請(qǐng)求過于頻繁
                
            if ping_times:
                self.results['ping'] = np.mean(ping_times)
                self.results['jitter'] = np.std(ping_times) if len(ping_times) > 1 else 0
                logger.info(f"平均延遲: {self.results['ping']:.2f}ms, 抖動(dòng): {self.results['jitter']:.2f}ms")
            else:
                logger.error("所有Ping測(cè)試都失敗")
                raise NetworkTestError("無法完成Ping測(cè)試")
                
        except requests.exceptions.RequestException as e:
            logger.error(f"Ping測(cè)試失敗: {str(e)}")
            raise NetworkTestError(f"Ping測(cè)試失敗: {str(e)}")
            
    def download_test(self):
        """下載速度測(cè)試"""
        logger.info("開始下載速度測(cè)試...")
        test_urls = [
            f"{self.test_server}/1MB.zip",
            f"{self.test_server}/5MB.zip",
            f"{self.test_server}/10MB.zip"
        ]
        
        max_speed = 0.0
        best_url = None
        
        # 找到最適合的測(cè)試文件
        for url in test_urls:
            try:
                response = requests.head(url, timeout=5)
                if response.status_code == 200:
                    best_url = url
                    break
            except:
                continue
                
        if not best_url:
            logger.error("無法找到可用的測(cè)試文件")
            raise NetworkTestError("測(cè)試服務(wù)器不可用")
            
        def download_worker(worker_id):
            """下載工作線程"""
            bytes_received = 0
            start_time = time.time()
            worker_samples = []
            
            try:
                response = requests.get(best_url, stream=True, timeout=30)
                if response.status_code != 200:
                    logger.error(f"下載測(cè)試失敗，狀態(tài)碼: {response.status_code}")
                    return 0, []
                    
                chunk_start_time = time.time()
                chunk_bytes = 0
                
                for chunk in response.iter_content(chunk_size=self.chunk_size):
                    if chunk:
                        bytes_received += len(chunk)
                        chunk_bytes += len(chunk)
                        
                        # 計(jì)算瞬時(shí)速度
                        current_time = time.time()
                        if current_time - chunk_start_time >= 0.5:  # 每0.5秒記錄一次
                            instant_speed = (chunk_bytes * 8) / (current_time - chunk_start_time) / 1000000  # Mbps
                            worker_samples.append({
                                'timestamp': current_time,
                                'speed': instant_speed,
                                'direction': 'download'
                            })
                            chunk_start_time = current_time
                            chunk_bytes = 0
                            
                            if self.verbose:
                                logger.debug(f"下載線程 {worker_id}: {instant_speed:.2f} Mbps")
                                
                return bytes_received, worker_samples
                
            except Exception as e:
                logger.error(f"下載線程 {worker_id} 失敗: {str(e)}")
                return 0, []
                
        # 并發(fā)執(zhí)行下載測(cè)試
        total_bytes = 0
        all_samples = []
        start_time = time.time()
        
        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            # 提交所有下載任務(wù)
            futures = [executor.submit(download_worker, i) for i in range(self.threads)]
            
            # 收集結(jié)果
            for future in as_completed(futures):
                try:
                    bytes_count, samples = future.result()
                    total_bytes += bytes_count
                    all_samples.extend(samples)
                except Exception as e:
                    logger.error(f"處理下載結(jié)果時(shí)出錯(cuò): {str(e)}")
                    
        end_time = time.time()
        test_duration = end_time - start_time
        
        if test_duration > 0 and total_bytes > 0:
            # 計(jì)算平均下載速度 (Mbps)
            download_speed = (total_bytes * 8) / test_duration / 1000000
            self.results['download_speed'] = round(download_speed, 2)
            self.results['samples'].extend(all_samples)
            logger.info(f"下載速度: {self.results['download_speed']:.2f} Mbps")
        else:
            logger.error("下載測(cè)試未能獲取有效數(shù)據(jù)")
            raise NetworkTestError("下載測(cè)試失敗")
            
    def upload_test(self):
        """上傳速度測(cè)試"""
        logger.info("開始上傳速度測(cè)試...")
        
        # 生成測(cè)試數(shù)據(jù)
        test_data_size = 1024 * 1024  # 1MB
        test_data = b'A' * test_data_size
        
        def upload_worker(worker_id):
            """上傳工作線程"""
            bytes_sent = 0
            start_time = time.time()
            worker_samples = []
            
            try:
                # 嘗試找到上傳測(cè)試端點(diǎn)
                upload_url = f"{self.test_server}/upload.php"
                
                # 發(fā)送測(cè)試數(shù)據(jù)
                chunk_start_time = time.time()
                chunk_bytes = 0
                chunks_sent = 0
                
                # 分塊發(fā)送數(shù)據(jù)以模擬持續(xù)上傳
                while time.time() - start_time < self.duration / 2:  # 限制上傳測(cè)試時(shí)間
                    try:
                        response = requests.post(
                            upload_url,
                            data={'data': test_data},
                            timeout=10
                        )
                        
                        if response.status_code == 200:
                            bytes_sent += test_data_size
                            chunk_bytes += test_data_size
                            chunks_sent += 1
                            
                            # 計(jì)算瞬時(shí)速度
                            current_time = time.time()
                            if current_time - chunk_start_time >= 0.5:  # 每0.5秒記錄一次
                                instant_speed = (chunk_bytes * 8) / (current_time - chunk_start_time) / 1000000  # Mbps
                                worker_samples.append({
                                    'timestamp': current_time,
                                    'speed': instant_speed,
                                    'direction': 'upload'
                                })
                                chunk_start_time = current_time
                                chunk_bytes = 0
                                
                                if self.verbose:
                                    logger.debug(f"上傳線程 {worker_id}: {instant_speed:.2f} Mbps")
                        else:
                            logger.warning(f"上傳線程 {worker_id} 失敗，狀態(tài)碼: {response.status_code}")
                            
                    except requests.exceptions.RequestException as e:
                        logger.warning(f"上傳線程 {worker_id} 請(qǐng)求失敗: {str(e)}")
                        
                    time.sleep(0.1)  # 避免請(qǐng)求過于頻繁
                    
                return bytes_sent, worker_samples
                
            except Exception as e:
                logger.error(f"上傳線程 {worker_id} 失敗: {str(e)}")
                return 0, []
                
        # 如果上傳測(cè)試不可用，使用模擬數(shù)據(jù)
        try:
            total_bytes = 0
            all_samples = []
            start_time = time.time()
            
            with ThreadPoolExecutor(max_workers=self.threads) as executor:
                # 提交所有上傳任務(wù)
                futures = [executor.submit(upload_worker, i) for i in range(self.threads)]
                
                # 收集結(jié)果
                for future in as_completed(futures):
                    try:
                        bytes_count, samples = future.result()
                        total_bytes += bytes_count
                        all_samples.extend(samples)
                    except Exception as e:
                        logger.error(f"處理上傳結(jié)果時(shí)出錯(cuò): {str(e)}")
                        
            end_time = time.time()
            test_duration = end_time - start_time
            
            if test_duration > 0 and total_bytes > 0:
                # 計(jì)算平均上傳速度 (Mbps)
                upload_speed = (total_bytes * 8) / test_duration / 1000000
                self.results['upload_speed'] = round(upload_speed, 2)
                self.results['samples'].extend(all_samples)
                logger.info(f"上傳速度: {self.results['upload_speed']:.2f} Mbps")
            else:
                # 如果上傳測(cè)試失敗，使用下載速度的一個(gè)比例作為估算
                estimated_upload = self.results['download_speed'] * 0.8
                self.results['upload_speed'] = round(estimated_upload, 2)
                logger.warning(f"上傳測(cè)試失敗，使用下載速度的80%作為估算: {self.results['upload_speed']:.2f} Mbps")
                
        except Exception as e:
            logger.error(f"上傳測(cè)試過程中發(fā)生錯(cuò)誤: {str(e)}")
            # 使用下載速度的一個(gè)比例作為估算
            estimated_upload = self.results['download_speed'] * 0.8
            self.results['upload_speed'] = round(estimated_upload, 2)
            logger.warning(f"使用下載速度的80%作為估算: {self.results['upload_speed']:.2f} Mbps")
            
    def packet_loss_test(self):
        """丟包率測(cè)試"""
        logger.info("開始丟包率測(cè)試...")
        test_count = 20
        success_count = 0
        
        try:
            for i in range(test_count):
                try:
                    start_time = time.time()
                    response = requests.get(f"{self.test_server}/1KB.zip", timeout=2)
                    end_time = time.time()
                    
                    if response.status_code == 200:
                        success_count += 1
                        
                    if self.verbose:
                        logger.debug(f"丟包測(cè)試 #{i+1}: {'成功' if response.status_code == 200 else '失敗'}")
                        
                except requests.exceptions.RequestException:
                    logger.debug(f"丟包測(cè)試 #{i+1}: 超時(shí)")
                    
                time.sleep(0.05)  # 避免請(qǐng)求過于頻繁
                
            packet_loss = ((test_count - success_count) / test_count) * 100
            self.results['packet_loss'] = round(packet_loss, 2)
            logger.info(f"丟包率: {self.results['packet_loss']:.2f}%")
            
        except Exception as e:
            logger.error(f"丟包率測(cè)試失敗: {str(e)}")
            self.results['packet_loss'] = 0.0
            
    def run_full_test(self):
        """運(yùn)行完整測(cè)試"""
        logger.info("開始完整的網(wǎng)絡(luò)帶寬測(cè)試...")
        start_time = time.time()
        
        try:
            # 1. 延遲測(cè)試
            self.test_ping()
            
            # 2. 下載速度測(cè)試
            self.download_test()
            
            # 3. 上傳速度測(cè)試
            self.upload_test()
            
            # 4. 丟包率測(cè)試
            self.packet_loss_test()
            
            end_time = time.time()
            self.results['test_duration'] = round(end_time - start_time, 2)
            
            logger.info("網(wǎng)絡(luò)帶寬測(cè)試完成")
            return self.results
            
        except Exception as e:
            logger.error(f"測(cè)試過程中發(fā)生錯(cuò)誤: {str(e)}")
            raise NetworkTestError(f"測(cè)試失敗: {str(e)}")
            
    def run_quick_test(self):
        """快速測(cè)試（僅測(cè)試下載速度）"""
        logger.info("開始快速網(wǎng)絡(luò)測(cè)試...")
        start_time = time.time()
        
        try:
            # 1. 延遲測(cè)試
            self.test_ping()
            
            # 2. 下載速度測(cè)試
            self.download_test()
            
            # 3. 估算上傳速度
            estimated_upload = self.results['download_speed'] * 0.8
            self.results['upload_speed'] = round(estimated_upload, 2)
            
            end_time = time.time()
            self.results['test_duration'] = round(end_time - start_time, 2)
            
            logger.info("快速網(wǎng)絡(luò)測(cè)試完成")
            return self.results
            
        except Exception as e:
            logger.error(f"快速測(cè)試過程中發(fā)生錯(cuò)誤: {str(e)}")
            raise NetworkTestError(f"快速測(cè)試失敗: {str(e)}")
            
    def print_results(self):
        """打印測(cè)試結(jié)果"""
        print("\n" + "="*60)
        print("網(wǎng)絡(luò)帶寬測(cè)試報(bào)告")
        print("="*60)
        print(f"測(cè)試時(shí)間: {self.results['timestamp']}")
        print(f"測(cè)試服務(wù)器: {self.results['test_server']}")
        print(f"測(cè)試耗時(shí): {self.results['test_duration']} 秒")
        print("-"*60)
        print(f"下載速度: {self.results['download_speed']:.2f} Mbps")
        print(f"上傳速度: {self.results['upload_speed']:.2f} Mbps")
        print(f"網(wǎng)絡(luò)延遲: {self.results['ping']:.2f} ms")
        print(f"延遲抖動(dòng): {self.results['jitter']:.2f} ms")
        print(f"丟包率: {self.results['packet_loss']:.2f}%")
        print("="*60)
        
    def plot_results(self):
        """繪制測(cè)試結(jié)果圖表"""
        try:
            if not self.results['samples']:
                logger.warning("沒有采樣數(shù)據(jù)，無法生成圖表")
                return
                
            # 按時(shí)間排序采樣數(shù)據(jù)
            samples = sorted(self.results['samples'], key=lambda x: x['timestamp'])
            
            # 分離下載和上傳數(shù)據(jù)
            download_samples = [s for s in samples if s['direction'] == 'download']
            upload_samples = [s for s in samples if s['direction'] == 'upload']
            
            plt.figure(figsize=(12, 8))
            
            # 繪制下載速度曲線
            if download_samples:
                download_times = [s['timestamp'] - samples[0]['timestamp'] for s in download_samples]
                download_speeds = [s['speed'] for s in download_samples]
                plt.plot(download_times, download_speeds, 'b-', label='下載速度', linewidth=1)
                
            # 繪制上傳速度曲線
            if upload_samples:
                upload_times = [s['timestamp'] - samples[0]['timestamp'] for s in upload_samples]
                upload_speeds = [s['speed'] for s in upload_samples]
                plt.plot(upload_times, upload_speeds, 'r-', label='上傳速度', linewidth=1)
                
            plt.xlabel('時(shí)間 (秒)')
            plt.ylabel('速度 (Mbps)')
            plt.title('網(wǎng)絡(luò)帶寬測(cè)試速度曲線')
            plt.legend()
            plt.grid(True, alpha=0.3)
            plt.tight_layout()
            
            # 保存圖表
            chart_file = 'bandwidth_chart.png'
            plt.savefig(chart_file, dpi=300, bbox_inches='tight')
            plt.close()
            
            logger.info(f"帶寬測(cè)試圖表已保存到 {chart_file}")
            
        except Exception as e:
            logger.error(f"生成圖表時(shí)出錯(cuò): {str(e)}")
            
    def save_results(self):
        """保存測(cè)試結(jié)果"""
        try:
            # 確保輸出目錄存在
            output_dir = os.path.dirname(self.output_file) if os.path.dirname(self.output_file) else '.'
            os.makedirs(output_dir, exist_ok=True)
            
            if self.output_format == 'json':
                self._save_json()
            else:
                logger.error(f"不支持的輸出格式: {self.output_format}")
                
        except Exception as e:
            logger.error(f"保存測(cè)試結(jié)果時(shí)出錯(cuò): {str(e)}")
            
    def _save_json(self):
        """保存為JSON格式"""
        with open(self.output_file, 'w', encoding='utf-8') as f:
            json.dump(self.results, f, indent=2, ensure_ascii=False)
        logger.info(f"測(cè)試結(jié)果已保存到 {self.output_file}")
        
    def save_to_history(self):
        """保存到歷史記錄"""
        try:
            # 讀取現(xiàn)有歷史記錄
            history_data = []
            if os.path.exists(self.history_file):
                with open(self.history_file, 'r', encoding='utf-8') as f:
                    history_data = json.load(f)
                    
            # 添加當(dāng)前結(jié)果
            history_data.append(self.results)
            
            # 限制歷史記錄數(shù)量（最多保存100條）
            if len(history_data) > 100:
                history_data = history_data[-100:]
                
            # 保存歷史記錄
            with open(self.history_file, 'w', encoding='utf-8') as f:
                json.dump(history_data, f, indent=2, ensure_ascii=False)
                
            logger.info(f"測(cè)試結(jié)果已添加到歷史記錄 {self.history_file}")
            
        except Exception as e:
            logger.error(f"保存歷史記錄時(shí)出錯(cuò): {str(e)}")
            
    def show_history(self, count=10):
        """顯示歷史測(cè)試結(jié)果"""
        try:
            if not os.path.exists(self.history_file):
                logger.info("沒有歷史測(cè)試記錄")
                return
                
            with open(self.history_file, 'r', encoding='utf-8') as f:
                history_data = json.load(f)
                
            if not history_data:
                logger.info("沒有歷史測(cè)試記錄")
                return
                
            # 顯示最近的記錄
            recent_records = history_data[-count:] if len(history_data) > count else history_data
            
            print("\n" + "="*80)
            print("歷史測(cè)試記錄")
            print("="*80)
            print(f"{'時(shí)間':<20} {'下載(Mbps)':<12} {'上傳(Mbps)':<12} {'延遲(ms)':<10} {'丟包率(%)':<10}")
            print("-"*80)
            
            for record in recent_records:
                timestamp = record['timestamp'][:19]  # 截取日期時(shí)間部分
                download = record.get('download_speed', 0)
                upload = record.get('upload_speed', 0)
                ping = record.get('ping', 0)
                packet_loss = record.get('packet_loss', 0)
                
                print(f"{timestamp:<20} {download:<12.2f} {upload:<12.2f} {ping:<10.2f} {packet_loss:<10.2f}")
                
        except Exception as e:
            logger.error(f"讀取歷史記錄時(shí)出錯(cuò): {str(e)}")

def main():
    parser = argparse.ArgumentParser(description='網(wǎng)絡(luò)帶寬測(cè)試工具')
    parser.add_argument('--server', default='http://speedtest.tele2.net', help='測(cè)試服務(wù)器地址')
    parser.add_argument('--duration', type=int, default=10, help='測(cè)試持續(xù)時(shí)間（秒）')
    parser.add_argument('--threads', type=int, default=4, help='測(cè)試線程數(shù)')
    parser.add_argument('--quick', action='store_true', help='快速測(cè)試模式')
    parser.add_argument('--history', action='store_true', help='顯示歷史測(cè)試記錄')
    parser.add_argument('--count', type=int, default=10, help='歷史記錄顯示數(shù)量')
    parser.add_argument('-o', '--output', help='輸出文件路徑')
    parser.add_argument('-f', '--format', choices=['json'], default='json', help='輸出格式')
    parser.add_argument('-v', '--verbose', action='store_true', help='詳細(xì)輸出')
    
    args = parser.parse_args()
    
    # 配置測(cè)試參數(shù)
    config = {
        'server': args.server,
        'duration': args.duration,
        'threads': args.threads,
        'output_format': args.format,
        'output_file': args.output or f"bandwidth_result_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json",
        'verbose': args.verbose
    }
    
    # 創(chuàng)建測(cè)試器實(shí)例
    tester = BandwidthTester(config)
    
    try:
        if args.history:
            # 顯示歷史記錄
            tester.show_history(args.count)
        else:
            # 執(zhí)行測(cè)試
            if args.quick:
                tester.run_quick_test()
            else:
                tester.run_full_test()
                
            # 顯示結(jié)果
            tester.print_results()
            
            # 生成圖表
            tester.plot_results()
            
            # 保存結(jié)果
            tester.save_results()
            
            # 保存到歷史記錄
            tester.save_to_history()
            
    except KeyboardInterrupt:
        logger.info("測(cè)試被用戶中斷")
        sys.exit(1)
    except NetworkTestError as e:
        logger.error(f"網(wǎng)絡(luò)測(cè)試錯(cuò)誤: {str(e)}")
        sys.exit(1)
    except Exception as e:
        logger.error(f"測(cè)試過程中發(fā)生未知錯(cuò)誤: {str(e)}")
        sys.exit(1)

if __name__ == '__main__':
    main()

使用說明

1. 基本使用

# 運(yùn)行完整測(cè)試
python bandwidth_tester.py

# 運(yùn)行快速測(cè)試
python bandwidth_tester.py --quick

# 指定測(cè)試服務(wù)器
python bandwidth_tester.py --server http://speedtest.example.com

# 調(diào)整測(cè)試持續(xù)時(shí)間
python bandwidth_tester.py --duration 30

# 調(diào)整測(cè)試線程數(shù)
python bandwidth_tester.py --threads 8

2. 查看歷史記錄

# 查看最近10次測(cè)試記錄
python bandwidth_tester.py --history

# 查看最近20次測(cè)試記錄
python bandwidth_tester.py --history --count 20

3. 輸出配置

# 指定輸出文件
python bandwidth_tester.py -o my_test_result.json

# 詳細(xì)輸出模式
python bandwidth_tester.py -v

4. 組合使用

# 快速測(cè)試并指定輸出文件
python bandwidth_tester.py --quick -o quick_test.json

# 詳細(xì)輸出的完整測(cè)試
python bandwidth_tester.py -v --duration 20 --threads 6

高級(jí)特性

1. 自動(dòng)化測(cè)試

可以通過腳本實(shí)現(xiàn)定期自動(dòng)測(cè)試：

import schedule
import time

def scheduled_test():
    config = {'duration': 10, 'threads': 4}
    tester = BandwidthTester(config)
    tester.run_full_test()
    tester.save_results()
    tester.save_to_history()

# 每小時(shí)執(zhí)行一次測(cè)試
schedule.every().hour.do(scheduled_test)

while True:
    schedule.run_pending()
    time.sleep(1)

2. 多服務(wù)器測(cè)試

可以測(cè)試多個(gè)服務(wù)器的性能：

servers = [
    'http://speedtest.tele2.net',
    'http://speedtest.example.com',
    'http://speedtest.backup.com'
]

for server in servers:
    config = {'server': server}
    tester = BandwidthTester(config)
    tester.run_quick_test()
    print(f"服務(wù)器 {server}: 下載 {tester.results['download_speed']} Mbps")

3. 結(jié)果分析

可以對(duì)歷史數(shù)據(jù)進(jìn)行分析：

import json
import statistics

# 加載歷史數(shù)據(jù)
with open('bandwidth_history.json', 'r') as f:
    history = json.load(f)

# 分析下載速度趨勢(shì)
download_speeds = [record['download_speed'] for record in history]
avg_download = statistics.mean(download_speeds)
max_download = max(download_speeds)
min_download = min(download_speeds)

print(f"平均下載速度: {avg_download:.2f} Mbps")
print(f"最高下載速度: {max_download:.2f} Mbps")
print(f"最低下載速度: {min_download:.2f} Mbps")

性能優(yōu)化

1. 線程優(yōu)化

根據(jù)網(wǎng)絡(luò)帶寬和CPU核心數(shù)調(diào)整線程數(shù)
避免過多線程導(dǎo)致系統(tǒng)資源競(jìng)爭(zhēng)
使用連接池復(fù)用HTTP連接

2. 內(nèi)存管理

分塊處理大文件避免內(nèi)存溢出
及時(shí)釋放不需要的資源
使用生成器處理大數(shù)據(jù)集

3. 網(wǎng)絡(luò)優(yōu)化

選擇就近的測(cè)試服務(wù)器減少延遲
使用HTTP/2提高傳輸效率
實(shí)現(xiàn)智能重試機(jī)制提高成功率

安全考慮

1. 數(shù)據(jù)安全

測(cè)試數(shù)據(jù)不包含敏感信息
結(jié)果文件權(quán)限設(shè)置合理
歷史記錄定期清理避免積累過多

2. 網(wǎng)絡(luò)安全

驗(yàn)證測(cè)試服務(wù)器的合法性
使用HTTPS加密傳輸
限制測(cè)試數(shù)據(jù)的大小避免濫用

3. 系統(tǒng)安全

避免測(cè)試過程中占用過多系統(tǒng)資源
實(shí)現(xiàn)合理的超時(shí)機(jī)制防止掛起
記錄詳細(xì)的日志便于問題追蹤

這個(gè)網(wǎng)絡(luò)帶寬測(cè)試工具是一個(gè)功能完整、準(zhǔn)確可靠的網(wǎng)絡(luò)性能測(cè)量工具，能夠幫助用戶全面了解網(wǎng)絡(luò)連接的質(zhì)量和性能。

到此這篇關(guān)于基于Python編寫一個(gè)網(wǎng)絡(luò)帶寬測(cè)試工具的文章就介紹到這了,更多相關(guān)Python網(wǎng)絡(luò)帶寬測(cè)試內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: