# 安装核心依赖库
pip install paramiko fabric3 pyyaml python-dotenv tqdm

# 依赖说明：
# paramiko：SSH连接核心库，实现远程命令执行
# fabric3：基于paramiko的封装，简化批量操作
# pyyaml：配置文件解析（服务器列表、部署参数）
# python-dotenv：环境变量管理（敏感信息如SSH密码）
# tqdm：进度条显示，监控部署进度

# servers.yaml 服务器配置文件
servers:
  # 生产环境服务器组
  production:
    - host: 192.168.1.101
      port: 22
      username: root
      password: ${PROD_SSH_PWD}  # 从环境变量读取密码
      tags: ["web", "nginx"]
    - host: 192.168.1.102
      port: 22
      username: root
      password: ${PROD_SSH_PWD}
      tags: ["web", "nginx"]
    # 更多生产服务器...
  
  # 测试环境服务器组
  test:
    - host: 192.168.2.101
      port: 22
      username: root
      password: ${TEST_SSH_PWD}
      tags: ["test", "nginx"]
    # 更多测试服务器...

# .env 环境变量文件（git忽略，避免泄露敏感信息）
PROD_SSH_PWD=你的生产服务器密码
TEST_SSH_PWD=你的测试服务器密码

#!/usr/bin/env python3
# ssh_key_deploy.py 批量配置SSH免密登录
import os
import paramiko
import yaml
from dotenv import load_dotenv

# 加载环境变量和配置文件
load_dotenv()
with open("servers.yaml", "r", encoding="utf-8") as f:
    config = yaml.safe_load(f)

# 本地公钥路径
PUBLIC_KEY_PATH = os.path.expanduser("~/.ssh/id_rsa.pub")

def deploy_ssh_key(host, port, username, password):
    """向单台服务器部署SSH公钥"""
    try:
        # 读取本地公钥
        with open(PUBLIC_KEY_PATH, "r") as f:
            pub_key = f.read().strip()
        
        # 建立SSH连接
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(host, port=port, username=username, password=password, timeout=10)
        
        # 执行命令：创建.ssh目录，添加公钥到authorized_keys
        commands = [
            "mkdir -p ~/.ssh && chmod 700 ~/.ssh",
            f'echo "{pub_key}" >> ~/.ssh/authorized_keys',
            "chmod 600 ~/.ssh/authorized_keys"
        ]
        
        for cmd in commands:
            stdin, stdout, stderr = ssh.exec_command(cmd)
            stdout.channel.recv_exit_status()  # 等待命令执行完成
            if stderr.read():
                print(f"[{host}] 执行命令失败: {cmd}")
                return False
        
        ssh.close()
        print(f"[{host}] SSH免密登录配置成功")
        return True
    except Exception as e:
        print(f"[{host}] 配置失败: {str(e)}")
        return False

if __name__ == "__main__":
    # 遍历所有生产服务器配置免密登录
    for server in config["servers"]["production"]:
        # 替换环境变量
        password = os.environ.get(server["password"].replace("${", "").replace("}", ""))
        deploy_ssh_key(
            host=server["host"],
            port=server["port"],
            username=server["username"],
            password=password
        )

#!/usr/bin/env python3
# batch_execute.py 批量执行命令
import os
import paramiko
import yaml
import threading
from dotenv import load_dotenv
from tqdm import tqdm  # 进度条

# 加载配置
load_dotenv()
with open("servers.yaml", "r", encoding="utf-8") as f:
    config = yaml.safe_load(f)

# 全局变量：进度条、锁
progress_bar = None
lock = threading.Lock()

def execute_command(server, commands, result_dict):
    """执行单台服务器的命令"""
    host = server["host"]
    try:
        # 替换密码环境变量
        password = os.environ.get(server["password"].replace("${", "").replace("}", ""))
        
        # 建立SSH连接
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(
            hostname=server["host"],
            port=server["port"],
            username=server["username"],
            password=password,
            timeout=15,
            allow_agent=False,
            look_for_keys=False
        )
        
        # 执行命令
        output = {}
        for cmd in commands:
            stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
            # 读取输出（避免缓冲区溢出）
            stdout_content = stdout.read().decode("utf-8", errors="ignore")
            stderr_content = stderr.read().decode("utf-8", errors="ignore")
            output[cmd] = {
                "stdout": stdout_content,
                "stderr": stderr_content,
                "exit_code": stdout.channel.recv_exit_status()
            }
        
        ssh.close()
        
        # 更新结果和进度
        with lock:
            result_dict[host] = {"status": "success", "output": output}
            progress_bar.update(1)
    except Exception as e:
        with lock:
            result_dict[host] = {"status": "failed", "error": str(e)}
            progress_bar.update(1)

def batch_execute(server_group, commands, max_threads=10):
    """批量执行命令（多线程）"""
    global progress_bar
    servers = config["servers"][server_group]
    result_dict = {}
    
    # 初始化进度条
    progress_bar = tqdm(total=len(servers), desc=f"批量执行命令 ({server_group})")
    
    # 多线程执行
    threads = []
    semaphore = threading.Semaphore(max_threads)  # 限制并发数
    
    def worker(server):
        with semaphore:
            execute_command(server, commands, result_dict)
    
    for server in servers:
        t = threading.Thread(target=worker, args=(server,))
        threads.append(t)
        t.start()
    
    # 等待所有线程完成
    for t in threads:
        t.join()
    
    progress_bar.close()
    return result_dict

if __name__ == "__main__":
    # 示例：批量在生产服务器执行系统更新和查看磁盘空间
    commands = [
        "yum update -y",
        "df -h",
        "free -m"
    ]
    
    # 执行批量命令
    results = batch_execute("production", commands, max_threads=10)
    
    # 输出执行结果
    print("\n===== 执行结果汇总 =====")
    success_count = 0
    failed_count = 0
    for host, result in results.items():
        if result["status"] == "success":
            success_count += 1
            print(f"✅ {host}: 执行成功")
        else:
            failed_count += 1
            print(f"❌ {host}: 执行失败 - {result['error']}")
    
    print(f"\n总计: {success_count} 台成功, {failed_count} 台失败")

