Python DevOps 完整实战指南

# DevOps 生命周期的 Python 表示
class DevOpsLifecycle:
    def __init__(self):
        self.stages = [
            "Plan",      # 规划
            "Code",      # 编码
            "Build",     # 构建
            "Test",      # 测试
            "Release",   # 发布
            "Deploy",    # 部署
            "Operate",   # 运维
            "Monitor"    # 监控
        ]

    def execute_stage(self, stage, artifacts=None):
        """执行 DevOps 生命周期的某个阶段"""
        if stage not in self.stages:
            raise ValueError(f"Invalid stage: {stage}")

        print(f"Executing {stage} stage...")

        # 根据不同阶段执行相应操作
        if stage == "Build":
            return self.build_application(artifacts)
        elif stage == "Test":
            return self.run_tests(artifacts)
        elif stage == "Deploy":
            return self.deploy_application(artifacts)
        elif stage == "Monitor":
            return self.monitor_application()

    def build_application(self, source_code):
        """构建应用程序"""
        # 模拟构建过程
        return {"status": "success", "artifact": "app.tar.gz"}

    def run_tests(self, build_artifact):
        """运行自动化测试"""
        # 模拟测试过程
        return {"status": "passed", "coverage": 85}

    def deploy_application(self, artifact):
        """部署应用程序"""
        # 模拟部署过程
        return {"status": "deployed", "url": "https://app.example.com"}

    def monitor_application(self):
        """监控应用程序"""
        # 模拟监控过程
        return {"status": "healthy", "uptime": "99.9%"}

# 使用示例
devops = DevOpsLifecycle()
for stage in devops.stages:
    result = devops.execute_stage(stage)
    print(f"{stage} result: {result}")

import datetime
from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class TeamMember:
    name: str
    role: str
    skills: List[str]
    availability: float  # 0.0 to 1.0

@dataclass
class Task:
    id: str
    title: str
    description: str
    required_skills: List[str]
    estimated_hours: int
    priority: str  # high, medium, low
    assigned_to: Optional[str] = None
    status: str = "todo"  # todo, in_progress, done

class DevOpsTeam:
    def __init__(self, name: str):
        self.name = name
        self.members: List[TeamMember] = []
        self.tasks: List[Task] = []
        self.sprint_capacity = 0

    def add_member(self, member: TeamMember):
        """添加团队成员"""
        self.members.append(member)
        print(f"Added {member.name} ({member.role}) to team {self.name}")

    def calculate_sprint_capacity(self, sprint_hours: int = 80):
        """计算团队冲刺容量"""
        total_capacity = 0
        for member in self.members:
            member_capacity = sprint_hours * member.availability
            total_capacity += member_capacity
            print(f"{member.name}: {member_capacity} hours")

        self.sprint_capacity = total_capacity
        print(f"Total team capacity: {total_capacity} hours")
        return total_capacity

    def assign_tasks(self):
        """智能任务分配"""
        # 按优先级排序任务
        sorted_tasks = sorted(self.tasks,
                            key=lambda x: {"high": 3, "medium": 2, "low": 1}[x.priority],
                            reverse=True)

        for task in sorted_tasks:
            if task.assigned_to:
                continue

            # 找到最适合的团队成员
            best_member = self.find_best_assignee(task)
            if best_member:
                task.assigned_to = best_member.name
                print(f"Assigned task '{task.title}' to {best_member.name}")

    def find_best_assignee(self, task: Task) -> Optional[TeamMember]:
        """找到最适合执行任务的团队成员"""
        candidates = []

        for member in self.members:
            # 检查技能匹配度
            skill_match = len(set(task.required_skills) & set(member.skills))
            if skill_match > 0:
                candidates.append((member, skill_match))

        if not candidates:
            return None

        # 选择技能匹配度最高的成员
        candidates.sort(key=lambda x: x[1], reverse=True)
        return candidates[0][0]

    def generate_daily_standup_report(self):
        """生成每日站会报告"""
        print(f"\n=== Daily Standup Report for {self.name} ===")
        print(f"Date: {datetime.datetime.now().strftime('%Y-%m-%d')}")

        for member in self.members:
            member_tasks = [t for t in self.tasks if t.assigned_to == member.name]
            in_progress = [t for t in member_tasks if t.status == "in_progress"]
            completed = [t for t in member_tasks if t.status == "done"]

            print(f"\n{member.name} ({member.role}):")
            print(f"  Yesterday: Completed {len(completed)} tasks")
            print(f"  Today: Working on {len(in_progress)} tasks")
            if in_progress:
                for task in in_progress:
                    print(f"    - {task.title}")

