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

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

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

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ù)文件格式錯誤")
    raise NetworkTestError("歷史數(shù)據(jù)文件損壞")

try:
    # 大文件下載時的內(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)存不足，無法完成測試")
    raise NetworkTestError("系統(tǒng)內(nèi)存不足")

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
網(wǎng)絡(luò)帶寬測試工具
功能：測量網(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ò)測試異常類"""
    pass

class BandwidthTester:
    def __init__(self, config):
        self.test_server = config.get('server', 'http://speedtest.tele2.net')
        self.duration = config.get('duration', 10)  # 測試持續(xù)時間（秒）
        self.threads = config.get('threads', 4)     # 測試線程數(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)
        
        # 測試結(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):
        """測試網(wǎng)絡(luò)延遲"""
        logger.info("開始測試網(wǎng)絡(luò)延遲...")
        ping_times = []
        
        try:
            for i in range(5):  # 測試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測試 #{i+1}: {ping_time:.2f}ms")
                else:
                    logger.warning(f"Ping測試 #{i+1} 失敗，狀態(tài)碼: {response.status_code}")
                
                time.sleep(0.1)  # 避免請求過于頻繁
                
            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, 抖動: {self.results['jitter']:.2f}ms")
            else:
                logger.error("所有Ping測試都失敗")
                raise NetworkTestError("無法完成Ping測試")
                
        except requests.exceptions.RequestException as e:
            logger.error(f"Ping測試失敗: {str(e)}")
            raise NetworkTestError(f"Ping測試失敗: {str(e)}")
            
    def download_test(self):
        """下載速度測試"""
        logger.info("開始下載速度測試...")
        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
        
        # 找到最適合的測試文件
        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("無法找到可用的測試文件")
            raise NetworkTestError("測試服務(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"下載測試失敗，狀態(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)
                        
                        # 計算瞬時速度
                        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í)行下載測試
        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é)果時出錯: {str(e)}")
                    
        end_time = time.time()
        test_duration = end_time - start_time
        
        if test_duration > 0 and total_bytes > 0:
            # 計算平均下載速度 (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("下載測試未能獲取有效數(shù)據(jù)")
            raise NetworkTestError("下載測試失敗")
            
    def upload_test(self):
        """上傳速度測試"""
        logger.info("開始上傳速度測試...")
        
        # 生成測試數(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:
                # 嘗試找到上傳測試端點
                upload_url = f"{self.test_server}/upload.php"
                
                # 發(fā)送測試數(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:  # 限制上傳測試時間
                    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
                            
                            # 計算瞬時速度
                            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} 請求失敗: {str(e)}")
                        
                    time.sleep(0.1)  # 避免請求過于頻繁
                    
                return bytes_sent, worker_samples
                
            except Exception as e:
                logger.error(f"上傳線程 {worker_id} 失敗: {str(e)}")
                return 0, []
                
        # 如果上傳測試不可用，使用模擬數(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é)果時出錯: {str(e)}")
                        
            end_time = time.time()
            test_duration = end_time - start_time
            
            if test_duration > 0 and total_bytes > 0:
                # 計算平均上傳速度 (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:
                # 如果上傳測試失敗，使用下載速度的一個比例作為估算
                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")
                
        except Exception as e:
            logger.error(f"上傳測試過程中發(fā)生錯誤: {str(e)}")
            # 使用下載速度的一個比例作為估算
            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):
        """丟包率測試"""
        logger.info("開始丟包率測試...")
        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"丟包測試 #{i+1}: {'成功' if response.status_code == 200 else '失敗'}")
                        
                except requests.exceptions.RequestException:
                    logger.debug(f"丟包測試 #{i+1}: 超時")
                    
                time.sleep(0.05)  # 避免請求過于頻繁
                
            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"丟包率測試失敗: {str(e)}")
            self.results['packet_loss'] = 0.0
            
    def run_full_test(self):
        """運行完整測試"""
        logger.info("開始完整的網(wǎng)絡(luò)帶寬測試...")
        start_time = time.time()
        
        try:
            # 1. 延遲測試
            self.test_ping()
            
            # 2. 下載速度測試
            self.download_test()
            
            # 3. 上傳速度測試
            self.upload_test()
            
            # 4. 丟包率測試
            self.packet_loss_test()
            
            end_time = time.time()
            self.results['test_duration'] = round(end_time - start_time, 2)
            
            logger.info("網(wǎng)絡(luò)帶寬測試完成")
            return self.results
            
        except Exception as e:
            logger.error(f"測試過程中發(fā)生錯誤: {str(e)}")
            raise NetworkTestError(f"測試失敗: {str(e)}")
            
    def run_quick_test(self):
        """快速測試（僅測試下載速度）"""
        logger.info("開始快速網(wǎng)絡(luò)測試...")
        start_time = time.time()
        
        try:
            # 1. 延遲測試
            self.test_ping()
            
            # 2. 下載速度測試
            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ò)測試完成")
            return self.results
            
        except Exception as e:
            logger.error(f"快速測試過程中發(fā)生錯誤: {str(e)}")
            raise NetworkTestError(f"快速測試失敗: {str(e)}")
            
    def print_results(self):
        """打印測試結(jié)果"""
        print("\n" + "="*60)
        print("網(wǎng)絡(luò)帶寬測試報告")
        print("="*60)
        print(f"測試時間: {self.results['timestamp']}")
        print(f"測試服務(wù)器: {self.results['test_server']}")
        print(f"測試耗時: {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"延遲抖動: {self.results['jitter']:.2f} ms")
        print(f"丟包率: {self.results['packet_loss']:.2f}%")
        print("="*60)
        
    def plot_results(self):
        """繪制測試結(jié)果圖表"""
        try:
            if not self.results['samples']:
                logger.warning("沒有采樣數(shù)據(jù)，無法生成圖表")
                return
                
            # 按時間排序采樣數(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('時間 (秒)')
            plt.ylabel('速度 (Mbps)')
            plt.title('網(wǎng)絡(luò)帶寬測試速度曲線')
            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"帶寬測試圖表已保存到 {chart_file}")
            
        except Exception as e:
            logger.error(f"生成圖表時出錯: {str(e)}")
            
    def save_results(self):
        """保存測試結(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"保存測試結(jié)果時出錯: {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"測試結(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)
                    
            # 添加當前結(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"測試結(jié)果已添加到歷史記錄 {self.history_file}")
            
        except Exception as e:
            logger.error(f"保存歷史記錄時出錯: {str(e)}")
            
    def show_history(self, count=10):
        """顯示歷史測試結(jié)果"""
        try:
            if not os.path.exists(self.history_file):
                logger.info("沒有歷史測試記錄")
                return
                
            with open(self.history_file, 'r', encoding='utf-8') as f:
                history_data = json.load(f)
                
            if not history_data:
                logger.info("沒有歷史測試記錄")
                return
                
            # 顯示最近的記錄
            recent_records = history_data[-count:] if len(history_data) > count else history_data
            
            print("\n" + "="*80)
            print("歷史測試記錄")
            print("="*80)
            print(f"{'時間':<20} {'下載(Mbps)':<12} {'上傳(Mbps)':<12} {'延遲(ms)':<10} {'丟包率(%)':<10}")
            print("-"*80)
            
            for record in recent_records:
                timestamp = record['timestamp'][:19]  # 截取日期時間部分
                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"讀取歷史記錄時出錯: {str(e)}")

