网络Python案例：浪潮交换机保存接口的状态（查看华为交换机的mac地址表，此案例同样适用于华为华三，修改下执行命令即可）

网络Python案例：浪潮交换机保存接口的状态

测试的Python版本：

程序特点：

1、无限循环，直到被手动关闭。

2、每个月的txt保存到同一个文件夹、每天的txt保存到一个文件夹、每个txt都带时间，精确到秒（获取当前时间戳，精确到秒，并构造输出文件夹名）

3、输出结果禁用分页

4、多个命令执行后的结果保存到同一个txt中

5、用户名密码是从另一个txt文件中读取

6、如果多台设备，放到username2.txt中，txt中的一行为一台设备的信息。

7、特别注意，此代码必须有2条命令以上被执行，否则输出的txt结果文件少了后面大部分输出。

代码开始，如下：

import socket
import time
import os

#PyCharm 2023.3.3
#Python3.9.12
#Anaconda Navigator (Anaconda3)

from datetime import datetime
#cjh,输出的文件夹目录自定义前缀
# 配置参数
input_file = r"D:\pythonDate1\username1\username2.txt" # 输入文件路径
base_output_dir = r"D:\pythonDate1\output1" # 基础输出目录
command_interval = 300 # 20秒间隔、可以自行修改为60秒
separator = "=" * 20 # 20个分隔符
commands = [
"show clock",
"show interface brief",
#特别注意，此代码必须有2条命令以上被执行，否则输出的txt结果文件少了后面大部分输出。
# "show logging file",
# "display users",
"display clock",
# "display mac-address flapping record",
"display arp",
# "display interface brief",
# "display interface GigabitEthernet 0/0/43",
# "ping -c 4 172.16.1.1",
# "ping -c 4 172.16.1.205",
# "display mac-address",
# "display arp | include 172.16.1.205",
# "display mac-address | include xxxx-xxxx-xxxx",
]

class TelnetClient:
    def __init__(self, host, port=23, timeout=10):
        self.host = host
        self.port = port
        self.timeout = timeout
        self.sock = None

    def connect(self):
        """建立Telnet连接"""
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.settimeout(self.timeout)
        self.sock.connect((self.host, self.port))
        return self

    def login(self, username, password):
        """登录设备"""
        self.read_until("Username:")
        self.write(username + "\n")
        self.read_until("Password:")
        self.write(password + "\n")
        self.read_until(">") # 等待登录完成
        self.write("screen-length 0 temporary\n") # 禁用分页
        time.sleep(1)
        return self

    def write(self, text):
        """发送命令"""
        self.sock.send(text.encode('ascii'))

    def read_until(self, prompt, timeout=None):
        """读取直到遇到指定提示"""
        buffer = ""
        end_time = time.time() + (timeout or self.timeout)

        while time.time() < end_time:
            try:
                data = self.sock.recv(4096).decode('ascii', errors='ignore')
                if not data:
                    break
                buffer += data
                if prompt in buffer:
                    return buffer
            except socket.timeout:
                break
        return buffer

    def execute(self, command):
        """执行命令并返回结果"""
        self.write(command + "\n")
        time.sleep(2) # 等待命令执行
        return self.read_until(">")

    def close(self):
        """关闭连接"""
        if self.sock:
            self.sock.close()

def create_output_directory():
    """创建带时间戳的输出目录"""
    timestamp = datetime.now().strftime("%Y年%m分%d日")
    output_dir = os.path.join(base_output_dir, f"FileSW-##-{timestamp}")  # 修改处：添加eee-前缀
    os.makedirs(output_dir, exist_ok=True)
    return output_dir

def generate_output_filename():
    """生成带时间戳的输出文件名"""
    return f"txt-sw-###-1_{datetime.now().strftime('%Y年%m月%d日_%H时%M分%S秒')}.txt"

def load_credentials(file_path):
    """从文件加载IP、用户名和密码"""
    devices = []
    try:
        with open(file_path, 'r') as f:
            for line in f:
                if line.strip():
                    parts = line.strip().split(',')
                    if len(parts) == 3:
                        ip, username, password = parts
                        devices.append({
                            'ip': ip.strip(),
                            'username': username.strip(),
                            'password': password.strip()
                        })
        return devices
    except Exception as e:
        print(f"读取凭证文件失败: {str(e)}")
        return []

def main():
    # 确保基础输出目录存在
    os.makedirs(base_output_dir, exist_ok=True)

    # 加载设备凭证
    devices = load_credentials(input_file)
    if not devices:
        print("没有可用的设备凭证，请检查输入文件")
        return

    while True:
        try:
            # 创建带时间戳的输出目录
            output_dir = create_output_directory()
            output_filename = generate_output_filename()
            output_path = os.path.join(output_dir, output_filename)

            with open(output_path, 'w', encoding='utf-8') as f:
                for device in devices:
                    try:
                        # 连接设备
                        telnet = TelnetClient(device['ip']).connect()
                        telnet.login(device['username'], device['password'])

                        # 写入设备信息
                        f.write(
                            f"\n{separator} {device['ip']} - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} {separator}\n")

                        # 执行命令
                        for cmd in commands:
                            result = telnet.execute(cmd)
                            f.write(f"\n{separator} {cmd} {separator}\n{result}\n")

                        telnet.close()
                        print(f"已完成 {device['ip']} 的命令执行")
                    except Exception as e:
                        error_msg = f"\n{separator} {device['ip']} 连接失败: {str(e)} {separator}\n"
                        f.write(error_msg)
                        print(error_msg.strip())

            print(f"结果已保存到 {output_path}，等待下一次执行...")
            time.sleep(command_interval)
        except KeyboardInterrupt:
            print("\n检测到中断信号，程序即将退出...")
            break
        except Exception as e:
            print(f"发生未知错误: {str(e)}")