# 使用示例
team = DevOpsTeam("Platform Engineering")

# 添加团队成员
team.add_member(TeamMember("Alice", "DevOps Engineer", ["Python", "Docker", "Kubernetes"], 0.8))
team.add_member(TeamMember("Bob", "SRE", ["Go", "Prometheus", "Grafana"], 0.9))
team.add_member(TeamMember("Charlie", "Platform Engineer", ["Python", "Terraform", "AWS"], 0.7))

# 添加任务
team.tasks.extend([
    Task("T001", "Setup CI/CD Pipeline", "Implement GitLab CI/CD", ["Python", "Docker"], 16, "high"),
    Task("T002", "Monitor Setup", "Setup Prometheus monitoring", ["Prometheus", "Grafana"], 12, "medium"),
    Task("T003", "Infrastructure as Code", "Terraform AWS resources", ["Terraform", "AWS"], 20, "high")
])

# 计算容量和分配任务
team.calculate_sprint_capacity()
team.assign_tasks()
team.generate_daily_standup_report()

#!/usr/bin/env python3
"""
服务器健康检查脚本
"""

import psutil
import requests
import subprocess
import socket
import time
from datetime import datetime
from typing import Dict, List, Any
import json
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

class HealthChecker:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.results = {}
        self.alerts = []

    def check_cpu_usage(self) -> Dict[str, Any]:
        """检查 CPU 使用率"""
        cpu_percent = psutil.cpu_percent(interval=1)
        threshold = self.config.get('cpu_threshold', 80)

        result = {
            'metric': 'cpu_usage',
            'value': cpu_percent,
            'threshold': threshold,
            'status': 'OK' if cpu_percent < threshold else 'CRITICAL',
            'timestamp': datetime.now().isoformat()
        }

        if result['status'] == 'CRITICAL':
            self.alerts.append(f"High CPU usage: {cpu_percent}%")

        return result

    def check_memory_usage(self) -> Dict[str, Any]:
        """检查内存使用率"""
        memory = psutil.virtual_memory()
        threshold = self.config.get('memory_threshold', 85)

        result = {
            'metric': 'memory_usage',
            'value': memory.percent,
            'threshold': threshold,
            'status': 'OK' if memory.percent < threshold else 'CRITICAL',
            'timestamp': datetime.now().isoformat()
        }

        if result['status'] == 'CRITICAL':
            self.alerts.append(f"High memory usage: {memory.percent}%")

        return result

    def check_disk_usage(self) -> List[Dict[str, Any]]:
        """检查磁盘使用率"""
        results = []
        threshold = self.config.get('disk_threshold', 90)

        for partition in psutil.disk_partitions():
            try:
                usage = psutil.disk_usage(partition.mountpoint)
                percent = (usage.used / usage.total) * 100

                result = {
                    'metric': 'disk_usage',
                    'mountpoint': partition.mountpoint,
                    'value': percent,
                    'threshold': threshold,
                    'status': 'OK' if percent < threshold else 'CRITICAL',
                    'timestamp': datetime.now().isoformat()
                }

                if result['status'] == 'CRITICAL':
                    self.alerts.append(f"High disk usage on {partition.mountpoint}: {percent:.1f}%")

                results.append(result)
            except PermissionError:
                continue

        return results

    def check_service_status(self) -> List[Dict[str, Any]]:
        """检查服务状态"""
        results = []
        services = self.config.get('services', [])

        for service in services:
            try:
                result = subprocess.run(
                    ['systemctl', 'is-active', service],
                    capture_output=True, text=True
                )

                status = result.stdout.strip()
                is_active = status == 'active'

                check_result = {
                    'metric': 'service_status',
                    'service': service,
                    'status': 'OK' if is_active else 'CRITICAL',
                    'value': status,
                    'timestamp': datetime.now().isoformat()
                }

                if not is_active:
                    self.alerts.append(f"Service {service} is not active: {status}")

                results.append(check_result)
            except Exception as e:
                results.append({
                    'metric': 'service_status',
                    'service': service,
                    'status': 'ERROR',
                    'error': str(e),
                    'timestamp': datetime.now().isoformat()
                })

        return results

    def check_port_connectivity(self) -> List[Dict[str, Any]]:
        """检查端口连通性"""
        results = []
        ports = self.config.get('ports', [])

        for port_config in ports:
            host = port_config.get('host', 'localhost')
            port = port_config['port']
            timeout = port_config.get('timeout', 5)

            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(timeout)
                result = sock.connect_ex((host, port))
                sock.close()

                is_open = result == 0

                check_result = {
                    'metric': 'port_connectivity',
                    'host': host,
                    'port': port,
                    'status': 'OK' if is_open else 'CRITICAL',
                    'timestamp': datetime.now().isoformat()
                }

                if not is_open:
                    self.alerts.append(f"Port {host}:{port} is not accessible")

                results.append(check_result)
            except Exception as e:
                results.append({
                    'metric': 'port_connectivity',
                    'host': host,
                    'port': port,
                    'status': 'ERROR',
                    'error': str(e),
                    'timestamp': datetime.now().isoformat()
                })

        return results

    def check_http_endpoints(self) -> List[Dict[str, Any]]:
        """检查 HTTP 端点"""
        results = []
        endpoints = self.config.get('http_endpoints', [])

        for endpoint in endpoints:
            url = endpoint['url']
            timeout = endpoint.get('timeout', 10)
            expected_status = endpoint.get('expected_status', 200)

            try:
                response = requests.get(url, timeout=timeout)

                is_healthy = response.status_code == expected_status

                result = {
                    'metric': 'http_endpoint',
                    'url': url,
                    'status_code': response.status_code,
                    'response_time': response.elapsed.total_seconds(),
                    'status': 'OK' if is_healthy else 'CRITICAL',
                    'timestamp': datetime.now().isoformat()
                }

                if not is_healthy:
                    self.alerts.append(f"HTTP endpoint {url} returned status {response.status_code}")

                results.append(result)
            except Exception as e:
                results.append({
                    'metric': 'http_endpoint',
                    'url': url,
                    'status': 'ERROR',
                    'error': str(e),
                    'timestamp': datetime.now().isoformat()
                })

        return results

    def run_all_checks(self) -> Dict[str, Any]:
        """运行所有健康检查"""
        print("Running health checks...")

        self.results = {
            'hostname': socket.gethostname(),
            'timestamp': datetime.now().isoformat(),
            'checks': {
                'cpu': self.check_cpu_usage(),
                'memory': self.check_memory_usage(),
                'disk': self.check_disk_usage(),
                'services': self.check_service_status(),
                'ports': self.check_port_connectivity(),
                'http_endpoints': self.check_http_endpoints()
            },
            'alerts': self.alerts
        }

        return self.results

    def send_alerts(self):
        """发送告警"""
        if not self.alerts:
            return

        email_config = self.config.get('email', {})
        if not email_config:
            print("No email configuration found, skipping alerts")
            return

        try:
            msg = MIMEMultipart()
            msg['From'] = email_config['from']
            msg['To'] = ', '.join(email_config['to'])
            msg['Subject'] = f"Health Check Alerts - {socket.gethostname()}"

            body = f"""
Health Check Alerts for {socket.gethostname()}
Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

Alerts:
{chr(10).join(f"- {alert}" for alert in self.alerts)}

Full report attached.
"""

            msg.attach(MIMEText(body, 'plain'))

            # 添加详细报告
            report_json = json.dumps(self.results, indent=2)
            attachment = MIMEText(report_json, 'plain')
            attachment.add_header('Content-Disposition', 'attachment', filename='health_report.json')
            msg.attach(attachment)

            # 发送邮件
            server = smtplib.SMTP(email_config['smtp_server'], email_config['smtp_port'])
            if email_config.get('use_tls'):
                server.starttls()
            if email_config.get('username'):
                server.login(email_config['username'], email_config['password'])

            server.send_message(msg)
            server.quit()

            print(f"Alert email sent to {email_config['to']}")
        except Exception as e:
            print(f"Failed to send alert email: {e}")

    def save_results(self, filename: str = None):
        """保存检查结果"""
        if not filename:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"health_check_{timestamp}.json"

        with open(filename, 'w') as f:
            json.dump(self.results, f, indent=2)

        print(f"Results saved to: {filename}")

