原创 Python 实战：构建轻量级远程服务器批量控制平台

 2025-7-2 09:17  31 0 分类: 物联网

一、背景与目标

在企业或个人 DevOps 场景中，我们常常需要：

同时向多台服务器下发命令（如同步脚本、重启服务）
实时查看每台服务器的执行结果
支持主机分组（按环境或角色划分）
不借助 Ansible 等大型工具，保持轻量灵活

本项目使用纯 Python 实现一个“迷你 SSH 命令控制台”，并具备以下能力：

支持多台主机批量命令下发
支持主机配置、分组、备注
支持并发执行命令与结果汇总
支持失败重连、错误提示、执行日志保存

二、项目结构设计

bash
复制编辑
ssh_center/
├── main.py              # 主入口，控制台界面
├── config.yaml          # 主机配置文件（YAML 格式）
├── ssh_client.py        # SSH 客户端封装
├── manager.py           # 控制逻辑，广播命令
├── logs/
│   └── exec_log.txt     # 执行日志记录

三、主机配置文件（config.yaml）

yaml
复制编辑
groups:dev:-name:dev1host:192.168.1.101port:22user:rootpassword:password1-name:dev2host:192.168.1.102port:22user:rootpassword:password2prod:-name:prod1host:10.0.0.11port:22user:adminpassword:secret

四、SSH 客户端封装（ssh_client.py）

python
复制编辑
import paramiko

classSSHClient:
    def__init__(self, host, port, user, password):
        self.info = f"{user}@{host}:{port}"
        self.client = paramiko.SSHClient()
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.client.connect(host, port=port, username=user, password=password, timeout=5)

    defexec_command(self, cmd):
        stdin, stdout, stderr = self.client.exec_command(cmd)
        out = stdout.read().decode()
        err = stderr.read().decode()
        return out.strip(), err.strip()

    defclose(self):
        self.client.close()

五、批量执行管理器（manager.py）

python
复制编辑
import yaml
import threading
from ssh_client import SSHClient
from datetime import datetime
from pathlib import Path

defload_config(path="config.yaml"):
    withopen(path, 'r') as f:
        return yaml.safe_load(f)

defsave_log(entry):
    Path("logs").mkdir(exist_ok=True)
    withopen("logs/exec_log.txt", "a") as f:
        f.write(f"[{datetime.now()}] {entry}\n")

defexec_on_host(info, cmd, results):
    try:
        ssh = SSHClient(info['host'], info['port'], info['user'], info['password'])
        out, err = ssh.exec_command(cmd)
        ssh.close()
        msg = f"{info['name']} ({info['host']})\n>>> {cmd}\n{out or err}"
        results.append(msg)
        save_log(msg)
    except Exception as e:
        results.append(f"{info['name']} ERROR: {e}")
        save_log(f"{info['name']} ERROR: {e}")

defbroadcast_command(group_name, cmd):
    config = load_config()
    targets = config['groups'].get(group_name, [])
    threads = []
    results = []

    for host in targets:
        t = threading.Thread(target=exec_on_host, args=(host, cmd, results))
        t.start()
        threads.append(t)

    for t in threads:
        t.join()
    return results

六、控制台交互入口（main.py）

python
复制编辑
from manager import broadcast_command, load_config

defmain():
    config = load_config()
    print("远程主机分组：")
    for group in config['groups']:
        print(f" - {group}")

    group = input("选择组名：")
    if group notin config['groups']:
        print("组名不存在")
        returnwhileTrue:
        cmd = input(f"[{group}] 请输入要执行的命令（输入 exit 退出）：\n> ")
        if cmd.strip().lower() == "exit":
            break
        results = broadcast_command(group, cmd)
        print("\n=== 执行结果 ===")
        for res in results:
            print(res)
            print("-" * 40)

if __name__ == "__main__":
    main()

七、运行示例

bash
复制编辑
$ python main.py

远程主机分组：
 - dev
 - prod
选择组名：dev
[dev] 请输入要执行的命令（输入 exit 退出）：
> uptime

=== 执行结果 ===
dev1 (192.168.1.101)
>>> uptime
14:23:12 up 3 days, 2 users, load average: 0.01, 0.05, 0.03
----------------------------------------
dev2 (192.168.1.102)
>>> uptime
14:23:13 up 1 day, 1 user, load average: 0.00, 0.01, 0.00
----------------------------------------

八、扩展方向建议

方向	描述
主机连接超时管理	添加每台主机的连接超时时间限制，避免阻塞
执行成功统计	成功 / 失败主机数量展示
支持密钥登录	可选择使用 ssh key 而非密码连接
命令别名支持	定义一批常用命令别名，如 `restart_nginx`
执行日志按日期归档	每天生成独立的日志文件
支持 YAML 外部任务定义	提前配置要运行的命令脚本集合

九、项目总结与价值

本项目实现了一个具备以下特性的“远程服务器批量控制平台”：

✅ YAML 配置简洁可维护，支持分组管理
✅ 使用标准库 + Paramiko，无需额外平台依赖
✅ 线程并发处理任务，性能良好
✅ 结构清晰，便于未来扩展为 GUI、Web 平台

小型企业内部服务器维护
云主机 SSH 命令分发

运维日常操作自动化

上一篇： Wi-Fi HaLow（802.11ah）技术原理与物联网工程应用实践
返回列表

PARTNER CONTENT

文章评论（0条评论）