如何写一个脚本统计网络使用量

wen 实用脚本 2026-07-28 88

本文目录导读：

如何写一个脚本统计网络使用量

Linux Shell 脚本
Python 脚本
安装和使用
功能特点

我来帮你写一个统计网络使用量的脚本,这里提供几个不同平台的版本：

Linux Shell 脚本

基础版（统计接口流量）

#!/bin/bash
# 网络流量统计脚本
# 使用方法: ./network_stats.sh [接口名]
INTERFACE=${1:-eth0}  # 默认统计 eth0
# 获取当前流量数据
get_traffic() {
    local rx=$(cat /sys/class/net/$INTERFACE/statistics/rx_bytes)
    local tx=$(cat /sys/class/net/$INTERFACE/statistics/tx_bytes)
    echo "$rx $tx"
}
# 格式化字节数
format_bytes() {
    local bytes=$1
    if [ $bytes -lt 1024 ]; then
        echo "${bytes}B"
    elif [ $bytes -lt 1048576 ]; then
        echo "$((bytes / 1024))KB"
    elif [ $bytes -lt 1073741824 ]; then
        echo "$((bytes / 1048576))MB"
    else
        echo "$((bytes / 1073741824))GB"
    fi
}
# 主循环
echo "统计接口: $INTERFACE"
echo "按 Ctrl+C 退出"
echo "------------------------"
# 获取初始值
read rx1 tx1 <<< $(get_traffic)
sleep 1
while true; do
    read rx2 tx2 <<< $(get_traffic)
    # 计算速率 (bytes/s)
    rx_speed=$((rx2 - rx1))
    tx_speed=$((tx2 - tx1))
    # 转换为可读格式
    rx_format=$(format_bytes $rx_speed)
    tx_format=$(format_bytes $tx_speed)
    clear
    echo "网络流量统计 (按 Ctrl+C 退出)"
    echo "接口: $INTERFACE"
    echo "------------------------"
    echo "下载速率: ${rx_format}/s"
    echo "上传速率: ${tx_format}/s"
    echo "------------------------"
    echo "总下载: $(format_bytes $rx2)"
    echo "总上传: $(format_bytes $tx2)"
    rx1=$rx2
    tx1=$tx2
    sleep 1
done

进阶版（增强功能）

#!/bin/bash
# 增强版网络流量统计脚本
# 功能：实时监控、日志记录、阈值告警
LOG_FILE="/var/log/network_stats.log"
THRESHOLD_HIGH=10485760  # 10MB/s 告警阈值
# 获取所有网络接口
get_interfaces() {
    ls /sys/class/net/ | grep -v lo
}
# 初始化统计
init_stats() {
    declare -gA rx_start tx_start
    for iface in $(get_interfaces); do
        rx_start[$iface]=$(cat /sys/class/net/$iface/statistics/rx_bytes)
        tx_start[$iface]=$(cat /sys/class/net/$iface/statistics/tx_bytes)
    done
}
# 主监控函数
monitor() {
    echo "开始监控网络流量..."
    echo "时间: $(date '+%Y-%m-%d %H:%M:%S')" | tee -a $LOG_FILE
    init_stats
    sleep 1
    while true; do
        local timestamp=$(date '+%H:%M:%S')
        clear
        echo "网络流量实时监控"
        echo "时间: $timestamp"
        echo "========================="
        for iface in $(get_interfaces); do
            local rx_now=$(cat /sys/class/net/$iface/statistics/rx_bytes)
            local tx_now=$(cat /sys/class/net/$iface/statistics/tx_bytes)
            local rx_speed=$((rx_now - rx_start[$iface]))
            local tx_speed=$((tx_now - tx_start[$iface]))
            # 检测是否超过阈值
            if [ $rx_speed -gt $THRESHOLD_HIGH ]; then
                echo "⚠️ $iface 下载速率过高: $(echo $rx_speed | numfmt --to=iec)/s"
            fi
            echo "接口: $iface"
            echo "  ↓ 下载: $(numfmt --to=iec $rx_speed)/s"
            echo "  ↑ 上传: $(numfmt --to=iec $tx_speed)/s"
            echo "---"
            # 更新起始值
            rx_start[$iface]=$rx_now
            tx_start[$iface]=$tx_now
            # 记录日志
            echo "$timestamp,$iface,$rx_speed,$tx_speed" >> $LOG_FILE
        done
        sleep 1
    done
}
# 生成报告
generate_report() {
    echo "=== 网络使用报告 ==="
    echo "统计周期: $(date -d @$(head -1 $LOG_FILE | cut -d',' -f1) '+%H:%M:%S') - $(date '+%H:%M:%S')"
    local total_rx=0
    local total_tx=0
    for iface in $(get_interfaces); do
        local rx=$(cat /sys/class/net/$iface/statistics/rx_bytes)
        local tx=$(cat /sys/class/net/$iface/statistics/tx_bytes)
        echo "接口: $iface"
        echo "  总下载: $(numfmt --to=iec $rx)"
        echo "  总上传: $(numfmt --to=iec $tx)"
        total_rx=$((total_rx + rx))
        total_tx=$((total_tx + tx))
    done
    echo "总计:"
    echo "  总下载: $(numfmt --to=iec $total_rx)"
    echo "  总上传: $(numfmt --to=iec $total_tx)"
}
# 命令行参数处理
case "$1" in
    monitor)
        monitor
        ;;
    report)
        generate_report
        ;;
    *)
        echo "使用方法: $0 {monitor|report}"
        echo "  monitor  - 实时监控网络流量"
        echo "  report   - 生成网络使用报告"
        ;;
