• 正文
    • 一、背景與目標
    • 二、項目結(jié)構(gòu)設(shè)計
    • 三、主機配置文件（config.yaml）
    • 四、SSH 客戶端封裝（ssh_client.py）
    • 五、批量執(zhí)行管理器（manager.py）
    • 六、控制臺交互入口（main.py）
    • 七、運行示例
    • 八、擴展方向建議
    • 九、項目總結(jié)與價值
  • 相關(guān)推薦
申請入駐 產(chǎn)業(yè)圖譜

Python 實戰(zhàn) 構(gòu)建輕量級遠程服務(wù)器批量控制平臺

18小時前
348
加入交流群
掃碼加入
獲取工程師必備禮包
參與熱點資訊討論

一、背景與目標

在企業(yè)或個人 DevOps 場景中，我們常常需要：

  • 同時向多臺服務(wù)器下發(fā)命令（如同步腳本、重啟服務(wù)）

  • 實時查看每臺服務(wù)器的執(zhí)行結(jié)果

  • 支持主機分組（按環(huán)境或角色劃分）

  • 不借助 Ansible 等大型工具，保持輕量靈活

本項目使用純 Python 實現(xiàn)一個“迷你 SSH 命令控制臺”，并具備以下能力：

  • 支持多臺主機批量命令下發(fā)

  • 支持主機配置、分組、備注

  • 支持并發(fā)執(zhí)行命令與結(jié)果匯總

  • 支持失敗重連、錯誤提示、執(zhí)行日志保存

二、項目結(jié)構(gòu)設(shè)計

bash
ssh_center/
├── main.py # 主入口，控制臺界面
├── config.yaml # 主機配置文件（YAML 格式）
├── ssh_client.py # SSH 客戶端封裝
├── manager.py # 控制邏輯，廣播命令
├── logs/
│ └── exec_log.txt # 執(zhí)行日志記錄

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

yaml
groups:
dev:
- name: dev1
host: 192.168.1.101
port: 22
user: root
password: password1
- name: dev2
host: 192.168.1.102
port: 22
user: root
password: password2
prod:
- name: prod1
host: 10.0.0.11
port: 22
user: admin
password: secret

四、SSH 客戶端封裝（ssh_client.py）

python

import paramiko

class SSHClient:
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)

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

def close(self):
self.client.close()

五、批量執(zhí)行管理器（manager.py）

python
import yaml
import threading
from ssh_client import SSHClient
from datetime import datetime
from pathlib import Pathdef load_config(path="config.yaml"):
with open(path, 'r') as f:
return yaml.safe_load(f)

def save_log(entry):
Path("logs").mkdir(exist_ok=True)
with open("logs/exec_log.txt", "a") as f:
f.write(f"[{datetime.now()}] {entry}n")

def exec_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}")

def broadcast_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

def main():
config = load_config()
print("遠程主機分組：")
for group in config['groups']:
print(f" - {group}")

group = input("選擇組名：")
if group not in config['groups']:
print("組名不存在")
return

while True:
cmd = input(f"[{group}] 請輸入要執(zhí)行的命令（輸入 exit 退出）：n> ")
if cmd.strip().lower() == "exit":
break
results = broadcast_command(group, cmd)
print("n=== 執(zhí)行結(jié)果 ===")
for res in results:
print(res)
print("-" * 40)

if __name__ == "__main__":
main()

七、運行示例

bash

$ python main.py

遠程主機分組：
- dev
- prod
選擇組名：dev
[dev] 請輸入要執(zhí)行的命令（輸入 exit 退出）：
> uptime

=== 執(zhí)行結(jié)果 ===
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
----------------------------------------

八、擴展方向建議

方向 描述
主機連接超時管理 添加每臺主機的連接超時時間限制，避免阻塞
執(zhí)行成功統(tǒng)計 成功 / 失敗主機數(shù)量展示
支持密鑰登錄 可選擇使用 ssh key 而非密碼連接
命令別名支持 定義一批常用命令別名，如 restart_nginx
執(zhí)行日志按日期歸檔 每天生成獨立的日志文件
支持 YAML 外部任務(wù)定義 提前配置要運行的命令腳本集合

九、項目總結(jié)與價值

本項目實現(xiàn)了一個具備以下特性的“遠程服務(wù)器批量控制平臺”：

  • ? YAML 配置簡潔可維護，支持分組管理

  • ? 使用標準庫 + Paramiko，無需額外平臺依賴

  • ? 線程并發(fā)處理任務(wù)，性能良好

  • ? 結(jié)構(gòu)清晰，便于未來擴展為 GUI、Web 平臺

可用于：

  • 小型企業(yè)內(nèi)部服務(wù)器維護

  • 云主機 SSH 命令分發(fā)

  • 運維日常操作自動化

相關(guān)推薦