• 正文
  • 相關(guān)推薦
申請(qǐng)入駐 產(chǎn)業(yè)圖譜

Python 實(shí)戰(zhàn) 構(gòu)建輕量級(jí)遠(yuǎn)程服務(wù)器批量控制平臺(tái)

12小時(shí)前
307
加入交流群
掃碼加入
獲取工程師必備禮包
參與熱點(diǎn)資訊討論

一、背景與目標(biāo)

在企業(yè)或個(gè)人 DevOps 場(chǎng)景中,我們常常需要:

  • 同時(shí)向多臺(tái)服務(wù)器下發(fā)命令(如同步腳本、重啟服務(wù))

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

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

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

本項(xiàng)目使用純 Python 實(shí)現(xiàn)一個(gè)“迷你 SSH 命令控制臺(tái)”,并具備以下能力:

  • 支持多臺(tái)主機(jī)批量命令下發(fā)

  • 支持主機(jī)配置、分組、備注

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

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

二、項(xiàng)目結(jié)構(gòu)設(shè)計(jì)

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

三、主機(jī)配置文件(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 Path
def 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

六、控制臺(tái)交互入口(main.py)

python

from manager import broadcast_command, load_config

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

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

while True:
cmd = input(f"[{group}] 請(qǐng)輸入要執(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()

七、運(yùn)行示例

bash

$ python main.py

遠(yuǎn)程主機(jī)分組:
- dev
- prod
選擇組名:dev
[dev] 請(qǐng)輸入要執(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
----------------------------------------

八、擴(kuò)展方向建議

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

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

本項(xiàng)目實(shí)現(xiàn)了一個(gè)具備以下特性的“遠(yuǎn)程服務(wù)器批量控制平臺(tái)”:

  • ? YAML 配置簡(jiǎn)潔可維護(hù),支持分組管理

  • ? 使用標(biāo)準(zhǔn)庫(kù) + Paramiko,無(wú)需額外平臺(tái)依賴

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

  • ? 結(jié)構(gòu)清晰,便于未來(lái)擴(kuò)展為 GUI、Web 平臺(tái)

可用于:

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

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

  • 運(yùn)維日常操作自動(dòng)化

相關(guān)推薦