def upload_file(server, local_path, remote_path):
    """上传文件到单台服务器"""
    host = server["host"]
    try:
        # 替换密码环境变量
        password = os.environ.get(server["password"].replace("${", "").replace("}", ""))
        
        # 建立SFTP连接
        transport = paramiko.Transport((host, server["port"]))
        transport.connect(username=server["username"], password=password)
        sftp = paramiko.SFTPClient.from_transport(transport)
        
        # 上传文件（支持目录递归上传）
        if os.path.isdir(local_path):
            # 递归创建远程目录
            sftp.mkdir(remote_path, mode=0o755)
            for root, dirs, files in os.walk(local_path):
                # 计算相对路径
                rel_path = os.path.relpath(root, local_path)
                remote_dir = os.path.join(remote_path, rel_path)
                
                # 创建远程子目录
                if rel_path != ".":
                    try:
                        sftp.mkdir(remote_dir, mode=0o755)
                    except IOError:
                        pass  # 目录已存在
                
                # 上传文件
                for file in files:
                    local_file = os.path.join(root, file)
                    remote_file = os.path.join(remote_dir, file)
                    sftp.put(local_file, remote_file)
        else:
            # 上传单个文件
            sftp.put(local_path, remote_path)
        
        sftp.close()
        transport.close()
        
        with lock:
            progress_bar.update(1)
        return True
    except Exception as e:
        print(f"[{host}] 文件上传失败: {str(e)}")
        with lock:
            progress_bar.update(1)
        return False

# 批量上传文件调用示例
def batch_upload(server_group, local_path, remote_path, max_threads=10):
    servers = config["servers"][server_group]
    global progress_bar
    progress_bar = tqdm(total=len(servers), desc=f"批量上传文件 ({server_group})")
    
    threads = []
    semaphore = threading.Semaphore(max_threads)
    
    def worker(server):
        with semaphore:
            upload_file(server, local_path, remote_path)
    
    for server in servers:
        t = threading.Thread(target=worker, args=(server,))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    progress_bar.close()

# 调用示例：批量上传Nginx配置文件到所有生产服务器
if __name__ == "__main__":
    batch_upload(
        server_group="production",
        local_path="./nginx_config",
        remote_path="/etc/nginx/conf.d",
        max_threads=10
    )

# fabfile.py Fabric3部署脚本
from fabric import Connection, task
from fabric.exceptions import GroupException
from invoke.exceptions import UnexpectedExit
import yaml
import os
from dotenv import load_dotenv

# 加载配置
load_dotenv()
with open("servers.yaml", "r", encoding="utf-8") as f:
    config = yaml.safe_load(f)

# 构建服务器连接列表
def get_connections(server_group):
    """获取服务器连接列表"""
    connections = []
    for server in config["servers"][server_group]:
        # 替换密码环境变量
        password = os.environ.get(server["password"].replace("${", "").replace("}", ""))
        # 创建连接
        conn = Connection(
            host=server["host"],
            port=server["port"],
            user=server["username"],
            connect_kwargs={"password": password}
        )
        connections.append(conn)
    return connections

# 批量安装Nginx
@task
def install_nginx(ctx, env="production"):
    """批量安装Nginx"""
    connections = get_connections(env)
    print(f"开始在{env}环境的{len(connections)}台服务器安装Nginx...")
    
    # 并行执行安装命令
    try:
        for conn in connections:
            # 安装依赖
            conn.run("yum install -y gcc pcre-devel zlib-devel openssl-devel", hide=True)
            # 下载并编译安装Nginx
            conn.run("wget http://nginx.org/download/nginx-1.24.0.tar.gz -P /tmp", hide=True)
            conn.run("cd /tmp && tar -zxvf nginx-1.24.0.tar.gz", hide=True)
            conn.run(
                "cd /tmp/nginx-1.24.0 && ./configure --prefix=/usr/local/nginx --with-http_ssl_module && make && make install",
                hide=True
            )
            # 创建系统服务
            conn.run("""
                cat > /usr/lib/systemd/system/nginx.service << EOF
[Unit]
Description=nginx - high performance web server
After=network.target remote-fs.target nss-lookup.target

[Service]
Type=forking
PIDFile=/usr/local/nginx/logs/nginx.pid
ExecStart=/usr/local/nginx/sbin/nginx -c /usr/local/nginx/conf/nginx.conf
ExecReload=/usr/local/nginx/sbin/nginx -s reload
ExecStop=/usr/local/nginx/sbin/nginx -s stop
PrivateTmp=true

[Install]
WantedBy=multi-user.target
EOF
            """, hide=True)
            # 启动并设置开机自启
            conn.run("systemctl daemon-reload && systemctl start nginx && systemctl enable nginx", hide=True)
            print(f"✅ {conn.host}: Nginx安装成功")
    except GroupException as e:
        for conn, exc in e.result.items():
            print(f"❌ {conn.host}: Nginx安装失败 - {exc}")
    except UnexpectedExit as e:
        print(f"命令执行失败: {e}")

# 批量部署Nginx配置
@task
def deploy_nginx_config(ctx, env="production"):
    """批量部署Nginx配置文件"""
    connections = get_connections(env)
    print(f"开始向{env}环境部署Nginx配置...")
    
    for conn in connections:
        # 上传配置文件
        conn.put("./nginx.conf", "/usr/local/nginx/conf/nginx.conf", preserve_mode=True)
        # 验证配置并重启
        result = conn.run("/usr/local/nginx/sbin/nginx -t", warn=True)
        if result.ok:
            conn.run("systemctl restart nginx")
            print(f"✅ {conn.host}: Nginx配置部署成功")
        else:
            print(f"❌ {conn.host}: Nginx配置验证失败 - {result.stderr}")

# 批量检查服务状态
@task
def check_nginx_status(ctx, env="production"):
    """批量检查Nginx状态"""
    connections = get_connections(env)
    print(f"检查{env}环境Nginx状态...")
    
    for conn in connections:
        result = conn.run("systemctl is-active nginx", warn=True)
        if result.stdout.strip() == "active":
            print(f"✅ {conn.host}: Nginx运行正常")
        else:
            print(f"❌ {conn.host}: Nginx未运行 - {result.stdout}")

# 命令行调用示例：
# fab install_nginx --env=production    # 生产环境批量安装Nginx
# fab deploy_nginx_config --env=test    # 测试环境部署配置
# fab check_nginx_status --env=production  # 检查生产环境Nginx状态

# 1. 安装Jinja2
# pip install jinja2

