servers:
- host: 192.168.1.10port: 22
user: root
password: your_password
- host: 192.168.1.11
port: 22
user: root
password: your_password
app:
name: myapp
deploy_dir: /opt/myapp
backup_dir: /opt/backup
version: 1.0.0
notify:
dingtalk_webhook: https://oapi.dingtalk.com/robot/send?access_token=xxx
enabled: true

import paramiko
from typing import Tuple
class SSHClient:
def __init__(self, host, port, user, password):
self.host = host
self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.ssh.connect(host, port, user, password, timeout=10)
def exec(self, command: str) -> Tuple[int, str, str]:
"""執(zhí)行命令，返回退出碼、輸出、錯(cuò)誤"""
stdin, stdout, stderr = self.ssh.exec_command(command)
exit_code = stdout.channel.recv_exit_status()
return exit_code, stdout.read().decode(), stderr.read().decode()
def upload(self, local_path, remote_path):
"""上傳文件"""
sftp = self.ssh.open_sftp()
sftp.put(local_path, remote_path)
sftp.close()
def close(self):
self.ssh.close()

import logging
from datetime import datetime
def get_logger(name: str) -> logging.Logger:
logger = logging.getLogger(name)
logger.setLevel(logging.INFO)
# 控制臺(tái)輸出
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
# 文件輸出
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
fh = logging.FileHandler(f'logs/deploy_{timestamp}.log')
fh.setLevel(logging.INFO)
# 格式
formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
)
ch.setFormatter(formatter)
fh.setFormatter(formatter)
logger.addHandler(ch)
logger.addHandler(fh)
return logger