# 配置示例
config = {
    'cpu_threshold': 80,
    'memory_threshold': 85,
    'disk_threshold': 90,
    'services': ['nginx', 'mysql', 'redis'],
    'ports': [
        {'host': 'localhost', 'port': 80},
        {'host': 'localhost', 'port': 3306},
        {'host': 'localhost', 'port': 6379}
    ],
    'http_endpoints': [
        {'url': 'http://localhost/health', 'expected_status': 200},
        {'url': 'https://api.example.com/status', 'expected_status': 200}
    ],
    'email': {
        'smtp_server': 'smtp.gmail.com',
        'smtp_port': 587,
        'use_tls': True,
        'username': 'alerts@example.com',
        'password': 'your_password',
        'from': 'alerts@example.com',
        'to': ['admin@example.com', 'ops@example.com']
    }
}

# 使用示例
if __name__ == "__main__":
    checker = HealthChecker(config)
    results = checker.run_all_checks()

    # 打印摘要
    print(f"\nHealth Check Summary for {results['hostname']}:")
    print(f"Time: {results['timestamp']}")
    print(f"Alerts: {len(results['alerts'])}")

    if results['alerts']:
        print("\nAlerts:")
        for alert in results['alerts']:
            print(f"  - {alert}")

        # 发送告警
        checker.send_alerts()
    else:
        print("All checks passed!")

    # 保存结果
    checker.save_results()

