Python 开发网络流量分析工具的方法

import psutil
import time
from concurrent.futures import ThreadPoolExecutor
import socket
import requests
import logging
from datetime import datetime

# 配置日志系统
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

class NetworkAnalyzer:
    def __init__(self, check_interval=5):
        self.check_interval = check_interval
        self.running = True
        
    def monitor_traffic(self):
        """监控网络流量统计"""
        prev_bytes_sent = psutil.net_io_counters().bytes_sent
        prev_bytes_recv = psutil.net_io_counters().bytes_recv
        
        while self.running:
            time.sleep(self.check_interval)
            
            current_stats = psutil.net_io_counters()
            bytes_sent = current_stats.bytes_sent - prev_bytes_sent
            bytes_recv = current_stats.bytes_recv - prev_bytes_recv
            
            sent_rate = bytes_sent / self.check_interval
            recv_rate = bytes_recv / self.check_interval
            
            logging.info(f"流量统计 - 上传: {sent_rate:.2f} B/s, "
                        f"下载: {recv_rate:.2f} B/s")
            
            prev_bytes_sent = current_stats.bytes_sent
            prev_bytes_recv = current_stats.bytes_recv
    
    def check_tcp_service(self, host, port):
        """检查TCP服务可用性"""
        try:
            with socket.create_connection((host, port), timeout=5):
                return True
        except socket.error:
            return False
    
    def monitor_services(self, services_config):
        """监控多个网络服务"""
        while self.running:
            for service in services_config:
                host, port = service['host'], service['port']
                is_available = self.check_tcp_service(host, port)
                
                status = "可用" if is_available else "不可用"
                logging.info(f"服务检测 - {host}:{port} - {status}")
            
            time.sleep(self.check_interval)
    
    def http_performance_test(self, url):
        """测试HTTP服务性能"""
        try:
            start_time = time.time()
            response = requests.get(url, timeout=10)
            response_time = time.time() - start_time
            
            return {
                'status_code': response.status_code,
                'response_time': response_time,
                'success': True
            }
        except requests.RequestException as e:
            return {'success': False, 'error': str(e)}
    
    def start_monitoring(self, services_config, http_urls):
        """启动综合监控"""
        with ThreadPoolExecutor(max_workers=3) as executor:
            # 提交监控任务
            futures = [
                executor.submit(self.monitor_traffic),
                executor.submit(self.monitor_services, services_config),
                executor.submit(self.monitor_http_performance, http_urls)
            ]
            
            try:
                # 等待任务完成
                for future in futures:
                    future.result()
            except KeyboardInterrupt:
                self.running = False
                logging.info("监控程序已停止")
    
    def monitor_http_performance(self, urls):
        """监控HTTP服务性能"""
        while self.running:
            for url in urls:
                result = self.http_performance_test(url)
                if result['success']:
                    logging.info(f"HTTP性能 - {url} - "
                               f"状态码: {result['status_code']}, "
                               f"响应时间: {result['response_time']:.3f}s")
                else:
                    logging.error(f"HTTP性能 - {url} - 错误: {result['error']}")
            
            time.sleep(self.check_interval)

# 使用示例
if __name__ == "__main__":
    analyzer = NetworkAnalyzer(check_interval=3)
    
    # 配置要监控的服务
    services = [
        {'host': 'google.com', 'port': 80},
        {'host': 'github.com', 'port': 443}
    ]
    
    # 配置要测试的HTTP服务
    http_urls = [
        ' https://httpbin.org/get',
        ' https://jsonplaceholder.typicode.com/posts/1'
    ]
    
    try:
        analyzer.start_monitoring(services, http_urls)
    except KeyboardInterrupt:
        analyzer.running = False