ginkgo-tools-batchssh 0.1.6

批量SSH操作和文件传输工具包，提升了解码兼容能力

GINKGO-TOOLS-BATCHSSH

一个用于批量SSH操作和文件传输的Python工具包。

功能特性

• 🔧 SSH连接管理 - 简化SSH连接建立和管理
• ⚡ 远程命令执行 - 支持超时控制的远程命令执行
• 📁 SFTP文件传输 - 支持通配符的批量文件上传下载
• 🎯 批量主机操作 - 支持多主机并行操作
• 📝 智能日志系统 - 自动记录操作日志便于追踪
• 🛠️ 便捷工具函数 - 提供常用的主机管理和过滤工具

安装

pip install ginkgo_tools_batchssh

快速开始

基本SSH连接和命令执行

from ginkgo_tools_batchssh import SSHWrapper

# 创建SSH连接
ssh = SSHWrapper(
    host_ip="192.168.1.100",
    username="your_username",
    password="your_password"
)

# 执行单个命令
result = ssh.execute_command("ls -la")
print(f"执行结果: {result['result']}")
print(f"输出内容: {result['stdout']}")

# 执行多个命令
commands = ["cd /tmp", "pwd", "ls"]
result = ssh.execute_shell_commands(commands)
print(result['stdout'])

文件传输操作

# 上传单个文件
upload_result = ssh.sftp_upload_file("local_file.txt", "/remote/path/file.txt")

# 批量上传文件（支持通配符）
upload_result = ssh.sftp_upload_to("*.log", "/remote/logs/")

# 下载文件
download_result = ssh.sftp_download_file("/remote/file.txt", "local_file.txt")

# 批量下载文件（支持通配符）
download_result = ssh.sftp_download_to("/remote/logs/*.log", "./local_logs/")

主机管理工具

from ginkgo_tools_batchssh import read_hosts_file, parse_and_filter_hosts

# 读取hosts文件
hosts_content = read_hosts_file()

# 过滤和解析主机
filter_list = ["web", "server"]  # 只选择包含这些关键词的主机
target_hosts = parse_and_filter_hosts(hosts_content, filter_list)

# target_hosts 格式: {"ip_address": ["hostname1", "hostname2"]}
for ip, hostnames in target_hosts.items():
    print(f"{ip}: {', '.join(hostnames)}")

核心组件

SSHWrapper 类

主要的SSH操作类，提供以下核心功能：

• ssh_connect() - 建立SSH连接
• execute_command() - 执行单个命令
• execute_shell_commands() - 执行多个命令
• sftp_upload_*() - 文件上传相关方法
• sftp_download_*() - 文件下载相关方法
• commands_to_shell_*() - Shell脚本执行相关方法

便捷工具函数

• read_hosts_file() - 读取hosts文件内容
• parse_and_filter_hosts() - 解析和过滤主机列表

日志系统

• setup_logging() - 配置日志系统
• get_logger() - 获取日志记录器

高级用法

超时控制

# 设置命令执行超时时间（秒）
result = ssh.execute_command("sleep 60", timeout=30)
if result['do_break']:
    print("命令执行超时")

Shell脚本执行

# 将命令封装为shell脚本并执行
commands = [
    "echo '开始执行任务'",
    "date",
    "ls -la"
]

# 传输脚本文件
trans_result = ssh.commands_to_shell_trans(command_list=commands)

# 执行脚本（可选sudo切换用户）
exec_result = ssh.commands_to_shell_run(trans_result, sudo_to_user="root")

批量主机操作示例

from ginkgo_tools_batchssh import SSHWrapper, read_hosts_file, parse_and_filter_hosts

# 获取目标主机列表
hosts_content = read_hosts_file()
target_hosts = parse_and_filter_hosts(hosts_content, ["production"])

# 批量执行操作
results = {}
for ip, hostnames in target_hosts.items():
    try:
        ssh = SSHWrapper(
            host_ip=ip,
            username="admin",
            password="password",
            host_name=hostnames[0]
        )
        
        result = ssh.execute_command("uptime")
        results[ip] = result
        
        ssh.close()
    except Exception as e:
        results[ip] = {"result": False, "error": str(e)}

# 查看结果
for ip, result in results.items():
    if result['result']:
        print(f"{ip}: {result['stdout'].strip()}")
    else:
        print(f"{ip}: 执行失败 - {result.get('error', '未知错误')}")

返回值格式

大多数方法返回统一的字典格式：

{
    "result": True/False,           # 操作是否成功
    "stdout": "输出内容",            # 标准输出
    "stderr": "错误内容",            # 错误输出
    "exit_status": 0,               # 退出状态码
    "error": "错误信息",             # 错误描述（失败时）
    # 其他特定字段...
}

依赖要求

• Python >= 3.8
• paramiko >= 2.7.0

许可证

MIT License

作者

霍城 [e-mail]

---

注意：请根据实际使用场景调整连接参数和安全设置