#!/usr/bin/env python3
"""
AWS 资源管理工具
"""

import boto3
import json
import time
from typing import Dict, List, Any, Optional
from botocore.exceptions import ClientError
import yaml

class AWSResourceManager:
    def __init__(self, region: str = 'us-west-2', profile: str = None):
        self.region = region
        self.session = boto3.Session(profile_name=profile)

        # 初始化各种 AWS 客户端
        self.ec2 = self.session.client('ec2', region_name=region)
        self.s3 = self.session.client('s3')
        self.rds = self.session.client('rds', region_name=region)
        self.elbv2 = self.session.client('elbv2', region_name=region)
        self.route53 = self.session.client('route53')
        self.cloudformation = self.session.client('cloudformation', region_name=region)

    def create_vpc(self, vpc_config: Dict[str, Any]) -> str:
        """创建 VPC"""
        try:
            # 创建 VPC
            vpc_response = self.ec2.create_vpc(
                CidrBlock=vpc_config['cidr_block'],
                TagSpecifications=[{
                    'ResourceType': 'vpc',
                    'Tags': [
                        {'Key': 'Name', 'Value': vpc_config['name']},
                        {'Key': 'Environment', 'Value': vpc_config.get('environment', 'dev')}
                    ]
                }]
            )

            vpc_id = vpc_response['Vpc']['VpcId']
            print(f"Created VPC: {vpc_id}")

            # 等待 VPC 可用
            self.ec2.get_waiter('vpc_available').wait(VpcIds=[vpc_id])

            # 启用 DNS 主机名
            self.ec2.modify_vpc_attribute(
                VpcId=vpc_id,
                EnableDnsHostnames={'Value': True}
            )

            # 创建子网
            for subnet_config in vpc_config.get('subnets', []):
                self.create_subnet(vpc_id, subnet_config)

            # 创建互联网网关
            if vpc_config.get('internet_gateway', True):
                self.create_internet_gateway(vpc_id, vpc_config['name'])

            return vpc_id

        except ClientError as e:
            print(f"Error creating VPC: {e}")
            raise

    def create_subnet(self, vpc_id: str, subnet_config: Dict[str, Any]) -> str:
        """创建子网"""
        try:
            subnet_response = self.ec2.create_subnet(
                VpcId=vpc_id,
                CidrBlock=subnet_config['cidr_block'],
                AvailabilityZone=subnet_config.get('availability_zone'),
                TagSpecifications=[{
                    'ResourceType': 'subnet',
                    'Tags': [
                        {'Key': 'Name', 'Value': subnet_config['name']},
                        {'Key': 'Type', 'Value': subnet_config.get('type', 'private')}
                    ]
                }]
            )

            subnet_id = subnet_response['Subnet']['SubnetId']
            print(f"Created subnet: {subnet_id}")

            # 如果是公共子网，启用自动分配公网 IP
            if subnet_config.get('type') == 'public':
                self.ec2.modify_subnet_attribute(
                    SubnetId=subnet_id,
                    MapPublicIpOnLaunch={'Value': True}
                )

            return subnet_id

        except ClientError as e:
            print(f"Error creating subnet: {e}")
            raise

    def create_internet_gateway(self, vpc_id: str, name: str) -> str:
        """创建互联网网关"""
        try:
            # 创建互联网网关
            igw_response = self.ec2.create_internet_gateway(
                TagSpecifications=[{
                    'ResourceType': 'internet-gateway',
                    'Tags': [{'Key': 'Name', 'Value': f"{name}-igw"}]
                }]
            )

            igw_id = igw_response['InternetGateway']['InternetGatewayId']
            print(f"Created Internet Gateway: {igw_id}")

            # 附加到 VPC
            self.ec2.attach_internet_gateway(
                InternetGatewayId=igw_id,
                VpcId=vpc_id
            )

            return igw_id

        except ClientError as e:
            print(f"Error creating Internet Gateway: {e}")
            raise

    def create_security_group(self, vpc_id: str, sg_config: Dict[str, Any]) -> str:
        """创建安全组"""
        try:
            sg_response = self.ec2.create_security_group(
                GroupName=sg_config['name'],
                Description=sg_config['description'],
                VpcId=vpc_id,
                TagSpecifications=[{
                    'ResourceType': 'security-group',
                    'Tags': [{'Key': 'Name', 'Value': sg_config['name']}]
                }]
            )

            sg_id = sg_response['GroupId']
            print(f"Created Security Group: {sg_id}")

            # 添加入站规则
            for rule in sg_config.get('ingress_rules', []):
                self.ec2.authorize_security_group_ingress(
                    GroupId=sg_id,
                    IpPermissions=[{
                        'IpProtocol': rule['protocol'],
                        'FromPort': rule['from_port'],
                        'ToPort': rule['to_port'],
                        'IpRanges': [{'CidrIp': rule['cidr']}]
                    }]
                )

            return sg_id

        except ClientError as e:
            print(f"Error creating Security Group: {e}")
            raise

    def launch_ec2_instance(self, instance_config: Dict[str, Any]) -> str:
        """启动 EC2 实例"""
        try:
            # 用户数据脚本
            user_data = instance_config.get('user_data', '')
            if isinstance(user_data, list):
                user_data = '\n'.join(user_data)

            response = self.ec2.run_instances(
                ImageId=instance_config['ami_id'],
                MinCount=1,
                MaxCount=1,
                InstanceType=instance_config['instance_type'],
                KeyName=instance_config.get('key_name'),
                SecurityGroupIds=instance_config.get('security_group_ids', []),
                SubnetId=instance_config.get('subnet_id'),
                UserData=user_data,
                TagSpecifications=[{
                    'ResourceType': 'instance',
                    'Tags': [
                        {'Key': 'Name', 'Value': instance_config['name']},
                        {'Key': 'Environment', 'Value': instance_config.get('environment', 'dev')}
                    ]
                }]
            )

            instance_id = response['Instances'][0]['InstanceId']
            print(f"Launched EC2 instance: {instance_id}")

            # 等待实例运行
            print("Waiting for instance to be running...")
            self.ec2.get_waiter('instance_running').wait(InstanceIds=[instance_id])

            return instance_id

        except ClientError as e:
            print(f"Error launching EC2 instance: {e}")
            raise

    def create_load_balancer(self, lb_config: Dict[str, Any]) -> str:
        """创建负载均衡器"""
        try:
            response = self.elbv2.create_load_balancer(
                Name=lb_config['name'],
                Subnets=lb_config['subnet_ids'],
                SecurityGroups=lb_config.get('security_group_ids', []),
                Scheme=lb_config.get('scheme', 'internet-facing'),
                Type=lb_config.get('type', 'application'),
                Tags=[
                    {'Key': 'Name', 'Value': lb_config['name']},
                    {'Key': 'Environment', 'Value': lb_config.get('environment', 'dev')}
                ]
            )

            lb_arn = response['LoadBalancers'][0]['LoadBalancerArn']
            print(f"Created Load Balancer: {lb_arn}")

            return lb_arn

        except ClientError as e:
            print(f"Error creating Load Balancer: {e}")
            raise

    def deploy_infrastructure(self, config_file: str):
        """部署完整基础设施"""
        with open(config_file, 'r') as f:
            config = yaml.safe_load(f)

        print(f"Deploying infrastructure from {config_file}")

        # 创建 VPC
        vpc_id = self.create_vpc(config['vpc'])

        # 创建安全组
        security_groups = {}
        for sg_config in config.get('security_groups', []):
            sg_id = self.create_security_group(vpc_id, sg_config)
            security_groups[sg_config['name']] = sg_id

        # 启动 EC2 实例
        instances = {}
        for instance_config in config.get('instances', []):
            # 解析安全组引用
            if 'security_groups' in instance_config:
                instance_config['security_group_ids'] = [
                    security_groups[sg_name] for sg_name in instance_config['security_groups']
                ]

            instance_id = self.launch_ec2_instance(instance_config)
            instances[instance_config['name']] = instance_id

        # 创建负载均衡器
        for lb_config in config.get('load_balancers', []):
            if 'security_groups' in lb_config:
                lb_config['security_group_ids'] = [
                    security_groups[sg_name] for sg_name in lb_config['security_groups']
                ]

            self.create_load_balancer(lb_config)

        print("Infrastructure deployment completed!")
        return {
            'vpc_id': vpc_id,
            'security_groups': security_groups,
            'instances': instances
        }