# 2. Nginx配置模板 (nginx_template.conf)
'''
worker_processes {{ worker_processes }};  # 动态参数：CPU核心数
error_log /var/log/nginx/error.log {{ log_level }};

events {
    worker_connections {{ worker_connections }};
}

http {
    include       mime.types;
    default_type  application/octet-stream;
    
    sendfile        on;
    keepalive_timeout  65;
    
    {% for server in servers %}
    server {
        listen       {{ server.port }};
        server_name  {{ server.domain }};
        
        location / {
            proxy_pass http://{{ server.upstream }};
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
        }
        
        access_log /var/log/nginx/{{ server.domain }}.log;
    }
    {% endfor %}
}
'''

# 3. 动态渲染配置文件脚本
import jinja2
import yaml

def render_config_template(template_path, config_path, output_path):
"""渲染配置模板"""
# 加载模板
env = jinja2.Environment(loader=jinja2.FileSystemLoader("."))
template = env.get_template(template_path)

# 加载配置（从YAML文件读取动态参数）
with open(config_path, "r", encoding="utf-8") as f:
    config_data = yaml.safe_load(f)

# 可选：动态获取服务器CPU核心数（用于worker_processes参数）
def get_cpu_cores():
    """获取本地CPU核心数（若部署到远程可通过SSH执行nproc命令）"""
    import multiprocessing
    return multiprocessing.cpu_count()

# 补充动态参数
config_data["worker_processes"] = config_data.get("worker_processes", get_cpu_cores())
config_data["log_level"] = config_data.get("log_level", "warn")
config_data["worker_connections"] = config_data.get("worker_connections", 1024)

# 渲染模板并输出文件
with open(output_path, "w", encoding="utf-8") as f:
    rendered_content = template.render(**config_data)
    f.write(rendered_content)

print(f"✅ 配置模板渲染完成，输出文件：{output_path}")

# ========== 配置文件示例 (nginx_config.yaml) ==========
'''
# nginx_config.yaml 动态配置参数
worker_processes: 4  # 若不指定则自动获取CPU核心数
log_level: info
worker_connections: 2048
servers:
  - port: 80
    domain: www.example.com
    upstream: 127.0.0.1:8080
  - port: 443
    domain: www.example.com
    upstream: 127.0.0.1:8080
    ssl: true
    ssl_cert: /etc/nginx/certs/example.crt
    ssl_key: /etc/nginx/certs/example.key
'''

# ========== 调用示例 ==========
if __name__ == "__main__":
    # 渲染Nginx配置模板
    render_config_template(
        template_path="nginx_template.conf",  # 模板文件路径
        config_path="nginx_config.yaml",      # 配置参数文件
        output_path="nginx.conf"              # 渲染后的输出文件
    )
    
    # 渲染完成后，可直接调用之前的批量上传函数部署到服务器
    # batch_upload("production", "./nginx.conf", "/usr/local/nginx/conf/nginx.conf")

# configs/ 配置目录结构
# configs/
# ├── base.yaml          # 基础通用配置
# ├── production.yaml    # 生产环境配置（继承base）
# ├── test.yaml          # 测试环境配置（继承base）
# └── development.yaml   # 开发环境配置（继承base）

# 1. 基础配置 (base.yaml)
'''
# 通用配置
nginx:
  worker_processes: auto
  worker_connections: 1024
  log_level: warn
  timeout: 60s
common:
  timezone: Asia/Shanghai
  charset: utf-8
  backup_dir: /data/backup
'''

# 2. 生产环境配置 (production.yaml)
'''
# 继承基础配置
extends: base.yaml

# 覆盖/新增生产环境配置
nginx:
  worker_processes: 8
  worker_connections: 2048
  log_level: info
  access_log: /var/log/nginx/access.log
common:
  backup_cron: "0 2 * * *"  # 每天凌晨2点备份
  monitor_enabled: true
'''

# 3. 配置加载工具类
import yaml
import os

class ConfigLoader:
    """多环境配置加载器"""
    def __init__(self, config_dir="configs"):
        self.config_dir = config_dir
        self.config_cache = {}
    
    def load_config(self, env):
        """加载指定环境的配置（支持继承）"""
        if env in self.config_cache:
            return self.config_cache[env]
        
        # 加载当前环境配置
        env_config_path = os.path.join(self.config_dir, f"{env}.yaml")
        if not os.path.exists(env_config_path):
            raise FileNotFoundError(f"配置文件 {env_config_path} 不存在")
        
        with open(env_config_path, "r", encoding="utf-8") as f:
            env_config = yaml.safe_load(f)
        
        # 处理继承
        if "extends" in env_config:
            parent_env = env_config.pop("extends")
            parent_config = self.load_config(parent_env.replace(".yaml", ""))
            # 合并配置（子配置覆盖父配置）
            config = self._merge_config(parent_config, env_config)
        else:
            config = env_config
        
        self.config_cache[env] = config
        return config
    
    def _merge_config(self, parent, child):
        """递归合并配置字典"""
        merged = parent.copy()
        for key, value in child.items():
            if key in merged and isinstance(merged[key], dict) and isinstance(value, dict):
                merged[key] = self._merge_config(merged[key], value)
            else:
                merged[key] = value
        return merged

# 配置加载示例
if __name__ == "__main__":
    config_loader = ConfigLoader()
    
    # 加载生产环境配置
    prod_config = config_loader.load_config("production")
    print("生产环境Nginx配置：", prod_config["nginx"])
    
    # 加载测试环境配置
    test_config = config_loader.load_config("test")
    print("测试环境通用配置：", test_config["common"])
    
    # 渲染配置时传入对应环境的配置
    # render_config_template(
    #     template_path="nginx_template.conf",
    #     config_path=f"configs/{env}.yaml",
    #     output_path=f"nginx_{env}.conf"
    # )

#!/usr/bin/env python3
# deploy_with_args.py 支持命令行参数的部署脚本
import argparse
import sys
from batch_execute import batch_execute
from config_loader import ConfigLoader