if __name__ == "__main__":
    main()

代码结束，如上。

用户密码文件样式：

ip地址,用户名,密码

如果有多台设备，以上的代码输出结果全部放到同一个txt中，如下代码是一台设备放到一个txt文件：

import socket
import time
import os
from datetime import datetime

# 配置参数
input_file = r"D:\Python\cjh-pythonProject\username1.txt"  # 输入文件路径
base_output_dir = r"D:\Python\cjh-pythonProject"  # 基础输出目录
command_interval = 10  # 执行间隔（秒）
separator = "=" * 50  # 分隔符
commands = [
    "display clock",
    "display mac-address"
]


class TelnetClient:
    def __init__(self, host, port=23, timeout=10):
        self.host = host
        self.port = port
        self.timeout = timeout
        self.sock = None

    def connect(self):
        """建立Telnet连接"""
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.settimeout(self.timeout)
        self.sock.connect((self.host, self.port))
        return self

    def login(self, username, password):
        """登录设备"""
        self.read_until("Username:")
        self.write(username + "\n")
        self.read_until("Password:")
        self.write(password + "\n")
        self.read_until(">")  # 等待登录完成
        self.write("screen-length 0 temporary\n")  # 禁用分页
        time.sleep(1)
        return self

    def write(self, text):
        """发送命令"""
        self.sock.send(text.encode('ascii'))

    def read_until(self, prompt, timeout=None):
        """读取直到遇到指定提示"""
        buffer = ""
        end_time = time.time() + (timeout or self.timeout)

        while time.time() < end_time:
            try:
                data = self.sock.recv(4096).decode('ascii', errors='ignore')
                if not data:
                    break
                buffer += data
                if prompt in buffer:
                    return buffer
            except socket.timeout:
                break
        return buffer

    def execute(self, command):
        """执行命令并返回结果"""
        self.write(command + "\n")
        time.sleep(2)  # 等待命令执行
        return self.read_until(">")

    def close(self):
        """关闭连接"""
        if self.sock:
            self.sock.close()


def create_output_directory():
    """创建带时间戳的输出目录"""
    timestamp = datetime.now().strftime("%Y年%m分%d日")
    output_dir = os.path.join(base_output_dir, f"FileSW-172.16.15.2-{timestamp}")
    os.makedirs(output_dir, exist_ok=True)
    return output_dir


def generate_output_filename(ip):
    """生成带时间戳的输出文件名，包含IP地址"""
    timestamp = datetime.now().strftime('%Y年%m月%d日_%H时%M分%S秒')
    # 从IP地址生成文件名的一部分
    ip_suffix = ip.replace('.', '')
    return f"txt-sw-{ip_suffix}-1_{timestamp}.txt"


def load_credentials(file_path):
    """从文件加载IP、用户名和密码"""
    devices = []
    try:
        with open(file_path, 'r') as f:
            for line in f:
                if line.strip():
                    parts = line.strip().split(',')
                    if len(parts) == 3:
                        ip, username, password = parts
                        devices.append({
                            'ip': ip.strip(),
                            'username': username.strip(),
                            'password': password.strip()
                        })
        return devices
    except Exception as e:
        print(f"读取凭证文件失败: {str(e)}")
        return []


def main():
    # 确保基础输出目录存在
    os.makedirs(base_output_dir, exist_ok=True)

    # 加载设备凭证
    devices = load_credentials(input_file)
    if not devices:
        print("没有可用的设备凭证，请检查输入文件")
        return

    while True:
        try:
            # 创建带时间戳的输出目录
            output_dir = create_output_directory()

            # 为每台设备执行命令并保存到独立文件
            for device in devices:
                try:
                    # 生成当前设备的输出文件名
                    output_filename = generate_output_filename(device['ip'])
                    output_path = os.path.join(output_dir, output_filename)

                    # 连接设备
                    telnet = TelnetClient(device['ip']).connect()
                    telnet.login(device['username'], device['password'])

                    # 写入结果到独立文件
                    with open(output_path, 'w', encoding='utf-8') as f:
                        # 写入设备信息
                        f.write(
                            f"\n{separator} {device['ip']} - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} {separator}\n")

                        # 执行命令
                        for cmd in commands:
                            result = telnet.execute(cmd)
                            f.write(f"\n{separator} {cmd} {separator}\n{result}\n")

                    telnet.close()
                    print(f"已完成 {device['ip']} 的命令执行，结果已保存到 {output_path}")

                except Exception as e:
                    # 即使单个设备失败，也继续处理其他设备
                    error_msg = f"设备 {device['ip']} 处理失败: {str(e)}"
                    print(error_msg)

                    # 保存错误信息到文件
                    try:
                        output_filename = generate_output_filename(device['ip'])
                        output_path = os.path.join(output_dir, output_filename)
                        with open(output_path, 'w', encoding='utf-8') as f:
                            f.write(f"\n{separator} {device['ip']} 执行失败: {str(e)} {separator}\n")
                    except:
                        pass

            print(f"本轮执行完成，等待下一次执行...")
            time.sleep(command_interval)

        except KeyboardInterrupt:
            print("\n检测到中断信号，程序即将退出...")
            break
        except Exception as e:
            print(f"发生未知错误: {str(e)}")
            time.sleep(command_interval)


if __name__ == "__main__":
    main()

完。