# 基础设施配置示例
infrastructure_config = """
vpc:
  name: "my-vpc"
  cidr_block: "10.0.0.0/16"
  environment: "production"
  internet_gateway: true
  subnets:
    - name: "public-subnet-1"
      cidr_block: "10.0.1.0/24"
      type: "public"
      availability_zone: "us-west-2a"
    - name: "private-subnet-1"
      cidr_block: "10.0.2.0/24"
      type: "private"
      availability_zone: "us-west-2a"

security_groups:
  - name: "web-sg"
    description: "Security group for web servers"
    ingress_rules:
      - protocol: "tcp"
        from_port: 80
        to_port: 80
        cidr: "0.0.0.0/0"
      - protocol: "tcp"
        from_port: 443
        to_port: 443
        cidr: "0.0.0.0/0"
      - protocol: "tcp"
        from_port: 22
        to_port: 22
        cidr: "10.0.0.0/16"

instances:
  - name: "web-server-1"
    ami_id: "ami-0c02fb55956c7d316"  # Amazon Linux 2
    instance_type: "t3.micro"
    key_name: "my-key-pair"
    security_groups: ["web-sg"]
    environment: "production"
    user_data: |
      #!/bin/bash
      yum update -y
      yum install -y httpd
      systemctl start httpd
      systemctl enable httpd
      echo "<h1>Hello from Web Server 1</h1>" > /var/www/html/index.html

load_balancers:
  - name: "web-lb"
    type: "application"
    scheme: "internet-facing"
    security_groups: ["web-sg"]
    environment: "production"
"""