def parse_args():
    """解析命令行参数"""
    parser = argparse.ArgumentParser(description="Python批量服务器部署工具")
    
    # 必选参数
    parser.add_argument(
        "--env", 
        required=True,
        choices=["development", "test", "production"],
        help="部署环境：development/test/production"
    )
    
    # 可选参数
    parser.add_argument(
        "--action",
        default="deploy",
        choices=["deploy", "install", "config", "check", "backup"],
        help="执行操作：deploy(部署)/install(安装)/config(配置)/check(检查)/backup(备份)"
    )
    
    parser.add_argument(
        "--servers",
        nargs="+",
        help="指定服务器IP列表（不指定则使用配置文件中的所有服务器）"
    )
    
    parser.add_argument(
        "--threads",
        type=int,
        default=10,
        help="并发线程数（默认10）"
    )
    
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="模拟执行（不实际操作服务器）"
    )
    
    return parser.parse_args()

def main():
    """主函数"""
    args = parse_args()
    print(f"开始执行部署任务 | 环境：{args.env} | 操作：{args.action} | 并发数：{args.threads}")
    
    # 加载对应环境配置
    config_loader = ConfigLoader()
    env_config = config_loader.load_config(args.env)
    
    # 模拟执行模式提示
    if args.dry_run:
        print("⚠️  模拟执行模式，不会实际操作服务器！")
    
    # 根据操作类型执行不同逻辑
    if args.action == "install":
        # 安装命令（从配置读取）
        install_commands = env_config.get("install_commands", [
            "yum install -y nginx",
            "systemctl enable nginx"
        ])
        
        if not args.dry_run:
            results = batch_execute(
                server_group=args.env,
                commands=install_commands,
                max_threads=args.threads
            )
            # 处理执行结果
            success = sum(1 for r in results.values() if r["status"] == "success")
            print(f"安装完成 | 成功：{success}/{len(results)} 台服务器")
    
    elif args.action == "deploy":
        print(f"开始部署 {args.env} 环境配置...")
        # 部署逻辑（省略具体实现）
    
    elif args.action == "check":
        print(f"检查 {args.env} 环境服务状态...")
        # 检查逻辑（省略具体实现）
    
    elif args.action == "backup":
        print(f"执行 {args.env} 环境备份任务...")
        # 备份逻辑（省略具体实现")

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"❌ 执行失败：{str(e)}")
        sys.exit(1)

# 命令行调用示例：
# python deploy_with_args.py --env production --action install --threads 15
# python deploy_with_args.py --env test --action deploy --dry-run
# python deploy_with_args.py --env production --action check --servers 192.168.1.101 192.168.1.102

import time
import functools
from functools import wraps

def retry(max_retries=3, delay=2, backoff=2, exceptions=(Exception,)):
    """
    重试装饰器
    :param max_retries: 最大重试次数
    :param delay: 初始延迟（秒）
    :param backoff: 延迟倍数（每次重试延迟 * backoff）
    :param exceptions: 需要重试的异常类型
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            _max_retries = max_retries
            _delay = delay
            
            while _max_retries > 0:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    _max_retries -= 1
                    if _max_retries == 0:
                        raise  # 最后一次失败，抛出异常
                    
                    # 打印重试信息
                    host = args[0] if args else "未知服务器"
                    print(f"⚠️  [{host}] {func.__name__} 执行失败: {str(e)} | 剩余重试次数: {_max_retries} | 延迟 {_delay} 秒")
                    
                    # 延迟后重试
                    time.sleep(_delay)
                    _delay *= backoff  # 指数退避
            return None
        return wrapper
    return decorator

# 使用示例：为SSH连接和命令执行添加重试机制
@retry(max_retries=3, delay=2, backoff=2, exceptions=(paramiko.SSHException, TimeoutError))
def execute_command_with_retry(server, command):
    """带重试机制的命令执行函数"""
    host = server["host"]
    password = os.environ.get(server["password"].replace("${", "").replace("}", ""))
    
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(
        hostname=host,
        port=server["port"],
        username=server["username"],
        password=password,
        timeout=10
    )
    
    stdin, stdout, stderr = ssh.exec_command(command)
    exit_code = stdout.channel.recv_exit_status()
    
    if exit_code != 0:
        error_msg = stderr.read().decode("utf-8", errors="ignore")
        raise RuntimeError(f"命令执行失败 (退出码: {exit_code}): {error_msg}")
    
    result = stdout.read().decode("utf-8", errors="ignore")
    ssh.close()
    return result

# 调用示例
if __name__ == "__main__":
    server = {"host": "192.168.1.101", "port": 22, "username": "root", "password": "${PROD_SSH_PWD}"}
    try:
        # 执行可能失败的命令（如网络波动导致连接失败）
        result = execute_command_with_retry(server, "systemctl restart nginx")
        print(f"✅  {server['host']} 命令执行成功: {result}")
    except Exception as e:
        print(f"❌  {server['host']} 命令执行最终失败: {str(e)}")

import smtplib
from email.mime.text import MIMEText
from email.header import Header

# 异常类型定义
class DeployError(Exception):
    """部署基类异常"""
    pass

class SSHConnectError(DeployError):
    """SSH连接异常"""
    pass

class CommandExecuteError(DeployError):
    """命令执行异常"""
    pass

class FileUploadError(DeployError):
    """文件上传异常"""
    pass

# 告警工具类
class AlertManager:
    """告警管理器"""
    def __init__(self, smtp_config):
        self.smtp_host = smtp_config["host"]
        self.smtp_port = smtp_config["port"]
        self.smtp_user = smtp_config["user"]
        self.smtp_password = smtp_config["password"]
        self.recipients = smtp_config["recipients"]
    
    def send_email_alert(self, subject, content):
        """发送邮件告警"""
        try:
            msg = MIMEText(content, "plain", "utf-8")
            msg["From"] = Header("服务器部署系统", "utf-8")
            msg["To"] = Header(",".join(self.recipients), "utf-8")
            msg["Subject"] = Header(subject, "utf-8")
            
            # 发送邮件
            server = smtplib.SMTP_SSL(self.smtp_host, self.smtp_port)
            server.login(self.smtp_user, self.smtp_password)
            server.sendmail(self.smtp_user, self.recipients, msg.as_string())
            server.quit()
            
            print("📧 告警邮件发送成功")
        except Exception as e:
            print(f"❌ 告警邮件发送失败: {str(e)}")
    
    def send_dingding_alert(self, webhook, content):
        """发送钉钉告警（可选）"""
        import requests
        try:
            data = {
                "msgtype": "text",
                "text": {
                    "content": content
                }
            }
            response = requests.post(webhook, json=data, timeout=10)
            if response.json()["errcode"] != 0:
                raise Exception(f"钉钉告警失败: {response.text}")
            print("📱 钉钉告警发送成功")
        except Exception as e:
            print(f"❌ 钉钉告警发送失败: {str(e)}")

# 异常处理主逻辑
def handle_deploy_exception(server, exc, alert_manager=None):
    """处理部署异常并发送告警"""
    host = server["host"]
    error_type = type(exc).__name__
    
    # 构建告警内容
    subject = f"【服务器部署失败】{host} - {error_type}"
    content = f"""