def main():
    parser = argparse.ArgumentParser(description='網(wǎng)絡(luò)帶寬測試工具')
    parser.add_argument('--server', default='http://speedtest.tele2.net', help='測試服務(wù)器地址')
    parser.add_argument('--duration', type=int, default=10, help='測試持續(xù)時間（秒）')
    parser.add_argument('--threads', type=int, default=4, help='測試線程數(shù)')
    parser.add_argument('--quick', action='store_true', help='快速測試模式')
    parser.add_argument('--history', action='store_true', help='顯示歷史測試記錄')
    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='詳細輸出')
    
    args = parser.parse_args()
    
    # 配置測試參數(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)建測試器實例
    tester = BandwidthTester(config)
    
    try:
        if args.history:
            # 顯示歷史記錄
            tester.show_history(args.count)
        else:
            # 執(zhí)行測試
            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("測試被用戶中斷")
        sys.exit(1)
    except NetworkTestError as e:
        logger.error(f"網(wǎng)絡(luò)測試錯誤: {str(e)}")
        sys.exit(1)
    except Exception as e:
        logger.error(f"測試過程中發(fā)生未知錯誤: {str(e)}")
        sys.exit(1)

if __name__ == '__main__':
    main()

# 運行完整測試
python bandwidth_tester.py

# 運行快速測試
python bandwidth_tester.py --quick

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

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

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

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

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

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

# 詳細輸出模式
python bandwidth_tester.py -v

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

# 詳細輸出的完整測試
python bandwidth_tester.py -v --duration 20 --threads 6

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()

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

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

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")

import json
import statistics

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

# 分析下載速度趨勢
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")