# 使用示例
if __name__ == "__main__":
    # 保存配置到文件
    with open('infrastructure.yaml', 'w') as f:
        f.write(infrastructure_config)

    # 部署基础设施
    manager = AWSResourceManager(region='us-west-2')
    result = manager.deploy_infrastructure('infrastructure.yaml')

    print(f"Deployment result: {json.dumps(result, indent=2)}")

#!/usr/bin/env python3
"""
GitLab CI/CD 管理工具
"""

import requests
import json
import time
from typing import Dict, List, Any, Optional
import yaml

class GitLabCIManager:
    def __init__(self, gitlab_url: str, access_token: str):
        self.gitlab_url = gitlab_url.rstrip('/')
        self.access_token = access_token
        self.headers = {
            'Authorization': f'Bearer {access_token}',
            'Content-Type': 'application/json'
        }

    def get_project(self, project_id: str) -> Dict[str, Any]:
        """获取项目信息"""
        url = f"{self.gitlab_url}/api/v4/projects/{project_id}"
        response = requests.get(url, headers=self.headers)
        response.raise_for_status()
        return response.json()

    def get_pipelines(self, project_id: str, status: str = None) -> List[Dict[str, Any]]:
        """获取流水线列表"""
        url = f"{self.gitlab_url}/api/v4/projects/{project_id}/pipelines"
        params = {}
        if status:
            params['status'] = status

        response = requests.get(url, headers=self.headers, params=params)
        response.raise_for_status()
        return response.json()

    def get_pipeline_jobs(self, project_id: str, pipeline_id: str) -> List[Dict[str, Any]]:
        """获取流水线的作业列表"""
        url = f"{self.gitlab_url}/api/v4/projects/{project_id}/pipelines/{pipeline_id}/jobs"
        response = requests.get(url, headers=self.headers)
        response.raise_for_status()
        return response.json()

    def trigger_pipeline(self, project_id: str, ref: str = 'main', variables: Dict[str, str] = None) -> Dict[str, Any]:
        """触发流水线"""
        url = f"{self.gitlab_url}/api/v4/projects/{project_id}/pipeline"
        data = {'ref': ref}

        if variables:
            data['variables'] = [
                {'key': key, 'value': value} for key, value in variables.items()
            ]

        response = requests.post(url, headers=self.headers, json=data)
        response.raise_for_status()
        return response.json()

    def wait_for_pipeline(self, project_id: str, pipeline_id: str, timeout: int = 1800) -> str:
        """等待流水线完成"""
        start_time = time.time()

        while time.time() - start_time < timeout:
            url = f"{self.gitlab_url}/api/v4/projects/{project_id}/pipelines/{pipeline_id}"
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()

            pipeline = response.json()
            status = pipeline['status']

            print(f"Pipeline {pipeline_id} status: {status}")

            if status in ['success', 'failed', 'canceled', 'skipped']:
                return status

            time.sleep(30)

        raise TimeoutError(f"Pipeline {pipeline_id} did not complete within {timeout} seconds")

    def get_job_log(self, project_id: str, job_id: str) -> str:
        """获取作业日志"""
        url = f"{self.gitlab_url}/api/v4/projects/{project_id}/jobs/{job_id}/trace"
        response = requests.get(url, headers=self.headers)
        response.raise_for_status()
        return response.text

    def create_deployment(self, project_id: str, deployment_config: Dict[str, Any]) -> Dict[str, Any]:
        """创建部署"""
        url = f"{self.gitlab_url}/api/v4/projects/{project_id}/deployments"
        response = requests.post(url, headers=self.headers, json=deployment_config)
        response.raise_for_status()
        return response.json()

    def generate_gitlab_ci_yaml(self, config: Dict[str, Any]) -> str:
        """生成 .gitlab-ci.yml 文件"""
        gitlab_ci = {
            'stages': config.get('stages', ['build', 'test', 'deploy']),
            'variables': config.get('variables', {}),
            'before_script': config.get('before_script', []),
            'after_script': config.get('after_script', [])
        }

        # 添加作业
        for job_name, job_config in config.get('jobs', {}).items():
            gitlab_ci[job_name] = job_config

        return yaml.dump(gitlab_ci, default_flow_style=False)

    def deploy_application(self, project_id: str, environment: str, variables: Dict[str, str] = None) -> Dict[str, Any]:
        """部署应用程序"""
        print(f"Deploying to {environment}...")

        # 触发部署流水线
        deploy_variables = variables or {}
        deploy_variables['ENVIRONMENT'] = environment

        pipeline = self.trigger_pipeline(project_id, variables=deploy_variables)
        pipeline_id = pipeline['id']

        print(f"Triggered pipeline {pipeline_id}")

        # 等待流水线完成
        status = self.wait_for_pipeline(project_id, pipeline_id)

        if status == 'success':
            print(f"Deployment to {environment} successful!")
        else:
            print(f"Deployment to {environment} failed with status: {status}")

            # 获取失败的作业日志
            jobs = self.get_pipeline_jobs(project_id, pipeline_id)
            for job in jobs:
                if job['status'] == 'failed':
                    print(f"\nFailed job: {job['name']}")
                    log = self.get_job_log(project_id, job['id'])
                    print(f"Log:\n{log[-1000:]}")  # 显示最后 1000 字符

        return {
            'pipeline_id': pipeline_id,
            'status': status,
            'environment': environment
        }