服务器部署失败详情：
服务器IP: {host}
异常类型: {error_type}
异常信息: {str(exc)}
部署时间: {time.strftime('%Y-%m-%d %H:%M:%S')}
服务器标签: {server.get('tags', '无')}
    """.strip()
    
    # 打印错误日志
    print(f"\n{content}\n")
    
    # 发送告警
    if alert_manager:
        alert_manager.send_email_alert(subject, content)
        # 可选：发送钉钉告警
        # alert_manager.send_dingding_alert("https://oapi.dingtalk.com/robot/send?access_token=xxx", content)

# 使用示例
if __name__ == "__main__":
    # 初始化告警管理器
    smtp_config = {
        "host": "smtp.163.com",
        "port": 465,
        "user": "your_email@163.com",
        "password": "your_email_password",
        "recipients": ["admin@example.com", "ops@example.com"]
    }
    alert_manager = AlertManager(smtp_config)
    
    # 模拟部署失败
    server = {"host": "192.168.1.101", "port": 22, "username": "root", "tags": ["web", "nginx"]}
    try:
        # 模拟SSH连接失败
        raise SSHConnectError("连接超时（服务器可能宕机）")
    except DeployError as e:
        handle_deploy_exception(server, e, alert_manager)

import logging
import json
import os
from datetime import datetime

# 配置日志
def setup_deploy_logger(log_dir="deploy_logs"):
    """配置部署日志记录器"""
    # 创建日志目录
    os.makedirs(log_dir, exist_ok=True)
    
    # 日志文件名（按日期）
    log_file = os.path.join(log_dir, f"deploy_{datetime.now().strftime('%Y%m%d')}.log")
    
    # 配置日志格式
    log_format = "%(asctime)s - %(levelname)s - %(module)s - %(message)s"
    logging.basicConfig(
        level=logging.INFO,
        format=log_format,
        handlers=[
            logging.FileHandler(log_file, encoding="utf-8"),  # 文件日志
            logging.StreamHandler()  # 控制台日志
        ]
    )
    
    return logging.getLogger("deploy")

# 初始化日志器
logger = setup_deploy_logger()

# 结构化日志记录函数
def log_deploy_operation(server, operation, status, details=None):
    """记录部署操作日志（结构化）"""
    log_data = {
        "timestamp": datetime.now().isoformat(),
        "server": server["host"],
        "port": server["port"],
        "username": server["username"],
        "tags": server.get("tags", []),
        "operation": operation,
        "status": status,  # success/failed/running
        "details": details or {}
    }
    
    # 记录JSON格式日志（便于后续分析）
    log_message = json.dumps(log_data, ensure_ascii=False)
    
    if status == "success":
        logger.info(log_message)
    elif status == "failed":
        logger.error(log_message)
    elif status == "running":
        logger.warning(log_message)

# 使用示例
if __name__ == "__main__":
    server = {"host": "192.168.1.101", "port": 22, "username": "root", "tags": ["web", "nginx"]}
    
    # 记录开始部署
    log_deploy_operation(
        server=server,
        operation="install_nginx",
        status="running",
        details={"version": "1.24.0"}
    )
    
    try:
        # 模拟部署成功
        log_deploy_operation(
            server=server,
            operation="install_nginx",
            status="success",
            details={"version": "1.24.0", "duration": "45秒"}
        )
    except Exception as e:
        # 记录部署失败
        log_deploy_operation(
            server=server,
            operation="install_nginx",
            status="failed",
            details={"error": str(e), "retry_count": 3}
        )

# 1. 安装依赖
# pip install flask flask-socketio eventlet

# 2. 进度监控服务 (deploy_monitor.py)
from flask import Flask, render_template
from flask_socketio import SocketIO, emit
import eventlet
import threading
import json
from batch_execute import batch_execute

eventlet.monkey_patch()

app = Flask(__name__)
app.config['SECRET_KEY'] = 'deploy_monitor_secret'
socketio = SocketIO(app, cors_allowed_origins="*")

# 全局进度存储
deploy_progress = {
    "total": 0,
    "completed": 0,
    "success": 0,
    "failed": 0,
    "servers": {}
}

def update_progress(host, status, details=None):
    """更新进度并推送至前端"""
    global deploy_progress
    
    if status == "success":
        deploy_progress["success"] += 1
    elif status == "failed":
        deploy_progress["failed"] += 1
    
    deploy_progress["completed"] += 1
    deploy_progress["servers"][host] = {
        "status": status,
        "details": details or {},
        "timestamp": datetime.now().isoformat()
    }
    
    # 推送进度到前端
    socketio.emit('progress_update', {
        "progress": deploy_progress,
        "percentage": (deploy_progress["completed"] / deploy_progress["total"]) * 100 if deploy_progress["total"] > 0 else 0
    }, broadcast=True)

@app.route('/')
def index():
    """监控页面"""
    return render_template('monitor.html')

@socketio.on('start_deploy')
def handle_start_deploy(data):
    """处理开始部署请求"""
    global deploy_progress
    
    # 重置进度
    deploy_progress = {
        "total": 0,
        "completed": 0,
        "success": 0,
        "failed": 0,
        "servers": {}
    }
    
    # 获取部署参数
    server_group = data.get("server_group", "production")
    commands = data.get("commands", [])
    max_threads = data.get("max_threads", 10)
    
    # 读取服务器列表
    with open("servers.yaml", "r", encoding="utf-8") as f:
        config = yaml.safe_load(f)
    servers = config["servers"][server_group]
    deploy_progress["total"] = len(servers)
    
    # 启动部署线程（避免阻塞Flask）
    def deploy_thread():
        results = batch_execute(server_group, commands, max_threads)
        
        # 更新进度
        for host, result in results.items():
            update_progress(
                host=host,
                status=result["status"],
                details=result.get("error") or result.get("output")
            )
        
        # 发送部署完成通知
        socketio.emit('deploy_complete', {
            "success": deploy_progress["success"],
            "failed": deploy_progress["failed"],
            "total": deploy_progress["total"]
        }, broadcast=True)
    
    threading.Thread(target=deploy_thread).start()

if __name__ == "__main__":
    socketio.run(app, host='0.0.0.0', port=5000, debug=True)

# 3. 监控页面模板 (templates/monitor.html)
'''



    
    
    
    
    


    
服务器批量部署监控
    开始部署
    

        

    
    
0%
    

    



'''

#!/usr/bin/env python3
# full_nginx_deploy.py 完整的Nginx批量部署脚本
import os
import yaml
import logging
from dotenv import load_dotenv
from tqdm import tqdm
from threading import Thread, Lock, Semaphore
from config_loader import ConfigLoader
from alert_manager import AlertManager
from retry import retry
import paramiko

# 初始化配置
load_dotenv()
logger = setup_deploy_logger()
config_loader = ConfigLoader()
alert_manager = AlertManager({
    "host": os.getenv("SMTP_HOST"),
    "port": int(os.getenv("SMTP_PORT", 465)),
    "user": os.getenv("SMTP_USER"),
    "password": os.getenv("SMTP_PASSWORD"),
    "recipients": os.getenv("ALERT_RECIPIENTS").split(",")
})

# 全局变量
lock = Lock()
progress_bar = None
deploy_results = {}

class NginxDeployer:
    """Nginx批量部署器"""
    def __init__(self, env="production", max_threads=20):
        self.env = env
        self.max_threads = max_threads
        self.config = config_loader.load_config(env)
        self.servers = self._load_servers()
        self.semaphore = Semaphore(max_threads)
    
    def _load_servers(self):
        """加载服务器列表"""
        with open("servers.yaml", "r", encoding="utf-8") as f:
            servers_config = yaml.safe_load(f)
        return servers_config["servers"][self.env]
    
    @retry(max_retries=3, delay=2, backoff=2, exceptions=(paramiko.SSHException, TimeoutError))
    def _connect_ssh(self, server):
        """建立SSH连接（带重试）"""
        password = os.environ.get(server["password"].replace("${", "").replace("}", ""))
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(
            hostname=server["host"],
            port=server["port"],
            username=server["username"],
            password=password,
            timeout=15,
            allow_agent=False,
            look_for_keys=False
        )
        return ssh
    
    def _execute_command(self, ssh, command, desc="执行命令"):
        """执行单个命令"""
        stdin, stdout, stderr = ssh.exec_command(command, get_pty=True)
        exit_code = stdout.channel.recv_exit_status()
        stdout_content = stdout.read().decode("utf-8", errors="ignore")
        stderr_content = stderr.read().decode("utf-8", errors="ignore")
        
        if exit_code != 0:
            raise RuntimeError(f"{desc}失败 (退出码: {exit_code}): {stderr_content}")
        
        return stdout_content
    
    def _deploy_single_server(self, server):
        """部署单台服务器"""
        host = server["host"]
        try:
            with self.semaphore:
                # 记录开始部署
                log_deploy_operation(server, "nginx_deploy", "running")
                logger.info(f"开始部署 {host} Nginx")
                
                # 建立SSH连接
                ssh = self._connect_ssh(server)
                
                # 1. 安装依赖
                logger.info(f"{host} 安装依赖包")
                self._execute_command(
                    ssh,
                    "yum install -y gcc pcre-devel zlib-devel openssl-devel wget",
                    "安装依赖包"
                )
                
                # 2. 下载并编译安装Nginx
                nginx_version = self.config["nginx"]["version"]
                logger.info(f"{host} 安装Nginx {nginx_version}")
                self._execute_command(
                    ssh,
                    f"wget http://nginx.org/download/nginx-{nginx_version}.tar.gz -P /tmp -q",
                    "下载Nginx源码"
                )
                self._execute_command(
                    ssh,
                    f"cd /tmp && tar -zxf nginx-{nginx_version}.tar.gz",
                    "解压Nginx源码"
                )
                self._execute_command(
                    ssh,
                    f"cd /tmp/nginx-{nginx_version} && ./configure {self.config['nginx']['configure_args']} && make -j$(nproc) && make install",
                    "编译安装Nginx"
                )
                
                # 3. 配置系统服务
                logger.info(f"{host} 配置Nginx系统服务")
                self._execute_command(
                    ssh,
                    f"echo '{self.config['nginx']['service_file']}' > /usr/lib/systemd/system/nginx.service",
                    "创建系统服务文件"
                )
                
                # 4. 渲染并上传配置文件
                logger.info(f"{host} 部署Nginx配置")
                # 本地渲染配置文件
                render_config_template(
                    template_path="nginx_template.conf",
                    config_path=f"configs/{self.env}.yaml",
                    output_path=f"temp/nginx_{host}.conf"
                )
                # 上传配置文件
                sftp = ssh.open_sftp()
                sftp.put(f"temp/nginx_{host}.conf", "/usr/local/nginx/conf/nginx.conf")
                sftp.close()
                
                # 5. 启动并验证
                logger.info(f"{host} 启动Nginx服务")
                self._execute_command(ssh, "systemctl daemon-reload", "重载系统服务")
                self._execute_command(ssh, "systemctl enable nginx --now", "启动Nginx服务")
                self._execute_command(ssh, "/usr/local/nginx/sbin/nginx -t", "验证Nginx配置")
                
                # 6. 检查服务状态
                status = self._execute_command(ssh, "systemctl is-active nginx", "检查Nginx状态")
                if status.strip() != "active":
                    raise RuntimeError(f"Nginx服务未正常启动: {status}")
                
                # 关闭连接
                ssh.close()
                
                # 记录部署成功
                log_deploy_operation(
                    server,
                    "nginx_deploy",
                    "success",
                    {"version": nginx_version, "status": "active"}
                )
                logger.info(f"{host} Nginx部署成功")
                
                with lock:
                    deploy_results[host] = {"status": "success"}
                    progress_bar.update(1)
                
        except Exception as e:
            # 记录部署失败
            log_deploy_operation(
                server,
                "nginx_deploy",
                "failed",
                {"error": str(e)}
            )
            logger.error(f"{host} Nginx部署失败: {str(e)}")
            
            # 发送告警
            handle_deploy_exception(server, e, alert_manager)
            
            with lock:
                deploy_results[host] = {"status": "failed", "error": str(e)}
                progress_bar.update(1)
    
    def deploy(self):
        """批量部署Nginx"""
        global progress_bar
        deploy_results.clear()
        
        # 初始化进度条
        progress_bar = tqdm(total=len(self.servers), desc=f"部署Nginx到{self.env}环境")
        
        # 创建临时目录
        os.makedirs("temp", exist_ok=True)
        
        # 启动部署线程
        threads = []
        for server in self.servers:
            t = Thread(target=self._deploy_single_server, args=(server,))
            threads.append(t)
            t.start()
        
        # 等待所有线程完成
        for t in threads:
            t.join()
        
        # 清理临时文件
        os.system("rm -rf temp/*")
        
        # 输出部署结果
        progress_bar.close()
        success_count = sum(1 for r in deploy_results.values() if r["status"] == "success")
        failed_count = len(deploy_results) - success_count
        
        logger.info(f"部署完成 | 总计: {len(deploy_results)} 台 | 成功: {success_count} | 失败: {failed_count}")
        print(f"\n===== 部署结果汇总 ({self.env}环境) =====")
        print(f"总服务器数: {len(self.servers)} 台")
        print(f"✅ 成功部署: {success_count} 台")
        print(f"❌ 部署失败: {failed_count} 台")

        # 输出成功列表
        if success_count > 0:
            print("\n【成功服务器列表】:")
            for host in [k for k, v in deploy_results.items() if v["status"] == "success"]:
                print(f"  - {host}")

        # 输出失败详情
        if failed_count > 0:
            print("\n【失败服务器详情】:")
            for host, result in deploy_results.items():
                if result["status"] == "failed":
                    print(f"\n  🚨 {host}:")
                    print(f"     错误原因: {result['error']}")
                    # 如果有命令执行输出，补充输出
                    if "output" in result:
                        for cmd, cmd_result in result["output"].items():
                            if cmd_result.get("stderr"):
                                print(f"     命令: {cmd}")
                                print(f"     错误输出: {cmd_result['stderr'][:200]}...")  # 截断过长输出

        # 生成部署报告文件
        report_path = f"./deploy_reports/nginx_deploy_{self.env}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
        os.makedirs("./deploy_reports", exist_ok=True)

        with open(report_path, "w", encoding="utf-8") as f:
            f.write(f"===== Nginx批量部署报告 ({self.env}环境) =====\n")
            f.write(f"部署时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write(f"总服务器数: {len(self.servers)}\n")
            f.write(f"成功数: {success_count}\n")
            f.write(f"失败数: {failed_count}\n\n")
            
            f.write("【成功服务器】:\n")
            for host in [k for k, v in deploy_results.items() if v["status"] == "success"]:
                f.write(f"  - {host}\n")
            
            f.write("\n【失败服务器】:\n")
            for host, result in deploy_results.items():
                if result["status"] == "failed":
                    f.write(f"\n  {host}:\n")
                    f.write(f"    错误: {result['error']}\n")

        logger.info(f"部署报告已生成: {report_path}")
        print(f"\n📄 部署报告已保存至: {report_path}")

        # 失败数超过阈值时发送告警（示例：邮件/钉钉/企业微信）
        if failed_count > 0 and failed_count / len(self.servers) > 0.1:  # 失败率超10%告警
            self.send_alert(deploy_results, success_count, failed_count)
            logger.warning(f"部署失败率超过10%，已发送告警通知")
            print("\n⚠️  部署失败率超过10%，已发送告警通知！")

        return {
            "total": len(self.servers),
            "success": success_count,
            "failed": failed_count,
            "failed_hosts": [k for k, v in deploy_results.items() if v["status"] == "failed"],
            "report_path": report_path
        }
    
    def send_alert(self, deploy_results, success_count, failed_count):
        """发送部署告警（以钉钉为例）"""
        try:
            import requests
            # 钉钉机器人webhook地址（替换为实际地址）
            webhook_url = os.getenv("DINGTALK_WEBHOOK")
            if not webhook_url:
                logger.warning("未配置钉钉webhook，跳过告警发送")
                return
            
            failed_hosts = [k for k, v in deploy_results.items() if v["status"] == "failed"]
            alert_msg = {
                "msgtype": "markdown",
                "markdown": {
                    "title": f"Nginx批量部署告警 ({self.env}环境)",
                    "text": f"""### Nginx批量部署告警
