Python实现Linux系统上CI/CD工作流的方法详解

#!/usr/bin/env python3
import subprocess
import os
import sys
import time
import logging
import argparse
import shlex
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("ci_cd.log"), logging.StreamHandler()]
)
logger = logging.getLogger("CI/CD")

class CICD:
    def __init__(self, repo_path, branch="main"):
        self.repo_path = Path(repo_path).resolve()
        self.branch = branch
        self.last_commit = None
        self._check_dependencies(["git", "docker"])
        
    def _check_dependencies(self, deps):
        for cmd in deps:
            try:
                subprocess.run([cmd, "--version"], check=True, 
                              stdout=subprocess.DEVNULL)
            except Exception:
                raise RuntimeError(f"缺少依赖: {cmd}")

    def check_updates(self):
        try:
            # 检查远程更新
            subprocess.run(["git", "-C", str(self.repo_path), "fetch"], check=True)
            
            # 比较本地与远程差异
            diff = subprocess.run(
                ["git", "-C", str(self.repo_path), "diff", "--shortstat", f"origin/{self.branch}"],
                stdout=subprocess.PIPE
            )
            if not diff.stdout.strip():
                return False

            # 拉取最新代码
            subprocess.run(
                ["git", "-C", str(self.repo_path), "pull", "origin", self.branch],
                check=True
            )
            
            # 获取最新提交ID
            new_commit = subprocess.check_output(
                ["git", "-C", str(self.repo_path), "rev-parse", "HEAD"]
            ).decode().strip()
            
            if new_commit != self.last_commit:
                self.last_commit = new_commit
                return True
            return False
            
        except subprocess.CalledProcessError as e:
            logger.error(f"代码更新失败: {str(e)}")
            raise

    def run_tests(self):
        try:
            subprocess.run(["pytest", "tests/"], cwd=self.repo_path, check=True)
            return True
        except subprocess.CalledProcessError:
            logger.error("测试失败")
            return False

    def build(self):
        try:
            subprocess.run(
                ["docker", "build", "-t", "myapp:latest", "."],
                cwd=self.repo_path, check=True
            )
            return True
        except subprocess.CalledProcessError:
            logger.error("构建失败")
            return False

    def deploy(self):
        try:
            # 停止旧容器
            subprocess.run(["docker", "stop", "myapp"], check=False)
            # 启动新容器
            subprocess.run(
                ["docker", "run", "--rm", "-d", "--name", "myapp", "-p", "8000:8000", "myapp:latest"],
                check=True
            )
            return True
        except subprocess.CalledProcessError:
            logger.error("部署失败")
            return False

    def run_pipeline(self):
        if not self.check_updates():
            logger.info("无代码更新")
            return True
            
        logger.info(f"检测到新提交: {self.last_commit[:8]}")
        return self.run_tests() and self.build() and self.deploy()

def main():
    parser = argparse.ArgumentParser(description="CI/CD 流水线")
    parser.add_argument("--repo", required=True, help="仓库路径")
    parser.add_argument("--branch", default="main", help="监控分支")
    parser.add_argument("--daemon", action="store_true", help="守护模式")
    parser.add_argument("--interval", type=int, default=60, help="检查间隔")
    
    args = parser.parse_args()
    
    try:
        ci = CICD(args.repo, args.branch)
        if args.daemon:
            while True:
                ci.run_pipeline()
                time.sleep(args.interval)
        else:
            success = ci.run_pipeline()
            sys.exit(0 if success else 1)
    except Exception as e:
        logger.error(f"流程异常: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()

pip install pytest docker

# 单次运行模式
python ci_cd.py --repo /path/to/repo --branch main

# 守护进程模式（每5分钟检查一次）
python ci_cd.py --repo ~/myapp --daemon --interval 300

# 使用自定义测试命令
python ci_cd.py --repo ~/project \
  --test-cmd "npm run test" \
  --build-cmd "docker build -t myapp:v1 ." \
  --deploy-cmd "kubectl apply -f deploy.yaml"

#!/bin/sh
python /path/to/ci_cd.py --repo $(pwd)

# 添加至 CICD 类
def send_notification(self, message):
    import requests
    requests.post(
        "https://api.alert.com/notify",
        json={"text": f"[CI/CD] {message}"}
    )

# 在部署成功后调用
self.send_notification(f"部署成功: {self.last_commit[:8]}")

def health_check(self):
    import requests
    try:
        resp = requests.get("http://localhost:8000/health", timeout=5)
        return resp.status_code == 200
    except Exception:
        return False

repo: ~/myapp
branch: dev
test_cmd: "npm test"
build_cmd: "docker build -t myapp:latest ."

sudo usermod -aG docker $USER