class DockerManager:
    def __init__(self):
        import docker
        self.client = docker.from_env()

    def build_image(self, dockerfile_path: str, image_name: str, tag: str = 'latest') -> str:
        """构建 Docker 镜像"""
        print(f"Building Docker image: {image_name}:{tag}")

        image, logs = self.client.images.build(
            path=dockerfile_path,
            tag=f"{image_name}:{tag}",
            rm=True
        )

        for log in logs:
            if 'stream' in log:
                print(log['stream'].strip())

        print(f"Image built successfully: {image.id}")
        return image.id

    def push_image(self, image_name: str, tag: str = 'latest', registry: str = None) -> bool:
        """推送镜像到仓库"""
        full_name = f"{image_name}:{tag}"
        if registry:
            full_name = f"{registry}/{full_name}"

        print(f"Pushing image: {full_name}")

        try:
            for line in self.client.images.push(full_name, stream=True, decode=True):
                if 'status' in line:
                    print(f"{line['status']}: {line.get('progress', '')}")

            print("Image pushed successfully!")
            return True
        except Exception as e:
            print(f"Failed to push image: {e}")
            return False

    def run_container(self, image_name: str, container_config: Dict[str, Any]) -> str:
        """运行容器"""
        print(f"Running container from image: {image_name}")

        container = self.client.containers.run(
            image_name,
            **container_config,
            detach=True
        )

        print(f"Container started: {container.id}")
        return container.id

    def generate_dockerfile(self, app_config: Dict[str, Any]) -> str:
        """生成 Dockerfile"""
        base_image = app_config.get('base_image', 'python:3.9-slim')
        working_dir = app_config.get('working_dir', '/app')

        dockerfile_content = f"""
FROM {base_image}

WORKDIR {working_dir}

# 安装系统依赖
RUN apt-get update && apt-get install -y \\
    {' '.join(app_config.get('system_packages', []))} \\
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件
COPY requirements.txt .

# 安装 Python 依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 设置环境变量
{chr(10).join(f'ENV {key}={value}' for key, value in app_config.get('env_vars', {}).items())}

# 暴露端口
EXPOSE {app_config.get('port', 8000)}

# 健康检查
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \\
    CMD {app_config.get('health_check', 'curl -f http://localhost:8000/health || exit 1')}

# 启动命令
CMD {json.dumps(app_config.get('cmd', ['python', 'app.py']))}
"""

        return dockerfile_content.strip()

