Python Paramiko SSH输出无法保存至文件及异常报错排查
解决Paramiko SSH连接的两个问题:报错处理与空文件修复
咱们先从那个让代码卡壳的报错说起,再搞定文件为空的问题——这俩都有明确的修复思路:
一、关于报错:except 10060: TypeError: catching classes that do not inherit from BaseException is not allowed
这个错误本质是你用错了Python异常捕获的规则:except后面必须跟继承自BaseException的异常类,直接写数字错误码10060肯定不行。
10060是Windows系统里的「连接超时」错误码,对应的是socket模块里的异常。正确的做法是:
- 先导入
socket模块; - 捕获对应的异常类(比如
socket.timeout、socket.error); - 如果需要判断具体错误码,再从异常对象里提取
errno属性。
二、关于文件创建但内容为空
空文件的常见原因有这几个,你可以逐一排查:
- 只读取了标准输出,忽略了标准错误流:很多命令的错误提示、状态信息会输出到
stderr,如果只读stdout就会漏掉; - 读取输出的方式不对:一次性
read()可能在通道未完全关闭时读不到完整内容,或者没循环读取直到输出结束; - 文件未正确关闭/刷新:如果没用
with语句或者close(),缓冲区的内容可能没写入磁盘; - 命令执行无输出:不过你说标准输出能显示,这个可能性很低。
修正后的完整示例代码
下面是整合了正确异常处理和文件写入的代码,你可以参考:
import paramiko import socket def ssh_exec_and_save(host, user, pwd, cmd, output_path): ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: # 建立连接,设置超时时间避免无限等待 ssh.connect(hostname=host, username=user, password=pwd, timeout=15) # 执行命令,同时获取stdout和stderr stdin, stdout, stderr = ssh.exec_command(cmd) # 读取所有输出(合并标准输出和错误输出) full_output = stdout.read().decode('utf-8') + stderr.read().decode('utf-8') # 写入文件:用with语句自动处理关闭,确保内容写入磁盘 with open(output_path, 'w', encoding='utf-8') as f: f.write(full_output) # 同时输出到标准输出,满足你的原有需求 print(f"=== {host} 输出内容 ===") print(full_output) # 捕获连接超时异常 except socket.timeout: print(f"❌ 连接 {host} 超时(对应错误码10060)") # 捕获认证失败 except paramiko.AuthenticationException: print(f"❌ {host} 用户名或密码错误") # 捕获其他socket错误(比如连接被拒绝) except socket.error as e: if hasattr(e, 'errno') and e.errno == 10060: print(f"❌ 连接 {host} 超时:{str(e)}") else: print(f"❌ Socket错误:{str(e)}") # 捕获其他未知异常 except Exception as e: print(f"❌ 未知错误:{str(e)}") finally: # 确保SSH连接关闭,避免资源泄漏 ssh.close() # 调用示例:替换成你的实际参数 ssh_exec_and_save( host="192.168.1.50", user="root", pwd="your_password", cmd="df -h", output_path="server_192.168.1.50_output.txt" )
额外提示:处理实时/大体积输出
如果你的命令会产生大量输出,或者需要实时查看并保存,建议用循环读取的方式代替一次性read(),避免内存占用过高或内容截断:
# 替换示例代码中的读取和写入部分 with open(output_path, 'w', encoding='utf-8') as f: print(f"=== {host} 实时输出 ===") # 循环读取直到通道关闭 while not stdout.channel.closed: # 读取标准输出 if stdout.channel.recv_ready(): line = stdout.readline().decode('utf-8') f.write(line) print(line, end='') # 读取标准错误 if stderr.channel.recv_ready(): line = stderr.readline().decode('utf-8') f.write(line) print(line, end='')
内容的提问来源于stack exchange,提问作者Gustavo
相关产品推荐
相关产品推荐

