一学一实践:经典自动化协议之Python采集Netflow数据并通过饼图呈现

pip install bokeh pandas

from bokeh.plotting import figure, output_file, savefrom bokeh.transform import cumsumfrom bokeh.palettes import Category10import pandas as pdfrom math import piimport osfrom pathlib import PathOUTPUTS_DIR = Path(__file__).resolve().parent / 'outputs'def bokeh_bing(name_list, count_list, bing_name, save_name=None):    """使用 Bokeh 绘制饼状图, 生成交互式 HTML 文件。"""    # 构建数据字典, 转换为 DataFrame    data_dict = dict(zip(name_list, [float(c) for c in count_list]))    data = pd.Series(data_dict).reset_index(name='bytes').rename(columns={'index': 'application'})    # 计算每个扇形的角度    data['angle'] = data['bytes'] / data['bytes'].sum() * 2 * pi    # 分配颜色    num = len(data_dict)    if num <= 2:        data['color'] = Category10[3][:num]    elif num <= 10:        data['color'] = Category10[num]    else:        data['color'] = (Category10[10] * ((num // 10) + 1))[:num]    # 计算百分比    data['percentage'] = (data['bytes'] / data['bytes'].sum() * 100).round(2).astype(str) + '%'    # 创建图表    p = figure(height=500, width=700, title=bing_name, toolbar_location="right",               tools="hover,pan,wheel_zoom,box_zoom,reset,save",               tooltips="@application: @bytes (@percentage)", x_range=(-0.5, 1.0))    # 绘制饼图    p.wedge(x=0, y=1, radius=0.4,            start_angle=cumsum('angle', include_zero=True), end_angle=cumsum('angle'),            line_color="white", fill_color='color', legend_field='application', source=data)    # 美化    p.axis.axis_label = None    p.axis.visible = False    p.grid.grid_line_color = None    p.title.text_font_size = '16pt'    p.legend.label_text_font_size = '12pt'    p.legend.location = "center_right"    # 输出到 outputs 目录    os.makedirs(OUTPUTS_DIR, exist_ok=True)    output_filename = save_name if save_name else str(OUTPUTS_DIR / f"{bing_name}.html")    output_file(output_filename, title=bing_name)    save(p)    print(f"[*] Bokeh 饼状图已生成: {output_filename}")测试代码:bokeh_bing(['名称1', '名称2', '名称3'], [1000, 123, 444], '测试饼图')

import reimport osimport sys# 添加tools目录到Python路径sys.path.append(os.path.join(os.path.dirname(__file__), 'tools'))from day3_ssh_single_cmd import ssh_runfrom day3_bokeh_bing import bokeh_bingdef get_netflow_app(host, username, password):    """SSH登录路由器, 采集Netflow数据, 正则提取, 绘制Bokeh饼状图。"""    # 1. SSH执行命令获取Netflow数据    show_result = ssh_run(host, username, password,                          'show flow monitor name qytang-monitor cache format table')    if show_result is None:        print("[!] 无法获取Netflow数据")        return    print("\n[*] Netflow数据采集结果:")    print(show_result)    # 2. ★ 正则提取APP NAME和bytes    app_name_list = []    app_bytes_list = []    # 定义正则表达式模式    # APP NAME由前缀(port / layer7 / prot)加空格加应用名组成    # bytes为行末的数字    pattern = r'^(port|layer7|prot)\s+(.+?)\s+(\d+)$'    for line in show_result.strip().split('\n'):        line = line.strip()        # 跳过标题行和分隔线        if 'APP NAME' in line or '=' in line:            continue        # 使用正则表达式匹配        match = re.match(pattern, line)        if match:            # 构建完整的APP NAME            app_name = f"{match.group(1)}{match.group(2)}"            app_bytes = match.group(3)            app_name_list.append(app_name)            app_bytes_list.append(app_bytes)    # 3. 打印提取结果    if app_name_list and app_bytes_list:        print(f"\n[*] 提取到 {len(app_name_list)} 条 Netflow 记录")        for name, byt in zip(app_name_list, app_bytes_list):            print(f"    {name:<25s}{byt} bytes")    else:        print("\n[!] 未提取到Netflow记录")        return    # 4. 调用bokeh_bing生成饼状图    bokeh_bing(app_name_list, app_bytes_list, 'Netflow应用流量分布')if __name__ == "__main__":    # 设置路由器信息    router_host = '10.10.1.1'  # 用户提供的路由器IP    router_username = 'XXX'  # 用户提供的正确用户名    router_password = 'XXX'  # 用户提供的正确密码    print("=== NetDevOps-任务2 ===")    print(f"正在从路由器 {router_host} 采集Netflow数据...")    get_netflow_app(router_host, router_username, router_password)