elinpf / net_inspect Goto Github PK

from net_inspect import NetInspect

net = NetInspect()
net.set_plugins(input_plugin='console')
net.run(input_path='log')

print('total devices:', len(net.cluster.devices))

for device in net.cluster.devices:
    info = device.info
    print(' | '.join([info.hostname, info.ip, info.vendor,
          info.version, info.model, info.cpu_usage]))

from net_inspect import NetInspect, vendor

net = NetInspect()
net.set_plugins(input_plugin='console')
net.set_external_templates('local_templates')  # 调用本地模板库
net.run(input_path='log')

print('total devices:', len(net.cluster.devices))

for device in net.cluster.devices:
    info = device.info
    clock = ''
    if device.vendor == vendor.Huawei:
        with device.search_cmd('display clock') as cmd:
            if cmd.parse_result:
                ps = cmd.parse_result[0]
                clock = f"{ps['year']}-{ps['month']}-{ps['day']} {ps['time']}"

    if device.vendor == vendor.H3C:
        with device.search_cmd('display clock') as cmd:
            if cmd.parse_result:
                ps = cmd.parse_result[0]
                clock = f"{ps['year']}-{ps['month']}-{ps['day']} {ps['time']}"

    print(' | '.join([info.hostname, clock]))

from net_inspect import NetInspect, EachVendorDeviceInfo, BaseInfo, Device


class AppendClock(BaseInfo):
    clock: str = ''  # 巡检时间


class EachVendorWithClock(EachVendorDeviceInfo):

    base_info_class = AppendClock  # 基本信息类

    def do_huawei_vrp_baseinfo_2(self, device: Device, info: AppendClock):
        # 添加do_<vendor_platform>_baseinfo_<something>方法，可以自动运行
        with device.search_cmd('display clock') as cmd:
            if cmd.parse_result:
                row = cmd.parse_result[0]
                info.clock = f'{row["year"]}-{row["month"]}-{row["day"]} {row["time"]}'

    def do_hp_comware_baseinfo_2(self, device: Device, info: AppendClock):
        with device.search_cmd('display clock') as cmd:
            if cmd.parse_result:
                row = cmd.parse_result[0]
                info.clock = f'{row["year"]}-{row["month"]}-{row["day"]} {row["time"]}'


if __name__ == '__main__':
    net = NetInspect()
    net.set_plugins(input_plugin='console')
    net.set_base_info_handler(EachVendorWithClock)  # 设置获取设备基本信息的处理类
    net.set_external_templates('local_templates')
    net.run(input_path='log')

    print('total devices:', len(net.cluster.devices))

    for device in net.cluster.devices:
        info = device.info  # type: AppendClock
        print(' | '.join([info.hostname, info.clock]))

from __future__ import annotations

from typing import TYPE_CHECKING
from net_inspect import NetInspect, vendor
from net_inspect.analysis_plugin import analysis, AnalysisPluginAbc

if TYPE_CHECKING:
    from net_inspect.analysis_plugin import TemplateInfo
    from net_inspect.domain import AnalysisResult


class AnalysisPluginWithOSPFStatus(AnalysisPluginAbc):
    """OSPF status 状态不能为Init"""

    @analysis.vendor(vendor.Huawei)
    @analysis.template_key('huawei_vrp_display_ospf_peer_brief.textfsm', ['neighbor', 'state'])
    def huawei_vrp(template: TemplateInfo, result: AnalysisResult):
        """华为状态检查"""
        for row in template['display ospf peer brief']:
            if row['state'].lower() == 'init':
                result.add_warning(f'{row["neighbor"]} is in init state')


if __name__ == '__main__':
    net = NetInspect()
    net.set_plugins(input_plugin='console')
    net.run(input_path='log')

    print('total devices:', len(net.cluster.devices))

    for device in net.cluster.devices:
        ospf_status = device.analysis_result.get('ospf status')
        warning_list = []
        for alarm in ospf_status:
            if alarm.is_warning:
                warning_list.append(alarm.message)

        print(' | '.join([device.info.hostname, ', '.join(warning_list)]))

from net_inspect import NetInspect, OutputPluginAbstract
from rich.table import Table
from rich.console import Console


class Output(OutputPluginAbstract):
    def main(self):
        self.check_args('title')  # 检查是否提供title参数
        console = Console()

        table = Table(title=self.args.output_params.get(
            'title'), show_lines=False)
        columns = ['hostname', 'ip', 'model', 'version', 'power status']
        for col in columns:
            table.add_column(col, justify='center')
        table.row_styles = ['green']

        for device in self.args.devices:
            info = device.info
            table.add_row(
                info.hostname,
                info.ip,
                info.model,
                info.version,
                'Abnormal' if info.analysis.power else 'Normal'
            )

        console.print(table)

if __name__ == '__main__':

    net = NetInspect()
    net.set_plugins(input_plugin='console', output_plugin=Output)
    cluster = net.run(input_path='log', output_plugin_params={
                    'title': '设备信息'})

from net_inspect import NetInspect, InputPluginResult, vendor

net = NetInspect()

d = InputPluginResult()
d.add_cmd('display clock', "2021-03-19 10:23:08+08:00")
d.hostname = 'Device'
d.vendor = vendor.Huawei

net.add_device(d)

net.run()

for device in net.cluster.devices:
    print(device.info.hostname)
    print(device.parse_result('dis clo'))

class AnalysisPluginWithFanStatus(AnalysisPluginAbc):
    """
    要求设备所有在位风扇模块运行在正常状态。
    """
    @analysis.vendor(vendor.H3C)
    @analysis.template_key('hp_comware_display_fan.textfsm', ['slot', 'id', 'status'])
    def hp_comware(template: TemplateInfo, result: AnalysisResult):
        """模块状态不为Normal的时候告警"""
        for row in template['display fan']:
            if row['status'].lower() != 'normal':
                result.add_warning(
                    f'Slot {row["slot"]} Fan {row["id"]} 状态异常' if row['slot'] else f'Fan {row["id"]} 状态异常')