网络流量助手是一个用于监控和分析网络流量的工具，能够提供详细的网络统计信息和异常检测。以下是优化后的代码和使用说明

代码优化与扩展

接口信息获取

• 使用 iputils 库获取更详细的接口信息：

  from iputils import iplookup
  # 获取接口信息
  try:
      nic_info = iplookup.get_interface_info('eth')
  except Exception as e:
      print(f"无法获取接口信息：{e}")
      exit(1)
  # 打印接口信息
  print(f"接口：{nic_info.name}")
  print(f"状态：{nic_info.status}")
  print(f"类型：{nic_info.type}")
  print(f"MAC地址：{nic_info.mac}")
  print("-" * 50)

• 显示所有可用接口：

  from iputils import ipinterfaces
  # 获取所有接口
  try:
      interfaces = ipinterfaces.get_list()
  except Exception as e:
      print(f"无法获取接口列表：{e}")
      exit(1)
  # 输出接口信息
  for intf in interfaces:
      print(f"接口：{intf.name}")
      print(f"状态：{intf.status}")
      print(f"类型：{intf.type}")
      print(f"MAC地址：{intf.mac}")
      print("-" * 50)

传输速率计算

• 优化测试时间和并发测试：

  import socket
  import threading
  import time
  # 测试超时时间
  TEST_TIMEOUT = 5  # 秒
  class TrafficTest:
      def __init__(self, host, port):
          self.host = host
          self.port = port
          self.socket = None
          self.start_time = None
          self.bytes_sent = 0
          self.bytes_received = 0
      def start(self):
          try:
              self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
              self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
              self.socket.bind(('...', self.port))
              self.socket.listen(5)
              self.socket.settimeout(TEST_TIMEOUT)
              self.start_time = time.time()
              self.socket.accept()
              self.socket.send(b'hello')
              self.bytes_sent += 5
          except socket.timeout:
              self.stop()
          except Exception as e:
              self.stop()
              print(f"连接错误：{e}")
      def stop(self):
          if self.socket:
              self.socket.close()
          self.bytes_sent = 0
          self.bytes_received = 0
          self.start_time = None
      def get_stats(self):
          return {
              'sent': self.bytes_sent,
              'received': self.bytes_received,
              'time': time.time() - self.start_time
          }
  # 创建测试线程
  def test_connection(host, port):
      try:
          test = TrafficTest(host, port)
          test.start()
          while True:
              stats = test.get_stats()
              yield stats
      except Exception as e:
          print(f"测试失败：{e}")
  # 初始化测试
  def main():
      print("开始测试...")
      targets = [(f'192.168.1.{i}', 80) for i in range(2,4)]  # 示例目标
      results = []
      threads = []
      for host, port in targets:
          print(f"测试目标：{host}:{port}")
          thread = threading.Thread(target=test_connection, args=(host, port))
          thread.start()
          threads.append(thread)
      while threads:
          time.sleep(.1)
          for thread in threads:
              if not thread.isAlive():
                  thread.join()
                  stats = next(thread.stats, None)
                  if stats:
                      results.append({
                          'target': (host, port),
                          'sent': stats['sent'],
                          'received': stats['received'],
                          'time': stats['time']
                      })
      print("测试完成，生成报告...")
      generate_report(results)
  if __name__ == "__main__":
      main()

• 错误处理和并发测试：

  通过并发测试，减少测试时间，并增加错误处理机制,以应对网络波动。

报告生成

• 使用 matplotlib 生成图表，并支持多种格式输出：

  import matplotlib.pyplot as plt
  from datetime import datetime
  def generate_report(results):
      # 生成发送速率图表
      plt.figure(figsize=(10, 6))
      send_rates = [(stat['sent'] / stat['time']) * 100 for stat in results]
      plt.plot([i for i in range(len(results))], send_rates, marker='o',
               markersize=8, linestyle='-', linewidth=2, color='blue')
      plt.xlabel('目标')
      plt.ylabel('发送速率 (KB/s)')
      plt.title('发送速率测试结果')
      plt.grid(True, alpha=.3)
      plt.savefig('send_rate.png', dpi=300, bbox_inches='tight')
      plt.close()
      # 生成接收速率图表
      recv_rates = [(stat['received'] / stat['time']) * 100 for stat in results]
      plt.figure(figsize=(10, 6))
      plt.plot([i for i in range(len(results))], recv_rates, marker='o',
               markersize=8, linestyle='-', linewidth=2, color='red')
      plt.xlabel('目标')
      plt.ylabel('接收速率 (KB/s)')
      plt.title('接收速率测试结果')
      plt.grid(True, alpha=.3)
      plt.savefig('recv_rate.png', dpi=300, bbox_inches='tight')
      plt.close()
      print("生成的图表已保存在当前目录：send_rate.png 和 recv_rate.png")

• 支持不同格式输出：

  # 在 `generate_report` 函数中添加格式选择
  def generate_report(results, format='png'):
      # 生成图表并保存
      plt.figure(figsize=(10, 6))
      send_rates = [(stat['sent'] / stat['time']) * 100 for stat in results]
      plt.plot([i for i in range(len(results))], send_rates, marker='o',
               markersize=8, linestyle='-', linewidth=2, color='blue')
      plt.xlabel('目标')
      plt.ylabel('发送速率 (KB/s)')
      plt.title('发送速率测试结果')
      plt.grid(True, alpha=.3)
      if format == 'png':
          plt.savefig('send_rate.png', dpi=300, bbox_inches='tight')
      elif format == 'pdf':
          plt.savefig('send_rate.pdf', dpi=300, bbox_inches='tight')
      plt.close()
      # 同样处理接收速率
      # ...

日志记录

• 使用 logging 模块，增加上下文信息：

  import logging
  from datetime import datetime
  # 配置日志格式
  logging.basicConfig(
      level=logging.INFO,
      format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
      handlers=[
          logging.FileHandler('network_traffic.log'),
          logging.StreamHandler()
      ]
  )
  # 创建日志记录类
  class TrafficLogger:
      def __init__(self):
          self.logger = logging.getLogger(__name__)
          self.logger.setLevel(logging.INFO)
      def info(self, msg):
          self.logger.info(f"网络流量信息：{msg}")
      def error(self, msg):
          self.logger.error(f"网络流量异常：{msg}")
      def debug(self, msg):
          self.logger.debug(f"网络流量调试信息：{msg}")

• 记录每次测试的详细信息：

  # 在 `main()` 函数中添加日志记录
  def main():
      print("开始测试...")
      targets = [(f'192.168.1.{i}', 80) for i in range(2,4)]
      results = []
      threads = []
      logger = TrafficLogger()
      for host, port in targets:
          print(f"测试目标：{host}:{port}")
          thread = threading.Thread(target=test_connection, args=(host, port))
          thread.start()
          threads.append(thread)
          logger.info(f"开始测试目标：{host}:{port}")
      while threads:
          time.sleep(.1)
          for thread in threads:
              if not thread.isAlive():
                  thread.join()
                  stats = next(thread.stats, None)
                  if stats:
                      results.append(stats)
                      logger.info(f"测试完成，发送速率：{stats['sent']} KB/s，接收速率：{stats['received']} KB/s")
      logger.info("测试完成，生成报告...")
      generate_report(results, 'png')

异常检测

• **使用移动平均