esac

Python 脚本

#!/usr/bin/env python3
import psutil
import time
import sys
from datetime import datetime
class NetworkMonitor:
    def __init__(self):
        self.old_stats = {}
        self.new_stats = {}
    def get_network_stats(self):
        """获取网络接口统计"""
        net_io = psutil.net_io_counters(pernic=True)
        stats = {}
        for interface, data in net_io.items():
            if interface != 'lo':  # 排除回环接口
                stats[interface] = {
                    'bytes_sent': data.bytes_sent,
                    'bytes_recv': data.bytes_recv,
                    'packets_sent': data.packets_sent,
                    'packets_recv': data.packets_recv,
                    'errin': data.errin,
                    'errout': data.errout,
                    'dropin': data.dropin,
                    'dropout': data.dropout
                }
        return stats
    def format_bytes(self, bytes_val):
        """格式化字节数"""
        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
            if bytes_val < 1024.0:
                return f"{bytes_val:.2f} {unit}"
            bytes_val /= 1024.0
        return f"{bytes_val:.2f} PB"
    def calculate_speed(self, interface):
        """计算传输速率"""
        if interface in self.old_stats and interface in self.new_stats:
            old = self.old_stats[interface]
            new = self.new_stats[interface]
            time_diff = self.time_diff
            rx_speed = (new['bytes_recv'] - old['bytes_recv']) / time_diff
            tx_speed = (new['bytes_sent'] - old['bytes_sent']) / time_diff
            return rx_speed, tx_speed
        return 0, 0
    def display_stats(self, interfaces=None):
        """显示网络统计"""
        os.system('clear' if sys.platform == 'linux' else 'cls')
        print(f"\n{'='*60}")
        print(f"网络流量监控 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"{'='*60}")
        self.new_stats = self.get_network_stats()
        if self.old_stats:
            for interface in self.new_stats:
                if interfaces and interface not in interfaces:
                    continue
                rx_speed, tx_speed = self.calculate_speed(interface)
                current = self.new_stats[interface]
                print(f"\n接口: {interface}")
                print(f"  {'↓ 下载:':<12} {self.format_bytes(rx_speed)}/s")
                print(f"  {'↑ 上传:':<12} {self.format_bytes(tx_speed)}/s")
                print(f"  {'总下载:':<12} {self.format_bytes(current['bytes_recv'])}")
                print(f"  {'总上传:':<12} {self.format_bytes(current['bytes_sent'])}")
                # 显示错误/丢包信息
                if current['errin'] > 0 or current['errout'] > 0:
                    print(f"  ⚠️  错误: RX {current['errin']}, TX {current['errout']}")
                if current['dropin'] > 0 or current['dropout'] > 0:
                    print(f"  ⚠️  丢包: RX {current['dropin']}, TX {current['dropout']}")
        self.old_stats = self.new_stats
    def monitor(self, interval=1, interfaces=None):
        """实时监控"""
        try:
            print("开始监控... (按 Ctrl+C 退出)")
            time.sleep(1)
            while True:
                self.time_diff = interval
                self.display_stats(interfaces)
                time.sleep(interval)
        except KeyboardInterrupt:
            print("\n\n监控已停止")
            self.save_report()
    def save_report(self):
        """保存使用报告"""
        if self.new_stats:
            report_file = f"network_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
            with open(report_file, 'w') as f:
                f.write("=== 网络使用报告 ===\n")
                f.write(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write("=" * 40 + "\n\n")
                for interface, stats in self.new_stats.items():
                    f.write(f"接口: {interface}\n")
                    f.write(f"  总下载: {self.format_bytes(stats['bytes_recv'])}\n")
                    f.write(f"  总上传: {self.format_bytes(stats['bytes_sent'])}\n")
                    f.write(f"  总数据包: {stats['packets_recv'] + stats['packets_sent']}\n\n")
            print(f"报告已保存: {report_file}")
def main():
    import os
    monitor = NetworkMonitor()
    if len(sys.argv) > 1:
        if sys.argv[1] == '--help' or sys.argv[1] == '-h':
            print("使用方法: python network_monitor.py [选项]")
            print("选项:")
            print("  -i, --interface  指定监控接口 (逗号分隔)")
            print("  -t, --time       更新时间间隔 (秒)")
            print("  示例: python network_monitor.py -i eth0,wlan0 -t 2")
            return
        # 解析参数
        interfaces = None
        interval = 1
        for i, arg in enumerate(sys.argv):
            if arg in ['-i', '--interface'] and i + 1 < len(sys.argv):
                interfaces = sys.argv[i + 1].split(',')
            elif arg in ['-t', '--time'] and i + 1 < len(sys.argv):
                interval = float(sys.argv[i + 1])
        monitor.monitor(interval, interfaces)
    else:
        monitor.monitor()
if __name__ == "__main__":
    main()

安装和使用

Linux Shell 脚本

# 保存脚本
chmod +x network_stats.sh
# 运行
./network_stats.sh eth0  # 监控指定接口
./network_stats.sh        # 默认监控 eth0

Python 脚本

# 安装依赖
pip install psutil
# 运行
python network_monitor.py                    # 监控所有接口
python network_monitor.py -i eth0,wlan0      # 监控指定接口
python network_monitor.py -t 2               # 每2秒更新