import requests
import logging
logger = logging.getLogger(__name__)
def send_dingtalk(webhook: str, message: str):
"""發(fā)送釘釘通知"""
try:
data = {
"msgtype": "text",
"text": {
"content": f"
部署通知\n\n{message}"
}
}
??
response = requests.post(webhook, json=data, timeout=10)if response.status_code == 200:
logger.info("釘釘通知發(fā)送成功")
else:
logger.error(f"釘釘通知發(fā)送失?。簕response.text}")
except Exception as e:
logger.error(f"釘釘通知異常：{str(e)}")
def send_wechat(webhook: str, message: str):
"""發(fā)送企業(yè)微信通知"""
try:
data = {
"msgtype": "text",
"text": {
"content": message
}
}
response = requests.post(webhook, json=data, timeout=10)
if response.status_code == 200:
logger.info("企業(yè)微信通知發(fā)送成功")
else:
logger.error(f"企業(yè)微信通知發(fā)送失敗：{response.text}")
except Exception as e:
logger.error(f"企業(yè)微信通知異常：{str(e)}")

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import yaml
import datetime
import sys
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
from utils.ssh_client import SSHClient
from utils.logger import get_logger
from utils.notifier import send_dingtalk
logger = get_logger(__name__)
class DeployTool:
def __init__(self, config_path='config.yaml'):
self.config_path = Path(config_path)
self.load_config()
def load_config(self):"""加載配置文件"""
if not self.config_path.exists():
logger.error(f"配置文件不存在：{self.config_path}")
raise FileNotFoundError(f"Config file not found: {self.config_path}")
with open(self.config_path, 'r', encoding='utf-8') as f:
self.config = yaml.safe_load(f)
logger.info(f"配置文件加載成功：{self.config_path}")
def deploy_server(self, server: dict) -> bool:
"""部署單臺(tái)服務(wù)器"""
host = server['host']
try:
logger.info(f"[{host}] 開始部署...")
ssh = SSHClient(
server['host'],
server['port'],
server['user'],
server.get('password', '')
)
# 1. 備份當(dāng)前版本
logger.info(f"[{host}] 正在備份...")
backup_dir = self.config['app']['backup_dir']
deploy_dir = self.config['app']['deploy_dir']
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
backup_cmd = f"cp -r {deploy_dir} {backup_dir}/backup_{timestamp}"
exit_code, _, stderr = ssh.exec(backup_cmd)
if exit_code != 0:
logger.error(f"[{host}] 備份失?。簕stderr}")
ssh.close()
return False
# 2. 執(zhí)行部署腳本
logger.info(f"[{host}] 正在部署應(yīng)用...")
deploy_script = self.config['app'].get('deploy_script',
'/opt/scripts/deploy_app.sh')
exit_code, output, stderr = ssh.exec(f"bash {deploy_script}")
if exit_code != 0:
logger.error(f"[{host}] 部署失敗：{stderr}")
self.rollback(server, backup_dir)
ssh.close()
return False# 3. 健康檢查
logger.info(f"[{host}] 執(zhí)行健康檢查...")
health_check = self.config['app'].get('health_check', 'systemctl
status myapp')
exit_code, _, stderr = ssh.exec(health_check)
if exit_code != 0:
logger.warning(f"[{host}] 健康檢查未通過(guò)：{stderr}")
logger.info(f"[{host}]
ssh.close()
return True
? 部署成功")
except Exception as e:
logger.error(f"[{host}] 部署異常：{str(e)}")
self.rollback(server, backup_dir)
return False
def rollback(self, server: dict, backup_dir: str):
"""回滾到上一個(gè)版本"""
host = server['host']
deploy_dir = self.config['app']['deploy_dir']
try:
logger.info(f"[{host}] 開始回滾...")
ssh = SSHClient(
server['host'],
server['port'],
server['user'],
server.get('password', '')
)
# 獲取最新備份
ls_cmd = f"ls -t {backup_dir} | head -1"
exit_code, latest_backup, _ = ssh.exec(ls_cmd)
if exit_code == 0 and latest_backup.strip():
rollback_cmd = f"rm -rf {deploy_dir} && cp -r
{backup_dir}/{latest_backup.strip()} {deploy_dir}"
ssh.exec(rollback_cmd)
logger.info(f"[{host}]
回滾完成")
else:
logger.error(f"[{host}] 未找到備份文件")
?
ssh.close()
except Exception as e:
logger.error(f"[{host}] 回滾異常：{str(e)}")def deploy_all(self) -> bool:
"""并行部署所有服務(wù)器"""
servers = self.config['servers']
total = len(servers)
logger.info(f"開始部署，共 {total} 臺(tái)服務(wù)器")
success_count = 0
failed_servers = []
with ThreadPoolExecutor(max_workers=5) as executor:
future_to_server = {
executor.submit(self.deploy_server, server): server
for server in servers
}
for future in as_completed(future_to_server):
server = future_to_server[future]
host = server['host']
try:
if future.result():
success_count += 1
else:
failed_servers.append(host)
except Exception as e:
logger.error(f"[{host}] 部署異常：{str(e)}")
failed_servers.append(host)
# 發(fā)送通知
self.send_notification(success_count, total, failed_servers)
# 返回結(jié)果
if success_count == total:
logger.info(f"
部署完成：{success_count}/{total} 成功")
return True
else:
logger.warning(f"
部署部分失?。簕success_count}/{total} 成功")
return False
??
??
def send_notification(self, success: int, total: int, failed: list):
"""發(fā)送部署通知"""
if not self.config.get('notify', {}).get('enabled', False):
return
webhook = self.config['notify'].get('dingtalk_webhook', '')
?
if success == total:
message = f"
部署成功\n\n 成功：{success}/{total}\n 時(shí)間：
{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"??
else:
message = f"
部署部分失敗\n\n 成功：{success}/{total}\n 失?。簕',
'.join(failed)}\n 時(shí)間：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
send_dingtalk(webhook, message)
if __name__ == '__main__':
try:
deploy_tool = DeployTool()
success = deploy_tool.deploy_all()
sys.exit(0 if success else 1)
except Exception as e:
logger.error(f"部署失?。簕str(e)}")
sys.exit(1)

#!/bin/bash
# 應(yīng)用部署腳本
set -e
APP_NAME="myapp"
APP_DIR="/opt/myapp"
VERSION="1.0.0"
echo "開始部署 $APP_NAME v$VERSION"
# 1. 停止服務(wù)
echo "停止服務(wù)..."
systemctl stop $APP_NAME
# 2. 備份當(dāng)前版本
echo "備份當(dāng)前版本..."
# (已在 Python 腳本中完成)
# 3. 下載新版本
echo "下載新版本..."
cd $APP_DIR
wget http://your-server.com/$APP_NAME-$VERSION.tar.gz
tar -xzf $APP_NAME-$VERSION.tar.gz
# 4. 啟動(dòng)服務(wù)
echo "啟動(dòng)服務(wù)..."
systemctl start $APP_NAME
# 5. 檢查狀態(tài)
echo "檢查服務(wù)狀態(tài)..."
systemctl status $APP_NAME
echo "部署完成！"```
---
#### 7. 依賴文件 (requirements.txt)

pip install -r requirements.txt

vim config.yaml

python deploy.py

app:
health_check: |
curl -f http://localhost:8080/health || exit 1

def deploy_canary(self):
"""灰度發(fā)布"""
servers = self.config['servers']
# 先部署第一臺(tái)
if not self.deploy_server(servers[0]):
logger.error("灰度部署失敗，停止")
return False
# 檢查通過(guò)，部署其他
for server in servers[1:]:
self.deploy_server(server)
return True

# .gitlab-ci.yml
deploy:
stage: deploy
script:
- pip install -r requirements.txt
- python deploy.py
only:
- master