Python自动化部署：从手忙脚乱到一键搞定

看着挺完整对吧？结果第二次部署就翻车了——服务器上的Python版本是3.6，但requirements.txt里有个包需要3.8以上，安装直接报错。而且我当时没注意，安装到一半脚本还继续往下跑，导致环境搞乱了。

这个设计真的反人类 —— 我后来加上环境检测才发现，很多坑其实在部署开始前就能发现。

改造后的环境检测模块：

def check_environment():
"""部署前环境检测，不通过直接退出"""
errors = []

# 1. Python版本检测
py_version = sys.version_info
if py_version.major < 3 or (py_version.major == 3 and py_version.minor < 8):
errors.append(f"Python版本需要 >= 3.8，当前: {py_version.major}.{py_version.minor}")

# 2. 磁盘空间检测（至少2GB）
import shutil
total, used, free = shutil.disk_usage("/")
free_gb = free // (2**30)
if free_gb < 2:
errors.append(f"磁盘空间不足，剩余: {free_gb}GB，需要至少2GB")

# 3. 端口检测
import socket
port = 8000
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex(('127.0.0.1', port))
if result == 0:
er

rors.append(f"端口 {port} 已被占用")
sock.close()

# 4. Git仓库状态检测
repo_path = Path("/app/myproject")
if repo_path.exists():
result = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True, text=True, cwd=str(repo_path)
)
if result.stdout.strip():
errors.append("Git工作区有未提交的修改，建议先处理")

第二个坑：部署失败要能自动回滚

第一次写部署脚本时，我天真地以为"只要我代码写对了，就不会失败"。现实是：网络超时、服务器磁盘满、数据库连接数上限...各种意外都能让部署到一半挂了。

最惨的一次：新版代码已经覆盖了旧版，但启动时报错，服务直接挂了。等我反应过来，已经是5分钟后了，用户骂声一片。

官方文档这段文档不够清晰 —— 关于回滚策略，很多教程只说要备份，但没说怎么高效备份。我后来用的是"软链接切换"方案：

class DeployManager:
def __init__(self, base_dir="/opt/apps/myapp"):
self.base_dir = Path(base_dir)
self.releases_dir = self.base_dir / "releases"
self.current_link = self.base_dir / "current"

# 版本号用时间戳，保证唯一且有序
self.version = datetime.now().strftime("%Y%m%d_%H%M%S")
self.release_dir = self.releases_dir / self.version

def prepare_release(self):
"""准备新版本的部署目录"""
self.release_dir.mkdir(parents=True, exist_ok=True)
print(f"📁 创建部署目录: {self.release_dir}")

def deploy(self):
"""执行部署，这里简化了构建过程"""
try:
# 1. 拉取代码到临时目录
temp_dir = self.release_dir / "source"
subprocess.run(["git", "clone", "git@github.com:myproject.git", str(temp_dir)], check=True)

# 2. 安装依赖
subprocess.run(["pip", "install", "-r", f"{temp_dir}/requirements.txt", "-t",
str(self.release_dir / "venv")], check=True)

# 3. 切换软链接（核心操作）
# 先创建临时链接，避免直接覆盖导致服务中断
temp_link = self.base_dir / "temp_current"
if temp_link.exists():
temp_link.unlink()
temp_link.symlink_to(self.release_dir)

# 原子操作：重命名，比直接删除+创建更安全
os.rename(str(temp_link), str(self.current_link))

# 4. 重启服务
result = subprocess.run(["systemctl", "restart", "myapp"],
capture_output=True, text=True, timeout=30)

if result.returncode != 0:
raise Exception(f"服务重启失败: {result.stderr}")

print(f"✅ 部署成功: {self.version}")
return True

except Exception as e:
print(f"❌ 部署失败: {e}")
self.rollback()
return False

def rollback(self):
"""回滚到上一个版本"""
# 获取所有版本列表（按时间排序）
versions = sorted([d for d in self.releases_dir.iterdir() if d.is_dir()])

if len(versions) < 2:
print("❌ 没有可回滚的版本")
return

# 上一个版本就是倒数第二个
prev_version = versions[-2]

# 切换软链接
self.current_link.unlink()
self.current_link.symlink_to(prev_version)

# 重启服务
subprocess.run(["systemctl", "restart", "myapp"])

# 清理失败的版本
shutil.rmtree(self.release_dir)

这个方案的精髓在于：

第三个坑：日志和通知

部署脚本跑到半夜，失败了但你睡着了，第二天才知道——这种事儿我经历过。所以一定要有通知机制。

先看我的日志模块：

class DeployLogger:
def __init__(self, log_dir="/var/log/deploy"):
self.log_dir = Path(log_dir)
self.log_dir.mkdir(parents=True, exist_ok=True)

# 设置日志格式
log_file = self.log_dir / f"deploy_{datetime.now().strftime('%Y%m%d')}.log"

self.logger = logging.getLogger("deploy")
self.logger.setLevel(logging.DEBUG)

# 文件handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setLevel(logging.DEBUG)

# 控制台handler
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)

# 格式化
formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
fh.setFormatter(formatter)
ch.setFormatter(formatter)

self.logger.addHandler(fh)
self.logger.addHandler(ch)

def log_deploy_event(self, event_type, status, details=None):
"""记录部署事件，同时写入文件用于后续审计"""
event = {
"timestamp": datetime.now().isoformat(),
"type": event_type, # 'start', 'success', 'failure', 'rollback'
"status": status,
"details": details or {}
}

if status == "success":
self.logger.info(f"[{event_type}] {json.dumps(details)}")
elif status == "failure":
self.logger.error(f"[{event_type}] {json.dumps(details)}")
else:
self.logger.warning(f"[{event_type}] {json.dumps(details)}")

通知我用的是钉钉机器人，简单粗暴：

def send_dingtalk_notify(message, webhook_url):
"""发送钉钉通知"""
headers = {"Content-Type": "application/json"}

# 支持Markdown格式
data = {
"msgtype": "markdown",
"markdown": {
"title": "部署通知",
"text": message
}
}

完整的部署脚本

把上面这些整合起来：

import argparse
import sys
from pathlib import Path

def main():
parser = argparse.ArgumentParser(description="自动化部署脚本")
parser.add_argument("--env", required=True, choices=["staging", "production"])
parser.add_argument("--branch", default="main", help="Git分支")
parser.add_argument("--skip-tests", action="store_true", help="跳过测试")
args = parser.parse_args()

# 初始化
logger = DeployLogger()
deploy_mgr = DeployManager()

try:
# 1. 环境检测
logger.log_deploy_event("check", "start")
check_environment()
logger.log_deploy_event("check", "success")

# 2. 准备部署
logger.log_deploy_event("prepare", "start")
deploy_mgr.prepare_release()
logger.log_deploy_event("prepare", "success")

# 3. 执行部署
logger.log_deploy_event("deploy", "start", {"branch": args.branch})
success = deploy_mgr.deploy()

if success:
logger.log_deploy_event("deploy", "success")
send_dingtalk_notify(f"## ✅ 部署成功\n环境: {args.env}\n分支: {args.branch}")
else:
logger.log_deploy_event("deploy", "failure")
send_dingtalk_notify(f"## ❌ 部署失败\n环境: {args.env}\n分支: {args.branch}")

except Exception as e:
logger.log_deploy_event("error", "failure", {"error": str(e)})
send_dingtalk_notify(f"## ❌ 部署异常\n错误: {str(e)}")
sys.exit(1)