# CI/CD 配置示例
ci_config = {
    'stages': ['build', 'test', 'security', 'deploy'],
    'variables': {
        'DOCKER_DRIVER': 'overlay2',
        'DOCKER_TLS_CERTDIR': '/certs'
    },
    'before_script': [
        'echo "Starting CI/CD pipeline"',
        'docker info'
    ],
    'jobs': {
        'build': {
            'stage': 'build',
            'image': 'docker:latest',
            'services': ['docker:dind'],
            'script': [
                'docker build -t $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA .',
                'docker push $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA'
            ],
            'only': ['main', 'develop']
        },
        'test': {
            'stage': 'test',
            'image': 'python:3.9',
            'script': [
                'pip install -r requirements-dev.txt',
                'pytest tests/ --cov=src --cov-report=xml',
                'flake8 src/'
            ],
            'artifacts': {
                'reports': {
                    'coverage_report': {
                        'coverage_format': 'cobertura',
                        'path': 'coverage.xml'
                    }
                }
            }
        },
        'security_scan': {
            'stage': 'security',
            'image': 'owasp/zap2docker-stable',
            'script': [
                'zap-baseline.py -t http://localhost:8000'
            ],
            'allow_failure': True
        },
        'deploy_staging': {
            'stage': 'deploy',
            'image': 'alpine/helm:latest',
            'script': [
                'helm upgrade --install myapp ./helm-chart --namespace staging',
                'kubectl rollout status deployment/myapp -n staging'
            ],
            'environment': {
                'name': 'staging',
                'url': 'https://staging.example.com'
            },
            'only': ['develop']
        },
        'deploy_production': {
            'stage': 'deploy',
            'image': 'alpine/helm:latest',
            'script': [
                'helm upgrade --install myapp ./helm-chart --namespace production',
                'kubectl rollout status deployment/myapp -n production'
            ],
            'environment': {
                'name': 'production',
                'url': 'https://example.com'
            },
            'when': 'manual',
            'only': ['main']
        }
    }
}

# 使用示例
if __name__ == "__main__":
    # GitLab CI/CD 管理
    gitlab_manager = GitLabCIManager(
        gitlab_url='https://gitlab.example.com',
        access_token='your_access_token'
    )

    # 生成 .gitlab-ci.yml
    gitlab_ci_yaml = gitlab_manager.generate_gitlab_ci_yaml(ci_config)
    with open('.gitlab-ci.yml', 'w') as f:
        f.write(gitlab_ci_yaml)
    print("Generated .gitlab-ci.yml")

    # Docker 管理
    docker_manager = DockerManager()

    # 应用配置
    app_config = {
        'base_image': 'python:3.9-slim',
        'working_dir': '/app',
        'system_packages': ['curl', 'git'],
        'port': 8000,
        'env_vars': {
            'FLASK_ENV': 'production',
            'DATABASE_URL': 'postgresql://user:pass@db:5432/myapp'
        },
        'health_check': 'curl -f http://localhost:8000/health || exit 1',
        'cmd': ['gunicorn', '--bind', '0.0.0.0:8000', 'app:app']
    }

    # 生成 Dockerfile
    dockerfile_content = docker_manager.generate_dockerfile(app_config)
    with open('Dockerfile', 'w') as f:
        f.write(dockerfile_content)
    print("Generated Dockerfile")