> 环境: {self.env}
> 部署时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
> 总服务器数: {len(self.servers)}
> 成功数: {success_count}
> 失败数: {failed_count}
> 失败率: {failed_count/len(self.servers)*100:.1f}%

#### 失败服务器列表:
{chr(10).join([f"- {host}" for host in failed_hosts])}

#### 建议操作:
1. 检查失败服务器网络连通性
2. 核对服务器SSH认证信息
3. 查看部署报告获取详细错误日志
"""
                }
            }
            
            response = requests.post(webhook_url, json=alert_msg, timeout=10)
            if response.status_code == 200:
                logger.info("钉钉告警发送成功")
            else:
                logger.error(f"钉钉告警发送失败: {response.text}")
        except Exception as e:
            logger.error(f"发送告警异常: {str(e)}")

# 使用示例
if __name__ == "__main__":
    # 初始化部署器
    deployer = NginxDeployer(env="production", max_threads=20)
    
    # 开始批量部署
    results = deployer.deploy()
    
    # 打印最终统计
    print(f"\n🎉 部署完成！成功率: {results['success']/results['total']*100:.1f}%")
    if results['failed'] > 0:
        print(f"⚠️  需要手动处理的失败服务器: {', '.join(results['failed_hosts'])}")

#!/usr/bin/env python3
# optimize_deploy.py 大规模部署性能优化
import asyncio
import aiohttp
import asyncssh
import yaml
from typing import List, Dict
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# 异步SSH执行（500+服务器时性能提升明显）
class AsyncSSHDeployer:
    """异步SSH部署器（支持500+服务器）"""
    def __init__(self, max_concurrent=50):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def execute_command_async(self, host, port, username, password, command):
        """异步执行命令"""
        async with self.semaphore:
            try:
                async with asyncssh.connect(
                    host=host,
                    port=port,
                    username=username,
                    password=password,
                    known_hosts=None
                ) as conn:
                    result = await conn.run(command, check=True)
                    return {"host": host, "status": "success", "output": result.stdout}
            except Exception as e:
                return {"host": host, "status": "failed", "error": str(e)}
    
    async def batch_execute_async(self, servers: List[Dict], commands: List[str]):
        """批量异步执行命令"""
        tasks = []
        for server in servers:
            for cmd in commands:
                task = self.execute_command_async(
                    host=server["host"],
                    port=server["port"],
                    username=server["username"],
                    password=server["password"],
                    command=cmd
                )
                tasks.append(task)
        
        # 使用asyncio.gather并行执行所有任务
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

# 连接池优化（避免频繁建立/断开连接）
class SSHConnectionPool:
    """SSH连接池（减少连接开销）"""
    def __init__(self, max_size=20):
        self.max_size = max_size
        self.pool = {}
        self.lock = asyncio.Lock()
    
    async def get_connection(self, host, port, username, password):
        """从连接池获取连接"""
        key = f"{host}:{port}:{username}"
        
        async with self.lock:
            if key in self.pool:
                conn = self.pool[key]
                try:
                    # 测试连接是否仍然有效
                    await conn.run("echo test", timeout=2)
                    return conn
                except:
                    # 连接失效，重新建立
                    del self.pool[key]
            
            # 创建新连接
            conn = await asyncssh.connect(
                host=host, port=port,
                username=username, password=password,
                known_hosts=None
            )
            self.pool[key] = conn
            return conn
    
    async def close_all(self):
        """关闭所有连接"""
        for conn in self.pool.values():
            conn.close()
        self.pool.clear()

# 智能批处理（根据服务器性能动态调整）
class SmartBatchDeployer:
    """智能批处理部署器"""
    def __init__(self, servers, test_batch_size=5):
        self.servers = servers
        self.test_batch_size = test_batch_size
        self.optimal_batch_size = 20  # 默认值
    
    def find_optimal_batch_size(self):
        """通过测试批次找到最优批量大小"""
        print("🔍 正在测试最优批量大小...")
        
        test_servers = self.servers[:self.test_batch_size]
        batch_sizes = [5, 10, 20, 30, 50]
        results = {}
        
        for batch_size in batch_sizes:
            start_time = time.time()
            # 模拟执行（实际中应执行轻量级命令）
            for i in range(0, len(test_servers), batch_size):
                batch = test_servers[i:i+batch_size]
                # 这里应该是实际的部署逻辑
                time.sleep(0.1 * len(batch))  # 模拟耗时
            
            total_time = time.time() - start_time
            results[batch_size] = total_time
            print(f"  批量大小 {batch_size}: {total_time:.2f}秒")
        
        # 选择最优批量大小（平衡并发和资源占用）
        optimal = min(results, key=results.get)
        print(f"✅ 推荐批量大小: {optimal}")
        self.optimal_batch_size = optimal
        return optimal
    
    def deploy_in_batches(self):
        """按最优批量大小分批部署"""
        optimal_size = self.find_optimal_batch_size()
        
        total_batches = (len(self.servers) + optimal_size - 1) // optimal_size
        print(f"📊 总服务器: {len(self.servers)} | 批量大小: {optimal_size} | 总批次数: {total_batches}")
        
        for i in range(0, len(self.servers), optimal_size):
            batch = self.servers[i:i+optimal_size]
            batch_num = i // optimal_size + 1
            print(f"\n🚀 开始部署第 {batch_num}/{total_batches} 批 ({len(batch)} 台服务器)")
            
            # 实际部署逻辑
            # self._deploy_batch(batch)
            
            # 模拟部署成功
            time.sleep(2)  # 模拟部署耗时
            print(f"✅ 第 {batch_num} 批部署完成")
            
            # 批次间间隔（避免资源峰值）
            if i + optimal_size < len(self.servers):
                print("⏳ 等待3秒后继续下一批...")
                time.sleep(3)

# 使用示例
if __name__ == "__main__":
    # 加载服务器配置
    with open("servers.yaml", "r", encoding="utf-8") as f:
        config = yaml.safe_load(f)
    
    servers = config["servers"]["production"][:100]  # 取前100台测试
    
    # 智能批处理部署
    smart_deployer = SmartBatchDeployer(servers)
    smart_deployer.deploy_in_batches()
    
    # 异步部署示例（需安装asyncssh: pip install asyncssh）
    # async_deployer = AsyncSSHDeployer(max_concurrent=50)
    # loop = asyncio.get_event_loop()
    # results = loop.run_until_complete(
    #     async_deployer.batch_execute_async(servers, ["hostname", "date"])
    # )

# 完整自动化部署项目结构
auto-deploy-project/
├── README.md                    # 项目说明文档
├── requirements.txt             # Python依赖包列表
├── .env.example                 # 环境变量示例（不含真实密码）
├── .gitignore                   # Git忽略文件配置
│
├── configs/                     # 配置文件目录
│   ├── base.yaml               # 基础通用配置
│   ├── production.yaml         # 生产环境配置
│   ├── test.yaml               # 测试环境配置
│   └── development.yaml        # 开发环境配置
│
├── scripts/                     # 核心脚本目录
│   ├── deploy_nginx.py         # Nginx部署主脚本
│   ├── batch_execute.py        # 批量命令执行
│   ├── file_uploader.py        # 文件批量上传
│   ├── config_loader.py        # 配置加载器
│   ├── alert_manager.py        # 告警管理器
│   └── monitor_server.py       # 部署监控服务
│
├── templates/                   # 配置文件模板
│   ├── nginx_template.conf     # Nginx配置模板
│   ├── service_template.service # 系统服务模板
│   └── app_config.json.j2      # 应用配置模板
│
├── deploy_logs/                # 部署日志目录（自动创建）
│   └── deploy_20231215.log     # 按日期自动分割日志
│
├── deploy_reports/             # 部署报告目录（自动创建）
│   └── nginx_deploy_20231215_143022.txt
│
├── temp/                       # 临时文件目录（自动清理）
│
├── fabfile.py                  # Fabric3部署脚本
│
└── servers.yaml                # 服务器清单配置文件