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

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

测试的Python版本：

程序特点：

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

2、获取当前时间戳，精确到秒，并构造输出文件夹名

3、输出结果禁用分布

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

代码开始，如下：

import socket
import time
import os
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",
# "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()